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()