105 lines
3.4 KiB
Python
105 lines
3.4 KiB
Python
"""CsvTelemetryLogger — writes telemetry snapshots to a CSV file with throttling.
|
|
|
|
Design decisions:
|
|
- Does NOT use the logging module — writes directly via csv.writer so the file
|
|
is readable independently of the text log level.
|
|
- Flushes after every row so the file is intact even on crash or force-quit.
|
|
- Throttle: only one row per TELEMETRY_CSV_INTERVAL_S seconds, even if
|
|
metrics_updated fires every 500 ms. This keeps the file manageable for
|
|
long sessions (8 h @ 5 s interval = 5 760 rows).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import time
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from app.config import TELEMETRY_CSV_INTERVAL_S
|
|
from app.telemetry.telemetry_collector import TelemetrySnapshot
|
|
|
|
_CSV_HEADER = [
|
|
"timestamp",
|
|
"fps_got",
|
|
"fps_req",
|
|
"frame_time_ms",
|
|
"dropped_frames",
|
|
"cpu_sys_pct",
|
|
"cpu_core_pct",
|
|
"mem_mb",
|
|
]
|
|
|
|
|
|
class CsvTelemetryLogger:
|
|
"""
|
|
Receives TelemetrySnapshot objects and writes throttled rows to a CSV file.
|
|
|
|
Usage:
|
|
logger = CsvTelemetryLogger(path)
|
|
telemetry_collector.metrics_updated.connect(logger.on_metrics_updated)
|
|
# call logger.close() on application exit
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
path: Path,
|
|
interval_s: float = TELEMETRY_CSV_INTERVAL_S,
|
|
) -> None:
|
|
self._interval_s = interval_s
|
|
self._last_write_time: float = 0.0
|
|
self._rows_written: int = 0
|
|
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._file = path.open("w", newline="", encoding="utf-8")
|
|
self._writer = csv.writer(self._file)
|
|
self._writer.writerow(_CSV_HEADER)
|
|
self._file.flush()
|
|
|
|
# ------------------------------------------------------------------
|
|
# Slot — connect to TelemetryCollector.metrics_updated
|
|
# ------------------------------------------------------------------
|
|
|
|
def on_metrics_updated(self, snapshot: TelemetrySnapshot) -> None:
|
|
"""Write a row if the throttle interval has elapsed."""
|
|
now = time.monotonic()
|
|
if now - self._last_write_time < self._interval_s:
|
|
return
|
|
self._last_write_time = now
|
|
self._write_row(snapshot)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Lifecycle
|
|
# ------------------------------------------------------------------
|
|
|
|
def close(self) -> None:
|
|
"""Flush and close the CSV file. Call on application shutdown."""
|
|
try:
|
|
self._file.flush()
|
|
self._file.close()
|
|
except OSError:
|
|
pass
|
|
|
|
@property
|
|
def rows_written(self) -> int:
|
|
return self._rows_written
|
|
|
|
# ------------------------------------------------------------------
|
|
# Private
|
|
# ------------------------------------------------------------------
|
|
|
|
def _write_row(self, snap: TelemetrySnapshot) -> None:
|
|
ts = datetime.now().strftime("%H:%M:%S.%f")[:-3] # HH:MM:SS.mmm
|
|
self._writer.writerow([
|
|
ts,
|
|
f"{snap.fps:.1f}",
|
|
f"{snap.target_fps:.1f}" if snap.target_fps is not None else "",
|
|
f"{snap.frame_time_ms:.2f}",
|
|
snap.dropped_frames,
|
|
f"{snap.cpu_percent_sys:.1f}",
|
|
f"{snap.cpu_percent_core:.1f}",
|
|
f"{snap.memory_mb:.1f}" if snap.memory_mb is not None else "",
|
|
])
|
|
self._file.flush()
|
|
self._rows_written += 1
|