206 lines
7.2 KiB
Python
206 lines
7.2 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
import threading
|
|
import time
|
|
import unittest
|
|
|
|
from app.core.controller import InfinityMirrorController
|
|
from app.output.base import OutputBackend, OutputDiagnostics, OutputResult
|
|
from app.output.manager import OutputManager
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SAMPLE_MAPPING = ROOT / "sample_data" / "infinity_mirror_mapping_clean.xml"
|
|
|
|
|
|
class _NamedBackend(OutputBackend):
|
|
backend_id = "preview"
|
|
display_name = "Preview Only"
|
|
supports_live_output = False
|
|
|
|
def send_frame(self, config, frame) -> OutputResult:
|
|
return OutputResult(ok=True)
|
|
|
|
|
|
class _LiveNamedBackend(OutputBackend):
|
|
backend_id = "live"
|
|
display_name = "Live Backend"
|
|
supports_live_output = True
|
|
|
|
def send_frame(self, config, frame) -> OutputResult:
|
|
return OutputResult(ok=True)
|
|
|
|
|
|
class RecordingOutputManager:
|
|
def __init__(self) -> None:
|
|
self.output_enabled = False
|
|
self.active_backend_id = "preview"
|
|
self._backend = _NamedBackend()
|
|
self.submitted_frames = []
|
|
|
|
def update_config(self, config) -> None:
|
|
self._config = config.clone()
|
|
|
|
def submit_frame(self, frame):
|
|
self.submitted_frames.append(frame)
|
|
return OutputResult(ok=True)
|
|
|
|
def drain_status_messages(self) -> list[str]:
|
|
return []
|
|
|
|
def diagnostics_snapshot(self) -> OutputDiagnostics:
|
|
return OutputDiagnostics(
|
|
backend_id=self.active_backend_id,
|
|
backend_name=self._backend.display_name,
|
|
output_enabled=self.output_enabled,
|
|
worker_running=False,
|
|
target_fps=25.0,
|
|
frames_submitted=len(self.submitted_frames),
|
|
controller_fps=22.5,
|
|
controller_live_devices=2,
|
|
controller_sampled_devices=2,
|
|
controller_total_devices=2,
|
|
controller_source="stub",
|
|
)
|
|
|
|
def set_active_backend(self, backend_id: str) -> None:
|
|
self.active_backend_id = backend_id
|
|
|
|
def set_output_enabled(self, enabled: bool) -> None:
|
|
self.output_enabled = bool(enabled)
|
|
|
|
def active_backend(self) -> OutputBackend:
|
|
return self._backend
|
|
|
|
def shutdown(self) -> None:
|
|
pass
|
|
|
|
|
|
class LiveRecordingOutputManager(RecordingOutputManager):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.active_backend_id = "live"
|
|
self._backend = _LiveNamedBackend()
|
|
|
|
|
|
class SlowBackend(OutputBackend):
|
|
backend_id = "slow"
|
|
display_name = "Slow Backend"
|
|
supports_live_output = True
|
|
|
|
def __init__(self, delay_s: float = 0.15) -> None:
|
|
self.delay_s = delay_s
|
|
self.send_event = threading.Event()
|
|
|
|
def send_frame(self, config, frame) -> OutputResult:
|
|
time.sleep(self.delay_s)
|
|
self.send_event.set()
|
|
return OutputResult(ok=True, packets_sent=1, device_count=1)
|
|
|
|
|
|
class RealtimeOutputTests(unittest.TestCase):
|
|
def test_foh_preview_does_not_submit_next_scene_to_hardware(self) -> None:
|
|
output_manager = RecordingOutputManager()
|
|
controller = InfinityMirrorController(ROOT, output_manager=output_manager)
|
|
try:
|
|
controller.timer.stop()
|
|
controller.load_mapping(SAMPLE_MAPPING)
|
|
controller.set_foh_mode(True)
|
|
controller.set_pattern("sparkle")
|
|
|
|
controller.render_once(timestamp=10.0)
|
|
|
|
self.assertTrue(output_manager.submitted_frames)
|
|
self.assertEqual(output_manager.submitted_frames[-1].pattern_id, "solid")
|
|
self.assertEqual(controller.preview_frame_for("next").pattern_id, "sparkle")
|
|
finally:
|
|
controller.shutdown()
|
|
|
|
def test_render_once_does_not_wait_for_slow_output_backend(self) -> None:
|
|
output_manager = OutputManager(target_fps=25.0)
|
|
slow_backend = SlowBackend(delay_s=0.15)
|
|
output_manager.backends[slow_backend.backend_id] = slow_backend
|
|
output_manager.set_active_backend(slow_backend.backend_id)
|
|
|
|
controller = InfinityMirrorController(ROOT, output_manager=output_manager)
|
|
try:
|
|
controller.timer.stop()
|
|
controller.load_mapping(SAMPLE_MAPPING)
|
|
controller.set_output_target_fps(20.0)
|
|
output_manager.set_output_enabled(True)
|
|
slow_backend.send_event.clear()
|
|
|
|
started_at = time.perf_counter()
|
|
controller.render_once(timestamp=10.0)
|
|
duration_s = time.perf_counter() - started_at
|
|
|
|
self.assertLess(duration_s, 0.08)
|
|
self.assertTrue(slow_backend.send_event.wait(0.5))
|
|
|
|
diagnostics = controller.realtime_diagnostics()
|
|
self.assertEqual(diagnostics.backend_id, slow_backend.backend_id)
|
|
self.assertEqual(diagnostics.target_output_fps, 20.0)
|
|
self.assertGreaterEqual(diagnostics.frames_rendered, 1)
|
|
self.assertEqual(diagnostics.packets_last_frame, 1)
|
|
finally:
|
|
controller.shutdown()
|
|
|
|
def test_realtime_diagnostics_include_controller_side_metrics_when_available(self) -> None:
|
|
output_manager = RecordingOutputManager()
|
|
controller = InfinityMirrorController(ROOT, output_manager=output_manager)
|
|
try:
|
|
controller.timer.stop()
|
|
controller.load_mapping(SAMPLE_MAPPING)
|
|
|
|
diagnostics = controller.realtime_diagnostics()
|
|
|
|
self.assertEqual(diagnostics.controller_fps, 22.5)
|
|
self.assertEqual(diagnostics.controller_live_devices, 2)
|
|
self.assertEqual(diagnostics.controller_total_devices, 2)
|
|
self.assertEqual(diagnostics.controller_source, "stub")
|
|
finally:
|
|
controller.shutdown()
|
|
|
|
def test_live_output_throttles_preview_signals_but_not_frame_submission(self) -> None:
|
|
output_manager = LiveRecordingOutputManager()
|
|
controller = InfinityMirrorController(ROOT, output_manager=output_manager)
|
|
try:
|
|
controller.timer.stop()
|
|
controller.load_mapping(SAMPLE_MAPPING)
|
|
controller.set_output_enabled(True)
|
|
|
|
emitted_timestamps: list[float] = []
|
|
controller.frame_ready.connect(lambda frame: emitted_timestamps.append(frame.timestamp))
|
|
output_manager.submitted_frames.clear()
|
|
|
|
for timestamp in (10.0, 10.02, 10.04, 10.09):
|
|
controller.render_once(timestamp=timestamp)
|
|
|
|
self.assertEqual(len(output_manager.submitted_frames), 4)
|
|
self.assertEqual(emitted_timestamps, [10.0, 10.09])
|
|
finally:
|
|
controller.shutdown()
|
|
|
|
def test_user_triggered_pattern_change_forces_preview_refresh_during_live_output(self) -> None:
|
|
output_manager = LiveRecordingOutputManager()
|
|
controller = InfinityMirrorController(ROOT, output_manager=output_manager)
|
|
try:
|
|
controller.timer.stop()
|
|
controller.load_mapping(SAMPLE_MAPPING)
|
|
controller.set_output_enabled(True)
|
|
controller.render_once(timestamp=10.0)
|
|
|
|
emitted_patterns: list[str] = []
|
|
controller.frame_ready.connect(lambda frame: emitted_patterns.append(frame.pattern_id))
|
|
|
|
controller.set_pattern("sparkle")
|
|
|
|
self.assertIn("sparkle", emitted_patterns)
|
|
finally:
|
|
controller.shutdown()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|