mufasa.io.outputs.streaming

Classes

StreamingOutputNode

Output node that delivers each result to a callable.

Module Contents

class mufasa.io.outputs.streaming.StreamingOutputNode(on_output=None)

Bases: mufasa.io.outputs.base.OutputNode

Output node that delivers each result to a callable.

out = StreamingOutputNode(my_callback)(prev_node)

The callback receives one Location (or Observation) per alarm. on_output can also be reassigned at any time after construction.

on_output = None

process(item: mufasa.location.Location) → 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:

item (mufasa.location.Location)

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