mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-08-01 16:58:06 +00:00
4d49b0f892
The display service PNG-encoded its frame to /tmp/led_matrix_preview.png at 5 fps, 24/7 — identical frames, no viewers, per-call imports and a chmod every write. On the devpi baseline the display service idles at ~92% CPU; this was one of its biggest fixed costs. - New pure policy (src/common/snapshot_policy.py, unit-tested off-Pi): WRITE changed frames at full rate only while a viewer is watching, at a 30s idle cadence otherwise; NEVER re-encode unchanged frames — bump mtime (os.utime) every 20s instead, keeping the health check's snapshot-age liveness proxy (60s threshold in api_v3) green. Cross- referencing comments guard the two constants. - Viewer detection: the web SSE display broadcaster (which only runs while browsers are subscribed) touches /tmp/led_matrix_preview_viewer each loop; the display service stats it at most 1/s. On viewer arrival the write clock resets so the first frame lands within ~1s. - Hoisted the per-call pathlib/permission_utils imports; directory permissions ensured once instead of every frame. Claude-Session: https://claude.ai/code/session_01FqzC1nzTWL4kaqgMaQZFam Co-authored-by: Chuck <chuck@example.com> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
1245 lines
57 KiB
Python
1245 lines
57 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 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 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
|
||
# When True, update_display() and clear() skip hardware writes (used during off-screen content capture)
|
||
self._capture_mode_active = False
|
||
# 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
|
||
|
||
# Scrolling state tracking for graceful updates
|
||
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
|
||
|
||
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 = hardware_config.get('pixel_mapper_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)
|
||
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)
|
||
# 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
|
||
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
|
||
|
||
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))
|
||
|
||
# Draw some text - changed from "TEST" to "Initializing" with smaller font
|
||
self.draw.text((10, 10), "Initializing", font=self.font, fill=(0, 0, 255))
|
||
|
||
# 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)
|
||
|
||
@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
|
||
|
||
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."""
|
||
try:
|
||
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
|
||
|
||
# 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
|
||
self.matrix.SwapOnVSync(self.offscreen_canvas)
|
||
|
||
# Swap our canvas references
|
||
self.offscreen_canvas, self.current_canvas = self.current_canvas, self.offscreen_canvas
|
||
|
||
# Write a snapshot for the web preview (throttled)
|
||
self._write_snapshot_if_due()
|
||
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)
|
||
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)
|
||
|
||
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.
|
||
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)
|
||
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}")
|
||
|
||
def set_scrolling_state(self, is_scrolling: bool):
|
||
"""Set the current scrolling state. Call this when a display starts/stops scrolling."""
|
||
current_time = time.time()
|
||
self._scrolling_state['is_scrolling'] = is_scrolling
|
||
if is_scrolling:
|
||
self._scrolling_state['last_scroll_activity'] = current_time
|
||
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) -> 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."""
|
||
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 = 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
|
||
tmp_path = f"{self._snapshot_path}.tmp"
|
||
self.image.save(tmp_path, format='PNG')
|
||
try:
|
||
os.replace(tmp_path, self._snapshot_path)
|
||
except Exception:
|
||
# 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}") |