Compare commits

..
Author SHA1 Message Date
ChuckandClaude Fable 5 dd5bc93d87 fix: initialise wifi-status throttle state in __init__
Codacy (pylint access-member-before-definition) on #403: the throttled
early-return read _wifi_status_last_result relying on the non-local
invariant that the first call always passes the throttle window and
assigns it. Correct at runtime, but fragile — initialise both throttle
fields in the constructor and drop the getattr fallback.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FqzC1nzTWL4kaqgMaQZFam
2026-07-13 09:03:52 -04:00
ChuckandClaude Fable 5 512e38647a perf: hot-path micro fixes in the render loop
- _check_wifi_status_message stat'd the status file on every render
  iteration (60+ fps) for a message whose lifetime is seconds; throttle
  the check to 1 Hz with a cached result.
- Demote the per-iteration "Display active, processing mode" INFO to
  DEBUG and convert the remaining eager f-string logs to lazy % args —
  the devpi baseline showed ~9 journald lines/sec, which is both noise
  and SD-card wear.
- Vegas cycle-end blank frame: hoist the inline PIL import and reuse a
  preallocated buffer instead of allocating per cycle wrap.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FqzC1nzTWL4kaqgMaQZFam
2026-07-12 12:24:31 -04:00
4 changed files with 41 additions and 202 deletions
+23 -19
View File
@@ -199,6 +199,10 @@ class DisplayController:
self.wifi_status_file = WIFI_STATUS_FILE
self.wifi_status_active = False
self.wifi_status_expires_at: Optional[float] = None
# _check_wifi_status_message throttle state (checked at frame rate,
# stat'd at most once per second)
self._wifi_status_check_ts = 0.0
self._wifi_status_last_result: Optional[Dict[str, Any]] = None
# Plugin display() signature cache — must be initialised before the plugin
# loading loop below so the .pop() invalidation at load time is always safe.
@@ -1635,7 +1639,7 @@ class DisplayController:
self._sleep_with_plugin_updates(60)
continue
logger.info(f"Display active, processing mode: {self.current_display_mode}")
logger.debug("Display active, processing mode: %s", self.current_display_mode)
# Plugins update on their own schedules - no forced sync updates needed
# Each plugin has its own update_interval and background services
@@ -1803,7 +1807,7 @@ class DisplayController:
if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
should_skip = self.plugin_manager.health_tracker.should_skip_plugin(plugin_id)
if should_skip:
logger.info(f"Skipping plugin {plugin_id} due to circuit breaker (mode: {active_mode})")
logger.info("Skipping plugin %s due to circuit breaker (mode: %s)", plugin_id, active_mode)
display_result = False
# Skip to next mode - let existing logic handle it
manager_to_display = None
@@ -1861,7 +1865,7 @@ class DisplayController:
if isinstance(result, bool):
display_result = result
if not display_result:
logger.info(f"Plugin {plugin_id} display() returned False for mode {active_mode}")
logger.info("Plugin %s display() returned False for mode %s", plugin_id, active_mode)
# Record success if display completed without exception
if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
@@ -2072,21 +2076,10 @@ class DisplayController:
# For plugins, call display multiple times to allow game rotation
if manager_to_display and hasattr(manager_to_display, 'display'):
# High-FPS decision, in precedence order:
# 1. A plugin that declares needs_high_fps knows best
# (e.g. static-image sets it False for still PNGs,
# True for animated GIFs).
# 2. Back-compat: older static-image versions without
# the attribute keep the historical forced high-FPS
# (GIF support).
# 3. Otherwise scrolling plugins get high FPS.
# Check if plugin needs high FPS (like stock ticker)
# Always enable high-FPS for static-image plugin (for GIF animation support)
plugin_id = getattr(manager_to_display, 'plugin_id', None)
declared = getattr(manager_to_display, 'needs_high_fps', None)
if declared is not None:
needs_high_fps = bool(declared)
logger.debug("FPS check for %s - plugin declares needs_high_fps=%s",
active_mode, needs_high_fps)
elif plugin_id == 'static-image':
if plugin_id == 'static-image':
needs_high_fps = True
logger.debug("FPS check - static-image plugin: forcing high-FPS mode for GIF support")
else:
@@ -2365,6 +2358,16 @@ class DisplayController:
Returns None on any error or if message is expired/invalid.
"""
try:
# Throttle the existence stat to ~1 Hz: this runs on every render
# iteration (60+ fps), and the file usually doesn't exist — the
# status message's lifetime is measured in seconds anyway.
# Both attributes are initialised in __init__.
now = time.time()
if (now - self._wifi_status_check_ts) < 1.0:
return self._wifi_status_last_result
self._wifi_status_check_ts = now
self._wifi_status_last_result = None
# Check if file exists
if not self.wifi_status_file or not self.wifi_status_file.exists():
return None
@@ -2415,13 +2418,14 @@ class DisplayController:
pass
return None
# Message is valid and not expired
return {
# Message is valid and not expired — cache for the throttle window
self._wifi_status_last_result = {
'message': message,
'timestamp': timestamp,
'duration': duration,
'expires_at': expires_at
}
return self._wifi_status_last_result
except Exception as e:
# Catch-all for any unexpected errors - log but don't break the display
+2 -44
View File
@@ -36,7 +36,6 @@ import time
from typing import Dict, Any, List, Optional
import logging
import math
import zlib
import freetype
# Get logger without configuring
@@ -189,12 +188,6 @@ class DisplayManager:
self._snapshot_path = "/tmp/led_matrix_preview.png" # nosec B108 - fixed path intentional; web UI reads same path
self._snapshot_min_interval_sec = 0.2 # max ~5 fps
self._last_snapshot_ts = 0.0
# Dirty tracking: (image digest, brightness) of the last frame pushed
# to the panel; update_display() skips identical pushes. Kill switch:
# display.dirty_tracking: false.
self._dirty_tracking_enabled = bool(
self.config.get('display', {}).get('dirty_tracking', True))
self._last_pushed_digest = None
# Scrolling state tracking for graceful updates
self._scrolling_state = {
@@ -425,10 +418,6 @@ class DisplayManager:
try:
# RGBMatrix accepts brightness as a property
self.matrix.brightness = brightness
# Brightness applies on the next swap — force a re-push even if
# the image itself is unchanged (belt-and-braces: brightness is
# also part of the dirty-tracking digest when readable).
self._last_pushed_digest = None
logger.info(f"[BRIGHTNESS] Display brightness set to {brightness}%")
return True
except AttributeError as e:
@@ -520,20 +509,7 @@ class DisplayManager:
return phys
def update_display(self):
"""Update the display using double buffering with proper sync.
Skips the panel push entirely when the frame is byte-identical to
the last pushed one (same image digest AND same brightness) static
content re-rendered every second, and 125 fps loops between actual
scroll steps, otherwise re-walk the full framebuffer for nothing.
The panel keeps refreshing the current frame from its own thread,
so skipping a swap never blanks or freezes the hardware.
Correctness hinges on invalidation: clear() resets the digest (it
writes to the matrix directly), and brightness is PART of the digest
so a dim-schedule change is never skipped. Disable via config
``display.dirty_tracking: false`` if a redraw issue is ever suspected.
"""
"""Update the display using double buffering with proper sync."""
try:
if self.matrix is None:
# Fallback mode - no actual hardware to update
@@ -545,19 +521,6 @@ class DisplayManager:
if self._capture_mode_active:
return # Skip hardware write — content is being captured off-screen
digest = None
if self._dirty_tracking_enabled:
try:
brightness = getattr(self.matrix, 'brightness', None)
except Exception:
brightness = None
digest = (zlib.adler32(self.image.tobytes()), brightness)
if digest == self._last_pushed_digest:
# Nothing changed since the last push — the panel is
# already showing exactly this frame.
self._write_snapshot_if_due()
return
# Copy the current image to the offscreen canvas. In double-sided
# mode the logical screen is first tiled across the full chain.
if self._double_sided is not None:
@@ -567,12 +530,10 @@ class DisplayManager:
# Swap buffers immediately
self.matrix.SwapOnVSync(self.offscreen_canvas)
# Swap our canvas references
self.offscreen_canvas, self.current_canvas = self.current_canvas, self.offscreen_canvas
self._last_pushed_digest = digest
# Write a snapshot for the web preview (throttled)
self._write_snapshot_if_due()
except Exception as e:
@@ -608,9 +569,6 @@ class DisplayManager:
# Clear both canvases and the underlying matrix to ensure no artifacts.
# Failures are non-fatal — the image buffer is already black above, so
# the next update_display() call will push clean content regardless.
# The matrix content no longer matches the last pushed digest,
# so dirty tracking must not skip the next push.
self._last_pushed_digest = None
try:
self.offscreen_canvas.Clear()
except (RuntimeError, OSError) as e:
+16 -4
View File
@@ -66,6 +66,10 @@ class RenderPipeline:
else display_manager.height
)
# Reusable blank frame for cycle-end pushes (allocated lazily,
# re-blacked before each reuse)
self._blank_frame = None
# ScrollHelper for optimized scrolling
self.scroll_helper = ScrollHelper(
self.display_width,
@@ -234,11 +238,19 @@ class RenderPipeline:
)
# Push blank immediately so the hardware never shows any
# post-wrap content while the coordinator recomposes the
# next cycle (~100 ms).
# next cycle (~100 ms). The blank is allocated once and
# reused across cycle wraps (fresh paste each time in case
# a consumer drew on the previous one).
try:
from PIL import Image as _Image
blank = _Image.new('RGB', (self.display_width, self.display_height))
self.display_manager.image = blank
if self._blank_frame is None or self._blank_frame.size != (
self.display_width, self.display_height):
self._blank_frame = Image.new(
'RGB', (self.display_width, self.display_height))
else:
self._blank_frame.paste(
(0, 0, 0),
(0, 0, self.display_width, self.display_height))
self.display_manager.image = self._blank_frame
self.display_manager.update_display()
except Exception:
logger.exception("Failed to write blank frame to display at cycle end")
-135
View File
@@ -1,135 +0,0 @@
"""Tests for update_display dirty tracking (src/display_manager.py).
Runs against RGBMatrixEmulator (EMULATOR=true), exercising the REAL
DisplayManager not a mock so the skip logic, its invalidation hooks,
and the kill switch are verified off-Pi.
The invariants:
- identical frames are pushed exactly once (SwapOnVSync not re-called)
- ANY pixel change pushes
- clear() and set_brightness() invalidate (the two paths that alter panel
state outside the digest's view)
- the kill switch (display.dirty_tracking: false) restores always-push
"""
import os
import sys
os.environ["EMULATOR"] = "true"
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
@pytest.fixture(scope="module")
def dm():
"""One real DisplayManager on the emulator (it's a process singleton)."""
from src.display_manager import DisplayManager
DisplayManager._instance = None
DisplayManager._initialized = False
manager = DisplayManager({
"display": {
"hardware": {"rows": 32, "cols": 64, "chain_length": 2,
"parallel": 1, "brightness": 90},
"runtime": {"gpio_slowdown": 0},
},
}, suppress_test_pattern=True)
yield manager
class _SwapSpy:
"""Counts SwapOnVSync calls through the real matrix object."""
def __init__(self, matrix):
self.matrix = matrix
self.count = 0
self._orig = matrix.SwapOnVSync
def __enter__(self):
def counting(canvas):
self.count += 1
return self._orig(canvas)
self.matrix.SwapOnVSync = counting
return self
def __exit__(self, *exc):
self.matrix.SwapOnVSync = self._orig
class TestDirtyTracking:
def test_identical_frames_push_once(self, dm):
dm.draw.rectangle([0, 0, 10, 10], fill=(255, 0, 0))
with _SwapSpy(dm.matrix) as spy:
dm.update_display()
dm.update_display()
dm.update_display()
assert spy.count == 1
def test_pixel_change_pushes(self, dm):
dm.update_display()
with _SwapSpy(dm.matrix) as spy:
dm.draw.point((5, 5), fill=(0, 255, 0))
dm.update_display()
dm.update_display() # unchanged again
assert spy.count == 1
def test_clear_invalidates(self, dm):
dm.draw.rectangle([0, 0, 20, 20], fill=(0, 0, 255))
dm.update_display()
dm.clear() # writes to the matrix directly; digest must reset
with _SwapSpy(dm.matrix) as spy:
dm.update_display() # black frame after clear must still push
assert spy.count == 1
def test_brightness_change_forces_push(self, dm):
dm.draw.rectangle([0, 0, 20, 20], fill=(200, 200, 200))
dm.update_display()
with _SwapSpy(dm.matrix) as spy:
dm.update_display() # identical -> skipped
assert spy.count == 0
dm.set_brightness(40) # dim schedule scenario
dm.update_display() # same image, new brightness -> push
assert spy.count == 1
dm.set_brightness(90)
def test_snapshot_still_written_on_skip(self, dm, tmp_path):
"""The web preview path must keep working through skipped pushes."""
dm._snapshot_path = str(tmp_path / "snap.png")
dm._last_snapshot_ts = 0.0
dm.draw.rectangle([0, 0, 30, 8], fill=(255, 255, 0))
dm.update_display() # push + snapshot
assert os.path.exists(dm._snapshot_path)
class TestKillSwitch:
def test_dirty_tracking_can_be_disabled(self, dm):
dm._dirty_tracking_enabled = False
try:
dm.draw.rectangle([0, 0, 10, 10], fill=(1, 2, 3))
with _SwapSpy(dm.matrix) as spy:
dm.update_display()
dm.update_display()
dm.update_display()
assert spy.count == 3 # always-push, exactly the old behavior
finally:
dm._dirty_tracking_enabled = True
dm._last_pushed_digest = None
def test_config_flag_wires_through(self):
from src.display_manager import DisplayManager
DisplayManager._instance = None
DisplayManager._initialized = False
manager = DisplayManager({
"display": {
"hardware": {"rows": 32, "cols": 64, "chain_length": 1,
"parallel": 1},
"runtime": {"gpio_slowdown": 0},
"dirty_tracking": False,
},
}, suppress_test_pattern=True)
assert manager._dirty_tracking_enabled is False
if __name__ == "__main__":
sys.exit(pytest.main([__file__, "-v"]))