forked from genewildish/Mainline
David Gwilliam
98a5862c74
This commit implements the Sideline/Mainline split with a clean plugin architecture: ## Core Changes ### Sideline Framework (New Directory) - Created directory containing the reusable pipeline framework - Moved pipeline core, controllers, adapters, and registry to - Moved display system to - Moved effects system to - Created plugin system with security and compatibility management in - Created preset pack system with ASCII art encoding in - Added default font (Corptic) to - Added terminal ANSI constants to ### Mainline Application (Updated) - Created for Mainline stage component registration - Updated to register Mainline stages at startup - Updated as a compatibility shim re-exporting from sideline ### Terminology Consistency - : Base class for all pipeline components (sources, effects, displays, cameras) - : Base class for distributable plugin packages (was ) - : Base class for visual effects (was ) - Backward compatibility aliases maintained for existing code ## Key Features - Plugin discovery via entry points and explicit registration - Security permissions system for plugins - Compatibility management with semantic version constraints - Preset pack system for distributable configurations - Default font bundled with Sideline (Corptic.otf) ## Testing - Updated tests to register Mainline stages before discovery - All StageRegistry tests passing Note: This is a major refactoring that separates the framework (Sideline) from the application (Mainline), enabling Sideline to be used by other applications.
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
|