mufasa.io.outputs.visualization#

Visualization output node with integrated rendering helpers.

Attributes#

BASEMAPS

Classes#

Visualization

Output node that collects Map snapshots and Locations for inspection.

Module Contents#

mufasa.io.outputs.visualization.BASEMAPS#

class mufasa.io.outputs.visualization.Visualization(snapshot_interval_s: float = 5.0)#

Bases: mufasa.io.outputs.base.OutputNode

Output node that collects Map snapshots and Locations for inspection.

Wire any number of Map or Location predecessors; access each by index or name after the run:

viz = Visualization(snapshot_interval_s=5)(pom_a, pom_b, threshold)
viz.show(0)                        # pom_a — latest snapshot
viz.show("threshold", end=100.0)   # threshold locations up to t=100
viz.animate(1, speed=2.0)          # pom_b at 2× real-time

snapshot_interval_s throttles how often Map state is captured. Locations are always stored without throttling.

Both show() and animate() default to basemap=True and latlon_labels=True and fix the axis extent to the pipeline bounding box. basemap also takes a style name from BASEMAPS ("streets", the default, or "satellite"). Pass basemap=False to suppress tile fetching (e.g. in tests).

Parameters:: snapshot_interval_s (float)

snapshot_interval_s = 5.0#

__call__(*inputs) → Visualization#

Return type:: Visualization

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() → 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

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

show(index: int | str, *, start: float | None = None, end: float | None = None, reduction: str = 'last', basemap: bool | str = True, latlon_labels: bool = True, **kwargs) → None#

Render the state of a predecessor, optionally filtered to a time window.

Parameters:

index (int | str)
start (float | None)
end (float | None)
reduction (str)
basemap (bool | str)
latlon_labels (bool)

Return type:

None

animate(index: int | str, *, start: float | None = None, end: float | None = None, speed: float = 1.0, basemap: bool | str = True, latlon_labels: bool = True, **kwargs)#

Animate snapshots or locations for a predecessor.

Returns (fig, anim) — keep a reference to anim to prevent garbage collection. speed scales playback: 1.0 = real-time, 2.0 = 2× faster.

Parameters:

index (int | str)
start (float | None)
end (float | None)
speed (float)
basemap (bool | str)
latlon_labels (bool)

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