mufasa.node

Classes

Node	Base class for all pipeline nodes.

Module Contents

class mufasa.node.Node

Bases: abc.ABC

Base class for all pipeline nodes.

Every concrete subclass must declare two class attributes:

class MyNode(Node):
    _input_types = [Location]       # types this node accepts; [] for source nodes
    _output_type = Observation      # type this node emits; None for sink nodes

Listing a parent type (e.g. Location) automatically covers all subclasses (e.g. Observation) because wiring uses issubclass.

Two methods must be implemented by every concrete subclass:

process(*args, **kwargs)

Called by the framework to push data through the node. Use the signature appropriate for this node’s role — see the docstring on process() for the standard patterns.

get_config() -> dict

Return all constructor arguments as a plain JSON-serializable dict so the pipeline can be saved and reconstructed.

Two further methods have no-op defaults and should be overridden by nodes that accumulate internal state across process() calls:

flush() -> None

Called by the framework after all inputs are exhausted (end of stream). Emit any observations or map snapshots remaining in an internal buffer. Stateless nodes can leave this as the no-op default.

reset() -> None

Called by the framework between pipeline runs. Clear all runtime state so the pipeline can be re-run cleanly without constructing a new instance. Stateless nodes can leave this as the no-op default.

name: str | None = None

__call__(*inputs: Node) → Node

Parameters:

inputs (Node)

Return type:

Node

configure(crs: str, bbox, resolution: tuple) → None

Receive spatial configuration from the Graph at construction time.

Override in nodes that need crs/bbox/resolution to initialise internal state (e.g. POM creates its map here). Default is a no-op.

Parameters:

• crs (str)

• resolution (tuple)

Return type:

None

property input_types: frozenset

Types this node accepts as input. Derived from _input_types.

Return type:

frozenset

property output_type: type | None

Type this node emits. Derived from _output_type.

Return type:

type | None

abstractmethod process(*args, **kwargs) → 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)

Return type:

None

flush() → None

Emit any buffered output remaining at end of stream.

Called by the framework after all inputs are exhausted. Override in nodes that accumulate observations or map data across multiple process() calls before emitting (e.g. scan-window trackers, batch clusterers). The default is a no-op — stateless nodes need not override this.

Return type:

None

reset() → None

Clear all runtime state, returning the node to its post-init condition.

Called by the framework between pipeline runs so the same node instance can process a fresh dataset without being reconstructed. Override in nodes that hold mutable state (buffers, accumulators, fitted models). The default is a no-op — stateless nodes need not override this.

Return type:

None

abstractmethod 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