session

session

Background-sampling telemetry session.

Uses Rust ring buffer — Rust backend is mandatory.

Classes

TelemetrySample dataclass

TelemetrySample(
    timestamp_ns: int,
    gpu_power_w: float = 0.0,
    cpu_power_w: float = 0.0,
    gpu_energy_j: float = 0.0,
    cpu_energy_j: float = 0.0,
    gpu_util_pct: float = 0.0,
    gpu_temp_c: float = 0.0,
    gpu_mem_gb: float = 0.0,
)

Single telemetry sample.

TelemetrySession

TelemetrySession(
    monitor: Optional[EnergyMonitor] = None,
    interval_ms: int = 100,
    buffer_size: int = 100000,
)

Background-sampling telemetry session.

Spawns a daemon thread that calls monitor.snapshot() at the configured interval. Stores samples in a ring buffer (Rust-backed if available, else pure-Python fallback).

Source code in src/diapason/telemetry/session.py

def __init__(
    self,
    monitor: Optional[EnergyMonitor] = None,
    interval_ms: int = 100,
    buffer_size: int = 100_000,
) -> None:
    self._monitor = monitor
    self._interval_ms = interval_ms
    self._buffer = _PythonRingBuffer(buffer_size)
    self._thread: Optional[threading.Thread] = None
    self._stop_event = threading.Event()

Methods:

start

start() -> None

Start background sampling thread.

Source code in src/diapason/telemetry/session.py

def start(self) -> None:
    """Start background sampling thread."""
    if self._monitor is None:
        return
    if self._thread is not None and self._thread.is_alive():
        return
    self._stop_event.clear()
    self._thread = threading.Thread(target=self._sample_loop, daemon=True)
    self._thread.start()

stop

stop() -> None

Stop sampling thread.

Source code in src/diapason/telemetry/session.py

def stop(self) -> None:
    """Stop sampling thread."""
    self._stop_event.set()
    if self._thread is not None:
        self._thread.join(timeout=2.0)
        self._thread = None