mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-04 14:25:08 +00:00
ChuckandClaude Opus 5
d12323e7f1
* perf(scroll): pace frames to the panel, not to a fixed sleep Scrolling ran at 44-46 fps on a 2x128x64 chain and 14-17% of frames took 41-53ms, which reads as judder. Four independent causes, each measured on the hardware; details and the diagnostic recipe are in docs/SCROLL_PERFORMANCE.md. The high-FPS loop slept a flat 8ms after every render. display() has already blocked on the panel's vsync by then, so that sleep was added to a wait that had happened: ~4ms of render plus 8ms put each iteration at ~12ms against a 10ms refresh grid, so every swap missed a refresh and the loop settled at 50fps while asking for 125 -- with no headroom, so a further 14% of frames slipped again. It now sleeps only the remainder, with a 1ms floor so plugin threads still get the GIL. ScrollHelper stepped position on a wall clock at 1/scroll_delay steps per second. Plugins set scroll_delay to the frame period, so that comparison sat exactly on its own threshold: a frame arriving a hair early moved zero pixels and rendered an identical frame, dirty-tracking skipped the swap, it returned in ~2ms, and the beat repeated. No scroll_delay value tunes that out -- a shorter delay trades stalled frames for periodic double-steps. Both modes now accumulate elapsed time at the same configured speed, so position stays proportional to real time. Sub-pixel blending goes back to off by default. It renders a half-step by mixing two adjacent columns, which on a coarse panel showing pixel-font text alternates crisp and smeared frames and reads as shimmer -- visibly worse than integer stepping on the hardware. Vegas mode still opts in. disk_cache uses orjson when importable, falling back to the stdlib. Encoding a ~1MB record drops from 14.8ms to 5.4ms end-to-end, and that work holds the GIL while a marquee is on screen. display_manager also checksummed the whole framebuffer twice per frame (dirty tracking, then the preview snapshot); the snapshot now takes the checksum the caller already computed. New src/common/scroll_config.py resolves scroll settings in one place. Five ticker plugins each hand-rolled this and disagreed: odds-ticker ranked the deprecated scroll_pixels_per_second above the documented scroll_speed/delay pair, and because that key carries a schema default the documented settings were dead for every user (ChuckBuilds/ledmatrix-plugins#408), while ledmatrix-leaderboard read the same key only as a fallback. The resolver also warns when a speed will not advance a whole number of pixels per refresh, which is the property that actually determines whether a scroll looks smooth. scripts/build_rgbmatrix_nogil.sh rebuilds the rgbmatrix binding so it releases the GIL. Upstream declares SwapOnVSync without nogil, unlike SetPixel/Clear/Fill beside it, so the render thread held the GIL for the whole vsync wait and starved background threads into long uninterruptible bursts. The script patches, builds and self-verifies into a scratch tree; --install backs up the original and rolls back if the service does not come back healthy. Measured after: 100 fps locked, no stalls observed, render thread down from 51% to 19% of one core. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(display): keep the panel swap locked to vsync while scrolling Dirty tracking skipped SwapOnVSync for byte-identical frames. That is the right call for static content, but SwapOnVSync is also what paces the render loop, so skipping it skips the wait for the panel: a duplicate frame returns in ~8ms instead of ~10ms on a 100Hz panel, advances the strip only 0.8px instead of 1.0px, and so makes the next frame more likely to repeat as well. The effect sustains itself once it starts. Measured over 20 minutes on a 2x128x64 chain, both scrollers configured identically at 100 px/s: leaderboard 10ms x35, 11ms x3 (clean) odds-ticker 10ms x26, 8ms x7, 15ms x5 (~20% duplicates mid-scroll) The duplicates were not end-of-cycle idling -- 38% of fast frames fell within 90s of a scroll completion against 35% of normal frames, a null result. The trigger is per-frame work: odds does more of it, and more variably, so it is first to land a frame that advances less than a whole pixel. Pushing an identical frame costs one canvas copy. Falling out of vsync lock costs smooth motion. Static content is untouched, because is_currently_scrolling() expires on its own inactivity threshold -- covered by test_stale_scrolling_state_stops_forcing_pushes so a plugin that stops scrolling without saying so cannot pin the panel into always-push. Also de-flakes test_snapshot_still_written_on_skip, which asserted a strict mtime increase between two writes that can land in the same filesystem tick; it failed about two runs in three on Windows regardless of the code under test. The file is now backdated before the check. 156 tests pass on the Pi. Not yet confirmed by eye on the panel. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(scroll): report the frame-time tail, and stop the row-major blit Two problems, both found by looking at the panel rather than the metric. The frame-stats line reported ONE instantaneous frame every 5 seconds -- about 1 frame in 500 -- printed beside a 100-frame average. Both hide exactly the fault they are used to chase: a 2ms duplicate and a 21ms double-wait average to precisely 10ms, so a ticker stalling on half its frames still reports a healthy "Avg FPS: 100.0". That reading cost several rounds of chasing the wrong layer. The line now aggregates every frame since the last log and reports median, p95, max, min, and explicit stall and skip rates (past 1.5x the median missed a refresh; under half never reached the panel, because dirty tracking skipped the swap so the frame never waited on vsync). On the hardware this now reads: leaderboard 100.0 fps over 501 frames | median 10.00ms p95 10.05ms max 10.34ms | stalls 0 (0.0%) skips 0 (0.0%) The binding rebuild's blit patch becomes opt-in (RGB_PATCH_BLIT=1, default off). Reordering that loop to row-major changes what a torn frame looks like: column-major tearing shows as a vertical seam, row-major as a horizontal split between the panel's upper and lower halves. On a 1/32 scan panel that reads as a one-pixel fold across the middle of every panel, which is what was reported on hardware and what went away when the blit was reverted. All of the measured gain comes from the SwapOnVSync change, so the risky half is simply not worth taking; the header says so. Also fixes --install resolving its paths against $HOME, which is /root under sudo, so it looked in /root/rgbmatrix-nogil-build and died with "no built module found" on a machine where the build had just succeeded. It now resolves SUDO_USER's home. Both build paths are verified on the Pi: default yields one GIL-release site, RGB_PATCH_BLIT=1 yields two. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * feat(scroll): let users pick a crisp speed for their own panel Whole-pixel motion was previously only available at multiples of the refresh rate -- 100, 200, 300 px/s on a 100Hz panel. 100 px/s crosses a 256px panel in 2.6s, which is brisk for reading, and everything slower had to blend (blur) or repeat frames unevenly (judder). There was no way to ask for 50 px/s and get clean motion. SwapOnVSync takes a framerate_fraction the display manager never passed. It holds each frame for N panel refreshes; the panel keeps refreshing at its full rate throughout, so holding costs nothing in flicker and only changes how often a NEW image is presented. That turns 50 px/s into one whole pixel every second refresh instead of half a pixel every refresh. The crisp speeds are therefore refresh_hz / hold * pixels_per_frame, and that ladder depends on the panel: a Pi Zero on a long chain has a different set of good speeds from a Pi 4 on a short one. crisp_ladder() enumerates them and solve_crisp() picks the best match for a requested speed. solve_crisp weights motion quality rather than picking the numerically nearest entry, which matters more than it sounds. Asked for 30 px/s, nearest-by-value answers 28.6 -- 2px jumps at 14fps -- over 33.3, which is single-pixel motion at 33fps and obviously better on the panel. The target is also clamped into the ladder's range first, because relative error saturates near 1.0 for a target far outside it and the quality penalty would otherwise answer "10000 px/s" with the slowest entry. configure() snaps to the ladder and applies the hold when given a display manager. Without one the hold silently cannot happen and motion falls back to fractional pixels, so it warns rather than failing quietly. set_frame_hold() resets to 1 when scrolling stops, so one plugin's pacing cannot leak into whatever is on screen next. scripts/scroll_speeds.py is the user-facing part: it prints the ladder for the configured rate, measures what the panel ACTUALLY manages (--measure, for hardware that cannot reach its configured limit), highlights the nearest option to a wanted speed, and demos one live. It never starts or stops the display service itself -- doing that inside a script stranded the panel twice today. Speeds below ~20 px/s remain stepped regardless. That is the pixel pitch, not a software limit. Also fixes the dirty-tracking test spy, which stubbed SwapOnVSync with a single-argument function and would have masked the new call as a failed push, and rewrites a configure() test that had started passing for the wrong reason: it asserted a judder warning, which snapping now prevents, and was matching the unrelated "hold could not be applied" warning instead. 183 tests pass on the Pi. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(scroll): tie the frame hold to the scroll, not the plugin The hold applied in configure() never reached the panel. Plugins share one display manager, and set_scrolling_state(False) -- fired whenever ANY other plugin finishes its scroll -- reset the hold to 1. A hold set once at plugin construction was therefore always gone by the time that plugin rendered. The symptom was a log line that lied. ledmatrix-stocks reported Scroll configured: 50.0 px/s (1px every 2 refreshes = 50.0 fps, smooth) while the panel measured 100.0 fps, median 10.00ms. Config, resolution and snapping were all correct; only the pacing silently was not applied. set_scrolling_state(is_scrolling, frame_hold=1) now carries it, so the hold lives exactly as long as the scroll that asked for it. configure() reports the value as ScrollSettings.frame_hold instead of applying it -- applying it behind the caller's back could never have been right on a shared display manager. Existing callers are unaffected; the default keeps one frame per refresh. Verified on hardware: stocks at 50 px/s now measures 50.0 fps over 251 frames | median 20.00ms p95 20.09ms | stalls 0 skips 0 20.00ms being exactly two refreshes, with the panel still refreshing at 100Hz underneath so flicker is unchanged. test_another_plugin_stopping_does_not_strand_a_hold pins the interaction that broke this. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(scroll,cache): resolve CodeRabbit review on #523 Eight findings, all reproduced before fixing. scroll_config.configure() read the refresh rate *after* resolve() had already used it. resolve() fills in target_fps, pixels_per_frame and the judder warning from that rate, so on a 60Hz panel every one of them described 100Hz -- and with snap_to_crisp=False nothing downstream corrected it, so set_target_fps() paced the helper to 100 FPS. The rate is now settled first, and falls back to the global config rather than straight to the default. refresh_hz_from_config() used `(cfg.get("display") or {}).get(...)`, which raises AttributeError when either level is truthy but not a mapping -- out of a function whose whole contract is a rate or a default. The frame-stats line reported the upper-middle sample as the median and the 96th sorted sample as p95 of 100. Both are also thresholds (stalls at 1.5x the median, skips at 0.5x), so the counts were biased too. The arithmetic is now in frame_stats()/format_frame_stats(), testable without a clock. configure()'s docstring and docs/SCROLL_PERFORMANCE.md still said it applies the frame hold and warns when it cannot. It deliberately does neither since "tie the frame hold to the scroll, not the plugin"; a caller following the old text would omit set_scrolling_state() and slow snapped speeds would still present every refresh. disk_cache had no policy for non-finite floats: orjson writes null, the stdlib writes NaN/Infinity, and orjson then rejects those legacy files so DiskCache.get deleted them as corrupt. One behaviour on both paths now -- write null, keep legacy records readable. allow_nan=False detects the values; the replacement walk runs only when there is one, so the ordinary write path is byte-identical and pays nothing. build_rgbmatrix_nogil.sh picked the build artifact with a glob piped to `head -1`, which sorts cpython-311 ahead of cpython-313, so a stale .so staged in from the source tree was installed as core.so while the GIL check -- which reads the generated core.cpp, not the .so -- still passed. It now requires the current interpreter's exact ABI name and fails closed. Its systemctl calls were also unchecked under `set -uo pipefail`: a failed stop left the old service running, the following start succeeded as a no-op, and the health check reported SUCCESS for a binding that was never loaded. orjson floor raised to 3.11.6 for CVE-2025-67221 (unbounded recursion in dumps); it covers the project's Python 3.10-3.13 range. Adds test/test_cache_nonfinite_floats.py (14) plus regression tests in test_scroll_config.py and test_scroll_helper.py. 9 of the cache tests and 9 of the scroll_config tests fail against the pre-fix code. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014RRtqXDCnvnY6EQwhT5CV9 * test(harness): keep the visual double's signature tied to production Moves set_scrolling_state's frame_hold into the test double here, where DisplayManager gains it, rather than in #534 where it arrived a PR early. CodeRabbit flagged the #534 version correctly: a double that accepts an argument production does not lets the call pass every harness run and raise TypeError on the panel, which is the one failure a safety harness exists to prevent. The drift has now gone both ways across two branches -- double behind production on this branch, double ahead of it on #534 -- so it is pinned instead of remembered. test_display_double_parity.py compares the two signatures and fails with the direction of the drift named. It reads the files with ast rather than importing them, because display_manager imports rgbmatrix at module scope and this check should hold on a laptop and in CI as well as on a Pi. Plugins begin passing frame_hold in ledmatrix-plugins#462, which is why production and the double both need it before that lands. Full suite: 3889 passed. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014RRtqXDCnvnY6EQwhT5CV9 --------- Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
1592 lines
75 KiB
Python
1592 lines
75 KiB
Python
"""
|
||
Display Manager — hardware abstraction layer for the RGB LED matrix.
|
||
|
||
This module provides :class:`DisplayManager`, the single interface between
|
||
application code and the physical (or emulated) LED panel.
|
||
|
||
Key responsibilities
|
||
--------------------
|
||
* Initialise the ``RGBMatrix`` (hardware) or ``RGBMatrixEmulator`` depending
|
||
on the ``EMULATOR`` environment variable.
|
||
* Expose a PIL ``Image``/``ImageDraw`` canvas that plugins draw into, then
|
||
flush it to the matrix via double-buffering (:meth:`DisplayManager.update_display`).
|
||
* Load and cache TTF/BDF fonts; expose ``draw_text`` for consistent text rendering.
|
||
* Provide ``width`` / ``height`` properties — always use these instead of
|
||
hard-coding display dimensions.
|
||
* Write periodic PNG snapshots to ``/tmp/led_matrix_preview.png`` for the
|
||
web-interface live preview.
|
||
* Track scrolling state and gate deferred updates so plugins don't race with
|
||
an in-progress scroll.
|
||
|
||
Singleton: only one ``DisplayManager`` instance exists per process. The
|
||
first call to ``DisplayManager(config)`` creates it; subsequent calls return
|
||
the same object.
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import socket
|
||
import tempfile
|
||
if os.getenv("EMULATOR", "false") == "true":
|
||
from RGBMatrixEmulator import RGBMatrix, RGBMatrixOptions
|
||
else:
|
||
from rgbmatrix import RGBMatrix, RGBMatrixOptions
|
||
from contextlib import contextmanager
|
||
from pathlib import Path
|
||
from PIL import Image, ImageDraw, ImageFont
|
||
import threading
|
||
import time
|
||
from collections import OrderedDict
|
||
from typing import Dict, Any, List, Optional, Tuple
|
||
import logging
|
||
import math
|
||
import zlib
|
||
import freetype
|
||
|
||
from src.common import snapshot_policy
|
||
from src.common.permission_utils import (
|
||
ensure_directory_permissions,
|
||
ensure_file_permissions,
|
||
get_assets_dir_mode,
|
||
get_assets_file_mode,
|
||
)
|
||
|
||
# Get logger without configuring
|
||
logger = logging.getLogger(__name__)
|
||
logger.setLevel(logging.INFO) # Set to INFO level
|
||
|
||
|
||
class _LogicalMatrix:
|
||
"""Proxy that reports a logical (per-screen) size for a physical matrix.
|
||
|
||
In double-sided mode the physical panel chain shows N identical copies of a
|
||
smaller logical screen. Plugins size themselves from ``matrix.width`` /
|
||
``matrix.height`` (the documented convention, used at 30+ call sites), so
|
||
this proxy reports the logical dimensions while delegating every real
|
||
operation — ``CreateFrameCanvas``, ``SwapOnVSync``, ``brightness``,
|
||
``Clear`` and so on — to the underlying physical matrix. The duplication
|
||
itself happens once per frame in :meth:`DisplayManager.update_display`.
|
||
"""
|
||
|
||
__slots__ = ("_logical_height", "_logical_width", "_matrix")
|
||
|
||
def __init__(self, matrix: RGBMatrix, logical_width: int, logical_height: int) -> None:
|
||
object.__setattr__(self, "_matrix", matrix)
|
||
object.__setattr__(self, "_logical_width", logical_width)
|
||
object.__setattr__(self, "_logical_height", logical_height)
|
||
|
||
@property
|
||
def width(self) -> int:
|
||
"""Logical (per-screen) width reported to plugins."""
|
||
return self._logical_width
|
||
|
||
@property
|
||
def height(self) -> int:
|
||
"""Logical (per-screen) height reported to plugins."""
|
||
return self._logical_height
|
||
|
||
def __getattr__(self, name: str) -> Any:
|
||
"""Forward any non-overridden attribute access to the physical matrix.
|
||
|
||
Reached only when normal lookup fails (i.e. not width/height/_*).
|
||
"""
|
||
return getattr(object.__getattribute__(self, "_matrix"), name)
|
||
|
||
def __setattr__(self, name: str, value: Any) -> None:
|
||
"""Forward attribute writes (e.g. ``matrix.brightness = 80``) to it."""
|
||
setattr(object.__getattribute__(self, "_matrix"), name, value)
|
||
|
||
|
||
def _resolve_double_sided(physical_width: int, physical_height: int,
|
||
ds_config: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
||
"""Validate the ``display.double_sided`` config against the physical size.
|
||
|
||
Returns a dict ``{copies, axis, logical_width, logical_height}`` when the
|
||
feature is enabled and the physical panel divides evenly into ``copies``
|
||
along the chosen axis, otherwise ``None`` (single-screen behaviour). Bad
|
||
config is logged and disabled rather than raised — a misconfigured panel
|
||
should still light up.
|
||
"""
|
||
if not isinstance(ds_config, dict) or not ds_config.get('enabled', False):
|
||
return None
|
||
|
||
copies = ds_config.get('copies', 2)
|
||
if not isinstance(copies, int) or copies < 2:
|
||
logger.warning(
|
||
"double_sided: 'copies' must be an integer >= 2 (got %r); "
|
||
"disabling double-sided mode", copies)
|
||
return None
|
||
|
||
axis = ds_config.get('axis', 'horizontal')
|
||
if axis not in ('horizontal', 'vertical'):
|
||
logger.warning(
|
||
"double_sided: 'axis' must be 'horizontal' or 'vertical' "
|
||
"(got %r); defaulting to 'horizontal'", axis)
|
||
axis = 'horizontal'
|
||
|
||
# Horizontal splits the chain (panels side by side); vertical splits the
|
||
# parallel outputs (panels stacked). The split axis must divide evenly.
|
||
if axis == 'horizontal':
|
||
if physical_width % copies != 0:
|
||
logger.warning(
|
||
"double_sided: physical width %d is not divisible by copies "
|
||
"%d; disabling double-sided mode", physical_width, copies)
|
||
return None
|
||
logical_width = physical_width // copies
|
||
logical_height = physical_height
|
||
else:
|
||
if physical_height % copies != 0:
|
||
logger.warning(
|
||
"double_sided: physical height %d is not divisible by copies "
|
||
"%d; disabling double-sided mode", physical_height, copies)
|
||
return None
|
||
logical_width = physical_width
|
||
logical_height = physical_height // copies
|
||
|
||
logger.info(
|
||
"double_sided enabled: %d copies on %s axis — logical screen %dx%d "
|
||
"tiled across physical %dx%d", copies, axis, logical_width,
|
||
logical_height, physical_width, physical_height)
|
||
return {
|
||
'copies': copies,
|
||
'axis': axis,
|
||
'logical_width': logical_width,
|
||
'logical_height': logical_height,
|
||
}
|
||
|
||
|
||
class DisplayManager:
|
||
"""
|
||
Singleton hardware abstraction layer for the RGB LED matrix.
|
||
|
||
Plugins should never interact with ``RGBMatrix`` directly; they use this
|
||
class to draw content and call :meth:`update_display` to push frames to
|
||
the panel.
|
||
|
||
Typical plugin usage::
|
||
|
||
canvas = Image.new('RGB', (self.display_manager.width,
|
||
self.display_manager.height), (0, 0, 0))
|
||
draw = ImageDraw.Draw(canvas)
|
||
# ... draw content ...
|
||
self.display_manager.image = canvas
|
||
self.display_manager.draw = ImageDraw.Draw(self.display_manager.image)
|
||
self.display_manager.update_display()
|
||
"""
|
||
|
||
_instance = None
|
||
_initialized = False
|
||
|
||
def __new__(cls, *args, **kwargs):
|
||
if cls._instance is None:
|
||
cls._instance = super(DisplayManager, cls).__new__(cls)
|
||
return cls._instance
|
||
|
||
def __init__(self, config: Dict[str, Any] = None, force_fallback: bool = False, suppress_test_pattern: bool = False):
|
||
start_time = time.time()
|
||
self.config = config or {}
|
||
self._force_fallback = force_fallback
|
||
self._suppress_test_pattern = suppress_test_pattern
|
||
# Per-thread capture state. update_display() and clear() skip hardware
|
||
# writes while the *calling* thread is capturing content off-screen.
|
||
#
|
||
# Thread-local rather than a plain flag because Vegas mode prepares
|
||
# upcoming content on a background thread: a shared flag set there would
|
||
# suppress the render loop's own frame pushes for the duration, freezing
|
||
# the panel exactly when the point was to avoid a freeze.
|
||
self._capture_state = threading.local()
|
||
# Double-sided mode state (resolved in _setup_matrix). When disabled,
|
||
# the logical image is blitted to the matrix unchanged.
|
||
self._double_sided = None # dict {copies, axis, logical_width, logical_height} or None
|
||
self._physical_image = None # full-chain buffer reused each frame when tiling
|
||
# Text-width measurement cache: (text, id(font)) -> (width, font_ref)
|
||
# Avoids re-measuring the same string+font on every display() call.
|
||
# LRU-bounded: keys embed the TEXT, so changing strings (a clock, a
|
||
# live score) would otherwise grow it forever on a 24/7 service.
|
||
# Entries hold a strong reference to the font so its id() can't be
|
||
# recycled by a different font object — an id-keyed cache without
|
||
# the reference can return the WRONG width after garbage collection.
|
||
# Cleared on _load_fonts() so stale entries don't survive a font reload.
|
||
self._text_width_cache: "OrderedDict[tuple, Tuple[int, Any]]" = OrderedDict()
|
||
self._TEXT_WIDTH_CACHE_MAX = 1024
|
||
# Snapshot mirror for web preview + health check (service writes, web
|
||
# reads). Cadence/skip decisions live in src/common/snapshot_policy.py:
|
||
# full rate only while the web SSE broadcaster keeps the viewer marker
|
||
# fresh; unchanged frames are never re-encoded, only mtime-touched.
|
||
self._snapshot_path = "/tmp/led_matrix_preview.png" # nosec B108 - fixed path intentional; web UI reads same path
|
||
self._viewer_marker_path = "/tmp/led_matrix_preview_viewer" # nosec B108 - touched by web SSE broadcaster
|
||
self._last_snapshot_ts = 0.0
|
||
self._last_snapshot_touch_ts = 0.0
|
||
self._last_snapshot_digest: Optional[int] = None
|
||
self._snapshot_dir_prepared = False
|
||
self._viewer_check_ts = 0.0
|
||
self._viewer_fresh = False
|
||
self._viewer_was_fresh = False
|
||
# Snapshot failures are logged as warnings, rate-limited so a
|
||
# persistent failure (e.g. an unwritable file) can't spam the log —
|
||
# but is never silent: the snapshot's mtime doubles as the web UI's
|
||
# hardware-liveness signal, so a quiet failure makes health checks lie.
|
||
self._snapshot_fail_log_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
|
||
# Serializes update_display(): plugins can call it directly from
|
||
# background threads (see docstring on update_display), not just the
|
||
# render loop. RLock in case a caller within the critical section
|
||
# ever re-enters (e.g. via a nested draw callback).
|
||
self._update_lock = threading.RLock()
|
||
|
||
# Scrolling state tracking for graceful updates
|
||
# How many panel refreshes each pushed frame is held for. 1 means a new
|
||
# frame every refresh. Higher values are how a scroll runs slower than
|
||
# one pixel per refresh WITHOUT fractional pixel positions: the panel
|
||
# keeps refreshing at full rate (so flicker is unchanged) but motion
|
||
# advances a whole pixel every Nth refresh instead of every one.
|
||
# See src/common/scroll_config.py and scripts/scroll_speeds.py.
|
||
self._frame_hold = 1
|
||
|
||
self._scrolling_state = {
|
||
'is_scrolling': False,
|
||
'last_scroll_activity': 0,
|
||
'scroll_inactivity_threshold': 2.0, # seconds of inactivity before considering "not scrolling"
|
||
'deferred_updates': [],
|
||
'max_deferred_updates': 50, # Limit queue size to prevent memory issues
|
||
'deferred_update_ttl': 300.0 # 5 minutes TTL for deferred updates
|
||
}
|
||
|
||
self._setup_matrix()
|
||
logger.info("Matrix setup completed in %.3f seconds", time.time() - start_time)
|
||
|
||
font_time = time.time()
|
||
self._load_fonts()
|
||
logger.info("Font loading completed in %.3f seconds", time.time() - font_time)
|
||
|
||
# Initialize managers
|
||
# Calendar manager is now initialized by DisplayController
|
||
|
||
# Orientation setting -> rpi-rgb-led-matrix "Rotate:<deg>" pixel-mapper suffix.
|
||
# "normal" needs no suffix since 0 degrees is the identity transform.
|
||
_ORIENTATION_ROTATE_DEGREES = {'normal': None, '90': 90, '180': 180, '270': 270}
|
||
|
||
def _build_pixel_mapper_config(self, hardware_config: dict) -> str:
|
||
"""Compose the raw pixel_mapper_config string with the orientation setting.
|
||
|
||
`pixel_mapper_config` stays available as a free-form advanced field (e.g.
|
||
for "U-mapper" chain layouts); `orientation` is the user-facing dropdown
|
||
for physical mounting (e.g. panels mounted upside down) and is appended as
|
||
a "Rotate:<deg>" mapper rather than overwriting any existing config.
|
||
"""
|
||
base_mapper = (hardware_config.get('pixel_mapper_config') or '').strip()
|
||
orientation = hardware_config.get('orientation', 'normal')
|
||
degrees = self._ORIENTATION_ROTATE_DEGREES.get(orientation)
|
||
if degrees is None:
|
||
return base_mapper
|
||
rotate_mapper = f'Rotate:{degrees}'
|
||
return f'{base_mapper};{rotate_mapper}' if base_mapper else rotate_mapper
|
||
|
||
def _setup_matrix(self):
|
||
"""Initialize the RGB matrix with configuration settings."""
|
||
_init_error_str = None
|
||
try:
|
||
# Allow callers (e.g., web UI) to force non-hardware fallback mode
|
||
if getattr(self, '_force_fallback', False):
|
||
raise RuntimeError('Forced fallback mode requested')
|
||
options = RGBMatrixOptions()
|
||
|
||
# Hardware configuration
|
||
hardware_config = self.config.get('display', {}).get('hardware', {})
|
||
runtime_config = self.config.get('display', {}).get('runtime', {})
|
||
|
||
# Basic hardware settings
|
||
options.rows = hardware_config.get('rows', 32)
|
||
options.cols = hardware_config.get('cols', 64)
|
||
options.chain_length = hardware_config.get('chain_length', 2)
|
||
options.parallel = hardware_config.get('parallel', 1)
|
||
options.hardware_mapping = hardware_config.get('hardware_mapping', 'adafruit-hat-pwm')
|
||
|
||
# Performance and stability settings
|
||
options.brightness = hardware_config.get('brightness', 90)
|
||
options.pwm_bits = hardware_config.get('pwm_bits', 10)
|
||
options.pwm_lsb_nanoseconds = hardware_config.get('pwm_lsb_nanoseconds', 150)
|
||
options.led_rgb_sequence = hardware_config.get('led_rgb_sequence', 'RGB')
|
||
options.pixel_mapper_config = self._build_pixel_mapper_config(hardware_config)
|
||
options.row_address_type = hardware_config.get('row_address_type', 0)
|
||
options.multiplexing = hardware_config.get('multiplexing', 0)
|
||
options.panel_type = hardware_config.get('panel_type', '')
|
||
options.disable_hardware_pulsing = hardware_config.get('disable_hardware_pulsing', False)
|
||
options.show_refresh_rate = hardware_config.get('show_refresh_rate', False)
|
||
options.limit_refresh_rate_hz = hardware_config.get('limit_refresh_rate_hz', 90)
|
||
options.gpio_slowdown = runtime_config.get('gpio_slowdown', 3)
|
||
|
||
# Disable internal privilege dropping - we manage this via systemd or remain root
|
||
# This prevents the library from dropping to 'daemon' user which breaks file permissions
|
||
options.drop_privileges = False
|
||
|
||
# Additional settings from config
|
||
if 'scan_mode' in hardware_config:
|
||
options.scan_mode = hardware_config.get('scan_mode')
|
||
if 'pwm_dither_bits' in hardware_config:
|
||
options.pwm_dither_bits = hardware_config.get('pwm_dither_bits')
|
||
if 'inverse_colors' in hardware_config:
|
||
options.inverse_colors = hardware_config.get('inverse_colors')
|
||
# Pi 5 only: 0=PIO/RP1 coprocessor (default, less CPU),
|
||
# 1=RIO/Registered IO (faster; gpio_slowdown effect is inverted in this mode)
|
||
if 'rp1_rio' in runtime_config:
|
||
if hasattr(options, 'rp1_rio'):
|
||
options.rp1_rio = runtime_config.get('rp1_rio')
|
||
else:
|
||
logger.warning(
|
||
"rp1_rio is set in config but the installed rgbmatrix library does "
|
||
"not support it — the library was likely built without Pi 5 RP1 "
|
||
"support (mmap to 0x3f000000 instead of RP1 chip). "
|
||
"Fix: sudo RPI_RGB_FORCE_REBUILD=1 ./first_time_install.sh"
|
||
)
|
||
|
||
logger.info(f"Initializing RGB Matrix with settings: rows={options.rows}, cols={options.cols}, chain_length={options.chain_length}, parallel={options.parallel}, hardware_mapping={options.hardware_mapping}")
|
||
|
||
# Initialize the matrix
|
||
self.matrix = RGBMatrix(options=options)
|
||
logger.info("RGB Matrix initialized successfully")
|
||
|
||
# Create double buffer for smooth updates. The canvases are always
|
||
# full physical size — they back the real chain regardless of mode.
|
||
self.offscreen_canvas = self.matrix.CreateFrameCanvas()
|
||
self.current_canvas = self.matrix.CreateFrameCanvas()
|
||
logger.info("Frame canvases created successfully")
|
||
|
||
# Double-sided mode: wrap the physical matrix so plugins see the
|
||
# logical (per-screen) size, and keep a full-chain buffer to tile
|
||
# the rendered screen into once per frame.
|
||
ds_config = self.config.get('display', {}).get('double_sided', {})
|
||
ds = _resolve_double_sided(self.matrix.width, self.matrix.height, ds_config)
|
||
self._double_sided = ds
|
||
if ds is not None:
|
||
self._physical_image = Image.new(
|
||
'RGB', (self.matrix.width, self.matrix.height))
|
||
self.matrix = _LogicalMatrix(
|
||
self.matrix, ds['logical_width'], ds['logical_height'])
|
||
|
||
# Create image with the (logical) display dimensions
|
||
self.image = Image.new('RGB', (self.matrix.width, self.matrix.height))
|
||
self.draw = ImageDraw.Draw(self.image)
|
||
self.draw.fontmode = "1" # 1-bit text: the panel has no partial brightness, so AA only smears glyphs.
|
||
logger.info(f"Image canvas created with dimensions: {self.matrix.width}x{self.matrix.height}")
|
||
|
||
# Initialize font with Press Start 2P
|
||
try:
|
||
self.font = ImageFont.truetype("assets/fonts/PressStart2P-Regular.ttf", 8)
|
||
logger.info("Initial Press Start 2P font loaded successfully")
|
||
except Exception as e:
|
||
logger.error(f"Failed to load initial font: {e}")
|
||
self.font = ImageFont.load_default()
|
||
|
||
# Draw a test pattern unless caller suppressed it (e.g., web on-demand)
|
||
if not getattr(self, '_suppress_test_pattern', False):
|
||
self._draw_test_pattern()
|
||
|
||
except Exception as e:
|
||
_init_error_str = str(e)
|
||
logger.error(f"Failed to initialize RGB Matrix: {e}", exc_info=True)
|
||
# Create a fallback image for web preview using configured dimensions when available
|
||
self.matrix = None
|
||
try:
|
||
hardware_config = self.config.get('display', {}).get('hardware', {}) if self.config else {}
|
||
rows = int(hardware_config.get('rows', 32))
|
||
cols = int(hardware_config.get('cols', 64))
|
||
chain_length = int(hardware_config.get('chain_length', 2))
|
||
parallel = int(hardware_config.get('parallel', 1))
|
||
fallback_width = max(1, cols * chain_length)
|
||
fallback_height = max(1, rows * parallel)
|
||
# Mirror double-sided in fallback so the preview shows one screen.
|
||
ds_config = self.config.get('display', {}).get('double_sided', {}) if self.config else {}
|
||
ds = _resolve_double_sided(fallback_width, fallback_height, ds_config)
|
||
self._double_sided = ds
|
||
if ds is not None:
|
||
fallback_width = ds['logical_width']
|
||
fallback_height = ds['logical_height']
|
||
except Exception:
|
||
fallback_width, fallback_height = 128, 32
|
||
|
||
self.image = Image.new('RGB', (fallback_width, fallback_height))
|
||
self.draw = ImageDraw.Draw(self.image)
|
||
self.draw.fontmode = "1" # 1-bit text: the panel has no partial brightness, so AA only smears glyphs.
|
||
# Simple fallback visualization so web UI shows a realistic canvas
|
||
try:
|
||
self.draw.rectangle([0, 0, fallback_width - 1, fallback_height - 1], outline=(255, 0, 0))
|
||
self.draw.line([0, 0, fallback_width - 1, fallback_height - 1], fill=(0, 255, 0))
|
||
self.draw.text((2, max(0, (fallback_height // 2) - 4)), "Simulation", fill=(0, 128, 255))
|
||
except Exception: # nosec B110 - best-effort fallback visualization; drawing errors must not crash startup
|
||
# Best-effort; ignore drawing errors in fallback
|
||
pass
|
||
logger.error(
|
||
f"Matrix initialization failed — running in fallback/simulation mode "
|
||
f"(size {fallback_width}x{fallback_height}). Error: {e}. "
|
||
"On Raspberry Pi 5: ensure rpi-rgb-led-matrix was built from the latest "
|
||
"submodule (re-run first_time_install.sh). gpio_slowdown of 2–3 is typical for Pi 5 PIO mode."
|
||
)
|
||
# Do not raise here; allow fallback mode so web preview and non-hardware environments work
|
||
|
||
# Write hardware status file so the web UI can surface init failures
|
||
_hw_status = {"ok": self.matrix is not None, "error": _init_error_str}
|
||
_status_path = "/tmp/led_matrix_hw_status.json" # nosec B108
|
||
try:
|
||
if os.path.islink(_status_path):
|
||
logger.warning("Skipping hardware status write: %s is a symlink", _status_path)
|
||
else:
|
||
_fd, _tmp_path = tempfile.mkstemp(dir="/tmp", prefix=".led_hw_") # nosec B108
|
||
try:
|
||
with os.fdopen(_fd, "w") as _f:
|
||
json.dump(_hw_status, _f)
|
||
_f.flush()
|
||
os.fsync(_f.fileno())
|
||
os.chmod(_tmp_path, 0o644)
|
||
os.replace(_tmp_path, _status_path)
|
||
except Exception:
|
||
try:
|
||
os.unlink(_tmp_path)
|
||
except OSError:
|
||
pass
|
||
raise
|
||
except Exception:
|
||
logger.error("Failed to write hardware status file", exc_info=True)
|
||
|
||
@property
|
||
def width(self):
|
||
"""Get the display width."""
|
||
if hasattr(self, 'matrix') and self.matrix is not None:
|
||
return self.matrix.width
|
||
elif hasattr(self, 'image'):
|
||
return self.image.width
|
||
else:
|
||
return 128 # Default fallback width
|
||
|
||
@property
|
||
def height(self):
|
||
"""Get the display height."""
|
||
if hasattr(self, 'matrix') and self.matrix is not None:
|
||
return self.matrix.height
|
||
elif hasattr(self, 'image'):
|
||
return self.image.height
|
||
else:
|
||
return 32 # Default fallback height
|
||
|
||
def set_brightness(self, brightness: int) -> bool:
|
||
"""
|
||
Set display brightness at runtime.
|
||
|
||
Args:
|
||
brightness: Brightness level (0-100)
|
||
|
||
Returns:
|
||
True if brightness was set successfully, False otherwise
|
||
"""
|
||
# Fail fast: validate input type
|
||
if not isinstance(brightness, (int, float)):
|
||
logger.error(f"[BRIGHTNESS] Invalid brightness type: {type(brightness).__name__}, expected int")
|
||
return False
|
||
|
||
if self.matrix is None:
|
||
logger.warning("[BRIGHTNESS] Cannot set brightness in fallback mode")
|
||
return False
|
||
|
||
# Clamp to valid range
|
||
brightness = max(0, min(100, int(brightness)))
|
||
|
||
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:
|
||
logger.error(f"[BRIGHTNESS] Matrix does not support brightness property: {e}", exc_info=True)
|
||
return False
|
||
except (TypeError, ValueError) as e:
|
||
logger.error(f"[BRIGHTNESS] Invalid brightness value rejected by hardware: {e}", exc_info=True)
|
||
return False
|
||
|
||
def get_brightness(self) -> int:
|
||
"""
|
||
Get current display brightness.
|
||
|
||
Returns:
|
||
Current brightness level (0-100), or -1 if unavailable
|
||
"""
|
||
if self.matrix is None:
|
||
logger.debug("[BRIGHTNESS] Cannot get brightness in fallback mode")
|
||
return -1
|
||
|
||
try:
|
||
return self.matrix.brightness
|
||
except AttributeError as e:
|
||
logger.warning(f"[BRIGHTNESS] Matrix does not support brightness property: {e}", exc_info=True)
|
||
return -1
|
||
|
||
@staticmethod
|
||
def _local_ip() -> Optional[str]:
|
||
"""This device's address on the network it routes through, or None.
|
||
|
||
Deliberately not `hostname -I` or a systemctl probe for AP mode, which
|
||
is how the web launcher does it: both spawn processes with multi-second
|
||
timeouts, and this runs on the startup path the rest of this change
|
||
exists to shorten. Connecting a UDP socket sends no packets -- it only
|
||
asks the kernel which source address it would use -- so it costs
|
||
microseconds and works with the network down, as long as a route
|
||
exists.
|
||
"""
|
||
sock = None
|
||
try:
|
||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||
sock.settimeout(0.2)
|
||
sock.connect(("8.8.8.8", 80)) # nosec B104 - no traffic; selects a route
|
||
ip = sock.getsockname()[0]
|
||
return ip if ip and not ip.startswith("127.") else None
|
||
except OSError:
|
||
return None
|
||
finally:
|
||
if sock is not None:
|
||
try:
|
||
sock.close()
|
||
except OSError:
|
||
pass
|
||
|
||
def _fitting_font(self, lines, width):
|
||
"""The largest font from the usual ladder that fits every line."""
|
||
candidates = [self.font,
|
||
("assets/fonts/4x6-font.ttf", 6)]
|
||
for candidate in candidates:
|
||
try:
|
||
font = candidate
|
||
if isinstance(candidate, tuple):
|
||
font = ImageFont.truetype(candidate[0], candidate[1])
|
||
if all(self.draw.textlength(t, font=font) <= width for t in lines):
|
||
return font
|
||
except (OSError, ValueError, AttributeError):
|
||
continue
|
||
return self.font
|
||
|
||
def _draw_startup_banner(self, lines, width: int, height: int) -> None:
|
||
"""Centre `lines` over whatever the test pattern already drew.
|
||
|
||
This screen stays on the panel for the whole initial plugin update, and
|
||
on a headless Pi it is the only place the device's address appears
|
||
without going looking for it -- so it has to be readable off a wall,
|
||
not merely present.
|
||
|
||
The font is chosen to fit rather than fixed at 8px: "Initializing" is
|
||
96px in PressStart2P, which ran off the side of a 64px panel even
|
||
before an address was added. And the pattern is punched out behind the
|
||
text, because the diagonal runs through the middle of the panel, which
|
||
is exactly where this sits.
|
||
|
||
The text stays blue. It is not decoration: the pattern draws one pure
|
||
channel per element -- red border, green diagonal, blue text -- so that
|
||
a glance at the panel says whether led_rgb_sequence is right. Swap the
|
||
wiring to BGR and the border comes up blue and this text red. Drawing
|
||
it white would light all three channels and destroy the only blue
|
||
reference on the screen, which is why it is worth a comment rather
|
||
than a quiet preference.
|
||
"""
|
||
if not lines:
|
||
return
|
||
font = self._fitting_font(lines, width - 2)
|
||
line_height = self.draw.textbbox((0, 0), "Ag", font=font)[3] + 1
|
||
block_height = line_height * len(lines)
|
||
block_top = max(1, (height - block_height) // 2)
|
||
block_width = max(self.draw.textlength(t, font=font) for t in lines)
|
||
block_left = max(0, (width - block_width) // 2)
|
||
|
||
self.draw.rectangle(
|
||
[block_left - 2, block_top - 1,
|
||
block_left + block_width + 1, block_top + block_height],
|
||
fill=(0, 0, 0))
|
||
|
||
for row, line in enumerate(lines):
|
||
line_width = self.draw.textlength(line, font=font)
|
||
self.draw.text(
|
||
(max(0, (width - line_width) // 2), block_top + row * line_height),
|
||
line, font=font, fill=(0, 0, 255))
|
||
|
||
def _draw_test_pattern(self):
|
||
"""Draw a test pattern to verify the display is working."""
|
||
try:
|
||
self.clear()
|
||
|
||
if self.matrix is None:
|
||
# Fallback mode - just draw on the image
|
||
self.draw.rectangle([0, 0, self.image.width-1, self.image.height-1], outline=(255, 0, 0))
|
||
self.draw.line([0, 0, self.image.width-1, self.image.height-1], fill=(0, 255, 0))
|
||
self.draw.text((10, 10), "Simulation", font=self.font, fill=(0, 0, 255))
|
||
logger.info("Drew test pattern in fallback mode")
|
||
return
|
||
|
||
# Draw a red rectangle border
|
||
self.draw.rectangle([0, 0, self.matrix.width-1, self.matrix.height-1], outline=(255, 0, 0))
|
||
|
||
# Draw a diagonal line
|
||
self.draw.line([0, 0, self.matrix.width-1, self.matrix.height-1], fill=(0, 255, 0))
|
||
|
||
lines = ["Initializing"]
|
||
ip = self._local_ip()
|
||
if ip:
|
||
lines.append(ip)
|
||
self._draw_startup_banner(lines, self.matrix.width, self.matrix.height)
|
||
|
||
# Update the display once after everything is drawn
|
||
self.update_display()
|
||
time.sleep(0.5) # Reduced from 1 second to 0.5 seconds for faster animation
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error drawing test pattern: {e}", exc_info=True)
|
||
|
||
@property
|
||
def _capture_mode_active(self) -> bool:
|
||
"""True while the calling thread is capturing content off-screen."""
|
||
return getattr(self._capture_state, 'active', False)
|
||
|
||
@_capture_mode_active.setter
|
||
def _capture_mode_active(self, value: bool) -> None:
|
||
self._capture_state.active = bool(value)
|
||
|
||
@contextmanager
|
||
def capture_mode(self):
|
||
"""Suppress hardware output during off-screen content capture.
|
||
|
||
Plugins call update_display() as part of their normal display() flow.
|
||
When fetching content for Vegas mode the render loop is still running,
|
||
so any incidental hardware write causes a visible flash on the matrix.
|
||
Entering this context prevents those writes without affecting the PIL
|
||
image buffer, which the adapter reads to extract content.
|
||
"""
|
||
self._capture_mode_active = True
|
||
try:
|
||
yield
|
||
finally:
|
||
self._capture_mode_active = False
|
||
|
||
@contextmanager
|
||
def render_size(self, width: int, height: Optional[int] = None):
|
||
"""Temporarily present a smaller logical canvas to plugins.
|
||
|
||
Plugins lay out against ``display_manager.matrix.width`` (and the
|
||
``width``/``height`` properties, which defer to it), so the only way to
|
||
get a *narrower layout* rather than a cropped one is to tell the plugin
|
||
the screen is narrower while it renders. Trimming after the fact cannot
|
||
fix a forecast spread across five columns or a progress bar drawn at
|
||
100% width — those need the plugin to make different layout decisions.
|
||
|
||
Vegas mode uses this so a plugin can occupy a fraction of a wide panel
|
||
and still look deliberately composed. Reuses the same _LogicalMatrix
|
||
indirection that double-sided mode relies on, so plugins see a
|
||
consistent size from every accessor.
|
||
|
||
Only meaningful inside :meth:`capture_mode` — this swaps the shared
|
||
image buffer, so the render loop must not be writing to it concurrently.
|
||
|
||
Args:
|
||
width: Logical width to report, clamped to at least 1 and to the
|
||
real panel width (a larger canvas would overflow the hardware).
|
||
height: Logical height, defaulting to the current height.
|
||
"""
|
||
real_matrix = self.matrix
|
||
prev_image = getattr(self, 'image', None)
|
||
prev_draw = getattr(self, 'draw', None)
|
||
|
||
current_w = self.width
|
||
current_h = self.height
|
||
target_w = max(1, min(int(width), current_w))
|
||
target_h = max(1, min(int(height) if height else current_h, current_h))
|
||
|
||
if target_w == current_w and target_h == current_h:
|
||
# Nothing to do; avoid pointless wrapping and buffer churn.
|
||
yield
|
||
return
|
||
|
||
try:
|
||
if real_matrix is not None:
|
||
self.matrix = _LogicalMatrix(real_matrix, target_w, target_h)
|
||
# With no hardware, the width/height properties fall through to
|
||
# self.image, so swapping the buffer below is enough on its own.
|
||
self.image = Image.new('RGB', (target_w, target_h))
|
||
self.draw = ImageDraw.Draw(self.image)
|
||
self.draw.fontmode = "1" # 1-bit text: the panel has no partial brightness, so AA only smears glyphs.
|
||
yield
|
||
finally:
|
||
self.matrix = real_matrix
|
||
if prev_image is not None:
|
||
self.image = prev_image
|
||
if prev_draw is not None:
|
||
self.draw = prev_draw
|
||
|
||
def _composite_double_sided(self):
|
||
"""Tile the logical screen across the full physical chain.
|
||
|
||
Renders once into ``self._physical_image`` by pasting the rendered
|
||
logical image ``copies`` times along the configured axis. The paste is
|
||
a single memcpy per copy, so the per-frame cost is negligible and the
|
||
plugin render path is untouched.
|
||
"""
|
||
ds = self._double_sided
|
||
phys = self._physical_image
|
||
lw = ds['logical_width']
|
||
lh = ds['logical_height']
|
||
for i in range(ds['copies']):
|
||
if ds['axis'] == 'vertical':
|
||
phys.paste(self.image, (0, i * lh))
|
||
else:
|
||
phys.paste(self.image, (i * lw, 0))
|
||
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.
|
||
|
||
Serialized via ``_update_lock``: plugins can call this directly from
|
||
background threads (e.g. sports base classes push an immediate
|
||
"live" refresh from inside update()), so without a lock two callers
|
||
could both pass the digest check before either writes it back,
|
||
double-pushing a frame, or interleave the offscreen/current canvas
|
||
swap below. The lock is scoped to this method, so callers never
|
||
need to know about it.
|
||
"""
|
||
try:
|
||
with self._update_lock:
|
||
if self.matrix is None:
|
||
# Fallback mode - no actual hardware to update
|
||
logger.debug("Update display called in fallback mode (no hardware)")
|
||
# Still write a snapshot so the web UI can preview
|
||
self._write_snapshot_if_due()
|
||
return
|
||
|
||
if self._capture_mode_active:
|
||
return # Skip hardware write — content is being captured off-screen
|
||
|
||
digest = None
|
||
frame_checksum = None
|
||
if self._dirty_tracking_enabled:
|
||
try:
|
||
brightness = getattr(self.matrix, 'brightness', None)
|
||
except AttributeError:
|
||
brightness = None
|
||
frame_checksum = zlib.adler32(self.image.tobytes())
|
||
digest = (frame_checksum, brightness)
|
||
if digest == self._last_pushed_digest and not self.is_currently_scrolling():
|
||
# Nothing changed since the last push — the panel is
|
||
# already showing exactly this frame.
|
||
#
|
||
# Never taken mid-scroll, and that exception is the
|
||
# point. SwapOnVSync is what paces the render loop, so
|
||
# skipping it also skips the wait: a duplicate frame
|
||
# returns in ~8ms instead of ~10ms on a 100Hz panel,
|
||
# advances only 0.8px instead of 1.0px, and so makes
|
||
# the *next* frame more likely to repeat as well. That
|
||
# is self-sustaining -- measured at ~20% duplicate
|
||
# frames mid-scroll on the odds ticker, against
|
||
# essentially zero on a lighter plugin with identical
|
||
# scroll settings. Swapping an identical frame costs
|
||
# one canvas copy and keeps the loop locked to the
|
||
# panel; falling out of that lock costs smooth motion.
|
||
# Static content is unaffected: is_currently_scrolling()
|
||
# expires on its own inactivity threshold.
|
||
self._write_snapshot_if_due(frame_checksum)
|
||
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:
|
||
self.offscreen_canvas.SetImage(self._composite_double_sided())
|
||
else:
|
||
self.offscreen_canvas.SetImage(self.image)
|
||
|
||
# Swap buffers immediately. framerate_fraction holds the frame
|
||
# for N refreshes; SwapOnVSync blocks for all of them, which is
|
||
# what paces the render loop to the chosen frame rate.
|
||
self.matrix.SwapOnVSync(self.offscreen_canvas, self._frame_hold)
|
||
|
||
# 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(frame_checksum)
|
||
except Exception as e:
|
||
logger.error(f"Error updating display: {e}")
|
||
|
||
def clear(self):
|
||
"""Clear the display completely."""
|
||
try:
|
||
if self.matrix is None:
|
||
# Fallback mode - just clear the image
|
||
# Explicitly clear old image reference to help garbage collection
|
||
old_image = getattr(self, 'image', None)
|
||
width = old_image.width if old_image else 64
|
||
height = old_image.height if old_image else 64
|
||
if old_image is not None:
|
||
del old_image
|
||
|
||
self.image = Image.new('RGB', (width, height))
|
||
self.draw = ImageDraw.Draw(self.image)
|
||
self.draw.fontmode = "1" # 1-bit text: the panel has no partial brightness, so AA only smears glyphs.
|
||
logger.debug("Cleared display in fallback mode")
|
||
return
|
||
|
||
# Explicitly clear old image reference to help garbage collection
|
||
old_image = getattr(self, 'image', None)
|
||
if old_image is not None:
|
||
del old_image
|
||
|
||
# Create a new black image
|
||
self.image = Image.new('RGB', (self.matrix.width, self.matrix.height))
|
||
self.draw = ImageDraw.Draw(self.image)
|
||
self.draw.fontmode = "1" # 1-bit text: the panel has no partial brightness, so AA only smears glyphs.
|
||
|
||
if not self._capture_mode_active:
|
||
# 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:
|
||
logger.error("Failed to clear offscreen canvas: %s", e)
|
||
try:
|
||
self.current_canvas.Clear()
|
||
except (RuntimeError, OSError) as e:
|
||
logger.error("Failed to clear current canvas: %s", e)
|
||
try:
|
||
self.matrix.Clear()
|
||
except (RuntimeError, OSError) as e:
|
||
logger.error("Failed to clear matrix front buffer: %s", e)
|
||
|
||
# Note: We do NOT call update_display() here to avoid black flashes.
|
||
# The caller should call update_display() after drawing new content.
|
||
# If an immediate clear is needed, the caller can explicitly call
|
||
# clear() followed by update_display().
|
||
except Exception as e:
|
||
logger.error(f"Error clearing display: {e}")
|
||
|
||
def _draw_bdf_text(self, text, x, y, color=(255, 255, 255), font=None):
|
||
"""Draw text using BDF font with proper bitmap handling."""
|
||
try:
|
||
# Use the passed font or fall back to calendar_font
|
||
face = font if font else self.calendar_font
|
||
|
||
# Compute baseline from font ascender so caller can pass top-left y
|
||
try:
|
||
ascender_px = face.size.ascender >> 6
|
||
except Exception:
|
||
ascender_px = 0
|
||
baseline_y = y + ascender_px
|
||
|
||
for char in text:
|
||
face.load_char(char)
|
||
bitmap = face.glyph.bitmap
|
||
|
||
# Get glyph metrics
|
||
glyph_left = face.glyph.bitmap_left
|
||
glyph_top = face.glyph.bitmap_top
|
||
|
||
# Draw the character
|
||
for i in range(bitmap.rows):
|
||
for j in range(bitmap.width):
|
||
byte_index = i * bitmap.pitch + (j // 8)
|
||
if byte_index < len(bitmap.buffer):
|
||
byte = bitmap.buffer[byte_index]
|
||
if byte & (1 << (7 - (j % 8))):
|
||
# Calculate actual pixel position
|
||
pixel_x = x + glyph_left + j
|
||
pixel_y = baseline_y - glyph_top + i
|
||
# Only draw if within bounds
|
||
if (0 <= pixel_x < self.width and 0 <= pixel_y < self.height):
|
||
self.draw.point((pixel_x, pixel_y), fill=color)
|
||
|
||
# Move to next character
|
||
x += face.glyph.advance.x >> 6
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error drawing BDF text: {e}", exc_info=True)
|
||
|
||
def _load_fonts(self):
|
||
"""Load fonts with proper error handling."""
|
||
# Font objects get new id()s after reload, so the text-width cache would
|
||
# return stale measurements keyed on the old ids. Clear it here.
|
||
self._text_width_cache.clear()
|
||
try:
|
||
# Load Press Start 2P font
|
||
self.regular_font = ImageFont.truetype("assets/fonts/PressStart2P-Regular.ttf", 8)
|
||
logger.info("Press Start 2P font loaded successfully")
|
||
|
||
# Use the same font for small text (currently same size; adjust size here if needed)
|
||
self.small_font = ImageFont.truetype("assets/fonts/PressStart2P-Regular.ttf", 8)
|
||
logger.info("Press Start 2P small font loaded successfully")
|
||
|
||
# Load 5x7 BDF font for calendar events
|
||
try:
|
||
self.calendar_font_path = "assets/fonts/5x7.bdf"
|
||
logger.info(f"Attempting to load 5x7 font from: {self.calendar_font_path}")
|
||
|
||
if not os.path.exists(self.calendar_font_path):
|
||
raise FileNotFoundError(f"Font file not found at {self.calendar_font_path}")
|
||
|
||
# Load with freetype for proper BDF handling
|
||
face = freetype.Face(self.calendar_font_path)
|
||
logger.info(f"5x7 calendar font loaded successfully from {self.calendar_font_path}")
|
||
logger.info(f"Calendar font size: {face.size.height >> 6} pixels")
|
||
|
||
# Store the face for later use
|
||
self.calendar_font = face
|
||
|
||
except Exception as font_err:
|
||
logger.error(f"Failed to load 5x7 font: {str(font_err)}", exc_info=True)
|
||
logger.error("Falling back to small font")
|
||
self.calendar_font = self.small_font
|
||
|
||
# Assign the loaded calendar_font (which should be 5x7 BDF or its fallback)
|
||
# to a new attribute for specific use, e.g., in MusicManager.
|
||
self.bdf_5x7_font = self.calendar_font
|
||
logger.info(f"Assigned calendar_font (type: {type(self.bdf_5x7_font).__name__}) to bdf_5x7_font.")
|
||
|
||
# Load 4x6 font as extra_small_font
|
||
try:
|
||
font_path = "assets/fonts/4x6-font.ttf"
|
||
logger.info(f"Attempting to load 4x6 TTF font from: {font_path} at size 6")
|
||
self.extra_small_font = ImageFont.truetype(font_path, 6)
|
||
logger.info(f"4x6 TTF extra small font loaded successfully from {font_path}")
|
||
except Exception as font_err:
|
||
logger.error(f"Failed to load 4x6 TTF font: {font_err}. Falling back.")
|
||
self.extra_small_font = self.small_font
|
||
|
||
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error in font loading: {e}", exc_info=True)
|
||
# Fallback to default font
|
||
self.regular_font = ImageFont.load_default()
|
||
self.small_font = self.regular_font
|
||
self.calendar_font = self.regular_font
|
||
if not hasattr(self, 'extra_small_font'):
|
||
self.extra_small_font = self.regular_font
|
||
if not hasattr(self, 'bdf_5x7_font'): # Ensure bdf_5x7_font also gets a fallback
|
||
self.bdf_5x7_font = self.regular_font
|
||
|
||
|
||
def get_text_width(self, text, font):
|
||
"""Get the width of text when rendered with the given font.
|
||
|
||
Results are cached by (text, font identity) so plugins that measure
|
||
the same string every frame (e.g. to centre a score) pay only one
|
||
measurement per unique (text, font) pair. The entry keeps the font
|
||
alive so its id() can't be recycled, and the cache is LRU-bounded so
|
||
ever-changing text (clocks, tickers) can't grow it without limit.
|
||
"""
|
||
cache_key = (text, id(font))
|
||
cached = self._text_width_cache.get(cache_key)
|
||
if cached is not None:
|
||
self._text_width_cache.move_to_end(cache_key)
|
||
return cached[0]
|
||
|
||
try:
|
||
if isinstance(font, freetype.Face):
|
||
width = 0
|
||
for char in text:
|
||
font.load_char(char)
|
||
width += font.glyph.advance.x >> 6
|
||
else:
|
||
bbox = self.draw.textbbox((0, 0), text, font=font)
|
||
width = bbox[2] - bbox[0]
|
||
except (AttributeError, TypeError, ValueError, OSError) as e:
|
||
logger.error("Error getting text width: %s", e)
|
||
return 0
|
||
|
||
self._text_width_cache[cache_key] = (width, font)
|
||
while len(self._text_width_cache) > self._TEXT_WIDTH_CACHE_MAX:
|
||
self._text_width_cache.popitem(last=False)
|
||
return width
|
||
|
||
def get_font_height(self, font):
|
||
"""Get the height of the given font for line spacing purposes."""
|
||
try:
|
||
if isinstance(font, freetype.Face):
|
||
# For FreeType faces (BDF), the 'height' metric gives the recommended line spacing.
|
||
return font.size.height >> 6
|
||
else:
|
||
# For PIL TTF fonts, getmetrics() provides ascent and descent.
|
||
# The line height is the sum of ascent and descent.
|
||
ascent, descent = font.getmetrics()
|
||
return ascent + descent
|
||
except Exception as e:
|
||
logger.error(f"Error getting font height for font type {type(font).__name__}: {e}")
|
||
# Fallback for TTF font if getmetrics() fails, or for other font types.
|
||
if hasattr(font, 'size'):
|
||
return font.size
|
||
return 8 # A reasonable default for an 8px font.
|
||
|
||
def draw_text(self, text: str, x: int = None, y: int = None, color: tuple = (255, 255, 255),
|
||
small_font: bool = False, font: ImageFont = None, centered: bool = False):
|
||
"""Draw text on the canvas with optional font selection.
|
||
|
||
Args:
|
||
text: Text to display
|
||
x: X position (None to auto-center, or used as center point if centered=True)
|
||
y: Y position (None defaults to 0)
|
||
color: RGB color tuple
|
||
small_font: Use small font if True
|
||
font: Custom font object (overrides small_font)
|
||
centered: If True, x is treated as center point; if False, x is left edge
|
||
"""
|
||
try:
|
||
# Select font based on parameters
|
||
if font:
|
||
current_font = font
|
||
else:
|
||
current_font = self.small_font if small_font else self.regular_font
|
||
|
||
# Calculate x position
|
||
if x is None:
|
||
# No x provided - center text
|
||
text_width = self.get_text_width(text, current_font)
|
||
x = (self.width - text_width) // 2
|
||
elif centered:
|
||
# x is provided as center point - adjust to left edge
|
||
text_width = self.get_text_width(text, current_font)
|
||
x = x - (text_width // 2)
|
||
|
||
# Set default y position if not provided
|
||
if y is None:
|
||
y = 0 # Default to top of display
|
||
|
||
# Draw the text
|
||
if isinstance(current_font, freetype.Face):
|
||
# For BDF fonts, _draw_bdf_text will compute the baseline from the
|
||
# provided top-left y using the font ascender. Do not adjust here.
|
||
self._draw_bdf_text(text, x, y, color, current_font)
|
||
else:
|
||
# For TTF fonts, use PIL's text drawing which expects top-left.
|
||
self.draw.text((x, y), text, font=current_font, fill=color)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error drawing text: {e}", exc_info=True)
|
||
|
||
def draw_sun(self, x: int, y: int, size: int = 16):
|
||
"""Draw a sun icon using yellow circles and lines."""
|
||
center = (x + size//2, y + size//2)
|
||
radius = size//3
|
||
|
||
# Draw the center circle
|
||
self.draw.ellipse([center[0]-radius, center[1]-radius,
|
||
center[0]+radius, center[1]+radius],
|
||
fill=(255, 255, 0)) # Yellow
|
||
|
||
# Draw the rays
|
||
ray_length = size//4
|
||
for angle in range(0, 360, 45):
|
||
rad = math.radians(angle)
|
||
start_x = center[0] + (radius * math.cos(rad))
|
||
start_y = center[1] + (radius * math.sin(rad))
|
||
end_x = center[0] + ((radius + ray_length) * math.cos(rad))
|
||
end_y = center[1] + ((radius + ray_length) * math.sin(rad))
|
||
self.draw.line([start_x, start_y, end_x, end_y], fill=(255, 255, 0), width=2)
|
||
|
||
def draw_cloud(self, x: int, y: int, size: int = 16, color=(200, 200, 200)):
|
||
"""Draw a cloud icon."""
|
||
# Draw multiple circles to form a cloud shape
|
||
self.draw.ellipse([x+size//4, y+size//3, x+size//4+size//2, y+size//3+size//2], fill=color)
|
||
self.draw.ellipse([x+size//2, y+size//3, x+size//2+size//2, y+size//3+size//2], fill=color)
|
||
self.draw.ellipse([x+size//3, y+size//6, x+size//3+size//2, y+size//6+size//2], fill=color)
|
||
|
||
def draw_rain(self, x: int, y: int, size: int = 16):
|
||
"""Draw rain icon with cloud and droplets."""
|
||
# Draw cloud
|
||
self.draw_cloud(x, y, size)
|
||
|
||
# Draw rain drops
|
||
drop_color = (0, 0, 255) # Blue
|
||
drop_size = size//6
|
||
for i in range(3):
|
||
drop_x = x + size//4 + (i * size//3)
|
||
drop_y = y + size//2
|
||
self.draw.line([drop_x, drop_y, drop_x, drop_y+drop_size],
|
||
fill=drop_color, width=2)
|
||
|
||
def draw_snow(self, x: int, y: int, size: int = 16):
|
||
"""Draw snow icon with cloud and snowflakes."""
|
||
# Draw cloud
|
||
self.draw_cloud(x, y, size)
|
||
|
||
# Draw snowflakes
|
||
snow_color = (200, 200, 255) # Light blue
|
||
for i in range(3):
|
||
center_x = x + size//4 + (i * size//3)
|
||
center_y = y + size//2 + size//4
|
||
# Draw a small star shape
|
||
for angle in range(0, 360, 60):
|
||
rad = math.radians(angle)
|
||
end_x = center_x + (size//8 * math.cos(rad))
|
||
end_y = center_y + (size//8 * math.sin(rad))
|
||
self.draw.line([center_x, center_y, end_x, end_y],
|
||
fill=snow_color, width=1)
|
||
|
||
# Weather icon color constants
|
||
WEATHER_COLORS = {
|
||
'sun': (255, 200, 0), # Bright yellow
|
||
'cloud': (200, 200, 200), # Light gray
|
||
'rain': (0, 100, 255), # Light blue
|
||
'snow': (220, 220, 255), # Ice blue
|
||
'storm': (255, 255, 0) # Lightning yellow
|
||
}
|
||
|
||
def _draw_sun(self, x: int, y: int, size: int) -> None:
|
||
"""Draw a sun icon with rays."""
|
||
center_x, center_y = x + size//2, y + size//2
|
||
radius = size//4
|
||
ray_length = size//3
|
||
|
||
# Draw the main sun circle
|
||
self.draw.ellipse([center_x - radius, center_y - radius,
|
||
center_x + radius, center_y + radius],
|
||
fill=self.WEATHER_COLORS['sun'])
|
||
|
||
# Draw sun rays
|
||
for angle in range(0, 360, 45):
|
||
rad = math.radians(angle)
|
||
start_x = center_x + int((radius + 2) * math.cos(rad))
|
||
start_y = center_y + int((radius + 2) * math.sin(rad))
|
||
end_x = center_x + int((radius + ray_length) * math.cos(rad))
|
||
end_y = center_y + int((radius + ray_length) * math.sin(rad))
|
||
self.draw.line([start_x, start_y, end_x, end_y],
|
||
fill=self.WEATHER_COLORS['sun'], width=2)
|
||
|
||
def _draw_cloud(self, x: int, y: int, size: int) -> None:
|
||
"""Draw a cloud using multiple circles."""
|
||
cloud_color = self.WEATHER_COLORS['cloud']
|
||
base_y = y + size//2
|
||
|
||
# Draw main cloud body (3 overlapping circles)
|
||
circle_radius = size//4
|
||
positions = [
|
||
(x + size//3, base_y), # Left circle
|
||
(x + size//2, base_y - size//6), # Top circle
|
||
(x + 2*size//3, base_y) # Right circle
|
||
]
|
||
|
||
for cx, cy in positions:
|
||
self.draw.ellipse([cx - circle_radius, cy - circle_radius,
|
||
cx + circle_radius, cy + circle_radius],
|
||
fill=cloud_color)
|
||
|
||
def _draw_rain(self, x: int, y: int, size: int) -> None:
|
||
"""Draw rain drops falling from a cloud."""
|
||
self._draw_cloud(x, y, size)
|
||
rain_color = self.WEATHER_COLORS['rain']
|
||
|
||
# Draw rain drops at an angle
|
||
drop_size = size//8
|
||
drops = [
|
||
(x + size//4, y + 2*size//3),
|
||
(x + size//2, y + 3*size//4),
|
||
(x + 3*size//4, y + 2*size//3)
|
||
]
|
||
|
||
for dx, dy in drops:
|
||
# Draw angled rain drops
|
||
self.draw.line([dx, dy, dx - drop_size//2, dy + drop_size],
|
||
fill=rain_color, width=2)
|
||
|
||
def _draw_snow(self, x: int, y: int, size: int) -> None:
|
||
"""Draw snowflakes falling from a cloud."""
|
||
self._draw_cloud(x, y, size)
|
||
snow_color = self.WEATHER_COLORS['snow']
|
||
|
||
# Draw snowflakes
|
||
flake_size = size//6
|
||
flakes = [
|
||
(x + size//4, y + 2*size//3),
|
||
(x + size//2, y + 3*size//4),
|
||
(x + 3*size//4, y + 2*size//3)
|
||
]
|
||
|
||
for fx, fy in flakes:
|
||
# Draw a snowflake (six-pointed star)
|
||
for angle in range(0, 360, 60):
|
||
rad = math.radians(angle)
|
||
end_x = fx + int(flake_size * math.cos(rad))
|
||
end_y = fy + int(flake_size * math.sin(rad))
|
||
self.draw.line([fx, fy, end_x, end_y],
|
||
fill=snow_color, width=1)
|
||
|
||
def _draw_storm(self, x: int, y: int, size: int) -> None:
|
||
"""Draw a storm cloud with lightning bolt."""
|
||
self._draw_cloud(x, y, size)
|
||
|
||
# Draw lightning bolt
|
||
bolt_color = self.WEATHER_COLORS['storm']
|
||
bolt_points = [
|
||
(x + size//2, y + size//2), # Top
|
||
(x + 3*size//5, y + 2*size//3), # Middle right
|
||
(x + 2*size//5, y + 2*size//3), # Middle left
|
||
(x + size//2, y + 5*size//6) # Bottom
|
||
]
|
||
self.draw.polygon(bolt_points, fill=bolt_color)
|
||
|
||
def draw_weather_icon(self, condition: str, x: int, y: int, size: int = 16) -> None:
|
||
"""Draw a weather icon based on the condition."""
|
||
if condition.lower() in ['clear', 'sunny']:
|
||
self._draw_sun(x, y, size)
|
||
elif condition.lower() in ['clouds', 'cloudy', 'partly cloudy']:
|
||
self._draw_cloud(x, y, size)
|
||
elif condition.lower() in ['rain', 'drizzle', 'shower']:
|
||
self._draw_rain(x, y, size)
|
||
elif condition.lower() in ['snow', 'sleet', 'hail']:
|
||
self._draw_snow(x, y, size)
|
||
elif condition.lower() in ['thunderstorm', 'storm']:
|
||
self._draw_storm(x, y, size)
|
||
else:
|
||
self._draw_sun(x, y, size)
|
||
# Note: No update_display() here - let the caller handle the update
|
||
|
||
def draw_text_with_icons(self, text: str, icons: List[tuple] = None, x: int = None, y: int = None,
|
||
color: tuple = (255, 255, 255)):
|
||
"""Draw text with weather icons at specified positions."""
|
||
# Draw the text
|
||
self.draw_text(text, x, y, color)
|
||
|
||
# Draw any icons
|
||
if icons:
|
||
for icon_type, icon_x, icon_y in icons:
|
||
self.draw_weather_icon(icon_type, icon_x, icon_y)
|
||
|
||
# Update the display once after everything is drawn
|
||
self.update_display()
|
||
|
||
def cleanup(self):
|
||
"""Clean up resources."""
|
||
if hasattr(self, 'matrix') and self.matrix is not None:
|
||
try:
|
||
self.matrix.Clear()
|
||
except Exception as e:
|
||
logger.warning(f"Error clearing matrix during cleanup: {e}")
|
||
# Ensure image/draw are reset to a blank state
|
||
if hasattr(self, 'image') and hasattr(self, 'draw'):
|
||
try:
|
||
self.image = Image.new('RGB', (self.width, self.height))
|
||
self.draw = ImageDraw.Draw(self.image)
|
||
self.draw.fontmode = "1" # 1-bit text: the panel has no partial brightness, so AA only smears glyphs.
|
||
except (OSError, RuntimeError, ValueError, MemoryError):
|
||
logger.debug("Canvas reset during cleanup failed", exc_info=True)
|
||
# Reset the singleton state when cleaning up
|
||
DisplayManager._instance = None
|
||
DisplayManager._initialized = False
|
||
|
||
def format_date_with_ordinal(self, dt):
|
||
"""Formats a datetime object into 'Mon Aug 30th' style."""
|
||
day = dt.day
|
||
if 11 <= day <= 13:
|
||
suffix = 'th'
|
||
else:
|
||
suffix = {1: 'st', 2: 'nd', 3: 'rd'}.get(day % 10, 'th')
|
||
|
||
return dt.strftime(f"%b %-d{suffix}")
|
||
|
||
@property
|
||
def refresh_hz(self) -> float:
|
||
"""The panel's refresh rate in Hz, from the hardware config.
|
||
|
||
The authoritative place to ask, because a plugin only receives its own
|
||
config section and cannot see display.hardware. Scroll pacing needs
|
||
this: the speeds a panel can show in whole pixels are refresh_hz
|
||
divided by the frame hold, so getting it wrong silently produces
|
||
fractional-pixel motion. See src/common/scroll_config.py.
|
||
|
||
Note this is the configured *cap*, not necessarily what the panel
|
||
achieves -- scripts/scroll_speeds.py --measure reports the real rate.
|
||
"""
|
||
hardware = (self.config.get('display') or {}).get('hardware') or {}
|
||
try:
|
||
value = float(hardware.get('limit_refresh_rate_hz') or 0)
|
||
except (TypeError, ValueError):
|
||
value = 0.0
|
||
return value if value > 0 else 100.0
|
||
|
||
def set_frame_hold(self, refreshes: int) -> None:
|
||
"""Hold each pushed frame for this many panel refreshes (>=1).
|
||
|
||
Set by the scroll configuration so a plugin can run at, say, 50px/s on
|
||
a 100Hz panel as one whole pixel every second refresh, rather than half
|
||
a pixel every refresh (which has to be blended or repeated unevenly).
|
||
|
||
Reset to 1 whenever scrolling stops, so one plugin's pacing cannot
|
||
leak into the next thing on screen.
|
||
"""
|
||
try:
|
||
value = int(refreshes)
|
||
except (TypeError, ValueError):
|
||
logger.warning("Ignoring unusable frame hold: %r", refreshes)
|
||
return
|
||
self._frame_hold = max(1, min(255, value))
|
||
|
||
def set_scrolling_state(self, is_scrolling: bool, frame_hold: int = 1):
|
||
"""Set the current scrolling state, and this scroll's frame pacing.
|
||
|
||
Call this when a display starts or stops scrolling. ``frame_hold`` is
|
||
how many panel refreshes each frame is held for -- 2 gives one whole
|
||
pixel every second refresh, which is how a scroll runs at half the
|
||
refresh rate without fractional pixel positions.
|
||
|
||
The hold is set here rather than once at plugin construction because
|
||
it must not outlive the scroll that asked for it: plugins share one
|
||
display manager, so a hold left set by whoever scrolled last would
|
||
silently re-pace the next plugin. Passing it alongside the state makes
|
||
the lifetime exactly the scroll, and the default of 1 means any caller
|
||
that does not care gets a new frame every refresh.
|
||
"""
|
||
current_time = time.time()
|
||
self._scrolling_state['is_scrolling'] = is_scrolling
|
||
if is_scrolling:
|
||
self._scrolling_state['last_scroll_activity'] = current_time
|
||
self.set_frame_hold(frame_hold)
|
||
else:
|
||
self._frame_hold = 1
|
||
logger.debug(f"Scrolling state set to: {is_scrolling}")
|
||
|
||
def is_currently_scrolling(self) -> bool:
|
||
"""Check if the display is currently in a scrolling state."""
|
||
current_time = time.time()
|
||
|
||
# If explicitly not scrolling, return False
|
||
if not self._scrolling_state['is_scrolling']:
|
||
return False
|
||
|
||
# If we've been inactive for the threshold period, consider it not scrolling
|
||
if current_time - self._scrolling_state['last_scroll_activity'] > self._scrolling_state['scroll_inactivity_threshold']:
|
||
self._scrolling_state['is_scrolling'] = False
|
||
return False
|
||
|
||
return True
|
||
|
||
def defer_update(self, update_func, priority: int = 0):
|
||
"""Defer an update function to be called when not scrolling.
|
||
|
||
Args:
|
||
update_func: Function to call when not scrolling
|
||
priority: Priority level (lower numbers = higher priority)
|
||
"""
|
||
current_time = time.time()
|
||
|
||
# Clean up expired updates before adding new ones
|
||
self._cleanup_expired_deferred_updates(current_time)
|
||
|
||
# Limit queue size to prevent memory issues
|
||
if len(self._scrolling_state['deferred_updates']) >= self._scrolling_state['max_deferred_updates']:
|
||
# Remove oldest update to make room
|
||
self._scrolling_state['deferred_updates'].pop(0)
|
||
logger.debug("Removed oldest deferred update due to queue size limit")
|
||
|
||
self._scrolling_state['deferred_updates'].append({
|
||
'func': update_func,
|
||
'priority': priority,
|
||
'timestamp': current_time
|
||
})
|
||
|
||
# Only sort if we have a reasonable number of updates to avoid excessive sorting
|
||
if len(self._scrolling_state['deferred_updates']) <= 20:
|
||
self._scrolling_state['deferred_updates'].sort(key=lambda x: x['priority'])
|
||
|
||
logger.debug(f"Deferred update added. Total deferred: {len(self._scrolling_state['deferred_updates'])}")
|
||
|
||
def process_deferred_updates(self):
|
||
"""Process any deferred updates if not currently scrolling."""
|
||
current_time = time.time()
|
||
|
||
# Always clean up expired updates, even if scrolling
|
||
# This prevents memory leaks from accumulated expired updates
|
||
self._cleanup_expired_deferred_updates(current_time)
|
||
|
||
if self.is_currently_scrolling():
|
||
return
|
||
|
||
if not self._scrolling_state['deferred_updates']:
|
||
return
|
||
|
||
if not self._scrolling_state['deferred_updates']:
|
||
return
|
||
|
||
# Process only a limited number of updates per call to avoid blocking
|
||
max_updates_per_call = min(5, len(self._scrolling_state['deferred_updates']))
|
||
updates_to_process = self._scrolling_state['deferred_updates'][:max_updates_per_call]
|
||
self._scrolling_state['deferred_updates'] = self._scrolling_state['deferred_updates'][max_updates_per_call:]
|
||
|
||
logger.debug(f"Processing {len(updates_to_process)} deferred updates (queue size: {len(self._scrolling_state['deferred_updates'])})")
|
||
|
||
failed_updates = []
|
||
for update_info in updates_to_process:
|
||
try:
|
||
# Check if update is still valid (not too old)
|
||
if current_time - update_info['timestamp'] > self._scrolling_state['deferred_update_ttl']:
|
||
logger.debug("Skipping expired deferred update")
|
||
continue
|
||
|
||
update_info['func']()
|
||
logger.debug("Deferred update executed successfully")
|
||
except Exception as e:
|
||
logger.error(f"Error executing deferred update: {e}")
|
||
# Only retry recent failures, and limit retries
|
||
if current_time - update_info['timestamp'] < 60.0: # Only retry for 1 minute
|
||
failed_updates.append(update_info)
|
||
|
||
# Re-add failed updates to the end of the queue (not the beginning)
|
||
if failed_updates:
|
||
self._scrolling_state['deferred_updates'].extend(failed_updates)
|
||
|
||
def _cleanup_expired_deferred_updates(self, current_time: float):
|
||
"""Remove expired deferred updates to prevent memory leaks."""
|
||
ttl = self._scrolling_state['deferred_update_ttl']
|
||
initial_count = len(self._scrolling_state['deferred_updates'])
|
||
|
||
# Filter out expired updates
|
||
self._scrolling_state['deferred_updates'] = [
|
||
update for update in self._scrolling_state['deferred_updates']
|
||
if current_time - update['timestamp'] <= ttl
|
||
]
|
||
|
||
removed_count = initial_count - len(self._scrolling_state['deferred_updates'])
|
||
if removed_count > 0:
|
||
logger.debug(f"Cleaned up {removed_count} expired deferred updates")
|
||
|
||
def get_scrolling_stats(self) -> dict:
|
||
"""Get current scrolling statistics for debugging."""
|
||
return {
|
||
'is_scrolling': self._scrolling_state['is_scrolling'],
|
||
'last_activity': self._scrolling_state['last_scroll_activity'],
|
||
'deferred_count': len(self._scrolling_state['deferred_updates']),
|
||
'inactivity_threshold': self._scrolling_state['scroll_inactivity_threshold'],
|
||
'max_deferred_updates': self._scrolling_state['max_deferred_updates'],
|
||
'deferred_update_ttl': self._scrolling_state['deferred_update_ttl']
|
||
}
|
||
|
||
def _viewer_is_fresh(self, now: float) -> bool:
|
||
"""True when a browser preview is watching (marker file touched by
|
||
the web SSE broadcaster). The marker is stat'd at most once per
|
||
second — at 125 fps loops a per-call stat would be pure overhead."""
|
||
if (now - self._viewer_check_ts) >= 1.0:
|
||
self._viewer_check_ts = now
|
||
try:
|
||
marker_age = now - os.stat(self._viewer_marker_path).st_mtime
|
||
self._viewer_fresh = marker_age < snapshot_policy.VIEWER_MARKER_FRESH_SEC
|
||
except OSError:
|
||
self._viewer_fresh = False
|
||
return self._viewer_fresh
|
||
|
||
def _write_snapshot_if_due(self, frame_checksum: Optional[int] = None) -> None:
|
||
"""Mirror the current frame to the preview snapshot when the policy
|
||
says it's worth it — see src/common/snapshot_policy.py. Unchanged
|
||
frames are never re-encoded; without viewers the cadence drops to
|
||
the idle keepalive.
|
||
|
||
Args:
|
||
frame_checksum: adler32 of the current frame, when the caller has
|
||
already computed one. Dirty tracking checksums every frame a
|
||
few lines above the call site, and re-deriving it here meant a
|
||
second tobytes() plus a second pass over the whole framebuffer
|
||
on every single frame — ~0.17ms per frame of the two combined
|
||
at 256x64, paid 100 times a second to reach the same number.
|
||
"""
|
||
try:
|
||
now = time.time()
|
||
viewer_fresh = self._viewer_is_fresh(now)
|
||
if viewer_fresh and not self._viewer_was_fresh:
|
||
# A preview just opened: let the next changed frame through
|
||
# immediately instead of waiting out the idle interval.
|
||
self._last_snapshot_ts = 0.0
|
||
self._viewer_was_fresh = viewer_fresh
|
||
|
||
digest = (frame_checksum if frame_checksum is not None
|
||
else zlib.adler32(self.image.tobytes()))
|
||
action = snapshot_policy.decide(
|
||
now, self._last_snapshot_ts, self._last_snapshot_touch_ts,
|
||
viewer_fresh, digest != self._last_snapshot_digest)
|
||
if action is snapshot_policy.SnapshotAction.SKIP:
|
||
return
|
||
if action is snapshot_policy.SnapshotAction.TOUCH:
|
||
# mtime bump only: keeps the health check (snapshot age)
|
||
# green without paying for a PNG encode of an unchanged frame
|
||
os.utime(self._snapshot_path, None)
|
||
self._last_snapshot_touch_ts = now
|
||
return
|
||
|
||
# WRITE: ensure directory permissions once, not per frame
|
||
snapshot_path_obj = Path(self._snapshot_path)
|
||
if not self._snapshot_dir_prepared:
|
||
# Never modify /tmp permissions - it has special system
|
||
# permissions (1777) that must not be changed or it breaks
|
||
# apt and other system tools
|
||
parent_dir = snapshot_path_obj.parent
|
||
if parent_dir and str(parent_dir) != '/tmp': # nosec B108 - guard to skip /tmp for permission ops
|
||
ensure_directory_permissions(parent_dir, get_assets_dir_mode())
|
||
self._snapshot_dir_prepared = True
|
||
# Write atomically: temp then replace. The temp name must be
|
||
# unique, not "<snapshot>.tmp": /tmp is world-writable and sticky,
|
||
# and this file is written by whichever user the display service
|
||
# runs as while tests and tooling run as someone else. A leftover
|
||
# fixed-name temp owned by another user is then unopenable even by
|
||
# root (fs.protected_regular refuses O_CREAT on a foreign file in a
|
||
# sticky dir), which froze the preview and the health check's
|
||
# liveness proxy until somebody deleted it by hand. Same pattern as
|
||
# the hardware-status write above.
|
||
_fd, tmp_path = tempfile.mkstemp(
|
||
dir=str(snapshot_path_obj.parent),
|
||
prefix=f".{snapshot_path_obj.name}.", suffix=".tmp")
|
||
try:
|
||
with os.fdopen(_fd, "wb") as _f:
|
||
self.image.save(_f, format='PNG')
|
||
os.chmod(tmp_path, 0o644)
|
||
os.replace(tmp_path, self._snapshot_path)
|
||
except Exception:
|
||
# Never leave the temp behind -- that is what made the failure
|
||
# permanent rather than transient.
|
||
try:
|
||
os.unlink(tmp_path)
|
||
except OSError:
|
||
pass
|
||
# Fallback to direct save if replace not supported
|
||
self.image.save(self._snapshot_path, format='PNG')
|
||
# Set proper file permissions after saving
|
||
try:
|
||
ensure_file_permissions(snapshot_path_obj, get_assets_file_mode())
|
||
except Exception:
|
||
pass
|
||
self._last_snapshot_ts = now
|
||
self._last_snapshot_touch_ts = now
|
||
self._last_snapshot_digest = digest
|
||
except Exception as e:
|
||
# Snapshot failures must never break display — but they must not
|
||
# be silent either: the snapshot's mtime is the web UI's display
|
||
# mirror AND its hardware-liveness proxy, so a quietly failing
|
||
# write freezes the mirror and makes health checks lie (seen in
|
||
# the field: a stale root-owned /tmp file froze it for a day).
|
||
# Warn at most once per 5 minutes to avoid log spam.
|
||
if (now - self._snapshot_fail_log_ts) > 300:
|
||
self._snapshot_fail_log_ts = now
|
||
logger.warning("Snapshot write failing (web preview/health "
|
||
"mirror is stale): %s", e)
|
||
else:
|
||
logger.debug(f"Snapshot write skipped: {e}") |