mufasa.nodes.mapping.heat#

Classes#

HeatMap

Base class for Observation-ingesting map nodes with temporal decay.

Module Contents#

class mufasa.nodes.mapping.heat.HeatMap(decay_s: float)#

Bases: mufasa.node.Node

Base class for Observation-ingesting map nodes with temporal decay.

Subclasses declare their output map type via the _map_class class variable (default Map). POM sets it to BayesianMap.

A single Map object is allocated in configure() and reused across all process() calls. self._data is an alias for self._map.data so subclass hooks (_decay, _model_update) operate directly on the live array with no copies.

Subclasses must implement:: _decay(delta_t_s): apply time decay to internal state _model_update(mask, confidence): update cells under the Observation footprint get_config() -> dict: constructor arguments for serialization

Parameters:: decay_s (float)

decay_s#

configure(crs, bbox, resolution) → 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.

Return type:: None

property map: mufasa.map.Map#

Current map — the same object is returned on every call.

Return type:: mufasa.map.Map

process(obs: mufasa.location.Observation)#

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)

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

flush() → None#

No buffered state to emit at end of stream.

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