ed/manual_slop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(sdm): Global pass with refined 'External Only' SDM tags. Pruned redundant internal references and fixed indentation logic in injector. Verified full project compilation.

Browse Source
This commit is contained in:
Edward R. Gonzalez
2026-05-09 14:32:44 -04:00
parent 696c08692e
commit 8c06c1767b
142 changed files with 2352 additions and 990 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/performance_monitor.py
+34 -7
View File
@@ -75,6 +75,9 @@ class PerformanceScope:
def get_monitor() -> PerformanceMonitor:
"""
[C: tests/test_perf_aggregate.py:test_build_tier3_context_scaling, tests/test_perf_dag.py:test_dag_performance]
"""
global _instance
if _instance is None:
_instance = PerformanceMonitor()
@@ -82,9 +85,10 @@ def get_monitor() -> PerformanceMonitor:
class PerformanceMonitor:
"""
Tracks application performance metrics like FPS, frame time, and CPU usage.
Supports thread-safe tracking for individual components with efficient moving averages.
"""
Tracks application performance metrics like FPS, frame time, and CPU usage.
Supports thread-safe tracking for individual components with efficient moving averages.
"""
def __init__(self, history_size: int = 300) -> None:
self.enabled: bool = False
self.history_size = history_size
@@ -154,6 +158,9 @@ class PerformanceMonitor:
return self._history_sums[key] / len(h)
def start_frame(self) -> None:
"""
[C: tests/test_performance_monitor.py:test_perf_monitor_basic_timing]
"""
now = time.perf_counter()
with self._lock:
if self._last_frame_start_time > 0:
@@ -165,6 +172,9 @@ class PerformanceMonitor:
self._frame_count += 1
def end_frame(self) -> None:
"""
[C: tests/test_performance_monitor.py:test_perf_monitor_basic_timing]
"""
if self._start_time is None:
return
now = time.perf_counter()
@@ -193,12 +203,18 @@ class PerformanceMonitor:
self._fps_timer = 0.0
def start_component(self, name: str) -> None:
"""
[C: tests/test_performance_monitor.py:test_perf_monitor_component_timing, tests/test_performance_monitor.py:test_perf_monitor_extended_metrics]
"""
if not self.enabled: return
now = time.perf_counter()
with self._lock:
self._component_starts[name] = now
def end_component(self, name: str) -> None:
"""
[C: tests/test_performance_monitor.py:test_perf_monitor_component_timing, tests/test_performance_monitor.py:test_perf_monitor_extended_metrics]
"""
if not self.enabled: return
now = time.perf_counter()
with self._lock:
@@ -215,7 +231,10 @@ class PerformanceMonitor:
self._add_to_history(f'comp_{name}', elapsed)
def get_metrics(self) -> dict[str, float]:
"""Returns current metrics and their moving averages. Thread-safe."""
"""
Returns current metrics and their moving averages. Thread-safe.
[C: tests/test_perf_aggregate.py:test_build_tier3_context_scaling, tests/test_perf_dag.py:test_dag_performance, tests/test_performance_monitor.py:test_perf_monitor_basic_timing, tests/test_performance_monitor.py:test_perf_monitor_component_timing, tests/test_performance_monitor.py:test_perf_monitor_extended_metrics, tests/test_performance_monitor.py:test_perf_monitor_scope_context_manager]
"""
with self._lock:
fps = self._fps
last_ft = self._last_frame_time
@@ -246,7 +265,10 @@ class PerformanceMonitor:
return metrics
def get_history(self, key: str) -> List[float]:
"""Returns a snapshot of the full history buffer for a specific metric key."""
"""
Returns a snapshot of the full history buffer for a specific metric key.
[C: tests/test_history.py:test_initial_state, tests/test_history.py:test_push_state]
"""
with self._lock:
if key in self._history:
return list(self._history[key])
@@ -255,11 +277,16 @@ class PerformanceMonitor:
return []
def scope(self, name: str) -> PerformanceScope:
"""Returns a context manager for timing a component."""
"""
Returns a context manager for timing a component.
[C: tests/test_perf_aggregate.py:test_build_tier3_context_scaling, tests/test_performance_monitor.py:test_perf_monitor_scope_context_manager]
"""
return PerformanceScope(self, name)
def stop(self) -> None:
"""
[C: tests/test_performance_monitor.py:test_perf_monitor_basic_timing, tests/test_performance_monitor.py:test_perf_monitor_component_timing, tests/test_performance_monitor.py:test_perf_monitor_extended_metrics, tests/test_performance_monitor.py:test_perf_monitor_scope_context_manager, tests/test_websocket_server.py:test_websocket_subscription_and_broadcast]
"""
self._stop_event.set()
if self._cpu_thread.is_alive():
self._cpu_thread.join(timeout=2.0)