mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-10 17:16:36 +00:00
Merge branch 'claude/hdpi-scroll-performance-antialiasing-4ae609' into claude/offscreen-rendering
This commit is contained in:
@@ -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):
|
||||
|
||||
Vendored
+43
@@ -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)
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
@@ -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)
|
||||
|
||||
@@ -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
@@ -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
@@ -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
@@ -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,
|
||||
@@ -1026,43 +1027,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)
|
||||
|
||||
@@ -1111,19 +1085,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
@@ -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
@@ -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."""
|
||||
|
||||
@@ -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
|
||||
@@ -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,
|
||||
|
||||
@@ -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]:
|
||||
"""
|
||||
|
||||
@@ -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'),
|
||||
|
||||
@@ -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, offscreen. 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
|
||||
@@ -328,41 +334,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
@@ -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:
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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."""
|
||||
|
||||
Reference in New Issue
Block a user