Merge remote-tracking branch 'origin/main' into claude/frame-timing-harness

# Conflicts:
#	CHANGELOG.md
#	docs/SCROLL_PERFORMANCE.md
This commit is contained in:
Chuck
2026-09-24 16:30:49 -04:00
62 changed files with 6170 additions and 3494 deletions
+2 -1
View File
@@ -26,6 +26,7 @@ from enum import Enum
from concurrent.futures import ThreadPoolExecutor
import pytz
from src.cache_manager import CacheManager
from src.common.json_body import response_json
from src.common.espn_dates import (
RANGE_RETRY_SECONDS,
_note_range_rejected,
@@ -389,7 +390,7 @@ class BackgroundDataService:
response.raise_for_status()
else:
response.raise_for_status()
data = response.json()
data = response_json(response)
# Validate data structure
if not isinstance(data, dict):
+43
View File
@@ -7,6 +7,7 @@ Handles persistent disk-based caching with atomic writes and error recovery.
import json
import math
import os
import re
import stat
import time
import tempfile
@@ -98,6 +99,40 @@ def _replace_nonfinite(obj: Any) -> Any:
# deleted. Both halves are covered by test/test_cache_nonfinite_floats.py.
#: Enough of a record to hold its header: ``{"timestamp":<float>,"ttl":<n>,``.
_HEAD_BYTES = 256
#: A record written with its header first (CacheManager.set does). Anything
#: else -- older files with "data" first, records from other writers -- does not
#: match and is parsed in full, as before.
_HEAD_RE = re.compile(
rb'\A\s*\{\s*"timestamp"\s*:\s*(-?[0-9][0-9.eE+-]*)\s*'
rb'(?:,\s*"ttl"\s*:\s*(-?[0-9][0-9.eE+-]*))?\s*[,}]'
)
def _stale_from_head(head: bytes, max_age: Optional[int], now: float) -> bool:
"""True when a record's header alone shows it has expired.
Mirrors the expiry rule in DiskCache.get: a per-entry ttl wins over the
caller's max_age, and no limit at all means never stale. False whenever the
header cannot be read, so the full parse decides as it always did.
"""
match = _HEAD_RE.match(head)
if not match:
return False
try:
timestamp = float(match.group(1))
limit = max_age
if match.group(2) is not None:
ttl = float(match.group(2))
if ttl >= 0:
limit = ttl
except ValueError:
return False
return limit is not None and (now - timestamp) > limit
if orjson is not None:
# Encoding the cache record dominated the background fetch worker: on a
# Pi 4, stdlib json.dumps runs ~12ms per MB and holds the GIL for all of
@@ -266,6 +301,14 @@ class DiskCache:
try:
with self._lock:
with open(cache_path, 'rb') as f:
# Decide staleness from the header before paying for the
# parse. A stale read is the common case for the biggest
# records (a season schedule is re-fetched when its cache
# expires), and parsing 53MB to throw it away held the GIL
# for ~1.8s -- a visible freeze on the panel.
if _stale_from_head(f.read(_HEAD_BYTES), max_age, time.time()):
return None
f.seek(0)
record = _loads(f.read())
# Determine record timestamp (prefer embedded, else file mtime)
+9 -5
View File
@@ -522,8 +522,9 @@ class CacheManager:
def update_cache(self, data_type: str, data: Dict[str, Any]) -> bool:
"""Update cache with new data."""
cache_data = {
# Header first; see DiskCache's stale check.
'timestamp': time.time(),
'data': data,
'timestamp': time.time()
}
return self.save_cache(data_type, cache_data)
@@ -556,12 +557,15 @@ class CacheManager:
from the key and is only a fallback for entries that did not
say. Omit it to keep that inferred behaviour.
"""
cache_data = {
'data': data,
'timestamp': time.time()
}
# timestamp and ttl before data, so they are the first bytes on disk:
# DiskCache.get reads them from the head of the file and can call a
# record stale without parsing it. That matters for the big ones -- a
# whole MLB season is 53MB and ~1.8s of orjson.loads with the GIL held,
# paid in full only to learn the record had expired.
cache_data: Dict[str, Any] = {'timestamp': time.time()}
if ttl is not None:
cache_data['ttl'] = ttl
cache_data['data'] = data
self.save_cache(key, cache_data)
@deprecated("3.7.0")
+9
View File
@@ -36,6 +36,15 @@ Utilities for loading and managing team logos.
Utilities for text processing and formatting.
## BDF Fonts (`bdf_font.py`)
The one way to load and draw BDF bitmap fonts. `load_bdf_face(path, size)`
returns `(face, realised_px)`, falling back to the file's native strike when
it has none at `size`; `draw_bdf_text(draw, text, x, y, face, color)` draws
top-left anchored onto a PIL `ImageDraw` exactly as the panel does.
`DisplayManager`, `FontManager`, `element_style` and the plugin test harness
all go through it.
## Scroll Helpers (`scroll_helper.py`)
Utilities for scrolling text on the display.
+261
View File
@@ -0,0 +1,261 @@
"""Loading and drawing BDF bitmap fonts: one loader, one rasterizer.
BDF fonts are fixed-size bitmap strikes. FreeType renders them at the size
baked into the file and rejects any other size, and PIL cannot draw a
``freetype.Face`` at all, so the core draws BDF text itself, glyph by glyph.
This used to be done in several places that drifted apart:
``FontManager``, ``element_style`` and ``DisplayManager`` each loaded faces
their own way, and ``DisplayManager`` and the plugin test harness
(``VisualTestDisplayManager``) each had a copy of the glyph drawing loop. The
harness renders plugin golden images and ``check_plugin`` / ``dev_server``
previews, so a copy that differs from the panel's shows something the panel
never draws. Everything now goes through the two functions here:
* :func:`load_bdf_face` -- a ``freetype.Face`` at the requested pixel size,
or at the file's native strike when the file has no strike at that size.
* :func:`draw_bdf_text` -- draw a string in a ``freetype.Face`` onto a PIL
``ImageDraw``, top-left anchored like ``ImageDraw.text``.
Only PIL and freetype-py are imported, so the module is as cheap to import
from the test harness as from core.
"""
from __future__ import annotations
import ctypes
import logging
import os
import threading
from collections import OrderedDict
from typing import Any, Optional, Sequence, Tuple
from PIL import Image
try:
import freetype
except ImportError: # pragma: no cover - freetype-py is a core requirement
freetype = None
logger = logging.getLogger(__name__)
__all__ = ["read_bdf_native_size", "load_bdf_face", "draw_bdf_text"]
# --------------------------------------------------------------------------
# Loading
# --------------------------------------------------------------------------
def read_bdf_native_size(bdf_path: str) -> Optional[int]:
"""A BDF file's one true pixel size, read from its header, or None.
Prefers the PIXEL_SIZE property, which states the real pixel height
directly; falls back to the SIZE line's point-size only if PIXEL_SIZE is
absent, since point-size only equals pixel height at exactly 100dpi --
several bundled fonts (e.g. 6x13.bdf, 5x8.bdf) are defined at 75dpi, where
the two values genuinely differ. Stops at the first STARTCHAR.
"""
size_line_value = None
try:
with open(bdf_path, "r", encoding="ascii", errors="ignore") as f:
for line in f:
if line.startswith("PIXEL_SIZE"):
parts = line.split()
if len(parts) >= 2:
return int(float(parts[1]))
elif line.startswith("SIZE") and size_line_value is None:
# Format: "SIZE <point_size> <xres> <yres>"
parts = line.split()
if len(parts) >= 2:
size_line_value = int(float(parts[1]))
elif line.startswith("STARTCHAR"):
break
except (OSError, ValueError):
return None
return size_line_value
#: Loaded faces, keyed on (absolute path, requested size, mtime_ns, file size)
#: so a font file replaced on disk under the same name is loaded afresh.
#: Bounded LRU: the display process runs for weeks and every config save can
#: introduce a new (font, size) pair, but a panel draws from a handful.
_FACE_CACHE_MAX = 256
_face_cache: "OrderedDict[tuple, Tuple[Any, int]]" = OrderedDict()
_face_cache_lock = threading.Lock()
def _face_at(path: str, size_px: int) -> Any:
face = freetype.Face(path)
# Character size in 1/64th points at 72dpi == pixel size.
face.set_char_size(size_px * 64, size_px * 64, 72, 72)
return face
def load_bdf_face(path: str, size_px: int) -> Tuple[Any, int]:
"""``(face, realised_px)`` for the BDF file at ``path``.
``realised_px`` is ``size_px`` when the file has a strike at that size,
otherwise the file's native size: FreeType refuses any other size for a
bitmap font, and answering that with some other typeface (which both
``FontManager`` and ``element_style`` once did) is worse than drawing the
font that was asked for at the size it can do. Callers that lay out by
size need ``realised_px``, not the size they asked for.
Faces are cached per thread. A ``freetype.Face`` holds per-glyph state
(``load_char`` rewrites its glyph slot), and FreeType does not allow two
threads to use one face at once, so the display thread and a plugin's
update thread must never be handed the same object. Within a thread the
face is shared by every caller. Raises if the file can't be loaded at
either size.
"""
if freetype is None:
raise RuntimeError("freetype-py is not installed; BDF fonts need it")
size_px = int(size_px)
abs_path = os.path.abspath(path)
try:
st = os.stat(abs_path)
key = (threading.get_ident(), abs_path, size_px,
st.st_mtime_ns, st.st_size)
except OSError:
key = None # let freetype raise its own error below
if key is not None:
with _face_cache_lock:
cached = _face_cache.get(key)
if cached is not None:
_face_cache.move_to_end(key)
return cached
try:
entry = (_face_at(abs_path, size_px), size_px)
except Exception:
native = read_bdf_native_size(abs_path)
if not native or native == size_px:
raise
# A fresh Face: the first one already took a failed set_char_size.
entry = (_face_at(abs_path, native), native)
logger.debug(
"BDF font %s requested at %spx renders at its native %spx "
"(the file has no strike at the requested size)",
abs_path, size_px, native,
)
if key is not None:
with _face_cache_lock:
_face_cache[key] = entry
_face_cache.move_to_end(key)
while len(_face_cache) > _FACE_CACHE_MAX:
_face_cache.popitem(last=False)
return entry
def clear_face_cache() -> None:
"""Drop every cached face (tests; a font directory swapped wholesale)."""
with _face_cache_lock:
_face_cache.clear()
# --------------------------------------------------------------------------
# Drawing
# --------------------------------------------------------------------------
def _bitmap_bytes(bitmap: Any, nbytes: int) -> bytes:
"""The first ``nbytes`` of a glyph bitmap's buffer, zero-padded.
``bitmap.buffer`` builds a Python list one byte at a time; reading the
underlying FT_Bitmap directly is the same bytes without that cost.
"""
raw = getattr(bitmap, "_FT_Bitmap", None)
if raw is not None and raw.buffer:
return ctypes.string_at(raw.buffer, nbytes)
buf = bytes(bitmap.buffer[:nbytes])
if len(buf) < nbytes:
buf += bytes(nbytes - len(buf))
return buf
def _glyph_points(bitmap: Any, left: int, top: int,
clip_w: int, clip_h: int) -> list:
"""Every lit pixel of a glyph, clipped, as ``(x, y)`` pairs.
The reference definition of which pixels a glyph lights: the MSB-first
bit ``j`` of byte ``i * pitch + j // 8``. Used only where the fast path
below can't express exactly the same thing.
"""
buffer = bitmap.buffer
pitch = bitmap.pitch
points = []
for i in range(bitmap.rows):
for j in range(bitmap.width):
byte_index = i * pitch + (j // 8)
if byte_index < len(buffer) and buffer[byte_index] & (1 << (7 - (j % 8))):
px = left + j
py = top + i
if 0 <= px < clip_w and 0 <= py < clip_h:
points.append((px, py))
return points
def draw_bdf_text(draw: Any, text: str, x: int, y: int, face: Any,
color: Any = (255, 255, 255),
clip: Optional[Sequence[int]] = None) -> int:
"""Draw ``text`` in a ``freetype.Face`` with ``draw``; return the pen x.
``(x, y)`` is the top-left of the line, as for ``ImageDraw.text``: the
baseline is ``y`` plus the face's ascender. Each glyph's lit bits are set
to ``color`` exactly -- no blending, no anti-aliasing -- and pixels
outside ``[0, clip_w) x [0, clip_h)`` are skipped (``clip`` defaults to
the image size). The pen advances by each glyph's advance width.
Glyphs are drawn as 1-bit masks with ``ImageDraw.bitmap`` rather than a
point at a time, which is pixel-identical and far faster. A ``draw`` that
blends (``ImageDraw.Draw(rgb_image, "RGBA")``) is drawn point by point, so
a translucent colour still blends exactly as it always has.
Errors (a non-BDF ``face``, a bad colour) propagate after any glyphs
before the failing one are drawn; callers decide whether to log them.
"""
try:
ascender_px = face.size.ascender >> 6
except Exception:
ascender_px = 0
baseline_y = y + ascender_px
if clip is None:
clip_w, clip_h = draw.im.size
else:
clip_w, clip_h = int(clip[0]), int(clip[1])
blending = draw.mode != draw.im.mode
for char in text:
face.load_char(char)
glyph = face.glyph
bitmap = glyph.bitmap
rows, width, pitch = bitmap.rows, bitmap.width, bitmap.pitch
left = x + glyph.bitmap_left
top = baseline_y - glyph.bitmap_top
if rows > 0 and width > 0:
if blending or pitch <= 0:
points = _glyph_points(bitmap, left, top, clip_w, clip_h)
if points:
draw.point(points, fill=color)
else:
# The visible part of the glyph box, in glyph coordinates.
x0, y0 = max(0, -left), max(0, -top)
x1, y1 = min(width, clip_w - left), min(rows, clip_h - top)
if x0 < x1 and y0 < y1:
# Raw mode "1" with stride=pitch reads exactly the bits
# _glyph_points does, whatever the glyph's pixel mode.
mask = Image.frombytes(
"1", (width, rows), _bitmap_bytes(bitmap, rows * pitch),
"raw", "1", pitch)
if (x0, y0, x1, y1) != (0, 0, width, rows):
mask = mask.crop((x0, y0, x1, y1))
# An all-blank glyph draws nothing -- and, as before,
# never touches the colour.
if mask.getbbox() is not None:
draw.bitmap((left + x0, top + y0), mask, fill=color)
x += glyph.advance.x >> 6
return x
+10 -2
View File
@@ -39,6 +39,14 @@ from datetime import date, timedelta
from functools import partial
from typing import Any, Dict, List, Optional, Tuple
try:
from src.common.json_body import response_json
except ImportError:
# Plugins bundle copies of this module for older cores, which predate
# json_body; the stdlib parse is what those cores always used.
def response_json(response: Any) -> Any:
return response.json()
# Above this, ESPN returns a truncated list instead of an error. See module
# docstring: 500 is the largest value measured to return complete data.
ESPN_MAX_LIMIT = 500
@@ -194,7 +202,7 @@ def _fetch_one_chunk(
timeout=timeout,
)
response.raise_for_status()
return response.json()
return response_json(response)
except Exception as exc: # noqa: BLE001 - see docstring
if logger:
logger.warning("ESPN chunk %s failed, skipping it: %s", chunk, exc)
@@ -371,4 +379,4 @@ def fetch_espn_scoreboard(
if data is not None:
return data
response.raise_for_status()
return response.json()
return response_json(response)
+29
View File
@@ -0,0 +1,29 @@
"""Parse an HTTP response body as JSON, with orjson when it is installed.
``requests``' ``response.json()`` uses the stdlib parser. For the payloads the
sports plugins fetch -- a season schedule is tens of MB -- that runs ~1.7x
slower than orjson on a Pi 4 (3.1s against 1.8s for the 53MB MLB season), and
both hold the GIL for the whole parse, which freezes the display for as long.
Nothing else changes: the result is the same Python objects.
"""
from __future__ import annotations
from typing import Any
try:
import orjson
except ImportError: # optional dependency; see docs/SCROLL_PERFORMANCE.md
orjson = None
def response_json(response: Any) -> Any:
"""``response.json()``, parsed by orjson when available."""
body = getattr(response, "content", None)
if orjson is None or not isinstance(body, (bytes, bytearray)):
return response.json()
try:
return orjson.loads(body)
except orjson.JSONDecodeError:
# Let requests raise its usual error, with its usual message.
return response.json()
+49 -16
View File
@@ -339,6 +339,18 @@ def format_game_date(config: Optional[Dict[str, Any]], logger, date_text: str,
if not raw:
return ""
fmt = str(scroll_card_option(config, "date_format", "abbrev") or "abbrev")
return _format_date_as(fmt, raw, lambda: weekday_for(config, logger, game))
def _format_date_as(fmt: str, raw: str, weekday, months=MONTH_ABBR) -> str:
"""Render a stripped, non-empty "M/D" *raw* in style *fmt*.
The body both date formatters share. They differ in which setting names the
style and in which zone the weekday is taken from (see
``SportsCoreSharedMixin._format_game_date``), so those arrive as arguments:
*weekday* is a zero-argument callable, only called for the "weekday" style.
*months* lets the mixin keep reading its (overridable) ``_MONTH_ABBR``.
"""
if fmt == "numeric":
return raw
parts = raw.replace("-", "/").split("/")
@@ -347,14 +359,14 @@ def format_game_date(config: Optional[Dict[str, Any]], logger, date_text: str,
month, day = int(parts[0]), int(parts[1])
if not 1 <= month <= 12:
return raw
name = MONTH_ABBR[month - 1]
name = months[month - 1]
if fmt == "numeric_day_first":
return f"{day}/{month}"
if fmt == "day_first":
return f"{day} {name}"
if fmt == "weekday":
weekday = weekday_for(config, logger, game)
return f"{weekday} {name} {day}" if weekday else f"{name} {day}"
day_name = weekday()
return f"{day_name} {name} {day}" if day_name else f"{name} {day}"
return f"{name} {day}"
@@ -388,6 +400,29 @@ def format_game_time(config: Optional[Dict[str, Any]], time_text: str) -> str:
_SCHEMA_FONT_SIZE_CACHE: Dict[str, Dict[str, int]] = {}
def _read_schema_font_sizes(schema_path: str) -> Dict[str, int]:
"""``{element: font_size default}`` from a config_schema.json. Raises.
The parse both schema-default lookups share. Each keeps its own cache --
this function per schema path, ``SportsCoreSharedMixin._schema_font_size``
per class -- because the lifetimes differ: a class is rebuilt when the
display service reloads a plugin, a module-level path cache is not. One
cache would change when a reloaded plugin sees an edited schema.
"""
import json
with open(schema_path) as fh:
schema = json.load(fh)
props = (schema.get('properties', {})
.get('customization', {})
.get('properties', {}))
sizes: Dict[str, int] = {}
for key, spec in props.items():
size = spec.get('properties', {}).get('font_size', {}).get('default')
if size is not None:
sizes[key] = int(size)
return sizes
def schema_font_size(schema_path: str, element_key) -> Optional[int]:
"""The font_size this plugin's config_schema.json declares, or None.
@@ -399,18 +434,8 @@ def schema_font_size(schema_path: str, element_key) -> Optional[int]:
return None
cache = _SCHEMA_FONT_SIZE_CACHE.get(schema_path)
if cache is None:
cache = {}
try:
import json
with open(schema_path) as fh:
schema = json.load(fh)
props = (schema.get('properties', {})
.get('customization', {})
.get('properties', {}))
for key, spec in props.items():
size = spec.get('properties', {}).get('font_size', {}).get('default')
if size is not None:
cache[key] = int(size)
cache = _read_schema_font_sizes(schema_path)
except Exception as exc:
# See sports_shared._schema_font_size: an unreadable schema
# silently disables the pixel-grid snap for every element.
@@ -444,7 +469,7 @@ def resolve_font_size(schema_path: str, element_config, element_key,
return crisp_size(font_name, default_size, aliases, grid_table)
def unshare_element_fonts(logger, fonts):
def unshare_element_fonts(logger, fonts, element_for_font=None):
"""Give each colourable element its own face object.
The colour a draw gets is resolved from the face it was handed, and
@@ -459,13 +484,21 @@ def unshare_element_fonts(logger, fonts):
the ability to tell two elements apart does. Faces that cannot be
rebuilt (a BDF loaded through freetype.Face, anything without a usable
path) are left shared, and their draws stay white as before.
*element_for_font* names the font keys to consider, in order (the first
holder of a face keeps it); it defaults to this module's
:data:`ELEMENT_FOR_FONT`. ``SportsCoreSharedMixin`` passes its own map,
which names different keys -- see ``resolve_font_color`` for why the two
vocabularies are kept apart.
"""
try:
from src.common.font_layout import load_truetype as _load
except ImportError: # pragma: no cover
return fonts
if element_for_font is None:
element_for_font = ELEMENT_FOR_FONT
seen = {}
for key in ELEMENT_FOR_FONT:
for key in element_for_font:
font = fonts.get(key)
if font is None:
continue
+60 -105
View File
@@ -64,14 +64,27 @@ live here. Only ``_SCORE_PROBE_TEXT`` varies -- afl and basketball reach three d
a side and override it, the same two that override ``_SCORE_PROBE`` on
``SportsGameRendererMixin``.
DELIBERATELY NOT MERGED WITH sports_card
----------------------------------------
Fourteen of these have same-named twins in ``src/common/sports_card.py``, which
the scoreboards' ``game_renderer.py`` already uses. They are NOT wired together
here. Only five are provably equivalent by source comparison; the other nine
differ in ways inspection cannot settle, and a wrong guess silently changes what
every scoreboard draws. Merging them needs differential testing against both
implementations, and is left for its own change.
TWINS IN sports_card
--------------------
Many of these have same-named twins in ``src/common/sports_card.py``, which the
scoreboards' ``game_renderer.py`` uses. ``test/test_sports_twins.py`` calls
each pair with the same inputs (the plugins' fixture games in every payload
shape, plus edge cases) and splits them in two:
- Identical: ``_card_option``, ``_vs_text``, ``_format_game_time``,
``_coerce_rgb``, ``_crisp_size``, ``_unshare_element_fonts`` (given the same
element map) and the constant tables. These are now thin wrappers over the
``sports_card`` function; ``_format_game_date`` and ``_schema_font_size``
share its body/parser while keeping their own setting, zone and cache.
``_resolve_font_size`` agrees too but keeps its body, because it dispatches
through the overridable ``_schema_font_size``/``_crisp_size``.
- Different, and pinned as they are: ``_side_is_favorite`` /
``_favorite_result`` / ``_recent_score_color`` (flat keys and the host's
favourites only), ``_weekday_for`` (the plugin's resolved zone, not
``config["timezone"]``), ``_font_color`` / ``_ELEMENT_FOR_FONT`` (another
element vocabulary), ``_element_color`` (passes ``SKIN_MODE``). Each shows
up in one display mode only, so which side is right is a product decision;
the test that pins it names the difference.
"""
from __future__ import annotations
@@ -87,6 +100,7 @@ import pytz
from src.common.espn_dates import fetch_espn_scoreboard
import requests
from PIL import Image, ImageDraw, ImageFont
from src.common import sports_card as _card
from src.common.font_layout import load_truetype
logger = logging.getLogger(__name__)
@@ -171,19 +185,17 @@ class SportsCoreSharedMixin:
_ELEMENT_FOR_FONT: ClassVar[Dict[str, str]] = {
"score": "score_text", "time": "period_text", "team": "team_text",
"detail": "detail_text", "status": "status_text"}
# The tables below are sports_card's (and font_layout's) values. The dicts
# are copies, so a caller that mutates one module's table -- or a subclass
# that replaces it -- does not reach into the other.
#: Default tint for a favourite team's finished game.
FAVORITE_RESULT_COLOR_DEFAULTS: ClassVar[Dict[str, Tuple[int, int, int]]] = {
"win": (0, 255, 0), "loss": (255, 0, 0), "tie": (255, 200, 0)}
_MONTH_ABBR: ClassVar[Tuple[str, ...]] = (
"Jan", "Feb", "Mar", "Apr", "May", "Jun",
"Jul", "Aug", "Sep", "Oct", "Nov", "Dec")
_WEEKDAY_ABBR: ClassVar[Tuple[str, ...]] = (
"Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun")
FAVORITE_RESULT_COLOR_DEFAULTS: ClassVar[Dict[str, Tuple[int, int, int]]] = dict(
_card.FAVORITE_RESULT_COLOR_DEFAULTS)
_MONTH_ABBR: ClassVar[Tuple[str, ...]] = _card.MONTH_ABBR
_WEEKDAY_ABBR: ClassVar[Tuple[str, ...]] = _card.WEEKDAY_ABBR
#: Bitmap fonts snap to their native pixel grid.
_FONT_PIXEL_GRID: ClassVar[Dict[str, int]] = {
"PressStart2P-Regular.ttf": 8, "4x6-font.ttf": 7}
_FONT_NAME_ALIASES: ClassVar[Dict[str, str]] = {
"press_start": "PressStart2P-Regular.ttf", "four_by_six": "4x6-font.ttf"}
_FONT_PIXEL_GRID: ClassVar[Dict[str, int]] = dict(_card.FONT_PIXEL_GRID)
_FONT_NAME_ALIASES: ClassVar[Dict[str, str]] = dict(_card.FONT_NAME_ALIASES)
#: Accepted values for the other-games quality filter.
_QUALITY_CHOICES: ClassVar[frozenset] = frozenset({"any", "ranked"})
#: How long to stay quiet between ranking-coverage warnings.
@@ -213,13 +225,11 @@ class SportsCoreSharedMixin:
"""Snap *desired* to the nearest size *font_file* renders crisply at.
A face with no known grid is returned unchanged, so a user-supplied
font is never second-guessed.
font is never second-guessed. The class's own tables are passed, so a
host that declares extra faces keeps them.
"""
font_file = cls._FONT_NAME_ALIASES.get(font_file, font_file)
grid = cls._FONT_PIXEL_GRID.get(font_file)
if not grid or not desired or desired <= 0:
return desired
return max(grid, int(round(float(desired) / grid)) * grid)
return _card.crisp_size(font_file, desired,
cls._FONT_NAME_ALIASES, cls._FONT_PIXEL_GRID)
#: Absolute path of this plugin's directory, declared by the plugin
#: itself. The mixin cannot work it out -- see _plugin_dir.
@@ -281,23 +291,19 @@ class SportsCoreSharedMixin:
"""The font_size this plugin's config_schema.json declares, or None."""
if not element_key:
return None
# Cached per class, not in sports_card's per-path cache: the display
# service rebuilds the class when it reloads a plugin, and that is
# what makes an edited schema take effect. Both caches parse through
# sports_card._read_schema_font_sizes.
cache = getattr(self.__class__, '_SCHEMA_FONT_SIZES', None)
if cache is None:
cache = {}
try:
import json
directory = self._plugin_dir()
if directory is None:
raise FileNotFoundError("no config_schema.json on the MRO")
with open(os.path.join(directory, 'config_schema.json')) as fh:
schema = json.load(fh)
props = (schema.get('properties', {})
.get('customization', {})
.get('properties', {}))
for key, spec in props.items():
size = spec.get('properties', {}).get('font_size', {}).get('default')
if size is not None:
cache[key] = int(size)
cache = _card._read_schema_font_sizes(
os.path.join(directory, 'config_schema.json'))
except Exception as exc:
# Say so. An unreadable schema is not cosmetic: every element's
# configured size then stops matching "the schema default", is
@@ -339,10 +345,7 @@ class SportsCoreSharedMixin:
def _card_option(self, key: str, default: Any = None) -> Any:
"""Read one key from the scroll_card config block."""
block = (self.config or {}).get("scroll_card")
if isinstance(block, dict) and block.get(key) is not None:
return block.get(key)
return default
return _card.scroll_card_option(self.config, key, default)
def _switch_upcoming_center(self) -> str:
"""Middle of the full-screen upcoming scorebug: 'vs', 'date_time' or 'none'."""
@@ -354,7 +357,7 @@ class SportsCoreSharedMixin:
def _vs_text(self) -> str:
"""Separator drawn between the teams -- "VS", "@", "at", anything."""
return str(self._card_option("vs_text", "VS"))
return _card.vs_text(self.config)
def _switch_date_format(self) -> str:
"""Date style for the full-screen scorebug.
@@ -374,28 +377,19 @@ class SportsCoreSharedMixin:
return fmt
def _format_game_date(self, date_text: str, game: Optional[Dict] = None) -> str:
"""Format an upcoming date per scroll_card.switch_date_format."""
"""Format an upcoming date per scroll_card.switch_date_format.
The formatting is sports_card's. What differs from the card's
``format_game_date`` is passed in: the setting (``switch_date_format``,
see :meth:`_switch_date_format`) and the weekday, which comes from
:meth:`_weekday_for` and so from this plugin's resolved timezone.
"""
raw = str(date_text or "").strip()
if not raw:
return raw
fmt = self._switch_date_format()
if fmt == "numeric":
return raw
parts = raw.replace("-", "/").split("/")
if not (len(parts) >= 2 and parts[0].strip().isdigit() and parts[1].strip().isdigit()):
return raw
month, day = int(parts[0]), int(parts[1])
if not 1 <= month <= 12:
return raw
name = self._MONTH_ABBR[month - 1]
if fmt == "numeric_day_first":
return f"{day}/{month}"
if fmt == "day_first":
return f"{day} {name}"
if fmt == "weekday":
weekday = self._weekday_for(game)
return f"{weekday} {name} {day}" if weekday else f"{name} {day}"
return f"{name} {day}"
return _card._format_date_as(self._switch_date_format(), raw,
lambda: self._weekday_for(game),
self._MONTH_ABBR)
def _weekday_for(self, game: Optional[Dict]) -> str:
"""Weekday abbreviation from the game's start time, or ''."""
@@ -413,22 +407,7 @@ class SportsCoreSharedMixin:
def _format_game_time(self, time_text: str) -> str:
"""Return the time as-is (12h) or converted to 24h."""
raw = str(time_text or "").strip()
if not raw or str(self._card_option("time_format", "12h")) != "24h":
return raw
cleaned = raw.upper().replace(" ", "")
meridiem = "AM" if cleaned.endswith("AM") else "PM" if cleaned.endswith("PM") else ""
if not meridiem:
return raw
try:
hh, _, mm = cleaned[:-2].partition(":")
hour, minute = int(hh), int(mm or 0)
except ValueError:
return raw
if not (0 <= hour <= 12 and 0 <= minute <= 59):
return raw
hour = hour % 12 + (12 if meridiem == "PM" else 0)
return f"{hour:02d}:{minute:02d}"
return _card.format_game_time(self.config, time_text)
def _scorebug_font(self, draw, text: str, width: int):
"""The face this scorebug draws its date and time in.
@@ -560,15 +539,7 @@ class SportsCoreSharedMixin:
@staticmethod
def _coerce_rgb(value, fallback):
"""Turn a configured [R, G, B] list into a clamped (r, g, b) tuple."""
# Checked before unpacking: a 3-character string ("123") would otherwise
# iterate into three digits and yield a colour rather than the fallback.
if not isinstance(value, (list, tuple)) or len(value) != 3:
return fallback
try:
r, g, b = (max(0, min(255, int(channel))) for channel in value)
except (TypeError, ValueError):
return fallback
return (r, g, b)
return _card.coerce_rgb(value, fallback)
@staticmethod
def _side_is_favorite(game: Dict, side: str, favorites: set) -> bool:
@@ -849,28 +820,12 @@ class SportsCoreSharedMixin:
the ability to tell two elements apart does. Faces that cannot be
rebuilt (a BDF loaded through freetype.Face, anything without a usable
path) are left shared, and their draws stay white as before.
The body is sports_card's; this class's own element map is passed, so
the keys considered are the ones this class colours by.
"""
try:
from src.common.font_layout import load_truetype as _load
except ImportError: # pragma: no cover
return fonts
seen = {}
for key in self._ELEMENT_FOR_FONT:
font = fonts.get(key)
if font is None:
continue
if id(font) not in seen:
seen[id(font)] = key
continue
path, size = getattr(font, "path", None), getattr(font, "size", None)
if not path or not size:
continue
try:
fonts[key] = _load(path, size)
except (OSError, ValueError, TypeError):
self.logger.debug(
"Could not un-share the %s face; it keeps the default colour", key)
return fonts
return _card.unshare_element_fonts(self.logger, fonts,
self._ELEMENT_FOR_FONT)
def _font_color(self, font, default: Tuple[int, int, int] = (255, 255, 255)):
"""Colour for whichever element owns this face.
+16 -48
View File
@@ -34,6 +34,7 @@ else:
from contextlib import contextmanager
from pathlib import Path
from PIL import Image, ImageDraw, ImageFont
from src.common.bdf_font import draw_bdf_text, load_bdf_face
from src.common.font_layout import crisp_size, load_truetype, resolve_asset_path
from src.display_geometry import (
DEFAULT_CHAIN_LENGTH, DEFAULT_COLS, DEFAULT_PARALLEL, DEFAULT_ROWS,
@@ -896,43 +897,16 @@ class DisplayManager:
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."""
"""Draw text in a BDF ``freetype.Face`` with (x, y) as its top-left.
Delegates to :func:`src.common.bdf_font.draw_bdf_text`, which the
plugin test harness uses too, so previews and golden images show the
pixels the panel does. Clipped to the logical display size.
"""
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
draw_bdf_text(self.draw, text, x, y, face, color,
clip=(self.width, self.height))
except Exception as e:
logger.error(f"Error drawing BDF text: {e}", exc_info=True)
@@ -981,19 +955,13 @@ class DisplayManager:
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)
# load_bdf_face sets the size: a Face built without
# set_char_size reports face.size.height 0, and every caller
# measuring the 5x7 face with get_font_height() got 0 and
# stacked rows on top of one another. 5x7.bdf is a fixed
# strike, so FreeType renders 7px whatever is asked for --
# the size sets the metrics, not the raster.
face, _ = load_bdf_face(self.calendar_font_path, _CALENDAR_FONT_PX)
logger.info(f"5x7 calendar font loaded successfully from {self.calendar_font_path}")
logger.info(f"Calendar font size: {face.size.height >> 6} pixels")
+12 -47
View File
@@ -53,13 +53,9 @@ from dataclasses import dataclass
from typing import Any, Dict, Optional, Tuple, Union
from PIL import ImageFont
from src.common.bdf_font import load_bdf_face, read_bdf_native_size
from src.common.font_layout import load_truetype
try:
import freetype
except ImportError: # pragma: no cover - freetype ships with the core
freetype = None
logger = logging.getLogger(__name__)
# Core install root (the directory that contains src/ and assets/fonts/),
@@ -164,56 +160,25 @@ def native_bdf_size(font_name: str) -> Optional[int]:
def _read_bdf_native_size(path: str) -> Optional[int]:
"""A BDF file's own pixel size, delegated to FontManager.
"""A BDF file's own pixel size (the web UI's fonts API imports this name).
Deliberately not reimplemented: FontManager's reader prefers PIXEL_SIZE
over the SIZE line's point-size (they differ on the several bundled
fonts defined at 75dpi) and stops at the first STARTCHAR. Core always
ships it; the guard is for the plugin test harnesses that stub the
module out.
See :func:`src.common.bdf_font.read_bdf_native_size`: it prefers
PIXEL_SIZE over the SIZE line's point-size, which differ on the several
bundled fonts defined at 75dpi.
"""
try:
from src.font_manager import FontManager
return FontManager._read_bdf_native_size(path)
except Exception: # pragma: no cover - defensive
return None
return read_bdf_native_size(path)
def _load_bdf(path: str, size: int) -> Tuple[Any, int]:
"""A ``freetype.Face`` for a BDF file at the closest size it can do.
BDF fonts are fixed-size bitmap strikes, not scalable outlines:
FreeType accepts only the exact pixel size baked into the file and
raises for anything else. 32 of the 35 shipped fonts are BDF, so a
size the user picked in the web UI usually is not a valid strike.
Retrying at the file's native size is the behaviour SportsCore already
has (``_load_custom_font_from_element_config``). Without it this
function fell through to the generic except below and returned
*PressStart2P* — so choosing 5x7.bdf at size 10 silently rendered a
completely different typeface rather than 5x7 at 7px.
BDF fonts are fixed-size bitmap strikes: FreeType accepts only the pixel
size baked into the file, and 32 of the 35 shipped fonts are BDF, so a
size the user picked in the web UI usually is not a valid strike. The
shared loader retries at the native size; without that, 5x7.bdf at size
10 used to fall through to *PressStart2P*, a different typeface.
"""
if freetype is None:
raise RuntimeError("freetype not available for BDF fonts")
def _face_at(px: int) -> Any:
face = freetype.Face(path)
# Character size in 1/64th points at 72dpi == pixel size.
face.set_char_size(px * 64, px * 64, 72, 72)
return face
try:
return _face_at(size), size
except Exception:
native = _read_bdf_native_size(path)
if not native or native == size:
raise
# A fresh Face: the first one already took a failed set_char_size.
face = _face_at(native)
logger.debug("BDF font %s loaded at its native size %s "
"(requested %s is not a strike in this file)",
path, native, size)
return face, native
return load_bdf_face(path, size)
def load_font(font_name: str, size: int) -> Any:
+11 -46
View File
@@ -38,6 +38,7 @@ import time
from collections import OrderedDict
from pathlib import Path
from PIL import ImageFont
from src.common.bdf_font import load_bdf_face, read_bdf_native_size
from src.common.font_layout import load_truetype, resolve_asset_path
from typing import Dict, Tuple, Optional, Union, Any, List
from src.deprecation import deprecated
@@ -517,29 +518,14 @@ class FontManager:
return font
def _load_bdf_font(self, font_path: str, size_px: int) -> freetype.Face:
"""Load a BDF font using FreeType."""
"""Load a BDF font through the shared loader.
A size the file has no strike for comes back at the native strike
rather than failing over to PIL's default font, a different typeface
(see :func:`src.common.bdf_font.load_bdf_face`).
"""
try:
native_size = self._read_bdf_native_size(font_path)
if native_size is not None and native_size != size_px:
# BDF is a fixed-strike bitmap format: FreeType renders the
# native size no matter what set_char_size asks for.
logger.debug(
"BDF font %s requested at %spx but renders at its native "
"%spx", font_path, size_px, native_size
)
face = freetype.Face(font_path)
try:
# Character size in 1/64th points at 72dpi == pixel size.
face.set_char_size(size_px * 64, size_px * 64, 72, 72)
except freetype.FT_Exception:
# FreeType rejects any size but the strike's own, and get_font
# used to answer that with PIL's default font -- a different
# typeface. Use the native strike, as element_style does.
if native_size is None or native_size == size_px:
raise
face = freetype.Face(font_path)
face.set_char_size(native_size * 64, native_size * 64, 72, 72)
return face
return load_bdf_face(font_path, size_px)[0]
except Exception as e:
logger.error(f"Error loading BDF font {font_path}: {e}")
raise
@@ -554,30 +540,9 @@ class FontManager:
@staticmethod
def _read_bdf_native_size(bdf_path: str) -> Optional[int]:
"""Read a BDF file's own header to find its one true pixel size.
Prefers the PIXEL_SIZE property, which states the real pixel height
directly; falls back to the SIZE line's point-size only if PIXEL_SIZE
is absent, since point-size only equals pixel height at exactly
100dpi — several bundled fonts (e.g. 6x13.bdf, 5x8.bdf) are defined
at 75dpi, where the two values genuinely differ."""
size_line_value = None
try:
with open(bdf_path, "r", encoding="ascii", errors="ignore") as f:
for line in f:
if line.startswith("PIXEL_SIZE"):
parts = line.split()
if len(parts) >= 2:
return int(float(parts[1]))
elif line.startswith("SIZE") and size_line_value is None:
# Format: "SIZE <point_size> <xres> <yres>"
parts = line.split()
if len(parts) >= 2:
size_line_value = int(float(parts[1]))
elif line.startswith("STARTCHAR"):
break
except (OSError, ValueError):
return None
return size_line_value
"""A BDF file's one true pixel size; see
:func:`src.common.bdf_font.read_bdf_native_size`."""
return read_bdf_native_size(bdf_path)
def _get_fallback_font(self) -> ImageFont.ImageFont:
"""Get a fallback font when loading fails."""
+345
View File
@@ -0,0 +1,345 @@
"""
One answer to "which directory holds plugin X?".
Five places used to answer it, each with its own rules and each re-parsing
every manifest per lookup: ``PluginManager`` discovery and
``get_plugin_directory``, ``PluginLoader.find_plugin_directory``,
``PluginStoreManager._find_plugin_path`` / ``list_installed_plugins`` and
``state_reconciliation.disk_plugin_ids``. The rules now live here once; what
still legitimately differs between callers (which directories to search,
whether a ``ledmatrix-`` prefix or a case difference counts as a match) is a
keyword argument at the call site, so a difference is always a visible choice
rather than an accident of which copy you read.
The rules
---------
* A directory is a *candidate* when it is a directory (a symlink to one counts:
dev plugins are symlinked in) and its name is neither hidden (leading ``.``)
nor carries :data:`BACKUP_MARKER`. store_manager renames a plugin aside with
that marker during install/rollback; the aside still holds a manifest, so
treating it as a plugin would resurrect a ghost.
* A plugin's id is its manifest ``id``. The directory name is only a fallback,
for callers that must still see a plugin whose manifest is missing an id.
* Resolving an id within one directory: a directory whose manifest declares
the id wins; among several, the one named exactly for the id, then
``ledmatrix-<id>``, then by name. Only when no manifest claims the id do
directory names count: ``<id>``, then ``ledmatrix-<id>`` (``prefix=True``),
then either of those ignoring case (``case_insensitive=True``). The name
fallback still returns a directory whose manifest is unreadable -- that is
how a broken plugin gets uninstalled or reinstalled. ``by_manifest=False``
(``PluginManager.get_plugin_directory``, whose discovery map already holds
the manifest answer) skips straight to the names.
* Several directories are searched one at a time, in the order given: the
first directory that resolves the id at all wins, by manifest or by name.
* The id must be one plain path segment (``safe_path_component``); anything
else resolves to nothing rather than being joined or truncated.
* Returned paths are ``search_dir / name`` and are not resolved, so a
symlinked dev plugin keeps the path that lies inside the search directory.
Manifests are read at most once per :class:`PluginDirectoryIndex`; one index is
one scan.
"""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Set, Union
from src.common.path_safety import safe_path_component
__all__ = [
'BACKUP_MARKER',
'PLUGIN_DIR_PREFIX',
'ManifestStatus',
'PluginDirEntry',
'PluginDirectoryIndex',
'is_ignored_dir_name',
'resolve_plugin_dir',
'store_search_dirs',
]
#: Substring store_manager embeds in a plugin directory it has set aside
#: (``<id>.standalone-backup-preinstall`` / ``-migrating``). Existing debris on
#: devices carries exactly this text, so it must never change.
BACKUP_MARKER = '.standalone-backup-'
#: Legacy repository naming (``ledmatrix-<id>``); some installs still use it
#: as the directory name.
PLUGIN_DIR_PREFIX = 'ledmatrix-'
PathLike = Union[str, Path]
class ManifestStatus:
"""What reading ``manifest.json`` in a candidate directory produced."""
OK = 'ok' # a JSON object with a non-empty "id"
MISSING = 'missing' # no manifest.json
UNREADABLE = 'unreadable' # I/O error or invalid JSON
NOT_OBJECT = 'not_object' # valid JSON, but not an object
NO_ID = 'no_id' # an object without a usable "id"
def is_ignored_dir_name(name: str) -> bool:
"""True for names that are never a plugin: hidden, or set aside mid-install."""
return name.startswith('.') or BACKUP_MARKER in name
@dataclass
class PluginDirEntry:
"""One candidate directory and its manifest, read once."""
path: Path
status: str
manifest: Optional[Any] = None
error: Optional[BaseException] = None
@property
def name(self) -> str:
return self.path.name
@property
def manifest_id(self) -> Optional[str]:
"""The manifest's ``id`` when the manifest is usable, else None."""
if self.status != ManifestStatus.OK:
return None
return self.manifest['id']
@property
def manifest_parses(self) -> bool:
"""The manifest exists and is valid JSON (of any shape)."""
return self.status in (ManifestStatus.OK, ManifestStatus.NOT_OBJECT,
ManifestStatus.NO_ID)
@property
def installed_id(self) -> str:
"""The manifest id, falling back to the directory name."""
return self.manifest_id or self.name
def _read_entry(path: Path) -> PluginDirEntry:
manifest_path = path / 'manifest.json'
if not manifest_path.is_file():
return PluginDirEntry(path, ManifestStatus.MISSING)
try:
with open(manifest_path, 'r', encoding='utf-8') as handle:
manifest = json.load(handle)
except (OSError, ValueError) as exc: # ValueError covers JSON + decode errors
return PluginDirEntry(path, ManifestStatus.UNREADABLE, error=exc)
if not isinstance(manifest, dict):
return PluginDirEntry(path, ManifestStatus.NOT_OBJECT, manifest)
plugin_id = manifest.get('id')
if not plugin_id or not isinstance(plugin_id, str):
return PluginDirEntry(path, ManifestStatus.NO_ID, manifest)
return PluginDirEntry(path, ManifestStatus.OK, manifest)
def _preference(plugin_id: str, name: str) -> tuple:
"""Sort key among directories that all claim ``plugin_id``."""
if name == plugin_id:
rank = 0
elif name == PLUGIN_DIR_PREFIX + plugin_id:
rank = 1
else:
rank = 2
return (rank, name)
@dataclass
class PluginDirectoryIndex:
"""Every candidate directory directly under ``root``, manifests read once.
Build one with :meth:`scan`. It is a snapshot: a directory added or
removed afterwards is not seen until the next scan.
"""
root: Path
entries: List[PluginDirEntry] = field(default_factory=list)
#: Set when ``root`` exists but could not be listed.
error: Optional[BaseException] = None
_plugins: Optional[Dict[str, PluginDirEntry]] = field(
default=None, init=False, repr=False, compare=False)
@classmethod
def scan(cls, root: PathLike) -> 'PluginDirectoryIndex':
root = Path(root)
index = cls(root)
try:
children = sorted(root.iterdir(), key=lambda p: p.name)
except FileNotFoundError:
return index
except OSError as exc:
index.error = exc
return index
for child in children:
if is_ignored_dir_name(child.name):
continue
try:
if not child.is_dir():
continue
except OSError:
continue
index.entries.append(_read_entry(child))
return index
# -- listing ----------------------------------------------------------
def plugins(self) -> Dict[str, PluginDirEntry]:
"""Manifest id -> entry, one entry per id.
When several directories declare the same id, the one named for it
wins, then ``ledmatrix-<id>``, then the first by name; see
:meth:`duplicates` for the losers.
"""
if self._plugins is not None:
return self._plugins
chosen: Dict[str, PluginDirEntry] = {}
for entry in self.entries:
plugin_id = entry.manifest_id
if plugin_id is None:
continue
current = chosen.get(plugin_id)
if current is None or (_preference(plugin_id, entry.name)
< _preference(plugin_id, current.name)):
chosen[plugin_id] = entry
self._plugins = chosen
return chosen
def duplicates(self) -> Dict[str, List[PluginDirEntry]]:
"""Ids declared by more than one directory -> every such entry."""
seen: Dict[str, List[PluginDirEntry]] = {}
for entry in self.entries:
if entry.manifest_id is not None:
seen.setdefault(entry.manifest_id, []).append(entry)
return {k: v for k, v in seen.items() if len(v) > 1}
def installed_ids(self, *, require_parseable_manifest: bool) -> Set[str]:
"""Ids of everything that counts as installed.
A directory counts when it has a manifest.json -- which must also be
valid JSON when ``require_parseable_manifest``. Its id is the manifest
id, or the directory name when the manifest does not carry one.
"""
ids: Set[str] = set()
for entry in self.entries:
if entry.status == ManifestStatus.MISSING:
continue
if require_parseable_manifest and not entry.manifest_parses:
continue
ids.add(entry.installed_id)
return ids
def entry_for_installed_id(self, plugin_id: str) -> Optional[PluginDirEntry]:
"""The entry :meth:`installed_ids` reported as ``plugin_id``."""
entry = self.plugins().get(plugin_id)
if entry is not None:
return entry
for entry in self.entries:
if entry.manifest_id is None and entry.name == plugin_id:
return entry
return None
# -- lookup -----------------------------------------------------------
def find(self, plugin_id: str, *, prefix: bool, case_insensitive: bool,
by_manifest: bool = True) -> Optional[Path]:
"""Resolve ``plugin_id`` within this directory (rules in the module doc)."""
plugin_id = _lookup_id(plugin_id)
if plugin_id is None:
return None
if by_manifest:
entry = self.plugins().get(plugin_id)
if entry is not None:
return entry.path
names = _candidate_names(plugin_id, prefix)
by_name = {e.name: e for e in self.entries}
for name in names:
if name in by_name:
return by_name[name].path
if case_insensitive:
for low in (n.lower() for n in names):
for entry in self.entries:
if entry.name.lower() == low:
return entry.path
return None
def _lookup_id(plugin_id: Any) -> Optional[str]:
"""``plugin_id`` if it can name a plugin directory at all, else None."""
plugin_id = safe_path_component(plugin_id)
if plugin_id is None or is_ignored_dir_name(plugin_id):
return None
return plugin_id
def _candidate_names(plugin_id: str, prefix: bool) -> List[str]:
names = [plugin_id]
if prefix:
names.append(PLUGIN_DIR_PREFIX + plugin_id)
return names
def _is_dir(path: Path) -> bool:
try:
return path.is_dir()
except OSError:
return False
def resolve_plugin_dir(plugin_id: Any, search_dirs: Iterable[PathLike], *,
prefix: bool, case_insensitive: bool = False,
by_manifest: bool = True) -> Optional[Path]:
"""The directory holding ``plugin_id``, searching ``search_dirs`` in order.
Each search directory is scanned once and each manifest in it read once.
``by_manifest=False`` skips the manifest pass and matches directory names
only, which reads no manifests at all.
Names are compared against the directory listing, never by probing
``search_dir / name``: on a case-insensitive filesystem that probe says
``Demo`` exists when the directory is ``demo``, which made the answer
depend on the platform.
"""
plugin_id = _lookup_id(plugin_id)
if plugin_id is None:
return None
for search_dir in search_dirs:
search_dir = Path(search_dir)
if by_manifest or case_insensitive:
found = PluginDirectoryIndex.scan(search_dir).find(
plugin_id, prefix=prefix, case_insensitive=case_insensitive,
by_manifest=by_manifest)
else:
found = _find_by_name(search_dir, _candidate_names(plugin_id, prefix))
if found is not None:
return found
return None
def _find_by_name(search_dir: Path, names: List[str]) -> Optional[Path]:
try:
present = {child.name for child in search_dir.iterdir()}
except OSError:
return None
for name in names:
if name in present and _is_dir(search_dir / name):
return search_dir / name
return None
def store_search_dirs(plugins_dir: PathLike) -> List[Path]:
"""Directories the plugin store searches: the configured one, then a
sibling ``plugins/`` (the legacy/dev location) when that is a different
directory. Discovery deliberately does NOT use this -- it scans only the
configured directory (see CLAUDE.md, test_discovery_path_contract.py)."""
plugins_dir = Path(plugins_dir)
dirs = [plugins_dir]
try:
base = plugins_dir if plugins_dir.is_absolute() else plugins_dir.resolve()
sibling = base.parent / 'plugins'
if sibling != base:
dirs.append(sibling)
except (OSError, ValueError):
pass
return dirs
+19 -68
View File
@@ -8,7 +8,6 @@ Extracted from PluginManager to improve separation of concerns.
import importlib
import importlib.metadata
import importlib.util
import json
import os
import sys
import subprocess
@@ -21,6 +20,7 @@ from packaging.requirements import InvalidRequirement, Requirement
from src.exceptions import PluginError
from src.logging_config import get_logger
from src.plugin_system.plugin_dirs import resolve_plugin_dir
def requirements_has_real_deps(requirements_file: str) -> bool:
@@ -214,85 +214,36 @@ class PluginLoader:
) -> Optional[Path]:
"""
Find the plugin directory for a given plugin ID.
Tries multiple strategies:
1. Use plugin_directories mapping if available
2. Direct path matching
3. Case-insensitive directory matching
4. Manifest-based search
1. The discovery mapping, when it has the id and the path exists.
2. ``plugins_dir`` only, by the shared rules in
``src/plugin_system/plugin_dirs.py``: a directory whose manifest
declares the id wins; otherwise ``<id>`` or ``ledmatrix-<id>``,
matched case-insensitively. Backup and hidden directories are
never matched.
Args:
plugin_id: Plugin identifier
plugins_dir: Base plugins directory
plugin_directories: Optional mapping of plugin_id to directory
Returns:
Path to plugin directory or None if not found
"""
# Sanitize plugin_id — os.path.basename is a CodeQL-recognized path sanitizer
plugin_id = os.path.basename(plugin_id or '')
if not plugin_id:
return None
Returns:
Path to plugin directory or None if not found. An id that is not
one plain path segment finds nothing.
"""
# Strategy 1: Use mapping from discovery
if plugin_directories and plugin_id in plugin_directories:
plugin_dir = plugin_directories[plugin_id]
if plugin_dir.exists():
self.logger.debug("Using plugin directory from discovery mapping: %s", plugin_dir)
return plugin_dir
# Strategy 2: Direct paths — resolve and validate they stay within plugins_dir
plugins_dir_resolved = plugins_dir.resolve()
for _candidate_name in (plugin_id, f"ledmatrix-{plugin_id}"):
_candidate = (plugins_dir_resolved / _candidate_name).resolve()
try:
_candidate.relative_to(plugins_dir_resolved)
except ValueError:
continue
if _candidate.exists():
return _candidate
# Strategy 3: Case-insensitive search
normalized_id = plugin_id.lower()
for item in plugins_dir.iterdir():
if not item.is_dir():
continue
item_name = item.name
if item_name.lower() == normalized_id:
return item
if item_name.lower() == f"ledmatrix-{plugin_id}".lower():
return item
# Strategy 4: Manifest-based search
self.logger.debug("Directory name search failed for %s, searching by manifest...", plugin_id)
for item in plugins_dir.iterdir():
if not item.is_dir():
continue
# Skip if already checked
if item.name.lower() == normalized_id or item.name.lower() == f"ledmatrix-{plugin_id}".lower():
continue
manifest_path = item / "manifest.json"
if manifest_path.exists():
try:
with open(manifest_path, 'r', encoding='utf-8') as f:
item_manifest = json.load(f)
item_manifest_id = item_manifest.get('id')
if item_manifest_id == plugin_id:
self.logger.info(
"Found plugin %s in directory %s (manifest ID matches)",
plugin_id,
item.name
)
return item
except (json.JSONDecodeError, Exception) as e:
self.logger.debug("Skipping %s due to manifest error: %s", item.name, e)
continue
return None
plugin_dir = resolve_plugin_dir(
plugin_id, [plugins_dir], prefix=True, case_insensitive=True)
if plugin_dir is not None and plugin_dir.name != plugin_id:
self.logger.debug("Found plugin %s in directory %s",
plugin_id, plugin_dir.name)
return plugin_dir
def install_dependencies(
self,
+78 -81
View File
@@ -25,7 +25,9 @@ from src.plugin_system.plugin_state import PluginStateManager, PluginState
from src.plugin_system.schema_manager import (
CORE_VEGAS_TUNING_KEYS, SchemaManager, normalize_legacy_booleans,
)
from src.common.path_safety import safe_path_component
from src.plugin_system.plugin_dirs import (
ManifestStatus, PluginDirectoryIndex, resolve_plugin_dir,
)
from src.deprecation import deprecated
from src.common.permission_utils import (
ensure_directory_permissions,
@@ -175,89 +177,88 @@ class PluginManager:
self.logger.error("Could not create plugins directory %s: %s", self.plugins_dir, e, exc_info=True)
raise PluginError(f"Could not create plugins directory: {self.plugins_dir}", context={'error': str(e)}) from e
def _report_skip_once(self, key: str, message: str, *args: Any) -> None:
"""Warn about a skipped directory once per process, not per scan.
Discovery runs on every web UI page load and every config reconcile,
so warning unconditionally would put a line in the journal each time
someone opened a page -- the same log-volume problem this is meant to
help diagnose.
"""
reported = self.__dict__.setdefault('_skip_reported', set())
if key in reported:
return
reported.add(key)
self.logger.warning(message, *args)
def _scan_directory_for_plugins(self, directory: Path) -> List[str]:
"""
Scan a directory for plugins.
Which directories count and how an id maps to one is decided by
:class:`PluginDirectoryIndex` (``src/plugin_system/plugin_dirs.py``),
shared with the loader, the store and reconciliation. Only
``directory`` is scanned: discovery has no fallback to ``plugins/``.
Directories set aside mid-install (``BACKUP_MARKER`` in the name) are
skipped so they don't overwrite live entries.
Args:
directory: Directory to scan
Returns:
List of plugin IDs found
"""
plugin_ids = []
if not directory.exists():
return plugin_ids
return []
# Build new state locally before acquiring lock
new_manifests: Dict[str, Dict[str, Any]] = {}
new_directories: Dict[str, Path] = {}
try:
for item in directory.iterdir():
if not item.is_dir():
continue
# Skip backup directories so they don't overwrite live entries
if '.standalone-backup-' in item.name:
continue
manifest_path = item / "manifest.json"
if not manifest_path.exists():
# Once per directory per process. Discovery runs on every
# web UI page load and every config reconcile, so warning
# unconditionally would put a line in the journal each
# time someone opened a page -- the same log-volume
# problem this is meant to help diagnose.
# A directory here that carries no manifest is not a
# plugin. Said once, because the alternative is a plugin
# that is enabled in config, enabled in plugin state,
# present on disk, and simply absent from the running
# process with nothing anywhere to say why. Working that
# out afterwards means reading cache-file mtimes.
if item.name not in self._skip_reported:
self._skip_reported.add(item.name)
self.logger.warning(
"Skipping %s: no manifest.json, so it cannot be "
"loaded as a plugin", item.name)
continue
try:
with open(manifest_path, 'r', encoding='utf-8') as f:
manifest = json.load(f)
except (json.JSONDecodeError, PermissionError, OSError) as e:
self.logger.warning("Error reading manifest from %s: %s", manifest_path, e, exc_info=True)
continue
index = PluginDirectoryIndex.scan(directory)
if index.error is not None:
self.logger.error("Error scanning directory %s: %s", directory,
index.error, exc_info=index.error)
for entry in index.entries:
if entry.status == ManifestStatus.MISSING:
# A directory here that carries no manifest is not a plugin.
# Said once, because the alternative is a plugin that is
# enabled in config, enabled in plugin state, present on disk,
# and simply absent from the running process with nothing
# anywhere to say why. Working that out afterwards means
# reading cache-file mtimes.
self._report_skip_once(
entry.name, "Skipping %s: no manifest.json, so it cannot be "
"loaded as a plugin", entry.name)
elif entry.status == ManifestStatus.UNREADABLE:
self.logger.warning("Error reading manifest from %s: %s",
entry.path / "manifest.json", entry.error,
exc_info=entry.error)
elif entry.status == ManifestStatus.NOT_OBJECT:
# json.load accepts any JSON value, so a manifest holding
# null, [] or "text" parses and then raises AttributeError on
# .get(). Nothing here catches that -- the outer handler takes
# OSError/PermissionError only -- so a single malformed
# manifest aborted the whole scan and every other plugin on
# null, [] or "text" parses. It once raised AttributeError on
# .get() and aborted the whole scan, so every other plugin on
# disk, however healthy, silently failed to register.
if not isinstance(manifest, dict):
if item.name not in self._skip_reported:
self._skip_reported.add(item.name)
self.logger.warning(
"Skipping %s: its manifest.json is %s, not a JSON "
"object", item.name, type(manifest).__name__)
continue
self._report_skip_once(
entry.name, "Skipping %s: its manifest.json is %s, not a "
"JSON object", entry.name, type(entry.manifest).__name__)
elif entry.status == ManifestStatus.NO_ID:
# Parsed but unusable. This was the quietest path of all: the
# manifest is read successfully and then dropped.
self._report_skip_once(
entry.name, "Skipping %s: its manifest.json has no \"id\", "
"so there is nothing to register it under", entry.name)
plugin_id = manifest.get('id')
if not plugin_id:
# Parsed but unusable. This was the quietest path of all:
# the manifest is read successfully and then dropped.
if item.name not in self._skip_reported:
self._skip_reported.add(item.name)
self.logger.warning(
"Skipping %s: its manifest.json has no \"id\", so "
"there is nothing to register it under", item.name)
continue
plugins = index.plugins()
for plugin_id, entries in index.duplicates().items():
self._report_skip_once(
"duplicate:" + plugin_id,
"Plugin id %r is declared by %d directories (%s); using %s",
plugin_id, len(entries), ", ".join(e.name for e in entries),
plugins[plugin_id].name)
plugin_ids.append(plugin_id)
new_manifests[plugin_id] = manifest
new_directories[plugin_id] = item
except (OSError, PermissionError) as e:
self.logger.error("Error scanning directory %s: %s", directory, e, exc_info=True)
new_manifests: Dict[str, Dict[str, Any]] = {
plugin_id: entry.manifest for plugin_id, entry in plugins.items()}
new_directories: Dict[str, Path] = {
plugin_id: entry.path for plugin_id, entry in plugins.items()}
# Replace shared state under lock so uninstalled plugins don't linger
with self._discovery_lock:
@@ -266,8 +267,8 @@ class PluginManager:
self.plugin_directories.clear()
self.plugin_directories.update(new_directories)
return plugin_ids
return list(plugins)
def discover_plugins(self) -> List[str]:
"""
Discover all plugins in the plugins directory.
@@ -772,24 +773,20 @@ class PluginManager:
not one plain path segment (``..``, ``a/b``, an absolute path) is
refused instead of being joined onto ``plugins_dir``. The join is not
resolved further: dev plugins are symlinks into ``plugins_dir``.
The discovery map is authoritative. For an id discovery has not seen,
only directory names are tried -- ``<id>`` then ``ledmatrix-<id>``,
in ``plugins_dir`` only -- so a miss on a web request never reads
every manifest on disk. Rules: ``src/plugin_system/plugin_dirs.py``.
"""
with self._discovery_lock:
if plugin_id in self.plugin_directories:
return str(self.plugin_directories[plugin_id])
plugin_id = safe_path_component(plugin_id)
if plugin_id is None:
return None
plugin_dir = self.plugins_dir / plugin_id
if plugin_dir.exists():
return str(plugin_dir)
plugin_dir = self.plugins_dir / f"ledmatrix-{plugin_id}"
if plugin_dir.exists():
return str(plugin_dir)
return None
plugin_dir = resolve_plugin_dir(
plugin_id, [self.plugins_dir], prefix=True, case_insensitive=False,
by_manifest=False)
return str(plugin_dir) if plugin_dir is not None else None
def get_plugin_display_modes(self, plugin_id: str) -> List[str]:
"""
+26 -32
View File
@@ -15,6 +15,7 @@ from enum import Enum
from pathlib import Path
from src.core_config_keys import CORE_CONFIG_KEYS
from src.plugin_system.plugin_dirs import PluginDirectoryIndex
from src.plugin_system.state_manager import PluginStateManager
from src.logging_config import get_logger
@@ -102,31 +103,25 @@ def config_plugin_ids(config: Dict[str, Any], ignored_keys: Set[str]) -> Set[str
def disk_plugin_ids(plugins_dir) -> Set[str]:
"""Plugin ids actually installed on disk.
A directory counts only when it is not a standalone backup and its
manifest.json parses. A corrupt manifest must not read as installed, or a
live "in config but not on disk" finding gets cleared on the strength of an
unreadable file.
A directory counts only when it is not a standalone backup (or hidden)
and its manifest.json parses. A corrupt manifest must not read as
installed, or a live "in config but not on disk" finding gets cleared on
the strength of an unreadable file.
The id is the manifest's ``id`` -- what discovery registers and what the
config is keyed by -- and the directory name only when the manifest has
none. Directory names alone made a plugin living in ``ledmatrix-stocks/``
with id ``stocks`` read as both "stocks in config but not on disk" and
"ledmatrix-stocks on disk but not in config".
"""
ids: Set[str] = set()
root = Path(plugins_dir)
try:
if not root.exists():
return ids
for entry in root.iterdir():
if not entry.is_dir() or '.standalone-backup-' in entry.name:
continue
manifest = entry / "manifest.json"
if not manifest.exists():
continue
try:
with open(manifest, 'r') as f:
json.load(f)
except (OSError, ValueError):
continue
ids.add(entry.name)
return _disk_index(plugins_dir).installed_ids(require_parseable_manifest=True)
except OSError:
return ids
return ids
return set()
def _disk_index(plugins_dir) -> PluginDirectoryIndex:
return PluginDirectoryIndex.scan(Path(plugins_dir))
def still_unresolved(entries: List[Dict[str, Any]],
@@ -326,16 +321,15 @@ class StateReconciliation:
"""Get plugin state from disk (installed plugins)."""
state = {}
try:
# Membership comes from the shared extractor so the web interface
# re-checks stored findings against this same definition; the
# manifest is then re-read here only for version/name.
for plugin_id in disk_plugin_ids(self.plugins_dir):
manifest_path = self.plugins_dir / plugin_id / "manifest.json"
try:
with open(manifest_path, 'r') as f:
manifest = json.load(f)
except (OSError, ValueError): # nosec B112 - raced or corrupt; skip
continue
# Membership uses the same index and rule as disk_plugin_ids, so
# the web interface re-checks stored findings against this same
# definition; each manifest is read once, by the scan.
index = _disk_index(self.plugins_dir)
for plugin_id in index.installed_ids(require_parseable_manifest=True):
entry = index.entry_for_installed_id(plugin_id)
manifest = entry.manifest if entry is not None else None
if not isinstance(manifest, dict):
manifest = {}
state[plugin_id] = {
'exists_on_disk': True,
'version': manifest.get('version'),
+48 -97
View File
@@ -28,6 +28,9 @@ from src.common.permission_utils import sudo_remove_directory, install_requireme
from src.plugin_system.plugin_loader import (
requirements_has_real_deps, requirements_are_satisfied, find_trusted_subdir
)
from src.plugin_system.plugin_dirs import (
BACKUP_MARKER, PluginDirectoryIndex, resolve_plugin_dir, store_search_dirs,
)
try:
from jsonschema import Draft7Validator, ValidationError
@@ -1233,9 +1236,9 @@ class PluginStoreManager:
Pass-through when nothing is installed, and when called from
`_reinstall_with_rollback`, which has already moved the old copy aside.
The aside name embeds '.standalone-backup-' so plugin discovery
(`plugin_manager._scan_directory_for_plugins`) skips it even though it
still holds a manifest.json.
The aside name embeds BACKUP_MARKER ('.standalone-backup-') so every
plugin directory lookup (src/plugin_system/plugin_dirs.py) skips it
even though it still holds a manifest.json.
Held under the per-plugin reinstall lock for the same reason
`_reinstall_with_rollback` is: the web UI runs Flask with
@@ -1250,7 +1253,7 @@ class PluginStoreManager:
return self._install_plugin_impl(plugin_id, branch)
backup_path = plugin_path.with_name(
f"{plugin_path.name}.standalone-backup-preinstall")
f"{plugin_path.name}{BACKUP_MARKER}preinstall")
if backup_path.exists() and not self._safe_remove_directory(backup_path):
# Can't stage a safety net. Better to attempt the install than
# to refuse outright, which is what callers got before this
@@ -2393,95 +2396,43 @@ class PluginStoreManager:
def _find_plugin_path(self, plugin_id: str) -> Optional[Path]:
"""
Find the plugin path by checking the configured directory and standard plugins directory.
Searches the configured directory, then a sibling ``plugins/`` (the
case where plugins sit in plugins/ but config says plugin-repos/) --
a store-only fallback; discovery scans the configured directory only.
Each directory is searched completely before the next, by the shared
rules in ``src/plugin_system/plugin_dirs.py``: a directory whose
manifest declares the id wins, then a directory named exactly for it.
The manifest match matters because a directory name can differ from
the id its manifest declares (a hand-made or legacy layout such as
`ledmatrix-stocks/` holding id `stocks`); a lookup by directory name
alone reported such a plugin as not installed, so update_plugin()
silently did nothing.
No ``ledmatrix-`` prefix and no case folding here, unlike the loader:
a store operation may delete what this returns, so it only accepts a
directory that names the id exactly or declares it. Note that this
leaves registry ids like `stocks` unresolved when the installed
plugin is `ledmatrix-stocks/` declaring `ledmatrix-stocks` (the
monorepo's leaderboard, music, stocks and weather); passing
``prefix=True`` would resolve them, but update_plugin()'s reinstall
path has not been checked against that yet.
Args:
plugin_id: Plugin identifier
Returns:
Path to plugin directory if found, None otherwise
"""
# First check the configured plugins directory
plugin_path = self.plugins_dir / plugin_id
if plugin_path.exists():
return plugin_path
# Also check the standard 'plugins/' directory if it's different
# This handles the case where plugins are in plugins/ but config says plugin-repos/
try:
if self.plugins_dir.is_absolute():
project_root = self.plugins_dir.parent
else:
project_root = self.plugins_dir.resolve().parent
standard_plugins_dir = project_root / 'plugins'
if standard_plugins_dir.exists() and standard_plugins_dir != self.plugins_dir:
plugin_path = standard_plugins_dir / plugin_id
if plugin_path.exists():
return plugin_path
except (OSError, ValueError):
pass
# Last resort: the directory name may differ from the id being looked
# up. install_plugin() deliberately renames a plugin's directory to the
# MANIFEST id when it differs from the REGISTRY id (see the rename near
# "doesn't match registry ID" above), so `stocks` in the registry lands
# in `ledmatrix-stocks/`. Every lookup above is by directory name, so
# update_plugin("stocks") found nothing and reported the plugin as not
# installed -- silently, and for good: the user sees no error and stays
# on a stale version. Four installed plugins hit this in practice
# (leaderboard, music, stocks, weather).
#
# Deliberately last so the two lookups above keep their exact meaning;
# this only runs when a direct hit already failed. See
# test_discovery_path_contract.py, which pins that ordering.
for search_dir in self._candidate_plugin_dirs():
match = self._find_by_manifest_id(search_dir, plugin_id)
if match is not None:
self.logger.debug(
"Resolved plugin '%s' to %s via its manifest id "
"(directory name differs from the id)", plugin_id, match)
return match
return None
return resolve_plugin_dir(
plugin_id, self._candidate_plugin_dirs(), prefix=False,
case_insensitive=False)
def _candidate_plugin_dirs(self) -> List[Path]:
"""Directories that may hold installed plugins, configured one first."""
dirs = [self.plugins_dir]
try:
base = self.plugins_dir if self.plugins_dir.is_absolute() else self.plugins_dir.resolve()
sibling = base.parent / 'plugins'
if sibling != self.plugins_dir:
dirs.append(sibling)
except (OSError, ValueError):
pass
return [d for d in dirs if d.exists()]
return [d for d in store_search_dirs(self.plugins_dir) if d.exists()]
@staticmethod
def _find_by_manifest_id(search_dir: Path, plugin_id: str) -> Optional[Path]:
"""A subdirectory of `search_dir` whose manifest declares `plugin_id`.
Skips half-finished installs: store_manager renames a directory aside
with '.standalone-backup-' during install and rollback, and treating
one as installed would resurrect a ghost plugin.
"""
try:
entries = sorted(search_dir.iterdir())
except (OSError, ValueError):
return None
for entry in entries:
if not entry.is_dir() or '.standalone-backup-' in entry.name:
continue
manifest = entry / 'manifest.json'
if not manifest.is_file():
continue
try:
with open(manifest, 'r', encoding='utf-8') as handle:
if json.load(handle).get('id') == plugin_id:
return entry
except (OSError, ValueError):
continue
return None
def uninstall_plugin(self, plugin_id: str) -> bool:
"""
Uninstall a plugin by removing its directory.
@@ -2600,9 +2551,9 @@ class PluginStoreManager:
field during the monorepo migration on a Pi with broken DNS — every
old-remote plugin was deleted and none could be re-downloaded).
The aside name embeds '.standalone-backup-' so plugin discovery
(plugin_manager._scan_directory_for_plugins) ignores it even though
it still contains a manifest.json.
The aside name embeds BACKUP_MARKER ('.standalone-backup-') so every
plugin directory lookup (src/plugin_system/plugin_dirs.py) ignores it
even though it still contains a manifest.json.
Held for the whole operation under a per-plugin_id lock: two
overlapping requests for the same plugin (double-click, two
@@ -2612,7 +2563,7 @@ class PluginStoreManager:
"""
with self._get_reinstall_lock(plugin_id):
backup_path = plugin_path.with_name(
f"{plugin_path.name}.standalone-backup-migrating")
f"{plugin_path.name}{BACKUP_MARKER}migrating")
# A stale aside from a previous crash would block the rename
if backup_path.exists():
if not self._safe_remove_directory(backup_path):
@@ -3084,16 +3035,16 @@ class PluginStoreManager:
def list_installed_plugins(self) -> List[str]:
"""
Get list of installed plugin IDs.
One entry per plugin directory in the configured directory that has a
manifest.json, named by the manifest's id (the directory name when
the manifest carries none, e.g. because it does not parse). Backup
and hidden directories are not plugins.
Returns:
List of plugin IDs
List of plugin IDs, sorted
"""
if not self.plugins_dir.exists():
return []
installed = []
for item in self.plugins_dir.iterdir():
if item.is_dir() and (item / "manifest.json").exists():
installed.append(item.name)
return installed
index = PluginDirectoryIndex.scan(self.plugins_dir)
return sorted(index.installed_ids(require_parseable_manifest=False))
@@ -14,13 +14,16 @@ PIL Image canvas and draws text using the actual project fonts.
MAINTENANCE WARNING: this class is a deliberate fork of
src/display_manager.py so it can run without hardware. It mirrors
these DisplayManager methods by name and behavior: _load_fonts,
_draw_bdf_text, get_font_height, get_text_width, draw_text,
get_font_height, get_text_width, draw_text,
draw_text_with_icons, draw_weather_icon (and the _draw_sun/_draw_cloud/
_draw_rain/_draw_snow/_draw_storm family), format_date_with_ordinal,
capture_mode, set_scrolling_state, is_currently_scrolling,
process_deferred_updates, update_display, render_size. A behavior
change to any of those in DisplayManager must be mirrored here, or
plugin visual tests will pass against stale behavior.
BDF text is not mirrored: both classes load BDF faces and draw BDF glyphs
through src/common/bdf_font.py, so those pixels cannot drift.
"""
import math
@@ -31,6 +34,7 @@ from pathlib import Path
from typing import Any, List, Optional, Tuple
from PIL import Image, ImageDraw, ImageFont
from src.common.bdf_font import draw_bdf_text, load_bdf_face
from src.common.font_layout import crisp_size, load_truetype
from src.logging_config import get_logger
@@ -147,16 +151,18 @@ class VisualTestDisplayManager:
self.small_font = load_truetype(ttf_path, crisp_size(press_start, 8))
self.font = self.regular_font # alias used by some code paths
# 5x7 BDF font via freetype
# 5x7 BDF font, loaded exactly as DisplayManager._load_fonts does
# (same loader, same 7px request as its _CALENDAR_FONT_PX). A bare
# freetype.Face has no active size, so its ascender reads 0 and
# every line drew a baseline too high.
try:
import freetype
bdf_path = str(fonts_dir / '5x7.bdf')
if not os.path.exists(bdf_path):
raise FileNotFoundError(f"BDF font not found: {bdf_path}")
face = freetype.Face(bdf_path)
face, _ = load_bdf_face(bdf_path, 7)
self.calendar_font = face
self.bdf_5x7_font = face
except (ImportError, FileNotFoundError, OSError) as e:
except Exception as e: # freetype missing or the file unloadable
logger.debug("BDF font not available, using small_font as fallback: %s", e)
self.calendar_font = self.small_font
self.bdf_5x7_font = self.small_font
@@ -300,41 +306,18 @@ class VisualTestDisplayManager:
logger.debug(f"Error drawing image: {e}")
def _draw_bdf_text(self, text, x, y, color=(255, 255, 255), font=None):
"""Draw text using BDF font with proper bitmap handling.
"""Draw text in a BDF ``freetype.Face`` with (x, y) as its top-left.
Replicated from DisplayManager._draw_bdf_text().
Not a copy: DisplayManager._draw_bdf_text calls the same
:func:`src.common.bdf_font.draw_bdf_text`, so what this draws is
what the panel draws.
"""
try:
if isinstance(color, list):
color = tuple(color)
face = font if font else self.calendar_font
# Compute baseline from font ascender
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
glyph_left = face.glyph.bitmap_left
glyph_top = face.glyph.bitmap_top
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))):
pixel_x = x + glyph_left + j
pixel_y = baseline_y - glyph_top + i
if 0 <= pixel_x < self.width and 0 <= pixel_y < self.height:
self.draw.point((pixel_x, pixel_y), fill=color)
x += face.glyph.advance.x >> 6
draw_bdf_text(self.draw, text, x, y, face, color,
clip=(self.width, self.height))
except Exception as e:
logger.debug(f"Error drawing BDF text: {e}")
+16 -3
View File
@@ -24,8 +24,13 @@ _REDACT_CREDENTIAL = re.compile(
# silently leak the ones nobody thought of. Not covered by the generic pattern
# above, whose value part stops at whitespace and so would keep the credential
# once a space follows the scheme.
#
# The opening quote and the whitespace after it are one optional unit. Written
# `\s*["\']?\s*`, a whitespace run with no quote in it could be split between
# the two `\s*` in every possible way, and a header with no credential after
# it tried them all: quadratic, 8s for 20k spaces.
_REDACT_AUTH_HEADER = re.compile(
r'((?:proxy-)?authorization["\']?\s*[=:]\s*["\']?\s*'
r'((?:proxy-)?authorization["\']?\s*[=:]\s*(?:["\']\s*)?'
r'(?:[A-Za-z][\w.+-]*[ \t]+)?)' # optional scheme name, kept
r'([^\s,"\'<>}]+)', # the credential, redacted
re.IGNORECASE,
@@ -34,8 +39,16 @@ _REDACT_AUTH_HEADER = re.compile(
# Credentials embedded in a URL: https://user:password@host. requests quotes
# the full URL in its exceptions, so this is a realistic leak. The username is
# kept -- it identifies which account failed without being the secret.
_REDACT_URL_USERINFO = re.compile(r'([a-z][a-z0-9+.-]*://[^/\s:@]+:)([^/\s@]+)(@)',
re.IGNORECASE)
#
# A match may only start where a run of scheme characters starts. Unanchored,
# `[a-z][a-z0-9+.-]*://` was tried from every letter of a long run (a hex
# digest, an ID, a blob of response body), each attempt reading to the end of
# the run: quadratic, 1.6s for 20k characters, all of it holding the GIL.
# Leading digits and `+.-` sit inside group 1 so the substitution puts them
# back; the scheme proper still has to start with a letter.
_REDACT_URL_USERINFO = re.compile(
r'((?<![a-z0-9+.-])[0-9+.-]*[a-z][a-z0-9+.-]*://[^/\s:@]+:)([^/\s@]+)(@)',
re.IGNORECASE)
def redact_credentials(text: str) -> str:
+37 -2
View File
@@ -9,7 +9,7 @@ from typing import Any, Optional, Dict, Tuple
from flask import jsonify, request
from src.web_interface.error_handler import create_error_response, create_success_response
from src.web_interface.errors import ErrorCode
from src.web_interface.errors import ErrorCode, WebInterfaceError
def success_response(
@@ -34,7 +34,8 @@ def success_response(
# metadata block for responses that have neither.
enriched = dict(metadata) if metadata is not None else {}
if hasattr(request, 'start_time'):
enriched['response_time_ms'] = int((time.time() - request.start_time) * 1000)
# request_logging stamps start_time from perf_counter, not the wall clock.
enriched['response_time_ms'] = int((time.perf_counter() - request.start_time) * 1000)
if metadata is not None or enriched:
response_data['metadata'] = enriched
@@ -74,6 +75,40 @@ def error_response(
)
def exception_error_response(
exc: Exception,
error_code: ErrorCode,
*,
with_context: bool = True,
status_code: int = 500
):
"""
error_response() for a caught exception, built by WebInterfaceError.
The message is the code's fixed, user-facing one -- never the exception
text. `details` comes from the exception's own `context` dict when it has
one, and `context` records the exception type. with_context=False leaves
the context out, as the operation-history routes always have.
Args:
exc: The exception being reported
error_code: Error code
with_context: Whether to include the context (exception type)
status_code: HTTP status code
Returns:
Flask jsonify response with status code
"""
error = WebInterfaceError.from_exception(exc, error_code)
return error_response(
error.error_code,
error.message,
details=error.details,
context=error.context if with_context else None,
status_code=status_code
)
def validate_request_json(required_fields: list, data: Optional[Dict] = None) -> Tuple[Optional[Dict], Optional[Any]]:
"""
Validate request JSON has required fields.
+34 -3
View File
@@ -7,9 +7,7 @@ Provides helpers for consistent error responses across API endpoints.
from typing import Any, Optional
from flask import jsonify
from src.web_interface.errors import (
WebInterfaceError, ErrorCode, ErrorCategory
)
from src.web_interface.errors import WebInterfaceError, ErrorCode
from src.logging_config import get_logger
from src.redaction import redact_credentials
@@ -72,6 +70,39 @@ def redact_text(text: str, max_length: int = _MAX_DETAIL_LENGTH) -> str:
return text
# What a failure nothing anticipated says. The detail beside it carries the
# actual diagnosis; this sentence only points at where the traceback went.
UNHANDLED_ERROR_MESSAGE = 'An error occurred; see logs for details'
def unhandled_exception_payload(exc: BaseException) -> dict:
"""JSON body for an exception no route handled: status, message, details.
Deliberately no `error_code`. The plugin API client (api_client.js) passes
a body that has one straight to the rich error modal, and wraps one that
has none as a plain API_ERROR toast; the api_v3 routes answered this shape
from their own catch-alls for years, so the UI is built around it.
"""
return {
'status': 'error',
'message': UNHANDLED_ERROR_MESSAGE,
'details': describe_exception(exc),
}
def http_exception_payload(error) -> dict:
"""JSON body for a werkzeug HTTPException (405, 400, 415, 413...).
Same shape web_interface/app.py's global handler returns, so a 4xx raised
inside an api_v3 route reads the same as one raised anywhere else.
"""
return {
'status': 'error',
'error_code': (error.name or 'HTTP_ERROR').upper().replace(' ', '_'),
'message': error.description,
}
def create_error_response(
error_code: ErrorCode,
message: str,
+5 -63
View File
@@ -1,7 +1,7 @@
"""
Structured error handling for web interface.
Provides error codes, categories, and consistent error response formatting.
Provides error codes and consistent error response formatting.
"""
from enum import Enum
@@ -9,17 +9,6 @@ from typing import Dict, Any, Optional, List
from dataclasses import dataclass
class ErrorCategory(Enum):
"""Error categories for classification."""
CONFIGURATION = "configuration"
PLUGIN = "plugin"
VALIDATION = "validation"
NETWORK = "network"
PERMISSION = "permission"
SYSTEM = "system"
UNKNOWN = "unknown"
class ErrorCode(Enum):
"""Error codes for specific error types."""
# Configuration errors
@@ -63,12 +52,11 @@ class WebInterfaceError:
"""
Structured error for web interface responses.
Provides consistent error format with error codes, categories,
messages, and context.
Provides consistent error format with error codes, messages, and
context.
"""
error_code: ErrorCode
message: str
category: ErrorCategory
details: Optional[str] = None
context: Optional[Dict[str, Any]] = None
suggested_fixes: Optional[List[str]] = None
@@ -78,7 +66,6 @@ class WebInterfaceError:
self,
error_code: ErrorCode,
message: str,
category: Optional[ErrorCategory] = None,
details: Optional[str] = None,
context: Optional[Dict[str, Any]] = None,
suggested_fixes: Optional[List[str]] = None,
@@ -86,7 +73,6 @@ class WebInterfaceError:
):
self.error_code = error_code
self.message = message
self.category = category or self._infer_category(error_code)
self.details = details
self.context = context or {}
# `is None`, not truthiness: an explicit [] means "this caller has
@@ -96,25 +82,6 @@ class WebInterfaceError:
else self._get_default_suggestions(error_code))
self.original_error = original_error
def _infer_category(self, error_code: ErrorCode) -> ErrorCategory:
"""Infer error category from error code."""
code_str = error_code.value
if code_str.startswith("CONFIG_"):
return ErrorCategory.CONFIGURATION
elif code_str.startswith("PLUGIN_"):
return ErrorCategory.PLUGIN
elif code_str.startswith("VALIDATION_") or code_str.startswith("SCHEMA_") or code_str == "INVALID_INPUT":
return ErrorCategory.VALIDATION
elif code_str.startswith("NETWORK_") or code_str == "API_ERROR" or code_str == "TIMEOUT":
return ErrorCategory.NETWORK
elif code_str.startswith("PERMISSION_") or code_str == "FILE_PERMISSION_ERROR":
return ErrorCategory.PERMISSION
elif code_str.startswith("SYSTEM_") or code_str == "SERVICE_UNAVAILABLE":
return ErrorCategory.SYSTEM
else:
return ErrorCategory.UNKNOWN
def _get_default_suggestions(self, error_code: ErrorCode) -> List[str]:
"""Get default suggested fixes for error code."""
suggestions_map = {
@@ -178,7 +145,6 @@ class WebInterfaceError:
result = {
"status": "error",
"error_code": self.error_code.value,
"error_category": self.category.value,
"message": self.message,
}
@@ -197,7 +163,7 @@ class WebInterfaceError:
def from_exception(
cls,
exception: Exception,
error_code: Optional[ErrorCode] = None,
error_code: ErrorCode,
context: Optional[Dict[str, Any]] = None
) -> 'WebInterfaceError':
"""
@@ -205,13 +171,9 @@ class WebInterfaceError:
Args:
exception: Exception to convert
error_code: Optional specific error code
error_code: The error code to report
context: Optional additional context
"""
# Infer error code from exception type if not provided
if not error_code:
error_code = cls._infer_error_code(exception)
# Build context
error_context = context or {}
error_context['exception_type'] = type(exception).__name__
@@ -252,26 +214,6 @@ class WebInterfaceError:
}
return messages.get(error_code, "An unexpected error occurred")
@classmethod
def _infer_error_code(cls, exception: Exception) -> ErrorCode:
"""Infer error code from exception type."""
exception_name = type(exception).__name__
if "Config" in exception_name:
return ErrorCode.CONFIG_LOAD_FAILED
elif "Plugin" in exception_name:
return ErrorCode.PLUGIN_LOAD_FAILED
elif "Permission" in exception_name or "Access" in exception_name:
return ErrorCode.PERMISSION_DENIED
elif "Validation" in exception_name or "Schema" in exception_name:
return ErrorCode.VALIDATION_ERROR
elif "Network" in exception_name or "Connection" in exception_name:
return ErrorCode.NETWORK_ERROR
elif "Timeout" in exception_name:
return ErrorCode.TIMEOUT
else:
return ErrorCode.UNKNOWN_ERROR
@classmethod
def _get_exception_details(cls, exception: Exception) -> Optional[str]:
"""Get additional details from exception."""