| from __future__ import annotations |
|
|
| from dataclasses import dataclass, field |
| from typing import Any, Dict, List |
| import time |
|
|
|
|
| @dataclass |
| class RuntimeMetrics: |
| events: List[Dict[str, Any]] = field(default_factory=list) |
| started_at: float = field(default_factory=time.perf_counter) |
|
|
| def record(self, event: str, **values: Any) -> None: |
| self.events.append({"event": event, "timestamp": time.time(), **values}) |
|
|
| def snapshot(self) -> Dict[str, Any]: |
| durations = { |
| event["event"]: event.get("duration_ms") |
| for event in self.events |
| if event.get("duration_ms") is not None |
| } |
| return { |
| "events": list(self.events), |
| "event_count": len(self.events), |
| "active_runtime_slots": sum(1 for event in self.events if event["event"] == "runtime.created") - sum(1 for event in self.events if event["event"] == "runtime.destroyed"), |
| "durations_ms": durations, |
| "elapsed_ms": round((time.perf_counter() - self.started_at) * 1000, 3), |
| } |
|
|
