feat(display): add reuse flag to Display protocol

- Add reuse parameter to Display.init() for all backends
- PygameDisplay: reuse existing SDL window via class-level flag
- TerminalDisplay: skip re-init when reuse=True
- WebSocketDisplay: skip server start when reuse=True
- SixelDisplay, KittyDisplay, NullDisplay: ignore reuse (not applicable)
- MultiDisplay: pass reuse to child displays
- Update benchmark.py to reuse pygame display for effect benchmarks
- Add test_websocket_e2e.py with e2e marker
- Register e2e marker in pyproject.toml
This commit is contained in:
2026-03-16 00:30:52 -07:00
parent f9991c24af
commit f5de2c62e0
13 changed files with 309 additions and 34 deletions

View File

@@ -62,9 +62,21 @@ def get_sample_buffer(width: int = 80, height: int = 24) -> list[str]:
def benchmark_display(
display_class, buffer: list[str], iterations: int = 100
display_class,
buffer: list[str],
iterations: int = 100,
display=None,
reuse: bool = False,
) -> BenchmarkResult | None:
"""Benchmark a single display."""
"""Benchmark a single display.
Args:
display_class: Display class to instantiate
buffer: Buffer to display
iterations: Number of iterations
display: Optional existing display instance to reuse
reuse: If True and display provided, use reuse mode
"""
old_stdout = sys.stdout
old_stderr = sys.stderr
@@ -72,8 +84,12 @@ def benchmark_display(
sys.stdout = StringIO()
sys.stderr = StringIO()
display = display_class()
display.init(80, 24)
if display is None:
display = display_class()
display.init(80, 24, reuse=False)
should_cleanup = True
else:
should_cleanup = False
times = []
chars = sum(len(line) for line in buffer)
@@ -84,7 +100,8 @@ def benchmark_display(
elapsed = (time.perf_counter() - t0) * 1000
times.append(elapsed)
display.cleanup()
if should_cleanup and hasattr(display, "cleanup"):
display.cleanup(quit_pygame=False)
except Exception:
return None
@@ -113,9 +130,17 @@ def benchmark_display(
def benchmark_effect_with_display(
effect_class, display, buffer: list[str], iterations: int = 100
effect_class, display, buffer: list[str], iterations: int = 100, reuse: bool = False
) -> BenchmarkResult | None:
"""Benchmark an effect with a display."""
"""Benchmark an effect with a display.
Args:
effect_class: Effect class to instantiate
display: Display instance to use
buffer: Buffer to process and display
iterations: Number of iterations
reuse: If True, use reuse mode for display
"""
old_stdout = sys.stdout
old_stderr = sys.stderr
@@ -149,7 +174,8 @@ def benchmark_effect_with_display(
elapsed = (time.perf_counter() - t0) * 1000
times.append(elapsed)
display.cleanup()
if not reuse and hasattr(display, "cleanup"):
display.cleanup(quit_pygame=False)
except Exception:
return None
@@ -206,6 +232,13 @@ def get_available_displays():
except Exception:
pass
try:
from engine.display.backends.pygame import PygameDisplay
displays.append(("pygame", PygameDisplay))
except Exception:
pass
return displays
@@ -255,6 +288,7 @@ def run_benchmarks(
if verbose:
print(f"Running benchmarks ({iterations} iterations each)...")
pygame_display = None
for name, display_class in displays:
if verbose:
print(f"Benchmarking display: {name}")
@@ -265,13 +299,44 @@ def run_benchmarks(
if verbose:
print(f" {result.fps:.1f} FPS, {result.avg_time_ms:.2f}ms avg")
if name == "pygame":
pygame_display = result
if verbose:
print()
pygame_instance = None
if pygame_display:
try:
from engine.display.backends.pygame import PygameDisplay
PygameDisplay.reset_state()
pygame_instance = PygameDisplay()
pygame_instance.init(80, 24, reuse=False)
except Exception:
pygame_instance = None
for effect_name, effect_class in effects:
for display_name, display_class in displays:
if display_name == "websocket":
continue
if display_name == "pygame":
if verbose:
print(f"Benchmarking effect: {effect_name} with {display_name}")
if pygame_instance:
result = benchmark_effect_with_display(
effect_class, pygame_instance, buffer, iterations, reuse=True
)
if result:
results.append(result)
if verbose:
print(
f" {result.fps:.1f} FPS, {result.avg_time_ms:.2f}ms avg"
)
continue
if verbose:
print(f"Benchmarking effect: {effect_name} with {display_name}")
@@ -285,6 +350,12 @@ def run_benchmarks(
if verbose:
print(f" {result.fps:.1f} FPS, {result.avg_time_ms:.2f}ms avg")
if pygame_instance:
try:
pygame_instance.cleanup(quit_pygame=True)
except Exception:
pass
summary = generate_summary(results)
return BenchmarkReport(