From 829c4ab63def460a7690e4a8bdde0d395f97b6dc Mon Sep 17 00:00:00 2001 From: David Gwilliam Date: Sun, 15 Mar 2026 22:25:28 -0700 Subject: [PATCH] refactor: modularize display backends and add benchmark runner MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Create engine/display/ package with registry pattern - Move displays to engine/display/backends/ (terminal, null, websocket, sixel) - Add DisplayRegistry with auto-discovery - Add benchmark.py for performance testing effects × displays matrix - Add mise tasks: benchmark, benchmark-json, benchmark-report - Update controller to use new display module --- engine/benchmark.py | 431 ++++++++++++++++++ engine/controller.py | 11 +- engine/display/__init__.py | 102 +++++ engine/display/backends/multi.py | 33 ++ engine/display/backends/null.py | 32 ++ .../{display.py => display/backends/sixel.py} | 131 +----- engine/display/backends/terminal.py | 48 ++ .../backends/websocket.py} | 18 +- mise.toml | 8 + tests/test_websocket.py | 30 +- 10 files changed, 694 insertions(+), 150 deletions(-) create mode 100644 engine/benchmark.py create mode 100644 engine/display/__init__.py create mode 100644 engine/display/backends/multi.py create mode 100644 engine/display/backends/null.py rename engine/{display.py => display/backends/sixel.py} (70%) create mode 100644 engine/display/backends/terminal.py rename engine/{websocket_display.py => display/backends/websocket.py} (95%) diff --git a/engine/benchmark.py b/engine/benchmark.py new file mode 100644 index 0000000..e4a3882 --- /dev/null +++ b/engine/benchmark.py @@ -0,0 +1,431 @@ +#!/usr/bin/env python3 +""" +Benchmark runner for mainline - tests performance across effects and displays. + +Usage: + python -m engine.benchmark + python -m engine.benchmark --output report.md + python -m engine.benchmark --displays terminal,websocket --effects glitch,fade +""" + +import argparse +import json +import sys +import time +from dataclasses import dataclass, field +from typing import Any + +import numpy as np + + +@dataclass +class BenchmarkResult: + """Result of a single benchmark run.""" + + name: str + display: str + effect: str | None + iterations: int + total_time_ms: float + avg_time_ms: float + std_dev_ms: float + min_ms: float + max_ms: float + fps: float + chars_processed: int + chars_per_sec: float + + +@dataclass +class BenchmarkReport: + """Complete benchmark report.""" + + timestamp: str + python_version: str + results: list[BenchmarkResult] = field(default_factory=list) + summary: dict[str, Any] = field(default_factory=dict) + + +def get_sample_buffer(width: int = 80, height: int = 24) -> list[str]: + """Generate a sample buffer for benchmarking.""" + lines = [] + for i in range(height): + line = f"\x1b[32mLine {i}\x1b[0m " + "A" * (width - 10) + lines.append(line) + return lines + + +def benchmark_display( + display_class, buffer: list[str], iterations: int = 100 +) -> BenchmarkResult: + """Benchmark a single display.""" + display = display_class() + display.init(80, 24) + + times = [] + chars = sum(len(line) for line in buffer) + + for _ in range(iterations): + t0 = time.perf_counter() + display.show(buffer) + elapsed = (time.perf_counter() - t0) * 1000 + times.append(elapsed) + + display.cleanup() + + times_arr = np.array(times) + + return BenchmarkResult( + name=f"display_{display_class.__name__}", + display=display_class.__name__, + effect=None, + iterations=iterations, + total_time_ms=sum(times), + avg_time_ms=np.mean(times_arr), + std_dev_ms=np.std(times_arr), + min_ms=np.min(times_arr), + max_ms=np.max(times_arr), + fps=1000.0 / np.mean(times_arr) if np.mean(times_arr) > 0 else 0, + chars_processed=chars * iterations, + chars_per_sec=(chars * iterations) / (sum(times) / 1000) + if sum(times) > 0 + else 0, + ) + + +def benchmark_effect_with_display( + effect_class, display, buffer: list[str], iterations: int = 100 +) -> BenchmarkResult: + """Benchmark an effect with a display.""" + effect = effect_class() + effect.configure(enabled=True, intensity=1.0) + + times = [] + chars = sum(len(line) for line in buffer) + + for _ in range(iterations): + processed = effect.process(buffer) + t0 = time.perf_counter() + display.show(processed) + elapsed = (time.perf_counter() - t0) * 1000 + times.append(elapsed) + + display.cleanup() + + times_arr = np.array(times) + + return BenchmarkResult( + name=f"effect_{effect_class.__name__}_with_{display.__class__.__name__}", + display=display.__class__.__name__, + effect=effect_class.__name__, + iterations=iterations, + total_time_ms=sum(times), + avg_time_ms=np.mean(times_arr), + std_dev_ms=np.std(times_arr), + min_ms=np.min(times_arr), + max_ms=np.max(times_arr), + fps=1000.0 / np.mean(times_arr) if np.mean(times_arr) > 0 else 0, + chars_processed=chars * iterations, + chars_per_sec=(chars * iterations) / (sum(times) / 1000) + if sum(times) > 0 + else 0, + ) + + +def get_available_displays(): + """Get available display classes.""" + from engine.display import ( + DisplayRegistry, + NullDisplay, + TerminalDisplay, + ) + from engine.display.backends.sixel import SixelDisplay + + DisplayRegistry.initialize() + + displays = [ + ("null", NullDisplay), + ("terminal", TerminalDisplay), + ] + + try: + from engine.display.backends.websocket import WebSocketDisplay + + displays.append(("websocket", WebSocketDisplay)) + except Exception: + pass + + try: + displays.append(("sixel", SixelDisplay)) + except Exception: + pass + + return displays + + +def get_available_effects(): + """Get available effect classes.""" + try: + from engine.effects.registry import get_effect_registry + except Exception: + return [] + + effects = [] + registry = get_effect_registry() + + for name in registry.list_effects(): + effect = registry.get(name) + if effect: + effects.append((name, effect)) + + return effects + + +def run_benchmarks( + displays: list[tuple[str, Any]] | None = None, + effects: list[tuple[str, Any]] | None = None, + iterations: int = 100, + output_format: str = "text", +) -> BenchmarkReport: + """Run all benchmarks and return report.""" + from datetime import datetime + + if displays is None: + displays = get_available_displays() + + if effects is None: + effects = get_available_effects() + + buffer = get_sample_buffer(80, 24) + results = [] + + print(f"Running benchmarks ({iterations} iterations each)...") + print() + + for name, display_class in displays: + print(f"Benchmarking display: {name}") + try: + result = benchmark_display(display_class, buffer, iterations) + results.append(result) + print(f" {result.fps:.1f} FPS, {result.avg_time_ms:.2f}ms avg") + except Exception as e: + print(f" Error: {e}") + + print() + + for effect_name, effect_class in effects: + for display_name, display_class in displays: + if display_name == "websocket": + continue + print(f"Benchmarking effect: {effect_name} with {display_name}") + try: + display = display_class() + display.init(80, 24) + result = benchmark_effect_with_display( + effect_class, display, buffer, iterations + ) + results.append(result) + print(f" {result.fps:.1f} FPS, {result.avg_time_ms:.2f}ms avg") + except Exception as e: + print(f" Error: {e}") + + summary = generate_summary(results) + + return BenchmarkReport( + timestamp=datetime.now().isoformat(), + python_version=sys.version, + results=results, + summary=summary, + ) + + +def generate_summary(results: list[BenchmarkResult]) -> dict[str, Any]: + """Generate summary statistics from results.""" + by_display: dict[str, list[BenchmarkResult]] = {} + by_effect: dict[str, list[BenchmarkResult]] = {} + + for r in results: + if r.display not in by_display: + by_display[r.display] = [] + by_display[r.display].append(r) + + if r.effect: + if r.effect not in by_effect: + by_effect[r.effect] = [] + by_effect[r.effect].append(r) + + summary = { + "by_display": {}, + "by_effect": {}, + "overall": { + "total_tests": len(results), + "displays_tested": len(by_display), + "effects_tested": len(by_effect), + }, + } + + for display, res in by_display.items(): + fps_values = [r.fps for r in res] + summary["by_display"][display] = { + "avg_fps": np.mean(fps_values), + "min_fps": np.min(fps_values), + "max_fps": np.max(fps_values), + "tests": len(res), + } + + for effect, res in by_effect.items(): + fps_values = [r.fps for r in res] + summary["by_effect"][effect] = { + "avg_fps": np.mean(fps_values), + "min_fps": np.min(fps_values), + "max_fps": np.max(fps_values), + "tests": len(res), + } + + return summary + + +def format_report_text(report: BenchmarkReport) -> str: + """Format report as human-readable text.""" + lines = [ + "# Mainline Performance Benchmark Report", + "", + f"Generated: {report.timestamp}", + f"Python: {report.python_version}", + "", + "## Summary", + "", + f"Total tests: {report.summary['overall']['total_tests']}", + f"Displays tested: {report.summary['overall']['displays_tested']}", + f"Effects tested: {report.summary['overall']['effects_tested']}", + "", + "## By Display", + "", + ] + + for display, stats in report.summary["by_display"].items(): + lines.append(f"### {display}") + lines.append(f"- Avg FPS: {stats['avg_fps']:.1f}") + lines.append(f"- Min FPS: {stats['min_fps']:.1f}") + lines.append(f"- Max FPS: {stats['max_fps']:.1f}") + lines.append(f"- Tests: {stats['tests']}") + lines.append("") + + if report.summary["by_effect"]: + lines.append("## By Effect") + lines.append("") + + for effect, stats in report.summary["by_effect"].items(): + lines.append(f"### {effect}") + lines.append(f"- Avg FPS: {stats['avg_fps']:.1f}") + lines.append(f"- Min FPS: {stats['min_fps']:.1f}") + lines.append(f"- Max FPS: {stats['max_fps']:.1f}") + lines.append(f"- Tests: {stats['tests']}") + lines.append("") + + lines.append("## Detailed Results") + lines.append("") + lines.append("| Display | Effect | FPS | Avg ms | StdDev ms | Min ms | Max ms |") + lines.append("|---------|--------|-----|--------|-----------|--------|--------|") + + for r in report.results: + effect_col = r.effect if r.effect else "-" + lines.append( + f"| {r.display} | {effect_col} | {r.fps:.1f} | {r.avg_time_ms:.2f} | " + f"{r.std_dev_ms:.2f} | {r.min_ms:.2f} | {r.max_ms:.2f} |" + ) + + return "\n".join(lines) + + +def format_report_json(report: BenchmarkReport) -> str: + """Format report as JSON.""" + data = { + "timestamp": report.timestamp, + "python_version": report.python_version, + "summary": report.summary, + "results": [ + { + "name": r.name, + "display": r.display, + "effect": r.effect, + "iterations": r.iterations, + "total_time_ms": r.total_time_ms, + "avg_time_ms": r.avg_time_ms, + "std_dev_ms": r.std_dev_ms, + "min_ms": r.min_ms, + "max_ms": r.max_ms, + "fps": r.fps, + "chars_processed": r.chars_processed, + "chars_per_sec": r.chars_per_sec, + } + for r in report.results + ], + } + return json.dumps(data, indent=2) + + +def main(): + parser = argparse.ArgumentParser(description="Run mainline benchmarks") + parser.add_argument( + "--displays", + help="Comma-separated list of displays to test (default: all)", + ) + parser.add_argument( + "--effects", + help="Comma-separated list of effects to test (default: all)", + ) + parser.add_argument( + "--iterations", + type=int, + default=100, + help="Number of iterations per test (default: 100)", + ) + parser.add_argument( + "--output", + help="Output file path (default: stdout)", + ) + parser.add_argument( + "--format", + choices=["text", "json"], + default="text", + help="Output format (default: text)", + ) + + args = parser.parse_args() + + displays = None + if args.displays: + display_map = dict(get_available_displays()) + displays = [ + (name, display_map[name]) + for name in args.displays.split(",") + if name in display_map + ] + + effects = None + if args.effects: + effect_map = dict(get_available_effects()) + effects = [ + (name, effect_map[name]) + for name in args.effects.split(",") + if name in effect_map + ] + + report = run_benchmarks(displays, effects, args.iterations, args.format) + + if args.format == "json": + output = format_report_json(report) + else: + output = format_report_text(report) + + if args.output: + with open(args.output, "w") as f: + f.write(output) + print(f"Report written to {args.output}") + else: + print(output) + + +if __name__ == "__main__": + main() diff --git a/engine/controller.py b/engine/controller.py index cd7c2e6..0d7bf6f 100644 --- a/engine/controller.py +++ b/engine/controller.py @@ -3,18 +3,25 @@ Stream controller - manages input sources and orchestrates the render stream. """ from engine.config import Config, get_config -from engine.display import MultiDisplay, NullDisplay, SixelDisplay, TerminalDisplay +from engine.display import ( + DisplayRegistry, + MultiDisplay, + NullDisplay, + SixelDisplay, + TerminalDisplay, + WebSocketDisplay, +) from engine.effects.controller import handle_effects_command from engine.eventbus import EventBus from engine.events import EventType, StreamEvent from engine.mic import MicMonitor from engine.ntfy import NtfyPoller from engine.scroll import stream -from engine.websocket_display import WebSocketDisplay def _get_display(config: Config): """Get the appropriate display based on config.""" + DisplayRegistry.initialize() display_mode = config.display.lower() displays = [] diff --git a/engine/display/__init__.py b/engine/display/__init__.py new file mode 100644 index 0000000..d092de1 --- /dev/null +++ b/engine/display/__init__.py @@ -0,0 +1,102 @@ +""" +Display backend system with registry pattern. + +Allows swapping output backends via the Display protocol. +Supports auto-discovery of display backends. +""" + +from typing import Protocol + +from engine.display.backends.multi import MultiDisplay +from engine.display.backends.null import NullDisplay +from engine.display.backends.sixel import SixelDisplay +from engine.display.backends.terminal import TerminalDisplay +from engine.display.backends.websocket import WebSocketDisplay + + +class Display(Protocol): + """Protocol for display backends.""" + + width: int + height: int + + def init(self, width: int, height: int) -> None: + """Initialize display with dimensions.""" + ... + + def show(self, buffer: list[str]) -> None: + """Show buffer on display.""" + ... + + def clear(self) -> None: + """Clear display.""" + ... + + def cleanup(self) -> None: + """Shutdown display.""" + ... + + +class DisplayRegistry: + """Registry for display backends with auto-discovery.""" + + _backends: dict[str, type[Display]] = {} + _initialized = False + + @classmethod + def register(cls, name: str, backend_class: type[Display]) -> None: + """Register a display backend.""" + cls._backends[name.lower()] = backend_class + + @classmethod + def get(cls, name: str) -> type[Display] | None: + """Get a display backend class by name.""" + return cls._backends.get(name.lower()) + + @classmethod + def list_backends(cls) -> list[str]: + """List all available display backend names.""" + return list(cls._backends.keys()) + + @classmethod + def create(cls, name: str, **kwargs) -> Display | None: + """Create a display instance by name.""" + backend_class = cls.get(name) + if backend_class: + return backend_class(**kwargs) + return None + + @classmethod + def initialize(cls) -> None: + """Initialize and register all built-in backends.""" + if cls._initialized: + return + + cls.register("terminal", TerminalDisplay) + cls.register("null", NullDisplay) + cls.register("websocket", WebSocketDisplay) + cls.register("sixel", SixelDisplay) + + cls._initialized = True + + +def get_monitor(): + """Get the performance monitor.""" + try: + from engine.effects.performance import get_monitor as _get_monitor + + return _get_monitor() + except Exception: + return None + + +__all__ = [ + "Display", + "DisplayRegistry", + "get_monitor", + "TerminalDisplay", + "NullDisplay", + "WebSocketDisplay", + "SixelDisplay", + "MultiDisplay", +] diff --git a/engine/display/backends/multi.py b/engine/display/backends/multi.py new file mode 100644 index 0000000..d37667d --- /dev/null +++ b/engine/display/backends/multi.py @@ -0,0 +1,33 @@ +""" +Multi display backend - forwards to multiple displays. +""" + + +class MultiDisplay: + """Display that forwards to multiple displays.""" + + width: int = 80 + height: int = 24 + + def __init__(self, displays: list): + self.displays = displays + self.width = 80 + self.height = 24 + + def init(self, width: int, height: int) -> None: + self.width = width + self.height = height + for d in self.displays: + d.init(width, height) + + def show(self, buffer: list[str]) -> None: + for d in self.displays: + d.show(buffer) + + def clear(self) -> None: + for d in self.displays: + d.clear() + + def cleanup(self) -> None: + for d in self.displays: + d.cleanup() diff --git a/engine/display/backends/null.py b/engine/display/backends/null.py new file mode 100644 index 0000000..1865f52 --- /dev/null +++ b/engine/display/backends/null.py @@ -0,0 +1,32 @@ +""" +Null/headless display backend. +""" + +import time + + +class NullDisplay: + """Headless/null display - discards all output.""" + + width: int = 80 + height: int = 24 + + def init(self, width: int, height: int) -> None: + self.width = width + self.height = height + + def show(self, buffer: list[str]) -> None: + from engine.display import get_monitor + + monitor = get_monitor() + if monitor: + t0 = time.perf_counter() + chars_in = sum(len(line) for line in buffer) + elapsed_ms = (time.perf_counter() - t0) * 1000 + monitor.record_effect("null_display", elapsed_ms, chars_in, chars_in) + + def clear(self) -> None: + pass + + def cleanup(self) -> None: + pass diff --git a/engine/display.py b/engine/display/backends/sixel.py similarity index 70% rename from engine/display.py rename to engine/display/backends/sixel.py index 912096a..56f3991 100644 --- a/engine/display.py +++ b/engine/display/backends/sixel.py @@ -1,132 +1,8 @@ """ -Display output abstraction - allows swapping output backends. - -Protocol: - - init(width, height): Initialize display with terminal dimensions - - show(buffer): Render buffer (list of strings) to display - - clear(): Clear the display - - cleanup(): Shutdown display +Sixel graphics display backend - renders to sixel graphics in terminal. """ import time -from typing import Protocol - - -class Display(Protocol): - """Protocol for display backends.""" - - def init(self, width: int, height: int) -> None: - """Initialize display with dimensions.""" - ... - - def show(self, buffer: list[str]) -> None: - """Show buffer on display.""" - ... - - def clear(self) -> None: - """Clear display.""" - ... - - def cleanup(self) -> None: - """Shutdown display.""" - ... - - -def get_monitor(): - """Get the performance monitor.""" - try: - from engine.effects.performance import get_monitor as _get_monitor - - return _get_monitor() - except Exception: - return None - - -class TerminalDisplay: - """ANSI terminal display backend.""" - - def __init__(self): - self.width = 80 - self.height = 24 - - def init(self, width: int, height: int) -> None: - from engine.terminal import CURSOR_OFF - - self.width = width - self.height = height - print(CURSOR_OFF, end="", flush=True) - - def show(self, buffer: list[str]) -> None: - import sys - - t0 = time.perf_counter() - sys.stdout.buffer.write("".join(buffer).encode()) - sys.stdout.flush() - elapsed_ms = (time.perf_counter() - t0) * 1000 - - monitor = get_monitor() - if monitor: - chars_in = sum(len(line) for line in buffer) - monitor.record_effect("terminal_display", elapsed_ms, chars_in, chars_in) - - def clear(self) -> None: - from engine.terminal import CLR - - print(CLR, end="", flush=True) - - def cleanup(self) -> None: - from engine.terminal import CURSOR_ON - - print(CURSOR_ON, end="", flush=True) - - -class NullDisplay: - """Headless/null display - discards all output.""" - - def init(self, width: int, height: int) -> None: - self.width = width - self.height = height - - def show(self, buffer: list[str]) -> None: - monitor = get_monitor() - if monitor: - t0 = time.perf_counter() - chars_in = sum(len(line) for line in buffer) - elapsed_ms = (time.perf_counter() - t0) * 1000 - monitor.record_effect("null_display", elapsed_ms, chars_in, chars_in) - - def clear(self) -> None: - pass - - def cleanup(self) -> None: - pass - - -class MultiDisplay: - """Display that forwards to multiple displays.""" - - def __init__(self, displays: list[Display]): - self.displays = displays - self.width = 80 - self.height = 24 - - def init(self, width: int, height: int) -> None: - self.width = width - self.height = height - for d in self.displays: - d.init(width, height) - - def show(self, buffer: list[str]) -> None: - for d in self.displays: - d.show(buffer) - - def clear(self) -> None: - for d in self.displays: - d.clear() - - def cleanup(self) -> None: - for d in self.displays: - d.cleanup() def _parse_ansi( @@ -303,6 +179,9 @@ def _encode_sixel(image) -> str: class SixelDisplay: """Sixel graphics display backend - renders to sixel graphics in terminal.""" + width: int = 80 + height: int = 24 + def __init__(self, cell_width: int = 9, cell_height: int = 16): self.width = 80 self.height = 24 @@ -373,6 +252,8 @@ class SixelDisplay: elapsed_ms = (time.perf_counter() - t0) * 1000 + from engine.display import get_monitor + monitor = get_monitor() if monitor: chars_in = sum(len(line) for line in buffer) diff --git a/engine/display/backends/terminal.py b/engine/display/backends/terminal.py new file mode 100644 index 0000000..a42c761 --- /dev/null +++ b/engine/display/backends/terminal.py @@ -0,0 +1,48 @@ +""" +ANSI terminal display backend. +""" + +import time + + +class TerminalDisplay: + """ANSI terminal display backend.""" + + width: int = 80 + height: int = 24 + + def __init__(self): + self.width = 80 + self.height = 24 + + def init(self, width: int, height: int) -> None: + from engine.terminal import CURSOR_OFF + + self.width = width + self.height = height + print(CURSOR_OFF, end="", flush=True) + + def show(self, buffer: list[str]) -> None: + import sys + + t0 = time.perf_counter() + sys.stdout.buffer.write("".join(buffer).encode()) + sys.stdout.flush() + elapsed_ms = (time.perf_counter() - t0) * 1000 + + from engine.display import get_monitor + + monitor = get_monitor() + if monitor: + chars_in = sum(len(line) for line in buffer) + monitor.record_effect("terminal_display", elapsed_ms, chars_in, chars_in) + + def clear(self) -> None: + from engine.terminal import CLR + + print(CLR, end="", flush=True) + + def cleanup(self) -> None: + from engine.terminal import CURSOR_ON + + print(CURSOR_ON, end="", flush=True) diff --git a/engine/websocket_display.py b/engine/display/backends/websocket.py similarity index 95% rename from engine/websocket_display.py rename to engine/display/backends/websocket.py index 6ff7d36..6f0117b 100644 --- a/engine/websocket_display.py +++ b/engine/display/backends/websocket.py @@ -1,11 +1,5 @@ """ -WebSocket display server - broadcasts frame buffer to connected web clients. - -Usage: - ws_display = WebSocketDisplay(host="0.0.0.0", port=8765) - ws_display.init(80, 24) - ws_display.show(["line1", "line2", ...]) - ws_display.cleanup() +WebSocket display backend - broadcasts frame buffer to connected web clients. """ import asyncio @@ -23,6 +17,9 @@ except ImportError: class Display(Protocol): """Protocol for display backends.""" + width: int + height: int + def init(self, width: int, height: int) -> None: """Initialize display with dimensions.""" ... @@ -53,6 +50,9 @@ def get_monitor(): class WebSocketDisplay: """WebSocket display backend - broadcasts to HTML Canvas clients.""" + width: int = 80 + height: int = 24 + def __init__( self, host: str = "0.0.0.0", @@ -177,7 +177,9 @@ class WebSocketDisplay: import os from http.server import HTTPServer, SimpleHTTPRequestHandler - client_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "client") + client_dir = os.path.join( + os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "client" + ) class Handler(SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): diff --git a/mise.toml b/mise.toml index b817122..a51b61c 100644 --- a/mise.toml +++ b/mise.toml @@ -44,6 +44,14 @@ run-client = { run = "mise run run-both & sleep 2 && $(open http://localhost:876 cmd = "uv run cmdline.py" cmd-stats = { run = "uv run cmdline.py -w \"/effects stats\"", depends = ["sync-all"] } +# ===================== +# Benchmark +# ===================== + +benchmark = { run = "uv run python -m engine.benchmark", depends = ["sync-all"] } +benchmark-json = { run = "uv run python -m engine.benchmark --format json --output benchmark.json", depends = ["sync-all"] } +benchmark-report = { run = "uv run python -m engine.benchmark --output BENCHMARK.md", depends = ["sync-all"] } + # Initialize ntfy topics (warm up before first use) topics-init = "curl -s -d 'init' https://ntfy.sh/klubhaus_terminal_mainline_cc_cmd > /dev/null && curl -s -d 'init' https://ntfy.sh/klubhaus_terminal_mainline_cc_resp > /dev/null && curl -s -d 'init' https://ntfy.sh/klubhaus_terminal_mainline > /dev/null" diff --git a/tests/test_websocket.py b/tests/test_websocket.py index 391f2d9..50a4641 100644 --- a/tests/test_websocket.py +++ b/tests/test_websocket.py @@ -1,12 +1,12 @@ """ -Tests for engine.websocket_display module. +Tests for engine.display.backends.websocket module. """ from unittest.mock import MagicMock, patch import pytest -from engine.websocket_display import WebSocketDisplay +from engine.display.backends.websocket import WebSocketDisplay class TestWebSocketDisplayImport: @@ -14,9 +14,9 @@ class TestWebSocketDisplayImport: def test_import_does_not_error(self): """Module imports without error.""" - from engine import websocket_display + from engine.display import backends - assert websocket_display is not None + assert backends is not None class TestWebSocketDisplayInit: @@ -24,7 +24,7 @@ class TestWebSocketDisplayInit: def test_default_init(self): """Default initialization sets correct defaults.""" - with patch("engine.websocket_display.websockets", None): + with patch("engine.display.backends.websocket.websockets", None): display = WebSocketDisplay() assert display.host == "0.0.0.0" assert display.port == 8765 @@ -34,7 +34,7 @@ class TestWebSocketDisplayInit: def test_custom_init(self): """Custom initialization uses provided values.""" - with patch("engine.websocket_display.websockets", None): + with patch("engine.display.backends.websocket.websockets", None): display = WebSocketDisplay(host="localhost", port=9000, http_port=9001) assert display.host == "localhost" assert display.port == 9000 @@ -60,7 +60,7 @@ class TestWebSocketDisplayProtocol: def test_websocket_display_is_display(self): """WebSocketDisplay satisfies Display protocol.""" - with patch("engine.websocket_display.websockets", MagicMock()): + with patch("engine.display.backends.websocket.websockets", MagicMock()): display = WebSocketDisplay() assert hasattr(display, "init") assert hasattr(display, "show") @@ -73,7 +73,7 @@ class TestWebSocketDisplayMethods: def test_init_stores_dimensions(self): """init stores terminal dimensions.""" - with patch("engine.websocket_display.websockets", MagicMock()): + with patch("engine.display.backends.websocket.websockets", MagicMock()): display = WebSocketDisplay() display.init(100, 40) assert display.width == 100 @@ -81,31 +81,31 @@ class TestWebSocketDisplayMethods: def test_client_count_initially_zero(self): """client_count returns 0 when no clients connected.""" - with patch("engine.websocket_display.websockets", MagicMock()): + with patch("engine.display.backends.websocket.websockets", MagicMock()): display = WebSocketDisplay() assert display.client_count() == 0 def test_get_ws_port(self): """get_ws_port returns configured port.""" - with patch("engine.websocket_display.websockets", MagicMock()): + with patch("engine.display.backends.websocket.websockets", MagicMock()): display = WebSocketDisplay(port=9000) assert display.get_ws_port() == 9000 def test_get_http_port(self): """get_http_port returns configured port.""" - with patch("engine.websocket_display.websockets", MagicMock()): + with patch("engine.display.backends.websocket.websockets", MagicMock()): display = WebSocketDisplay(http_port=9001) assert display.get_http_port() == 9001 def test_frame_delay_defaults_to_zero(self): """get_frame_delay returns 0 by default.""" - with patch("engine.websocket_display.websockets", MagicMock()): + with patch("engine.display.backends.websocket.websockets", MagicMock()): display = WebSocketDisplay() assert display.get_frame_delay() == 0.0 def test_set_frame_delay(self): """set_frame_delay stores the value.""" - with patch("engine.websocket_display.websockets", MagicMock()): + with patch("engine.display.backends.websocket.websockets", MagicMock()): display = WebSocketDisplay() display.set_frame_delay(0.05) assert display.get_frame_delay() == 0.05 @@ -116,7 +116,7 @@ class TestWebSocketDisplayCallbacks: def test_set_client_connected_callback(self): """set_client_connected_callback stores callback.""" - with patch("engine.websocket_display.websockets", MagicMock()): + with patch("engine.display.backends.websocket.websockets", MagicMock()): display = WebSocketDisplay() callback = MagicMock() display.set_client_connected_callback(callback) @@ -124,7 +124,7 @@ class TestWebSocketDisplayCallbacks: def test_set_client_disconnected_callback(self): """set_client_disconnected_callback stores callback.""" - with patch("engine.websocket_display.websockets", MagicMock()): + with patch("engine.display.backends.websocket.websockets", MagicMock()): display = WebSocketDisplay() callback = MagicMock() display.set_client_disconnected_callback(callback)