babylon.engine.observers.metrics

TickStateRecorder observer for unified simulation metrics.

Implements SimulationObserver protocol to record tick-by-tick state during simulation runs. Supports two modes:

“interactive”: Rolling window of recent ticks (for dashboard)
“batch”: Accumulates all history (for parameter sweeps)

Note: This class was renamed from MetricsCollector to TickStateRecorder to avoid namespace collision with src/babylon/metrics/collector.py (RAG telemetry collection).

Sprint 4.1: Phase 4 Dashboard/Sweeper unification. Sprint 4.1C: Add JSON export for DAG structure preservation.

Classes

TickStateRecorder([mode, rolling_window])

Observer that records simulation state at each tick for analysis.

class babylon.engine.observers.metrics.TickStateRecorder(mode='interactive', rolling_window=50)[source]

Bases: object

Observer that records simulation state at each tick for analysis.

Implements SimulationObserver protocol. Extracts entity and edge metrics at each tick, with optional rolling window for memory efficiency in interactive mode.

Renamed from MetricsCollector to avoid namespace collision with babylon.metrics.collector.MetricsCollector (RAG telemetry).

Parameters:

mode (Literal['interactive', 'batch'])
rolling_window (int)

__init__(mode='interactive', rolling_window=50)[source]

Initialize the collector.

Parameters:

mode (Literal['interactive', 'batch']) – “interactive” uses rolling window, “batch” keeps all history
rolling_window (int) – Maximum ticks to keep in interactive mode

Return type:

None

property name: str: Return observer identifier.

property latest: TickMetrics | None: Return most recent tick metrics, or None if empty.

property history: list[TickMetrics]: Return metrics history as a list.

property summary: SweepSummary | None: Return sweep summary, or None if no data collected.

on_simulation_start(initial_state, config)[source]

Called when simulation begins. Clears history and records tick 0.

Return type:

None

Parameters:

initial_state (WorldState)
config (SimulationConfig)

on_tick(previous_state, new_state)[source]

Called after each tick completes. Records new state.

Return type:

None

Parameters:

previous_state (WorldState)
new_state (WorldState)

on_simulation_end(final_state)[source]

Called when simulation ends. No-op for MetricsCollector.

Return type:: None
Parameters:: final_state (WorldState)

to_csv_rows()[source]

Export metrics history as list of dicts for CSV output.

Return type:: list[dict[str, Any]]

to_json(defines, config, csv_path=None)[source]

Export run metadata as structured JSON for reproducibility.

Captures the causal DAG hierarchy: - Level 1 (Fundamental): GameDefines parameters - Level 2 (Config): SimulationConfig settings - Level 3 (Emergent): SweepSummary computed from simulation

Parameters:

defines (GameDefines) – GameDefines with fundamental parameters
config (SimulationConfig) – SimulationConfig with run settings
csv_path (Path | None) – Optional path to associated CSV time-series file

Return type:

dict[str, Any]

Returns:

Structured dict ready for JSON serialization

export_json(path, defines, config, csv_path=None)[source]

Write JSON metadata to file.

Parameters:

path (Path) – Output path for JSON file
defines (GameDefines) – GameDefines with fundamental parameters
config (SimulationConfig) – SimulationConfig with run settings
csv_path (Path | None) – Optional path to associated CSV time-series file

Return type:

None