Ex4pm.Engine.Discovery.Incremental (ex4pm v26.9.9)

Copy Markdown View Source

Real incremental (streaming) directly-follows-graph discovery.

Maintains a running DFG state and folds one new %Ex4pm.Event{} at a time via update/2 — no reprocessing of prior events. Distinct from a "batch" discovery that re-scans the whole log on every call: this module's update/2 performs O(1) amortized work per event (map updates only), and from_events/1 is provided purely as a Chicago-style cross-check oracle (fold the batch list through the same update/2 step function) so the incremental and batch paths are proven equivalent by construction, not by two independently written implementations that could silently drift.

Events are correlated into per-case sequences via a caller-supplied case-id extractor (defaults to the event's first object_id, falling back to a single global case when an event carries no object ids) so multi-case interleaving is handled the same way both online and offline miners in this app treat OCEL case correlation.

Summary

Types

edge()
t()

Functions

dfg(incremental)

Directly-follows edge frequency map, {from, to} => count.

finalize(state)

Finalize a state that has consumed a bounded stream by recording per-case end activities (the trailing activity of each case seen so far). Idempotent and non-destructive w.r.t. edges/activities — only touches ends.

from_events(events, opts \\ [])

Batch-computed DFG over a full event list, used only as a cross-check oracle. Implemented as a fold through the same update/2 step used for streaming, so equivalence with the incremental path is structural, not coincidental.

new(opts \\ [])

New empty incremental DFG state. opts[:case_id_fun] overrides case correlation.

update(state, event)

Fold exactly one new event into the running DFG state.

Types

edge()

@type edge() :: {String.t(), String.t()}

t()

@type t() :: %Ex4pm.Engine.Discovery.Incremental{
  activities: %{optional(String.t()) => non_neg_integer()},
  case_id_fun: (Ex4pm.Event.t() -> term()),
  edges: %{optional(edge()) => non_neg_integer()},
  ends: %{optional(String.t()) => non_neg_integer()},
  event_count: non_neg_integer(),
  last_activity: %{optional(term()) => String.t()},
  starts: %{optional(String.t()) => non_neg_integer()}
}

Functions

dfg(incremental)

@spec dfg(t()) :: %{optional(edge()) => non_neg_integer()}

Directly-follows edge frequency map, {from, to} => count.

finalize(state)

@spec finalize(t()) :: t()

Finalize a state that has consumed a bounded stream by recording per-case end activities (the trailing activity of each case seen so far). Idempotent and non-destructive w.r.t. edges/activities — only touches ends.

from_events(events, opts \\ [])

@spec from_events([Ex4pm.Event.t()], keyword()) :: t()

Batch-computed DFG over a full event list, used only as a cross-check oracle. Implemented as a fold through the same update/2 step used for streaming, so equivalence with the incremental path is structural, not coincidental.

new(opts \\ [])

@spec new(keyword()) :: t()

New empty incremental DFG state. opts[:case_id_fun] overrides case correlation.

update(state, event)

@spec update(t(), Ex4pm.Event.t()) :: t()

Fold exactly one new event into the running DFG state.

Real incremental update: touches only the edges/activity counters implicated by this single event and the case's previous activity — never rescans prior events.