Background thread: periodically publishes a full situation snapshot. ``engine_ref_getter`` is a zero-argument callable that returns the current ``SimulationEngine`` instance so we avoid a circular import.
(engine_ref_getter)
| 82 | # --------------------------------------------------------------------------- |
| 83 | |
| 84 | def _snapshot_loop(engine_ref_getter) -> None: |
| 85 | """ |
| 86 | Background thread: periodically publishes a full situation snapshot. |
| 87 | |
| 88 | ``engine_ref_getter`` is a zero-argument callable that returns the |
| 89 | current ``SimulationEngine`` instance so we avoid a circular import. |
| 90 | """ |
| 91 | last_push = 0.0 |
| 92 | while True: |
| 93 | now = time.monotonic() |
| 94 | if now - last_push >= _SNAPSHOT_INTERVAL: |
| 95 | try: |
| 96 | engine = engine_ref_getter() |
| 97 | if engine is not None: |
| 98 | snap = _build_snapshot(engine) |
| 99 | event_bus.publish("snapshot", snap) |
| 100 | except Exception: # noqa: BLE001 |
| 101 | pass # never crash the background thread |
| 102 | last_push = time.monotonic() |
| 103 | time.sleep(0.1) |
| 104 | |
| 105 | |
| 106 | def _build_snapshot(engine) -> dict: |
nothing calls this directly
no test coverage detected