Real threshold-crossing abstraction from raw numeric sensor samples into
canonical %Ex4pm.Event{} observations.
A raw sensor reading is {value, timestamp, sensor_id}. This module applies
a real (if simple) abstraction rule: an %Ex4pm.Event{} is emitted only on a
crossing edge of a configured threshold (value goes from at-or-below the
threshold to strictly above it, or vice versa for a falling-edge rule) — not
on every sample. This is the discretization step that turns a continuous
signal into discrete process-mining events, matching the canonical event IR
from Ex4pm.Event (apps/ex4pm_core/lib/ex4pm/ocel.ex) so downstream
discovery/conformance can consume sensor-originated events identically to
any other event source.
Usable directly as a Broadway context.sink-style callback (handle_message/2)
matching Ex4pm.Stream.Pipeline's sink contract, or driven synchronously via
sample/2/sample_all/2 for a plain message-handling loop.
Summary
Functions
Broadway sink-compatible message handler: unwraps a %Broadway.Message{data: reading},
abstracts it against the state held in context, and forwards any emitted event to
context.forward.(event). Matches Ex4pm.Stream.Pipeline's "sinks receive
observations only" contract — no DO authority is exercised here.
New sensor abstraction state. Options: :threshold, :rising_activity, :falling_activity.
Fold one raw sample into the abstraction state. Returns {new_state, event_or_nil}
— nil when the sample did not cross the threshold (the common case: most raw
samples are abstraction-irrelevant noise around a steady state).
Fold a real sequence of raw samples, returning the final state and the ordered list of abstracted events.
Types
@type reading() :: {number(), DateTime.t() | non_neg_integer(), String.t()}
@type t() :: %Ex4pm.Stream.SensorSink{ emitted: non_neg_integer(), falling_activity: String.t(), last_state: %{optional(String.t()) => :above | :at_or_below}, rising_activity: String.t(), threshold: number() }
Functions
@spec handle_message(Broadway.Message.t(), %{ sink_state: pid() | t(), forward: (Ex4pm.Event.t() -> any()) }) :: Broadway.Message.t()
Broadway sink-compatible message handler: unwraps a %Broadway.Message{data: reading},
abstracts it against the state held in context, and forwards any emitted event to
context.forward.(event). Matches Ex4pm.Stream.Pipeline's "sinks receive
observations only" contract — no DO authority is exercised here.
New sensor abstraction state. Options: :threshold, :rising_activity, :falling_activity.
Fold one raw sample into the abstraction state. Returns {new_state, event_or_nil}
— nil when the sample did not cross the threshold (the common case: most raw
samples are abstraction-irrelevant noise around a steady state).
Fold a real sequence of raw samples, returning the final state and the ordered list of abstracted events.