Output System

4.1 Output System

The SDK provides two complementary output mechanisms: OutputRecorder for direct programmatic recording, and SimulationRecorder for automatic multi-level capture via the hook system. Both support multiple export formats and can be used independently or together.

OutputRecorder

The OutputRecorder is a low-level time series recorder that stores per-step key-value data with timestamp mapping. It is used when the model author wants explicit control over what is recorded and when.

class OutputRecorder:
    def __init__(
        self,
        time_config: Optional[TimeConfig] = None,  # Timestamp mapping
        output_dir: str = "output",                 # Output directory
        format: OutputFormat = OutputFormat.CSV,     # Default export format
    ) -> None

The TimeConfig dataclass maps simulation ticks to wall-clock timestamps:

@dataclass
class TimeConfig:
    start: datetime      # Start timestamp (default: 2000-01-01 UTC)
    step_duration: timedelta  # Duration per tick (default: 1 second)
    timezone_name: str = "UTC"

    def timestamp_for_tick(self, tick: int) -> datetime

Recording and export methods:

Method	Signature	Description
record	(tick: int, seed: int = 1, **fields: float) -> None	Record arbitrary key-value pairs for a tick
record_from_model	(model, seed: int = 1) -> None	Record all accumulator values from a model
add_channel	(channel: OutputChannel) -> None	Add a named output channel with field filter
flush_channels	(output_dir=None) -> Dict[str, List[Path]]	Write all channels to disk
to_csv	(combined=False) -> List[Path]	Export as CSV
to_gzip	(combined=False) -> List[Path]	Export as gzip-compressed CSV
to_parquet	(combined=False) -> List[Path]	Export as Apache Parquet
to_duckdb	(path=None, table_name="simulation_output") -> Path	Export to DuckDB
to_s3	(bucket, prefix="", format="csv") -> List[str]	Upload to S3
to_database	(connection_string, table_name="output") -> int	Write to PostgreSQL/ClickHouse
clear	() -> None	Discard all recorded data

The OutputFormat enum supports: CSV, CSV_GZ, PARQUET, DATABASE, S3.

SimulationRecorder (Multi-Level)

The SimulationRecorder is a higher-level recorder that attaches to a model via the AFTER_STEP hook and automatically captures output at multiple levels of granularity. It requires zero changes to model code --- just attach it before running.

class SimulationRecorder:
    def __init__(self, config: Optional[RecorderConfig] = None) -> None
    def attach(self, model) -> None     # Wire into model's AFTER_STEP hook
    def detach(self) -> None            # Remove hook
    def flush(self) -> Dict[str, Path]  # Write all data + plots to disk
    def clear(self) -> None             # Discard recorded data

The RecorderConfig dataclass controls what is captured:

@dataclass
class RecorderConfig:
    output_dir: str = "output"
    levels: List[RecordLevel] = [RecordLevel.MODEL, RecordLevel.AGENT_GROUP]
    format: str = "csv.gz"
    plots: bool = True
    plot_style: str = "modern"
    plot_dpi: int = 200
    plot_size: Tuple[int, int] = (10, 5)
    agent_sample_rate: float = 1.0     # 1.0 = all agents, 0.1 = 10% sample
    agent_filter: Optional[Callable] = None  # Custom agent filter predicate
    agent_fields: Optional[Set[str]] = None  # Subset of state fields to record

The four recording levels:

Level	What's Captured	Output File	Content
MODEL	Accumulator values per step	model_output.csv.gz	One row per tick, one column per accumulator
AGENT_GROUP	Per-type aggregate statistics	agents_[TypeName].csv.gz	Per-tick mean, std, min, max of each state field
AGENT	Individual agent state	agent_[id].csv.gz	Per-tick values of all (or selected) state fields
ENVIRONMENT	GlobalState numeric values	environment.csv.gz	Per-tick values of all numeric environment fields

When plots=True, the recorder automatically generates PNG time series plots for every recorded variable --- one plot per accumulator (MODEL level), one plot per aggregate statistic (AGENT_GROUP level), and one plot per environment variable. These plots are saved alongside the CSV files.

The auto_record() convenience function creates and attaches a recorder in one call:

from simudyne.engine.recorder import auto_record

recorder = auto_record(model, output_dir="output",
                       levels=["model", "agent_group", "environment"],
                       plots=True, format="csv.gz")

---