Skip to content

simulatecraft.viewers.log

simulatecraft.viewers.log

JSONL event logger for replay/analysis, plus a replay utility.

JsonlLogger 

JsonlLogger(path: str | Path, bus: EventBus)

Append every outbound event to a .jsonl file.

Source code in src/simulatecraft/viewers/log.py 
45
46
47
48
49
50
def __init__(self, path: str | Path, bus: EventBus) -> None:
    self.path = Path(path)
    self.path.parent.mkdir(parents=True, exist_ok=True)
    self._fh: TextIO = self.path.open("a", encoding="utf-8")
    self._bus = bus
    self._unsub = bus.subscribe(self.handle_event)

Replayer 

Replayer(path: str | Path)

Replay logged events onto a (fresh) EventBus at controllable speed.

speed multiplier: 1.0 = real time per recorded timestamps, N>1 = N times faster, 0 = as fast as possible.

Source code in src/simulatecraft/viewers/log.py 
87
88
def __init__(self, path: str | Path) -> None:
    self.events: list[Event] = load_events(path)

load_events 

load_events(path: str | Path) -> list[Event]

Load logged JSONL events back into typed models.

Source code in src/simulatecraft/viewers/log.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
def load_events(path: str | Path) -> list[Event]:
    """Load logged JSONL events back into typed models."""
    out: list[Event] = []
    with open(path, encoding="utf-8") as fh:
        for line in fh:
            line = line.strip()
            if not line:
                continue
            raw: dict[str, Any] = json.loads(line)
            event_type = _EVENT_TYPES.get(raw.get("kind", ""))
            if event_type is not None:
                out.append(event_type.model_validate(raw))
    return out