mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-04 14:25:08 +00:00
ChuckandClaude Opus 5
f8e2e89edc
* refactor(sports): put the scoreboards on the shared scroll resolver Eight sports scoreboards -- afl, baseball, basketball, football, hockey, lacrosse, nrl, soccer -- scrolled through this module's own pacing while the other eleven scrolling plugins went through src/common/scroll_config. Two implementations of the same job, and this one was on the losing side of every difference. It never called set_scrolling_state. Two consequences, both of which this release's work was about: - The frame hold is applied through that call, so a speed the crisp ladder could render in whole pixels still presented a new frame every refresh. - Core only runs deferred updates while nothing is scrolling. Believing nothing was, it ran blocking work in the middle of these scrolls. The default is non-crisp today: scroll_speed 50.0 with scroll_delay 0.01 is 50 px/s, which on a 100Hz panel is half a pixel per refresh. That cannot render as motion -- it alternates 0px and 1px steps and judders at a 50Hz beat, on every scoreboard, out of the box. Resolved through the ladder it stays 50 px/s and holds each frame for two refreshes: same speed, whole-pixel motion. The stepping disagreement that used to justify a separate module is gone. scroll_config avoided frame-based mode because it stepped on a wall clock at 1/scroll_delay with scroll_delay set to the frame period, so the decision sat on its own threshold and flipped on sub-millisecond jitter. That branch now accumulates elapsed time, identical arithmetic to the time-based one, so the two differ only in the units the speed arrives in. What is NOT shared, and must not be: the two modules read identically-named keys with different meanings. Here scroll_speed is px/SECOND and scroll_delay only converts to px/frame; in scroll_config scroll_speed is px per STEP, so px/s is speed/delay. Passing this module's settings dict to the resolver turns 50 px/s into 5000, clamped to 500 -- a tenfold speed-up everywhere. So _get_scroll_settings keeps sole ownership of reading sports config, including the league merging, and hands the resolver a plain px/s. A test pins that specific number, because it is the mistake the refactor invites. MIN/MAX_PIXELS_PER_FRAME are gone; the resolver bounds speed and the helper clamps FPS. _resolve_target_fps stays, re-purposed: under the old model that key was the rate frames were presented at, so it is the faithful translation into the refresh the ladder is computed against, used when no hardware refresh is configured. Speed changes for panels that are not 100Hz: 50 px/s becomes 60 at 60Hz (+20%) and 48 at 120Hz (-4%). At 100Hz it is unchanged. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(display): drop the frame hold when a scroll times out, not just when it says so set_scrolling_state(False) clears the hold. The other way a scroll ends is is_currently_scrolling() deciding, after scroll_inactivity_threshold of silence, that it is over -- which is what happens when the rotation moves on mid-scroll or a plugin is torn down. That path cleared the flag and kept the hold, so every later plugin, scrolling or static, was presented at refresh/N by whoever scrolled last, until something called the explicit stop. The method's own docstring already states the rule this breaks: the hold "must not outlive the scroll that asked for it". The timeout was the exception it did not cover. Pre-existing, but reachable by three plugins before and eleven after the sports scoreboards moved onto the shared resolver, so it belongs with that change. The test ages the activity timestamp past the threshold rather than sleeping. Also adds scripts/sports_scroll_check.py. The sports scroll path is per-league opt-in, so a rig showing static game cards never constructs a SportsScrollDisplay and none of its pacing can be observed from a normal run -- which is exactly what happened when this change was first put on hardware: 26 minutes, zero sports scroll lines. The script drives the path directly with synthetic games and asserts the three things the resolver is meant to buy: the speed lands on whole pixels, the hold is published, and it is released after. It never starts or stops the display service, matching scroll_speeds.py, so a crash here cannot leave the panel dark. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(scripts): refuse to grab the panel while the display service has it The module docstring already said to stop ledmatrix first. Nothing enforced it, and running the script against a live service is not a harmless mistake: rpi-rgb-led-matrix configures GPIO directions and the hardware PWM inside RGBMatrix(), and when the root check fails it calls exit() from C with no cleanup. The service keeps rendering and swapping onto pins that have been reconfigured underneath it, so the panel goes black while every diagnostic says the display is healthy -- fresh framebuffer, every pixel lit, "RGB Matrix initialized successfully", nothing in the log. A restart fixes it, once you work out that is what happened. Found the hard way: this is what took the panel down on the test rig, not the change the script was written to verify. --fallback skips the check, since it never opens the matrix. --force is there for anyone who means it. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * chore(scripts): annotate the subprocess call the way this repo already does Codacy fails a PR on one new issue, and bandit B404 fires on any subprocess import. scripts/run_plugin_tests.py carries the same suppression with the same justification -- list-form argv, no shell -- so this follows it rather than inventing a second convention. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
1657 lines
78 KiB
Python
1657 lines
78 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
|
||
from src.common.font_layout import load_truetype
|
||
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
|
||
|
||
#: The strike 5x7.bdf is drawn at. FreeType renders a BDF at its own fixed
|
||
#: size regardless, but a Face needs an active size before its metrics --
|
||
#: and therefore get_font_height() -- report anything but 0.
|
||
_CALENDAR_FONT_PX = 7
|
||
|
||
|
||
def _bdf_native_size(face) -> int:
|
||
"""The pixel height a BDF Face declares, or 0 if it does not say.
|
||
|
||
Used only to rescue a Face that was built without ``set_char_size``, so a
|
||
zero line height never reaches layout code.
|
||
"""
|
||
try:
|
||
sizes = getattr(face, "available_sizes", None) or []
|
||
if sizes:
|
||
return int(getattr(sizes[0], "height", 0) or 0)
|
||
except (AttributeError, IndexError, TypeError, ValueError) as exc:
|
||
# This runs on the measurement path for a face the caller already
|
||
# holds, so a malformed strike table must degrade to "unknown" rather
|
||
# than take the display down. Say which face, so a font that is
|
||
# actually broken is diagnosable rather than silently 8px.
|
||
logger.debug("Could not read BDF strike size from %r: %s", face, exc)
|
||
return 0
|
||
|
||
|
||
|
||
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 = load_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.
|
||
|
||
The ladder ends at 4x6 at 5px because a full dotted-quad address --
|
||
"255.255.255.255", the widest this screen ever shows -- is 66px at
|
||
6px and a 64px panel has 62 to give it. That used to squeak in only
|
||
because the measurement depended on which text layout engine the host
|
||
Pillow had; with the engine pinned it does not, so the rung the
|
||
worst case actually needs is here rather than implied.
|
||
"""
|
||
candidates = [self.font,
|
||
("assets/fonts/4x6-font.ttf", 6),
|
||
("assets/fonts/4x6-font.ttf", 5)]
|
||
narrowest = None
|
||
for candidate in candidates:
|
||
try:
|
||
font = candidate
|
||
if isinstance(candidate, tuple):
|
||
font = load_truetype(candidate[0], candidate[1])
|
||
narrowest = font
|
||
if all(self.draw.textlength(t, font=font) <= width for t in lines):
|
||
return font
|
||
except (OSError, ValueError, AttributeError):
|
||
continue
|
||
# Nothing fit. Return the smallest face that loaded, not self.font --
|
||
# falling back to the widest option is how "Initializing" ran off the
|
||
# side of a 64px panel in the first place.
|
||
return narrowest or 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 = load_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 = load_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)
|
||
# A freshly constructed Face has no active size, so
|
||
# face.size.height is 0 until set_char_size is called -- and
|
||
# get_font_height() reads exactly that. Without this, every
|
||
# caller measuring the 5x7 face got 0 and stacked rows on top
|
||
# of one another; the "Calendar font size: 0 pixels" line
|
||
# below has been printing the symptom on every start-up.
|
||
# font_manager._load_bdf_font already does this; the two paths
|
||
# disagreed about whether a Face was usable for measurement.
|
||
# 5x7.bdf is a fixed strike, so FreeType renders 7px whatever
|
||
# is asked for -- this sets the metrics, not the raster.
|
||
face.set_char_size(_CALENDAR_FONT_PX * 64, _CALENDAR_FONT_PX * 64, 72, 72)
|
||
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 = load_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.
|
||
height = font.size.height >> 6
|
||
if height:
|
||
return height
|
||
# A Face constructed without set_char_size reports 0, and a
|
||
# zero line height collapses every stacked row onto one line.
|
||
# Fall back to the strike the file declares.
|
||
return _bdf_native_size(font) or 8
|
||
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
|
||
# Drop the hold with the state, exactly as set_scrolling_state(False)
|
||
# does. This path is the one a scroll takes when it ends without
|
||
# saying so -- the rotation moves on mid-scroll, or the plugin is
|
||
# torn down -- and leaving the hold set there means every later
|
||
# plugin, scrolling or static, is presented at refresh/N until
|
||
# somebody calls set_scrolling_state(False). The hold must not
|
||
# outlive the scroll that asked for it, however that scroll ends.
|
||
self._frame_hold = 1
|
||
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}") |