mufasa.nodes.tracking.kalman#

Multi-target Kalman tracker node backed by Stone Soup.

Classes#

KalmanTracker

Base class for all pipeline nodes.

Module Contents#

class mufasa.nodes.tracking.kalman.KalmanTracker(buffer_s: float = 1.0, max_late_s: float = 5.0, process_noise: float = 0.05, measurement_noise: float = 10.0, max_missed_frames: int = 3, min_detections_to_confirm: int = 1)#

Bases: mufasa.node.Node

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.

Parameters:

buffer_s (float)
max_late_s (float)
process_noise (float)
measurement_noise (float)
max_missed_frames (int)
min_detections_to_confirm (int)

buffer_s = 1.0#

max_late_s = 5.0#

process_noise = 0.05#

measurement_noise = 10.0#

max_missed_frames = 3#

min_detections_to_confirm = 1#

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

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

flush() → None#

Emit any detections remaining in the buffer at end of stream.

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

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