openg2g.events¶

@dataclass(frozen=True)
class SimEvent:
    """Structured simulation event with canonical clock metadata."""

    tick: int
    t_s: float
    source: EventSource
    topic: str
    data: dict[str, Any] = field(default_factory=dict)

class EventSink(Protocol):
    """Receives simulation events from components."""

    def emit(self, event: SimEvent) -> None:
        """Consume one event."""

@dataclass
class EventEmitter:
    """Source-bound event helper that stamps events with clock metadata."""

    clock: SimulationClock
    sink: EventSink
    source: EventSource

    def emit(self, topic: str, data: dict[str, Any] | None = None) -> None:
        """Emit one event with current clock metadata."""
        t_s = float(self.clock.time_s)
        self.sink.emit(
            SimEvent(
                tick=int(self.clock.step),
                t_s=t_s,
                source=self.source,
                topic=str(topic),
                data={} if data is None else dict(data),
            )
        )