backend/app/decidron/stats.py

"""Performance statistics for the Decidron simulator.

Aggregates per-sensor, per-command, and per-output counts plus latency
and last-run timestamps -- a deterministic stand-in for the patent's
"results determined and scored" step.
"""

from __future__ import annotations

import time
from collections import defaultdict


class StatsCollector:
    def __init__(self) -> None:
        self.sensors: dict[str, dict] = defaultdict(self._blank)
        self.commands: dict[str, dict] = defaultdict(self._blank)
        self.outputs: dict[str, dict] = defaultdict(self._blank)
        self.total_runs: int = 0
        self.total_matches: int = 0

    @staticmethod
    def _blank() -> dict:
        return {"count": 0, "matches": 0, "last_run": None, "avg_latency_ms": 0.0}

    @staticmethod
    def _update(bucket: dict, matched: bool, latency_ms: float) -> None:
        n = bucket["count"]
        bucket["count"] = n + 1
        if matched:
            bucket["matches"] += 1
        # incremental average latency
        bucket["avg_latency_ms"] = (bucket["avg_latency_ms"] * n + latency_ms) / (n + 1)
        bucket["last_run"] = time.time()

    def record(
        self,
        sensor: str,
        command: str,
        output: str | None,
        matched: bool,
        latency_ms: float,
    ) -> None:
        self.total_runs += 1
        if matched:
            self.total_matches += 1
        self._update(self.sensors[sensor], matched, latency_ms)
        self._update(self.commands[command], matched, latency_ms)
        if matched and output:
            self._update(self.outputs[output], True, latency_ms)

    def reset(self) -> None:
        self.__init__()

    def as_dict(self) -> dict:
        return {
            "total_runs": self.total_runs,
            "total_matches": self.total_matches,
            "per_sensor": dict(self.sensors),
            "per_command": dict(self.commands),
            "per_output": dict(self.outputs),
        }