104 lines
2.9 KiB
Python
104 lines
2.9 KiB
Python
from collections import deque
|
|
from dataclasses import dataclass
|
|
|
|
|
|
@dataclass
|
|
class EffectTiming:
|
|
name: str
|
|
duration_ms: float
|
|
buffer_chars_in: int
|
|
buffer_chars_out: int
|
|
|
|
|
|
@dataclass
|
|
class FrameTiming:
|
|
frame_number: int
|
|
total_ms: float
|
|
effects: list[EffectTiming]
|
|
|
|
|
|
class PerformanceMonitor:
|
|
"""Collects and stores performance metrics for effect pipeline."""
|
|
|
|
def __init__(self, max_frames: int = 60):
|
|
self._max_frames = max_frames
|
|
self._frames: deque[FrameTiming] = deque(maxlen=max_frames)
|
|
self._current_frame: list[EffectTiming] = []
|
|
|
|
def start_frame(self, frame_number: int) -> None:
|
|
self._current_frame = []
|
|
|
|
def record_effect(
|
|
self, name: str, duration_ms: float, chars_in: int, chars_out: int
|
|
) -> None:
|
|
self._current_frame.append(
|
|
EffectTiming(
|
|
name=name,
|
|
duration_ms=duration_ms,
|
|
buffer_chars_in=chars_in,
|
|
buffer_chars_out=chars_out,
|
|
)
|
|
)
|
|
|
|
def end_frame(self, frame_number: int, total_ms: float) -> None:
|
|
self._frames.append(
|
|
FrameTiming(
|
|
frame_number=frame_number,
|
|
total_ms=total_ms,
|
|
effects=self._current_frame,
|
|
)
|
|
)
|
|
|
|
def get_stats(self) -> dict:
|
|
if not self._frames:
|
|
return {"error": "No timing data available"}
|
|
|
|
total_times = [f.total_ms for f in self._frames]
|
|
avg_total = sum(total_times) / len(total_times)
|
|
min_total = min(total_times)
|
|
max_total = max(total_times)
|
|
|
|
effect_stats: dict[str, dict] = {}
|
|
for frame in self._frames:
|
|
for effect in frame.effects:
|
|
if effect.name not in effect_stats:
|
|
effect_stats[effect.name] = {"times": [], "total_chars": 0}
|
|
effect_stats[effect.name]["times"].append(effect.duration_ms)
|
|
effect_stats[effect.name]["total_chars"] += effect.buffer_chars_out
|
|
|
|
for name, stats in effect_stats.items():
|
|
times = stats["times"]
|
|
stats["avg_ms"] = sum(times) / len(times)
|
|
stats["min_ms"] = min(times)
|
|
stats["max_ms"] = max(times)
|
|
del stats["times"]
|
|
|
|
return {
|
|
"frame_count": len(self._frames),
|
|
"pipeline": {
|
|
"avg_ms": avg_total,
|
|
"min_ms": min_total,
|
|
"max_ms": max_total,
|
|
},
|
|
"effects": effect_stats,
|
|
}
|
|
|
|
def reset(self) -> None:
|
|
self._frames.clear()
|
|
self._current_frame = []
|
|
|
|
|
|
_monitor: PerformanceMonitor | None = None
|
|
|
|
|
|
def get_monitor() -> PerformanceMonitor:
|
|
global _monitor
|
|
if _monitor is None:
|
|
_monitor = PerformanceMonitor()
|
|
return _monitor
|
|
|
|
|
|
def set_monitor(monitor: PerformanceMonitor) -> None:
|
|
global _monitor
|
|
_monitor = monitor
|