mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-09-11 20:27:44 +02:00
27 lines
678 B
Python
27 lines
678 B
Python
"""Fixed-size rolling event log for support diagnostics."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
from collections import deque
|
|
|
|
P_MAX_SIZE = 50
|
|
P_LOCK = threading.Lock()
|
|
P_BUFFER: deque[dict] = deque(maxlen=P_MAX_SIZE)
|
|
|
|
|
|
def record(label: str, **meta: str | int | float | None) -> None:
|
|
"""Append an entry. Oldest drops when full."""
|
|
with P_LOCK:
|
|
P_BUFFER.append({
|
|
"l": label,
|
|
"t": time.time(),
|
|
**{k: v for k, v in meta.items() if v is not None},
|
|
})
|
|
|
|
|
|
def snapshot() -> list[dict]:
|
|
"""Return a copy of the current buffer, oldest first."""
|
|
with P_LOCK:
|
|
return list(P_BUFFER) |