mufasa.nodes.util.filter

Classes

ObservationFilter

Pass-through node that drops Observations not matching the specified criteria.

Module Contents

class mufasa.nodes.util.filter.ObservationFilter(min_confidence: float = 0.0, start_time: float | None = None, end_time: float | None = None, properties: dict | None = None)

Bases: mufasa.node.Node

Pass-through node that drops Observations not matching the specified criteria.

All criteria are AND-combined: an Observation must satisfy every specified criterion to be forwarded to successors.

Parameters:

• min_confidence (float) – Minimum confidence (inclusive). Observations below this value are dropped. Default 0.0 passes everything.

• start_time (float | None) – Only pass observations with timestamp >= start_time.

• end_time (float | None) – Only pass observations with timestamp < end_time.

• properties (dict[str, str] | None) – Map of property key -> regex pattern. An Observation passes only if, for every key, str(obs.properties.get(key, "")) fully matches the corresponding pattern. None (default) applies no property filter.

min_confidence = 0.0

start_time = None

end_time = None

process(obs: mufasa.location.Observation) → None

Process incoming data and forward results to successors.

Override with the signature appropriate for this node’s role:

Observation-consuming nodes receive one observation per call:

def process(self, obs: Observation) -> None:
    result = ...                    # compute result
    for succ in self._successors:
        succ.process(result)        # push to downstream nodes

Map-consuming nodes are called with no arguments and read their input from the predecessor via its .map attribute:

def process(self) -> None:
    m = self._predecessors[0].map   # read upstream map
    ...                             # update self.map in place
    self._trigger_successors()      # notify downstream nodes

Sink nodes collect data without emitting anything further:

def process(self, obs: Observation) -> None:
    self._collected.append(obs)

Parameters:

obs (mufasa.location.Observation)

Return type:

None

get_config() → dict

Return all constructor arguments as a plain JSON-serializable dict.

Used by Graph.to_config() to serialise the pipeline. Every key must be a string; every value must be JSON-serializable.

Return type:

dict