mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-08-01 16:58:06 +00:00
Compare commits
9
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3869e34911 | ||
|
|
efe76d3add | ||
|
|
273d9962d1 | ||
|
|
9e3b5f366e | ||
|
|
62fbed3471 | ||
|
|
6edd80d9f3 | ||
|
|
1c7a0cef66 | ||
|
|
6052a60d22 | ||
|
|
7f7f0d6464 |
@@ -0,0 +1,234 @@
|
|||||||
|
# Adaptive Layout & Font Scaling
|
||||||
|
|
||||||
|
`src/adaptive_layout.py` lets a plugin render legibly on **any** panel size
|
||||||
|
(64x32, 128x32, 96x48, 128x64, 256x64, ...) without hand-tuned per-display
|
||||||
|
layouts. It is **opt-in**: nothing changes for plugins that don't use it.
|
||||||
|
|
||||||
|
It generalizes three patterns proven in the plugin ecosystem:
|
||||||
|
|
||||||
|
| Pattern | Origin | Core API |
|
||||||
|
|---|---|---|
|
||||||
|
| Geometry scale factor vs. a design size | f1-scoreboard | `ctx.px(base)` / `ctx.scale` |
|
||||||
|
| Breakpoint tiers | masters-tournament | `ctx.tier` / `ctx.by_tier({...})` |
|
||||||
|
| "Largest crisp font that fits" ladder | baseball-scoreboard | `ctx.fit_text(...)` and friends |
|
||||||
|
|
||||||
|
## Quick start
|
||||||
|
|
||||||
|
Every `BasePlugin` has a lazy `self.layout` (a `LayoutContext` for the
|
||||||
|
current logical display size, rebuilt automatically if the size changes)
|
||||||
|
and a one-liner `self.draw_fit(...)`:
|
||||||
|
|
||||||
|
```python
|
||||||
|
def display(self, force_clear=False):
|
||||||
|
from src.adaptive_layout import LADDER_ARCADE
|
||||||
|
|
||||||
|
b = self.layout.bounds.inset(1) # Region(0,0,W,H) minus 1px margin
|
||||||
|
rows = b.split_v(3, 1, 1, gap=1) # 3/5 for time, 1/5 each for the rest
|
||||||
|
|
||||||
|
self.draw_fit(self.time_str, rows[0], ladder=LADDER_ARCADE)
|
||||||
|
self.draw_fit(self.weekday, rows[1]) # default LADDER_GRID
|
||||||
|
self.draw_fit(self.date_str, rows[2])
|
||||||
|
self.display_manager.update_display()
|
||||||
|
```
|
||||||
|
|
||||||
|
On 128x64 the time renders at press_start 24px; on 64x32 it steps down to
|
||||||
|
8px. The rows partition the height, so bands can never overlap — no more
|
||||||
|
`y = height - 7` magic numbers.
|
||||||
|
|
||||||
|
## Region — rect algebra
|
||||||
|
|
||||||
|
`Region(x, y, w, h)` is a frozen dataclass. All carving clamps to
|
||||||
|
non-negative dimensions, so degenerate panels behave.
|
||||||
|
|
||||||
|
- Carving: `inset(dx, dy)`, `top_band(h)`, `bottom_band(h)`,
|
||||||
|
`middle(top_h, bottom_h)`, `left_col(w)`, `right_col(w)`,
|
||||||
|
`split_h(*weights, gap=0)`, `split_v(*weights, gap=0)`
|
||||||
|
- Placement: `align_xy(w, h, align, valign)`, `center_xy(w, h)`,
|
||||||
|
`contains(w, h)`, `.center`, `.right`, `.bottom`
|
||||||
|
|
||||||
|
Scoreboard-style layout:
|
||||||
|
|
||||||
|
```python
|
||||||
|
b = self.layout.bounds
|
||||||
|
status = b.top_band(self.layout.px(7))
|
||||||
|
detail = b.bottom_band(self.layout.px(7))
|
||||||
|
score_area = b.middle(status.h, detail.h)
|
||||||
|
away_slot, home_slot = b.left_col(b.h), b.right_col(b.h)
|
||||||
|
```
|
||||||
|
|
||||||
|
## Font ladders — discrete, never fractional
|
||||||
|
|
||||||
|
Pixel fonts (BDF, PressStart2P) only look right at native/integer sizes, so
|
||||||
|
fonts are never scaled continuously. A `FontLadder` is an ordered tuple of
|
||||||
|
`FontStep(family, size_px)` rungs, largest first; fitting walks down until
|
||||||
|
the measured text fits.
|
||||||
|
|
||||||
|
- `LADDER_GRID` (default): X11 BDFs at native sizes — 10x20 → 9x18 → 9x15 →
|
||||||
|
8x13 → 7x13 → 6x13 → 6x12 → 6x10 → 6x9 → 5x8 → 5x7 → 4x6 → tom-thumb.
|
||||||
|
Body text, labels, multi-row content.
|
||||||
|
- `LADDER_ARCADE`: PressStart2P at 32/24/16/8 (integer multiples of its 8px
|
||||||
|
grid). Headline text: clocks, scores.
|
||||||
|
|
||||||
|
Custom ladders are just tuples — e.g. to add your plugin's registered font
|
||||||
|
on top: `(FontStep("myplugin::digits", 16),) + LADDER_GRID`.
|
||||||
|
|
||||||
|
## LayoutContext
|
||||||
|
|
||||||
|
Built per (width, height); exposes facts and fit queries:
|
||||||
|
|
||||||
|
- `bounds`, `width`, `height`, `aspect`
|
||||||
|
- `tier` by height (`xs`≤16, `sm`≤32, `md`≤48, `lg`≤64, `xl`) and
|
||||||
|
`width_tier` (`narrow`≤64, `normal`≤128, `wide`≤256, `ultrawide`)
|
||||||
|
- `is_wide_short` — aspect ≥ 2.5 and height ≤ 32 (the classic 128x32 shape)
|
||||||
|
- `scale` — `min(w/design_w, h/design_h)` vs. your manifest's
|
||||||
|
`display.design_size` (default 128x32). **Geometry only** — gaps, icon
|
||||||
|
and logo sizes via `px(base, minimum, maximum)`; fonts use ladders.
|
||||||
|
- `by_tier({"sm": 10, "lg": 18})` — value for the nearest defined tier
|
||||||
|
at-or-below the panel's tier.
|
||||||
|
- `fit_text(text, box, ladder, ellipsis=True)` → `FitResult` — largest rung
|
||||||
|
that fits; ellipsizes as a last resort. Cached per (text, box, ladder).
|
||||||
|
- `fit_text_proportional(text, box, base_size_px, ladder, ellipsis=True, scale=None)` —
|
||||||
|
rung closest to (not exceeding) `base_size_px * scale`, still capped to
|
||||||
|
what fits the box. Use this instead of `fit_text` when several
|
||||||
|
independently-fitted elements need to stay visually harmonious as the
|
||||||
|
panel grows — `fit_text` maximizes *each one* within its own region,
|
||||||
|
which can make one element (e.g. a score with a generous box) balloon
|
||||||
|
out of proportion to a neighbor that scales by geometry (e.g. logos
|
||||||
|
sized via `px()`), even though each individual pick is "correct" in
|
||||||
|
isolation. `base_size_px` is normally the element's existing classic/
|
||||||
|
fixed font size. `scale` defaults to `self.scale` (the conservative
|
||||||
|
min-of-both-axes factor `px()` uses); pass an axis-specific value when
|
||||||
|
the surrounding composition already scales that way — e.g. a scoreboard
|
||||||
|
whose logo slots track height alone (`min(height, width // 2)`) should
|
||||||
|
size its text by `height / design_height` too, or the text reads as
|
||||||
|
under-scaled next to bigger logos on a panel that only grew taller.
|
||||||
|
- `fit_lines(lines, box, ladder, spacing)` — every line fits the width and
|
||||||
|
the stack fits the height (measures the actual strings).
|
||||||
|
- `font_for_rows(rows, box_h, ladder)` — largest rung whose line height
|
||||||
|
fits `rows` rows.
|
||||||
|
|
||||||
|
`FitResult` carries the ready-to-use `font` (drops straight into
|
||||||
|
`display_manager.draw_text(font=...)`), the possibly-ellipsized `text`,
|
||||||
|
ink `width`/`height`, `baseline`, `y_offset`, `line_height`, and `fits`.
|
||||||
|
|
||||||
|
## Adaptive images
|
||||||
|
|
||||||
|
`src/adaptive_images.py` is the image counterpart to `fit_text`, exposed as
|
||||||
|
`self.layout.fit_image(...)` (cached per panel size) and the one-liner
|
||||||
|
`self.draw_image(...)`:
|
||||||
|
|
||||||
|
```python
|
||||||
|
# Team logo: trim its transparent padding, fill the slot height (the
|
||||||
|
# football/hockey pattern), cached across frames by a stable key
|
||||||
|
self.draw_image(logo, regs.away_slot, mode="fill_height",
|
||||||
|
crop_to_ink=True, cache_key=f"logo:{abbr}")
|
||||||
|
|
||||||
|
# Album art: cover-crop a square, faces kept by the top anchor
|
||||||
|
self.draw_image(art, row.art, mode="cover", anchor="top")
|
||||||
|
|
||||||
|
# Pixel flags / sprite icons: NEAREST keeps hard edges
|
||||||
|
from src.adaptive_images import RESAMPLE_NEAREST
|
||||||
|
self.draw_image(flag, box, resample=RESAMPLE_NEAREST)
|
||||||
|
```
|
||||||
|
|
||||||
|
Modes: `contain` (letterbox, default), `cover` (crop-to-fill),
|
||||||
|
`fill_height` (logo-style), `stretch`. Unlike PIL's `thumbnail()`
|
||||||
|
(downscale-only — why imagery stays tiny on big panels) fitting **upscales
|
||||||
|
by default**; pass `upscale=False` for the legacy behavior. Results are
|
||||||
|
cached per (image, box size, options) with a bounded LRU — always pass a
|
||||||
|
stable `cache_key` (e.g. `"logo:KC"`) for images you reload. The module
|
||||||
|
also exports the Pillow-compat `RESAMPLE_LANCZOS`/`RESAMPLE_NEAREST`
|
||||||
|
constants so plugins can drop their local shims.
|
||||||
|
|
||||||
|
## Composite layouts
|
||||||
|
|
||||||
|
Pre-carved Region arrangements for the layouts plugins keep rebuilding:
|
||||||
|
|
||||||
|
```python
|
||||||
|
from src.adaptive_layout import scoreboard_regions, media_row
|
||||||
|
|
||||||
|
regs = scoreboard_regions(self.layout.bounds, ctx=self.layout)
|
||||||
|
# regs.away_slot / home_slot — logo slots (logo_slot = min(H, W // 2),
|
||||||
|
# capped so a center reserve always exists —
|
||||||
|
# see below)
|
||||||
|
# regs.status_band — top band (replaces the magic y = 1)
|
||||||
|
# regs.score_area — center gap, plus a controlled bleed into
|
||||||
|
# each logo slot (replaces y = H//2 - 3)
|
||||||
|
# regs.detail_band — bottom band (replaces y = H - 7)
|
||||||
|
# regs.bottom_left / bottom_right — record/timeout corners
|
||||||
|
|
||||||
|
row = media_row(self.layout.bounds, ctx=self.layout) # art left, text right
|
||||||
|
```
|
||||||
|
|
||||||
|
Both work on the full panel or on a scroll-mode card Region. They return
|
||||||
|
Regions and never draw — compose them with `draw_fit`/`draw_image`.
|
||||||
|
|
||||||
|
**`scoreboard_regions`'s center reserve.** The raw `logo_slot = min(H, W//2)`
|
||||||
|
formula has a blind spot: at exactly 2:1 aspect ratio (width = 2×height —
|
||||||
|
two, four, or more square modules stacked into a taller panel, e.g.
|
||||||
|
96x48, 128x64, 256x128) the two logo slots mathematically claim the
|
||||||
|
*entire* width, leaving zero pixels for a center column no matter how
|
||||||
|
big the panel gets. Wide panels (the 128x32 design baseline, 192x48,
|
||||||
|
256x32) never hit this, since height is already the tighter constraint
|
||||||
|
there. Two parameters fix it without any plugin-side code:
|
||||||
|
`min_center_fraction`/`min_center_design_px` guarantee a real minimum
|
||||||
|
center reserve at any aspect ratio, and `score_bleed_fraction` lets the
|
||||||
|
score's *fit box* extend a controlled amount into each logo slot — the
|
||||||
|
same way a real broadcast scoreboard's numbers cross slightly into the
|
||||||
|
team marks flanking them — so a short score string never has to truncate
|
||||||
|
even on the tightest aspect ratios. All three have sane defaults; override
|
||||||
|
them per call if a plugin's card proportions genuinely differ.
|
||||||
|
|
||||||
|
## Preserving user customization
|
||||||
|
|
||||||
|
Adaptive layout supplies *defaults*; explicit user configuration wins:
|
||||||
|
|
||||||
|
- **User-set fonts win.** If the plugin's config has an explicit
|
||||||
|
`font`/`font_size` for an element, load it as before and skip the ladder —
|
||||||
|
fit only when the user hasn't overridden (see the football-scoreboard
|
||||||
|
`_resolve_element_fit` pattern).
|
||||||
|
- **Offsets apply on top.** `customization.layout.<element>.{x_offset,y_offset}`
|
||||||
|
style knobs translate the *computed* region as a final step:
|
||||||
|
`region.offset(user_dx, user_dy)`. `draw_image(..., offset=(dx, dy))`
|
||||||
|
does the same for images.
|
||||||
|
- **Colors pass through.** `draw_fit`/`draw_fitted_text` take explicit
|
||||||
|
`color=` params; adaptive mode never repaints semantic or user-chosen
|
||||||
|
colors.
|
||||||
|
|
||||||
|
## Manifest declaration
|
||||||
|
|
||||||
|
Declare the size your layout was authored against so `ctx.scale` means
|
||||||
|
something:
|
||||||
|
|
||||||
|
```json
|
||||||
|
"display": { "design_size": { "width": 128, "height": 32 } }
|
||||||
|
```
|
||||||
|
|
||||||
|
Also available under `requires.display_size`: `min_width`, `min_height`,
|
||||||
|
`max_width`, `max_height`.
|
||||||
|
|
||||||
|
## Performance notes (Pi)
|
||||||
|
|
||||||
|
Fit queries are cached, so cost is O(unique strings). For per-second text
|
||||||
|
(clocks, live scores), fit on a **shape placeholder** and reuse the font:
|
||||||
|
|
||||||
|
```python
|
||||||
|
fit = self.layout.fit_text("00:00", box, ladder=LADDER_ARCADE) # cached once
|
||||||
|
self.display_manager.draw_text(current_time, font=fit.font, ...)
|
||||||
|
```
|
||||||
|
|
||||||
|
## Testing across sizes
|
||||||
|
|
||||||
|
The harness already renders every plugin at a spread of sizes (now
|
||||||
|
including 96x48):
|
||||||
|
|
||||||
|
```bash
|
||||||
|
python scripts/check_plugin.py <plugin-dir> --sizes 64x32,128x32,96x48,128x64,256x64
|
||||||
|
python scripts/render_plugin.py <plugin-dir> --width 96 --height 48
|
||||||
|
```
|
||||||
|
|
||||||
|
`BoundsCheckingDisplayManager` flags right/bottom overflow and now records
|
||||||
|
mediated draw calls with negative coordinates in
|
||||||
|
`negative_coordinate_calls` (raw-PIL draws remain uncovered).
|
||||||
|
|
||||||
|
Reference migration: the **text-display** plugin's `font_mode: "auto"`.
|
||||||
@@ -2,6 +2,12 @@
|
|||||||
|
|
||||||
Advanced patterns, examples, and best practices for developing LEDMatrix plugins.
|
Advanced patterns, examples, and best practices for developing LEDMatrix plugins.
|
||||||
|
|
||||||
|
> **Adaptive layout:** for plugins that should render legibly on any panel
|
||||||
|
> size (fonts that grow on big panels, layouts that degrade gracefully on
|
||||||
|
> small ones), use the adaptive layout system — `self.layout`, `draw_fit`,
|
||||||
|
> `draw_image`, `scoreboard_regions` — documented in
|
||||||
|
> [ADAPTIVE_LAYOUT.md](ADAPTIVE_LAYOUT.md).
|
||||||
|
|
||||||
## Table of Contents
|
## Table of Contents
|
||||||
|
|
||||||
- [Using Weather Icons](#using-weather-icons)
|
- [Using Weather Icons](#using-weather-icons)
|
||||||
|
|||||||
@@ -48,6 +48,12 @@ display_manager.draw_text("Centered", centered=True) # Auto-center
|
|||||||
width = display_manager.get_text_width("Text", font)
|
width = display_manager.get_text_width("Text", font)
|
||||||
height = display_manager.get_font_height(font)
|
height = display_manager.get_font_height(font)
|
||||||
|
|
||||||
|
# Adaptive layout (recommended for multi-size support — text and images
|
||||||
|
# that scale to any panel; see docs/ADAPTIVE_LAYOUT.md)
|
||||||
|
rows = self.layout.bounds.inset(1).split_v(3, 1, gap=1)
|
||||||
|
self.draw_fit("12:34", rows[0]) # largest crisp font that fits
|
||||||
|
self.draw_image(logo, rows[1], mode="fill_height", crop_to_ink=True)
|
||||||
|
|
||||||
# Weather icons
|
# Weather icons
|
||||||
display_manager.draw_weather_icon("rain", x=10, y=10, size=16)
|
display_manager.draw_weather_icon("rain", x=10, y=10, size=16)
|
||||||
|
|
||||||
|
|||||||
@@ -6,6 +6,12 @@ Tools for rapid plugin development without deploying to the RPi.
|
|||||||
|
|
||||||
Interactive web UI for tweaking plugin configs and seeing the rendered display in real time.
|
Interactive web UI for tweaking plugin configs and seeing the rendered display in real time.
|
||||||
|
|
||||||
|
The size inputs have a preset dropdown with the harness's standard panel
|
||||||
|
sizes, and the **All Sizes** button renders the current config at every
|
||||||
|
harness size in a side-by-side gallery (`POST /api/render-matrix`) — the
|
||||||
|
quickest way to eyeball adaptive-layout behavior across panels
|
||||||
|
(see [ADAPTIVE_LAYOUT.md](ADAPTIVE_LAYOUT.md)).
|
||||||
|
|
||||||
### Quick Start
|
### Quick Start
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
|
|||||||
@@ -1,5 +1,12 @@
|
|||||||
# FontManager Usage Guide
|
# FontManager Usage Guide
|
||||||
|
|
||||||
|
> **Picking a size automatically:** if you want the *largest font that fits
|
||||||
|
> a given area* rather than a fixed size, use the adaptive layout system's
|
||||||
|
> font ladders, which resolve through this FontManager. `BasePlugin`
|
||||||
|
> subclasses get this as `self.layout.fit_text(...)`; other code can build
|
||||||
|
> a `LayoutContext(width, height, font_manager)` directly — see
|
||||||
|
> [ADAPTIVE_LAYOUT.md](ADAPTIVE_LAYOUT.md).
|
||||||
|
|
||||||
## Overview
|
## Overview
|
||||||
|
|
||||||
The enhanced FontManager provides comprehensive font management for the LEDMatrix application with support for:
|
The enhanced FontManager provides comprehensive font management for the LEDMatrix application with support for:
|
||||||
|
|||||||
@@ -2,6 +2,11 @@
|
|||||||
|
|
||||||
Complete API reference for plugin developers. This document describes all methods and properties available to plugins through the Display Manager, Cache Manager, and Plugin Manager.
|
Complete API reference for plugin developers. This document describes all methods and properties available to plugins through the Display Manager, Cache Manager, and Plugin Manager.
|
||||||
|
|
||||||
|
> **Adaptive layout:** every `BasePlugin` also exposes `self.layout`,
|
||||||
|
> `self.draw_fit(text, region)` and `self.draw_image(img, region, ...)` —
|
||||||
|
> the recommended way to render text and images that scale to any panel
|
||||||
|
> size. See [ADAPTIVE_LAYOUT.md](ADAPTIVE_LAYOUT.md).
|
||||||
|
|
||||||
## Table of Contents
|
## Table of Contents
|
||||||
|
|
||||||
- [BasePlugin](#baseplugin)
|
- [BasePlugin](#baseplugin)
|
||||||
|
|||||||
@@ -2,6 +2,14 @@
|
|||||||
|
|
||||||
This guide explains how to set up a development workflow for plugins that are maintained in separate Git repositories while still being able to test them within the LEDMatrix project.
|
This guide explains how to set up a development workflow for plugins that are maintained in separate Git repositories while still being able to test them within the LEDMatrix project.
|
||||||
|
|
||||||
|
> **Rendering guidance:** plugins should read the display size dynamically
|
||||||
|
> (`self.display_manager.matrix.width/height`) rather than hardcoding one
|
||||||
|
> panel. For plugins that want to *scale* their layout to any panel, the
|
||||||
|
> opt-in adaptive layout system ([ADAPTIVE_LAYOUT.md](ADAPTIVE_LAYOUT.md))
|
||||||
|
> provides the shared helpers — fonts, images, and composite layouts that
|
||||||
|
> scale. Existing plugins keep their classic rendering unless they adopt
|
||||||
|
> those APIs; nothing migrates automatically.
|
||||||
|
|
||||||
## Overview
|
## Overview
|
||||||
|
|
||||||
When developing plugins in separate repositories, you need a way to:
|
When developing plugins in separate repositories, you need a way to:
|
||||||
|
|||||||
@@ -90,11 +90,40 @@
|
|||||||
"min_height": {
|
"min_height": {
|
||||||
"type": "integer",
|
"type": "integer",
|
||||||
"minimum": 1
|
"minimum": 1
|
||||||
|
},
|
||||||
|
"max_width": {
|
||||||
|
"type": "integer",
|
||||||
|
"minimum": 1
|
||||||
|
},
|
||||||
|
"max_height": {
|
||||||
|
"type": "integer",
|
||||||
|
"minimum": 1
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
|
"display": {
|
||||||
|
"type": "object",
|
||||||
|
"properties": {
|
||||||
|
"design_size": {
|
||||||
|
"type": "object",
|
||||||
|
"properties": {
|
||||||
|
"width": {
|
||||||
|
"type": "integer",
|
||||||
|
"minimum": 8
|
||||||
|
},
|
||||||
|
"height": {
|
||||||
|
"type": "integer",
|
||||||
|
"minimum": 8
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"required": ["width", "height"],
|
||||||
|
"description": "Panel size the plugin's layout was authored against; core derives the adaptive-layout scale factor from it. Defaults to 128x32 when omitted."
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"description": "Display/layout hints for the adaptive layout system"
|
||||||
|
},
|
||||||
"config_schema": {
|
"config_schema": {
|
||||||
"type": "string",
|
"type": "string",
|
||||||
"description": "Path to configuration schema file"
|
"description": "Path to configuration schema file"
|
||||||
|
|||||||
+56
-20
@@ -37,10 +37,11 @@ os.environ['EMULATOR'] = 'true'
|
|||||||
|
|
||||||
from src.logging_config import get_logger # noqa: E402
|
from src.logging_config import get_logger # noqa: E402
|
||||||
from src.plugin_system.testing.loading import ( # noqa: E402
|
from src.plugin_system.testing.loading import ( # noqa: E402
|
||||||
find_plugin_dir, load_config_defaults, load_harness_spec,
|
find_plugin_dir, load_config_defaults, load_harness_spec, load_manifest,
|
||||||
)
|
)
|
||||||
from src.plugin_system.testing.harness import ( # noqa: E402
|
from src.plugin_system.testing.harness import ( # noqa: E402
|
||||||
RenderResult, render_plugin_matrix, compare_to_goldens, write_goldens,
|
RenderResult, render_plugin_matrix, compare_to_goldens, write_goldens,
|
||||||
|
check_scale_up,
|
||||||
)
|
)
|
||||||
from src.plugin_system.testing.sizes import ( # noqa: E402
|
from src.plugin_system.testing.sizes import ( # noqa: E402
|
||||||
parse_size_token, resolve_test_sizes, safe_mode_filename, size_label,
|
parse_size_token, resolve_test_sizes, safe_mode_filename, size_label,
|
||||||
@@ -110,28 +111,55 @@ def check_one(plugin_id: str, search_dirs: List[str], sizes, mock_data: Dict,
|
|||||||
effective_freeze = freeze_time or spec.get("freeze_time")
|
effective_freeze = freeze_time or spec.get("freeze_time")
|
||||||
effective_run_update = run_update and not spec.get("skip_update", False)
|
effective_run_update = run_update and not spec.get("skip_update", False)
|
||||||
|
|
||||||
results = render_plugin_matrix(
|
# The plugin's declared design size drives the scale-up fill check
|
||||||
plugin_id=plugin_id, plugin_dir=plugin_dir, config=full_config,
|
# (panels >= 2x the design size must not be left mostly empty).
|
||||||
mock_data=effective_mock_data, sizes=effective_sizes,
|
declared = load_manifest(plugin_dir).get("display", {}).get("design_size", {})
|
||||||
run_update=effective_run_update, freeze_time=effective_freeze,
|
design_size = (int(declared.get("width", 128)), int(declared.get("height", 32)))
|
||||||
)
|
fill_strict = spec.get("fill_check") == "strict"
|
||||||
|
|
||||||
golden_dir = golden_dir_override or (plugin_dir / 'test' / 'golden')
|
# Every run: the base config, plus one per harness.json "variant" —
|
||||||
if update_golden:
|
# a config overlay with its own golden dir (e.g. adaptive layout mode
|
||||||
written = write_goldens(results, golden_dir)
|
# tested alongside the classic default).
|
||||||
logger.info("Wrote %d golden image(s) for %s to %s", written, plugin_id, golden_dir)
|
runs = [(None, {}, golden_dir_override or (plugin_dir / 'test' / 'golden'))]
|
||||||
else:
|
for variant in spec.get("variants", []):
|
||||||
compare_to_goldens(results, golden_dir)
|
name = variant.get("name") or "variant"
|
||||||
|
vdir = plugin_dir / variant.get("golden_dir", f"test/golden-{name}")
|
||||||
|
runs.append((name, variant.get("config", {}), vdir))
|
||||||
|
|
||||||
if out_dir:
|
all_run_results: List[RenderResult] = []
|
||||||
for r in results:
|
for variant_name, overlay, golden_dir in runs:
|
||||||
if r.image is None:
|
run_config = {**full_config, **overlay}
|
||||||
continue
|
results = render_plugin_matrix(
|
||||||
dest = out_dir / plugin_id / size_label(r.width, r.height)
|
plugin_id=plugin_id, plugin_dir=plugin_dir, config=run_config,
|
||||||
dest.mkdir(parents=True, exist_ok=True)
|
mock_data=effective_mock_data, sizes=effective_sizes,
|
||||||
r.image.save(dest / f"{safe_mode_filename(r.mode)}.png", format="PNG")
|
run_update=effective_run_update, freeze_time=effective_freeze,
|
||||||
|
)
|
||||||
|
|
||||||
return results
|
if update_golden:
|
||||||
|
written = write_goldens(results, golden_dir)
|
||||||
|
logger.info("Wrote %d golden image(s) for %s%s to %s", written, plugin_id,
|
||||||
|
f" [{variant_name}]" if variant_name else "", golden_dir)
|
||||||
|
else:
|
||||||
|
compare_to_goldens(results, golden_dir)
|
||||||
|
|
||||||
|
check_scale_up(results, design_size=design_size, strict=fill_strict)
|
||||||
|
|
||||||
|
# Tag variant runs so the report and PNG dumps stay distinguishable.
|
||||||
|
if variant_name:
|
||||||
|
for r in results:
|
||||||
|
r.mode = f"{r.mode}@{variant_name}"
|
||||||
|
|
||||||
|
if out_dir:
|
||||||
|
for r in results:
|
||||||
|
if r.image is None:
|
||||||
|
continue
|
||||||
|
dest = out_dir / plugin_id / size_label(r.width, r.height)
|
||||||
|
dest.mkdir(parents=True, exist_ok=True)
|
||||||
|
r.image.save(dest / f"{safe_mode_filename(r.mode)}.png", format="PNG")
|
||||||
|
|
||||||
|
all_run_results.extend(results)
|
||||||
|
|
||||||
|
return all_run_results
|
||||||
|
|
||||||
|
|
||||||
def print_report(all_results: Dict[str, List[RenderResult]]) -> bool:
|
def print_report(all_results: Dict[str, List[RenderResult]]) -> bool:
|
||||||
@@ -147,6 +175,10 @@ def print_report(all_results: Dict[str, List[RenderResult]]) -> bool:
|
|||||||
detail = " (golden ✓)"
|
detail = " (golden ✓)"
|
||||||
if r.update_error is not None:
|
if r.update_error is not None:
|
||||||
detail += f" (update warn: {r.update_error})"
|
detail += f" (update warn: {r.update_error})"
|
||||||
|
if r.fill_checked and r.fill_ok is None and r.fill_extent:
|
||||||
|
# warn-only underfill: big panel left mostly empty
|
||||||
|
ex, ey = r.fill_extent
|
||||||
|
detail += f" (fill warn: extent {ex:.0%}x{ey:.0%})"
|
||||||
else:
|
else:
|
||||||
everything_ok = False
|
everything_ok = False
|
||||||
if r.error is not None:
|
if r.error is not None:
|
||||||
@@ -156,6 +188,10 @@ def print_report(all_results: Dict[str, List[RenderResult]]) -> bool:
|
|||||||
elif r.golden_ok is False:
|
elif r.golden_ok is False:
|
||||||
status = "FAIL"
|
status = "FAIL"
|
||||||
detail = f" golden drift: {r.golden_diff_pixels}px (max Δ={r.golden_max_delta})"
|
detail = f" golden drift: {r.golden_diff_pixels}px (max Δ={r.golden_max_delta})"
|
||||||
|
elif r.fill_ok is False:
|
||||||
|
ex, ey = r.fill_extent or (0.0, 0.0)
|
||||||
|
status = "FAIL"
|
||||||
|
detail = f" fill: extent {ex:.0%}x{ey:.0%} below required coverage"
|
||||||
else:
|
else:
|
||||||
status, detail = "FAIL", ""
|
status, detail = "FAIL", ""
|
||||||
print(f" [{status}] {r.size_label:>7} {r.mode}{detail}")
|
print(f" [{status}] {r.size_label:>7} {r.mode}{detail}")
|
||||||
|
|||||||
+198
-72
@@ -16,6 +16,7 @@ Opens at http://localhost:5001
|
|||||||
import sys
|
import sys
|
||||||
import os
|
import os
|
||||||
import json
|
import json
|
||||||
|
import re
|
||||||
import time
|
import time
|
||||||
import argparse
|
import argparse
|
||||||
import logging
|
import logging
|
||||||
@@ -44,6 +45,10 @@ MAX_HEIGHT = 512
|
|||||||
MIN_WIDTH = 1
|
MIN_WIDTH = 1
|
||||||
MIN_HEIGHT = 1
|
MIN_HEIGHT = 1
|
||||||
|
|
||||||
|
# plugin_id arrives in request input and is used to build filesystem paths —
|
||||||
|
# allowlist it (same pattern the web UI's pages_v3 uses)
|
||||||
|
_SAFE_PLUGIN_ID_RE = re.compile(r'^[a-zA-Z0-9_-]{1,64}$')
|
||||||
|
|
||||||
|
|
||||||
# --------------------------------------------------------------------------
|
# --------------------------------------------------------------------------
|
||||||
# Plugin discovery
|
# Plugin discovery
|
||||||
@@ -106,15 +111,30 @@ def discover_plugins() -> List[Dict[str, Any]]:
|
|||||||
|
|
||||||
|
|
||||||
def find_plugin_dir(plugin_id: str) -> Optional[Path]:
|
def find_plugin_dir(plugin_id: str) -> Optional[Path]:
|
||||||
"""Find a plugin directory by ID."""
|
"""Find a plugin directory by ID.
|
||||||
|
|
||||||
|
plugin_id comes from request input: it must pass an allowlist match,
|
||||||
|
and the resulting directory is normalized and required to live inside
|
||||||
|
one of the plugin search dirs, so a crafted id can never name a path
|
||||||
|
outside them.
|
||||||
|
"""
|
||||||
|
if not isinstance(plugin_id, str) or not _SAFE_PLUGIN_ID_RE.match(plugin_id):
|
||||||
|
return None
|
||||||
from src.plugin_system.plugin_loader import PluginLoader
|
from src.plugin_system.plugin_loader import PluginLoader
|
||||||
loader = PluginLoader()
|
loader = PluginLoader()
|
||||||
for search_dir in get_search_dirs():
|
for search_dir in get_search_dirs():
|
||||||
if not search_dir.exists():
|
if not search_dir.exists():
|
||||||
continue
|
continue
|
||||||
result = loader.find_plugin_directory(plugin_id, search_dir)
|
result = loader.find_plugin_directory(plugin_id, search_dir)
|
||||||
if result:
|
if not result:
|
||||||
return Path(result)
|
continue
|
||||||
|
# Normalize WITHOUT following symlinks (dev plugins are often
|
||||||
|
# symlinked into plugins/) and require lexical containment in the
|
||||||
|
# search dir, so no id can ever name a path outside it.
|
||||||
|
result_abs = os.path.abspath(str(result))
|
||||||
|
root_abs = os.path.abspath(str(search_dir))
|
||||||
|
if os.path.commonpath([result_abs, root_abs]) == root_abs:
|
||||||
|
return Path(result_abs)
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
@@ -176,6 +196,118 @@ def api_plugin_defaults(plugin_id):
|
|||||||
return jsonify({'defaults': defaults})
|
return jsonify({'defaults': defaults})
|
||||||
|
|
||||||
|
|
||||||
|
def _render_once(plugin_id, plugin_dir, manifest, config, mock_data, width, height,
|
||||||
|
skip_update):
|
||||||
|
"""Render one plugin at one size. Returns the /api/render response dict.
|
||||||
|
|
||||||
|
A fresh plugin instance per call, mirroring the safety harness, so sizes
|
||||||
|
never share state.
|
||||||
|
"""
|
||||||
|
from src.plugin_system.testing import VisualTestDisplayManager, MockCacheManager, MockPluginManager
|
||||||
|
from src.plugin_system.plugin_loader import PluginLoader
|
||||||
|
|
||||||
|
display_manager = VisualTestDisplayManager(width=width, height=height)
|
||||||
|
cache_manager = MockCacheManager()
|
||||||
|
plugin_manager = MockPluginManager()
|
||||||
|
|
||||||
|
# Pre-populate cache with mock data
|
||||||
|
for key, value in mock_data.items():
|
||||||
|
cache_manager.set(key, value)
|
||||||
|
|
||||||
|
loader = PluginLoader()
|
||||||
|
errors = []
|
||||||
|
warnings = []
|
||||||
|
|
||||||
|
plugin_instance, _module = loader.load_plugin(
|
||||||
|
plugin_id=plugin_id,
|
||||||
|
manifest=manifest,
|
||||||
|
plugin_dir=plugin_dir,
|
||||||
|
config=config,
|
||||||
|
display_manager=display_manager,
|
||||||
|
cache_manager=cache_manager,
|
||||||
|
plugin_manager=plugin_manager,
|
||||||
|
install_deps=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
start_time = time.time()
|
||||||
|
|
||||||
|
# Run update()
|
||||||
|
if not skip_update:
|
||||||
|
try:
|
||||||
|
plugin_instance.update()
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning("update() raised for plugin %s", plugin_id, exc_info=True)
|
||||||
|
warnings.append(f"update() raised: {type(e).__name__} — see server log")
|
||||||
|
|
||||||
|
# Run display()
|
||||||
|
try:
|
||||||
|
plugin_instance.display(force_clear=True)
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning("display() raised for plugin %s", plugin_id, exc_info=True)
|
||||||
|
errors.append(f"display() raised: {type(e).__name__} — see server log")
|
||||||
|
|
||||||
|
render_time_ms = round((time.time() - start_time) * 1000, 1)
|
||||||
|
|
||||||
|
return {
|
||||||
|
'image': f'data:image/png;base64,{display_manager.get_image_base64()}',
|
||||||
|
'width': width,
|
||||||
|
'height': height,
|
||||||
|
'render_time_ms': render_time_ms,
|
||||||
|
'errors': errors,
|
||||||
|
'warnings': warnings,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _trusted_plugin_dir(plugin_dir: Path) -> Optional[Path]:
|
||||||
|
"""Re-derive a plugin directory from the search dirs' own listings.
|
||||||
|
|
||||||
|
Path-injection barrier: unlike ``Path.iterdir()`` (which CodeQL doesn't
|
||||||
|
recognize as a taint-clearing enumeration), ``os.scandir()`` is. The
|
||||||
|
returned Path is built from a trusted root plus a name the filesystem
|
||||||
|
itself produced under that root via scandir — request-derived strings
|
||||||
|
never enter its construction — so a crafted plugin id can never make
|
||||||
|
downstream file access leave the plugin search dirs. Comparison is by
|
||||||
|
name, deliberately without symlink resolution (dev plugins are
|
||||||
|
commonly symlinked into plugins/).
|
||||||
|
"""
|
||||||
|
wanted_name = Path(os.path.normpath(str(plugin_dir))).name
|
||||||
|
for search_dir in get_search_dirs():
|
||||||
|
search_dir_str = str(search_dir)
|
||||||
|
try:
|
||||||
|
with os.scandir(search_dir_str) as entries:
|
||||||
|
for entry in entries:
|
||||||
|
if entry.name == wanted_name and entry.is_dir():
|
||||||
|
return Path(search_dir_str) / entry.name
|
||||||
|
except OSError:
|
||||||
|
continue
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _parse_render_request(data):
|
||||||
|
"""Shared /api/render* request prep. Returns (plugin_dir, manifest, config,
|
||||||
|
mock_data, skip_update) or raises ValueError with a client message."""
|
||||||
|
plugin_id = data['plugin_id']
|
||||||
|
candidate_dir = find_plugin_dir(plugin_id)
|
||||||
|
# Never reuse `candidate_dir` past this point: it's built from
|
||||||
|
# request-derived input, and a variable reassigned only on some paths
|
||||||
|
# isn't a barrier CodeQL's flow analysis honors. `trusted_dir` is the
|
||||||
|
# sole name used below, always the scandir-sourced result.
|
||||||
|
trusted_dir = _trusted_plugin_dir(candidate_dir) if candidate_dir else None
|
||||||
|
if not trusted_dir:
|
||||||
|
raise LookupError(f'Plugin not found: {plugin_id}')
|
||||||
|
|
||||||
|
manifest_path = trusted_dir / 'manifest.json'
|
||||||
|
with open(manifest_path, 'r') as f:
|
||||||
|
manifest = json.load(f)
|
||||||
|
|
||||||
|
# Build config: schema defaults + user overrides
|
||||||
|
config = {'enabled': True}
|
||||||
|
config.update(load_config_defaults(trusted_dir))
|
||||||
|
config.update(data.get('config', {}))
|
||||||
|
|
||||||
|
return trusted_dir, manifest, config, data.get('mock_data', {}), data.get('skip_update', False)
|
||||||
|
|
||||||
|
|
||||||
@app.route('/api/render', methods=['POST'])
|
@app.route('/api/render', methods=['POST'])
|
||||||
def api_render():
|
def api_render():
|
||||||
"""Render a plugin and return the display as base64 PNG."""
|
"""Render a plugin and return the display as base64 PNG."""
|
||||||
@@ -183,11 +315,6 @@ def api_render():
|
|||||||
if not data or 'plugin_id' not in data:
|
if not data or 'plugin_id' not in data:
|
||||||
return jsonify({'error': 'plugin_id is required'}), 400
|
return jsonify({'error': 'plugin_id is required'}), 400
|
||||||
|
|
||||||
plugin_id = data['plugin_id']
|
|
||||||
user_config = data.get('config', {})
|
|
||||||
mock_data = data.get('mock_data', {})
|
|
||||||
skip_update = data.get('skip_update', False)
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
width = int(data.get('width', 128))
|
width = int(data.get('width', 128))
|
||||||
height = int(data.get('height', 32))
|
height = int(data.get('height', 32))
|
||||||
@@ -199,78 +326,77 @@ def api_render():
|
|||||||
if not (MIN_HEIGHT <= height <= MAX_HEIGHT):
|
if not (MIN_HEIGHT <= height <= MAX_HEIGHT):
|
||||||
return jsonify({'error': f'height must be between {MIN_HEIGHT} and {MAX_HEIGHT}'}), 400
|
return jsonify({'error': f'height must be between {MIN_HEIGHT} and {MAX_HEIGHT}'}), 400
|
||||||
|
|
||||||
# Find plugin
|
try:
|
||||||
plugin_dir = find_plugin_dir(plugin_id)
|
plugin_dir, manifest, config, mock_data, skip_update = _parse_render_request(data)
|
||||||
if not plugin_dir:
|
except LookupError:
|
||||||
return jsonify({'error': f'Plugin not found: {plugin_id}'}), 404
|
return jsonify({'error': f"Plugin not found: {data['plugin_id']}"}), 404
|
||||||
|
except Exception:
|
||||||
# Load manifest
|
# Bad manifest.json / schema / fixture — details go to the dev's
|
||||||
manifest_path = plugin_dir / 'manifest.json'
|
# console, not the HTTP response
|
||||||
with open(manifest_path, 'r') as f:
|
app.logger.exception('render request preparation failed')
|
||||||
manifest = json.load(f)
|
return jsonify({'error': 'Could not prepare render request; see server log'}), 400
|
||||||
|
|
||||||
# Build config: schema defaults + user overrides
|
|
||||||
config_defaults = load_config_defaults(plugin_dir)
|
|
||||||
config = {'enabled': True}
|
|
||||||
config.update(config_defaults)
|
|
||||||
config.update(user_config)
|
|
||||||
|
|
||||||
# Create display manager and mocks
|
|
||||||
from src.plugin_system.testing import VisualTestDisplayManager, MockCacheManager, MockPluginManager
|
|
||||||
from src.plugin_system.plugin_loader import PluginLoader
|
|
||||||
|
|
||||||
display_manager = VisualTestDisplayManager(width=width, height=height)
|
|
||||||
cache_manager = MockCacheManager()
|
|
||||||
plugin_manager = MockPluginManager()
|
|
||||||
|
|
||||||
# Pre-populate cache with mock data
|
|
||||||
for key, value in mock_data.items():
|
|
||||||
cache_manager.set(key, value)
|
|
||||||
|
|
||||||
# Load plugin
|
|
||||||
loader = PluginLoader()
|
|
||||||
errors = []
|
|
||||||
warnings = []
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
plugin_instance, module = loader.load_plugin(
|
result = _render_once(data['plugin_id'], plugin_dir, manifest, config,
|
||||||
plugin_id=plugin_id,
|
mock_data, width, height, skip_update)
|
||||||
manifest=manifest,
|
except Exception:
|
||||||
plugin_dir=plugin_dir,
|
app.logger.exception('plugin load failed during render')
|
||||||
config=config,
|
return jsonify({'error': 'Failed to load plugin; see server log'}), 500
|
||||||
display_manager=display_manager,
|
return jsonify(result)
|
||||||
cache_manager=cache_manager,
|
|
||||||
plugin_manager=plugin_manager,
|
|
||||||
install_deps=False,
|
|
||||||
)
|
|
||||||
except Exception as e:
|
|
||||||
return jsonify({'error': f'Failed to load plugin: {e}'}), 500
|
|
||||||
|
|
||||||
start_time = time.time()
|
|
||||||
|
|
||||||
# Run update()
|
@app.route('/api/sizes')
|
||||||
if not skip_update:
|
def api_sizes():
|
||||||
|
"""The representative panel-size sample the safety harness renders at."""
|
||||||
|
from src.plugin_system.testing.sizes import DEFAULT_TEST_SIZES
|
||||||
|
return jsonify({'sizes': [list(s) for s in DEFAULT_TEST_SIZES]})
|
||||||
|
|
||||||
|
|
||||||
|
MAX_MATRIX_SIZES = 12
|
||||||
|
|
||||||
|
|
||||||
|
@app.route('/api/render-matrix', methods=['POST'])
|
||||||
|
def api_render_matrix():
|
||||||
|
"""Render a plugin at a list of sizes (default: the harness sample) so the
|
||||||
|
UI can show a side-by-side multi-resolution gallery."""
|
||||||
|
data = request.get_json()
|
||||||
|
if not data or 'plugin_id' not in data:
|
||||||
|
return jsonify({'error': 'plugin_id is required'}), 400
|
||||||
|
|
||||||
|
from src.plugin_system.testing.sizes import DEFAULT_TEST_SIZES
|
||||||
|
sizes = data.get('sizes') or [list(s) for s in DEFAULT_TEST_SIZES]
|
||||||
|
if len(sizes) > MAX_MATRIX_SIZES:
|
||||||
|
return jsonify({'error': f'at most {MAX_MATRIX_SIZES} sizes per request'}), 400
|
||||||
|
parsed_sizes = []
|
||||||
|
for pair in sizes:
|
||||||
try:
|
try:
|
||||||
plugin_instance.update()
|
w, h = int(pair[0]), int(pair[1])
|
||||||
except Exception as e:
|
except (TypeError, ValueError, IndexError):
|
||||||
warnings.append(f"update() raised: {e}")
|
return jsonify({'error': f'invalid size entry {pair!r} (expected [w, h])'}), 400
|
||||||
|
if not (MIN_WIDTH <= w <= MAX_WIDTH and MIN_HEIGHT <= h <= MAX_HEIGHT):
|
||||||
|
return jsonify({'error': f'size {w}x{h} out of bounds'}), 400
|
||||||
|
parsed_sizes.append((w, h))
|
||||||
|
|
||||||
# Run display()
|
|
||||||
try:
|
try:
|
||||||
plugin_instance.display(force_clear=True)
|
plugin_dir, manifest, config, mock_data, skip_update = _parse_render_request(data)
|
||||||
except Exception as e:
|
except LookupError:
|
||||||
errors.append(f"display() raised: {e}")
|
return jsonify({'error': f"Plugin not found: {data['plugin_id']}"}), 404
|
||||||
|
except Exception:
|
||||||
|
app.logger.exception('render request preparation failed')
|
||||||
|
return jsonify({'error': 'Could not prepare render request; see server log'}), 400
|
||||||
|
|
||||||
render_time_ms = round((time.time() - start_time) * 1000, 1)
|
results = []
|
||||||
|
for w, h in parsed_sizes:
|
||||||
return jsonify({
|
try:
|
||||||
'image': f'data:image/png;base64,{display_manager.get_image_base64()}',
|
results.append(_render_once(data['plugin_id'], plugin_dir, manifest,
|
||||||
'width': width,
|
config, mock_data, w, h, skip_update))
|
||||||
'height': height,
|
except Exception:
|
||||||
'render_time_ms': render_time_ms,
|
app.logger.exception('plugin load failed during %dx%d render', w, h)
|
||||||
'errors': errors,
|
results.append({'image': None, 'width': w, 'height': h,
|
||||||
'warnings': warnings,
|
'render_time_ms': 0,
|
||||||
})
|
'errors': ['Failed to load plugin; see server log'],
|
||||||
|
'warnings': []})
|
||||||
|
return jsonify({'results': results})
|
||||||
|
|
||||||
|
|
||||||
# --------------------------------------------------------------------------
|
# --------------------------------------------------------------------------
|
||||||
|
|||||||
@@ -209,6 +209,11 @@
|
|||||||
onchange="onConfigChange()">
|
onchange="onConfigChange()">
|
||||||
<span class="text-xs ml-2" style="color: var(--text-secondary);">px</span>
|
<span class="text-xs ml-2" style="color: var(--text-secondary);">px</span>
|
||||||
</div>
|
</div>
|
||||||
|
<select id="sizePreset" onchange="applySizePreset()"
|
||||||
|
class="w-full mt-2 px-2 py-1.5 rounded text-xs"
|
||||||
|
style="background: var(--bg-primary); color: var(--text-secondary); border: 1px solid var(--border-color);">
|
||||||
|
<option value="">Preset sizes…</option>
|
||||||
|
</select>
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
<!-- Config form -->
|
<!-- Config form -->
|
||||||
@@ -242,13 +247,18 @@
|
|||||||
</div>
|
</div>
|
||||||
</details>
|
</details>
|
||||||
|
|
||||||
<!-- Render button -->
|
<!-- Render buttons -->
|
||||||
<div class="flex gap-2">
|
<div class="flex gap-2">
|
||||||
<button onclick="renderPlugin()" id="renderBtn"
|
<button onclick="renderPlugin()" id="renderBtn"
|
||||||
class="flex-1 px-4 py-2.5 rounded-lg text-sm font-medium text-white"
|
class="flex-1 px-4 py-2.5 rounded-lg text-sm font-medium text-white"
|
||||||
style="background: var(--accent);">
|
style="background: var(--accent);">
|
||||||
Render
|
Render
|
||||||
</button>
|
</button>
|
||||||
|
<button onclick="renderAllSizes()" id="renderAllBtn" title="Render at every harness test size"
|
||||||
|
class="px-4 py-2.5 rounded-lg text-sm font-medium"
|
||||||
|
style="background: var(--bg-tertiary); color: var(--text-primary); border: 1px solid var(--border-color);">
|
||||||
|
All Sizes
|
||||||
|
</button>
|
||||||
</div>
|
</div>
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
@@ -311,6 +321,15 @@
|
|||||||
<div id="messagesPanel" class="panel p-3 hidden">
|
<div id="messagesPanel" class="panel p-3 hidden">
|
||||||
<div id="messagesList" class="text-xs font-mono space-y-1"></div>
|
<div id="messagesList" class="text-xs font-mono space-y-1"></div>
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
|
<!-- Multi-size gallery -->
|
||||||
|
<div id="galleryPanel" class="panel p-4 hidden">
|
||||||
|
<div class="flex items-center justify-between mb-3">
|
||||||
|
<span class="text-xs font-medium" style="color: var(--text-secondary);">All Sizes</span>
|
||||||
|
<span class="text-xs" style="color: var(--text-secondary);" id="galleryStatus"></span>
|
||||||
|
</div>
|
||||||
|
<div id="galleryGrid" class="flex flex-wrap gap-4 items-start"></div>
|
||||||
|
</div>
|
||||||
</div>
|
</div>
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
@@ -340,8 +359,30 @@
|
|||||||
opt.textContent = `${p.name} (${p.id})`;
|
opt.textContent = `${p.name} (${p.id})`;
|
||||||
select.appendChild(opt);
|
select.appendChild(opt);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
// Load harness size presets
|
||||||
|
try {
|
||||||
|
const sizesRes = await fetch('/api/sizes');
|
||||||
|
const sizesData = await sizesRes.json();
|
||||||
|
const preset = document.getElementById('sizePreset');
|
||||||
|
(sizesData.sizes || []).forEach(([w, h]) => {
|
||||||
|
const opt = document.createElement('option');
|
||||||
|
opt.value = `${w}x${h}`;
|
||||||
|
opt.textContent = `${w} x ${h}`;
|
||||||
|
preset.appendChild(opt);
|
||||||
|
});
|
||||||
|
} catch (e) { /* presets are a convenience; ignore */ }
|
||||||
});
|
});
|
||||||
|
|
||||||
|
function applySizePreset() {
|
||||||
|
const value = document.getElementById('sizePreset').value;
|
||||||
|
if (!value) return;
|
||||||
|
const [w, h] = value.split('x');
|
||||||
|
document.getElementById('displayWidth').value = w;
|
||||||
|
document.getElementById('displayHeight').value = h;
|
||||||
|
onConfigChange();
|
||||||
|
}
|
||||||
|
|
||||||
// ---------- Plugin selection ----------
|
// ---------- Plugin selection ----------
|
||||||
async function onPluginChange() {
|
async function onPluginChange() {
|
||||||
const pluginId = document.getElementById('pluginSelect').value;
|
const pluginId = document.getElementById('pluginSelect').value;
|
||||||
@@ -485,6 +526,89 @@
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ---------- Multi-size gallery ----------
|
||||||
|
async function renderAllSizes() {
|
||||||
|
if (!currentPluginId) return;
|
||||||
|
|
||||||
|
const btn = document.getElementById('renderAllBtn');
|
||||||
|
const panel = document.getElementById('galleryPanel');
|
||||||
|
const grid = document.getElementById('galleryGrid');
|
||||||
|
const status = document.getElementById('galleryStatus');
|
||||||
|
btn.disabled = true;
|
||||||
|
btn.textContent = 'Rendering…';
|
||||||
|
panel.classList.remove('hidden');
|
||||||
|
grid.innerHTML = '';
|
||||||
|
status.textContent = 'Rendering at all harness sizes…';
|
||||||
|
|
||||||
|
const config = jsonEditor ? jsonEditor.getValue() : {};
|
||||||
|
config.enabled = true;
|
||||||
|
let mockData = {};
|
||||||
|
const mockInput = document.getElementById('mockDataInput').value.trim();
|
||||||
|
if (mockInput) {
|
||||||
|
try { mockData = JSON.parse(mockInput); }
|
||||||
|
catch (e) { showMessages([], [`Mock data JSON error: ${e.message}`]); }
|
||||||
|
}
|
||||||
|
|
||||||
|
try {
|
||||||
|
const res = await fetch('/api/render-matrix', {
|
||||||
|
method: 'POST',
|
||||||
|
headers: { 'Content-Type': 'application/json' },
|
||||||
|
body: JSON.stringify({
|
||||||
|
plugin_id: currentPluginId,
|
||||||
|
config: config,
|
||||||
|
mock_data: mockData,
|
||||||
|
}),
|
||||||
|
});
|
||||||
|
const data = await res.json();
|
||||||
|
if (data.error) {
|
||||||
|
status.textContent = data.error;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
let failures = 0;
|
||||||
|
(data.results || []).forEach(r => {
|
||||||
|
const cell = document.createElement('div');
|
||||||
|
cell.style.cssText = 'display:flex;flex-direction:column;gap:4px;';
|
||||||
|
const failed = (r.errors || []).length > 0 || !r.image;
|
||||||
|
if (failed) failures++;
|
||||||
|
|
||||||
|
const label = document.createElement('span');
|
||||||
|
label.className = 'text-xs font-mono';
|
||||||
|
label.style.color = failed ? '#f87171' : 'var(--text-secondary)';
|
||||||
|
label.textContent = `${r.width}x${r.height} · ${r.render_time_ms}ms`;
|
||||||
|
cell.appendChild(label);
|
||||||
|
|
||||||
|
if (r.image) {
|
||||||
|
const img = document.createElement('img');
|
||||||
|
img.src = r.image;
|
||||||
|
// Small panels get 2x zoom so they stay legible in the grid
|
||||||
|
const zoom = r.height >= 128 ? 1 : 2;
|
||||||
|
img.style.cssText =
|
||||||
|
`image-rendering: pixelated; width:${r.width * zoom}px; ` +
|
||||||
|
`height:${r.height * zoom}px; ` +
|
||||||
|
`border:1px solid ${failed ? '#f87171' : 'var(--border-color)'};`;
|
||||||
|
cell.appendChild(img);
|
||||||
|
}
|
||||||
|
if (failed) {
|
||||||
|
const err = document.createElement('span');
|
||||||
|
err.className = 'text-xs font-mono';
|
||||||
|
err.style.color = '#f87171';
|
||||||
|
err.textContent = (r.errors || ['render failed']).join('; ');
|
||||||
|
cell.appendChild(err);
|
||||||
|
}
|
||||||
|
grid.appendChild(cell);
|
||||||
|
});
|
||||||
|
status.textContent = failures
|
||||||
|
? `${failures} size(s) failed`
|
||||||
|
: `${(data.results || []).length} sizes rendered`;
|
||||||
|
} catch (e) {
|
||||||
|
status.textContent = `Network error: ${e.message}`;
|
||||||
|
} finally {
|
||||||
|
btn.disabled = false;
|
||||||
|
btn.textContent = 'All Sizes';
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// ---------- Zoom ----------
|
// ---------- Zoom ----------
|
||||||
function updateZoom() {
|
function updateZoom() {
|
||||||
const zoom = parseInt(document.getElementById('zoomSlider').value);
|
const zoom = parseInt(document.getElementById('zoomSlider').value);
|
||||||
|
|||||||
+1
-1
@@ -4,5 +4,5 @@ LEDMatrix Display System
|
|||||||
Core source package for the LED Matrix Display project.
|
Core source package for the LED Matrix Display project.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
__version__ = "1.0.0"
|
__version__ = "3.1.0"
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,174 @@
|
|||||||
|
"""
|
||||||
|
Adaptive image fitting for plugins — the image counterpart to
|
||||||
|
src/adaptive_layout.py's text fitting.
|
||||||
|
|
||||||
|
Promotes the proven in-field image patterns into one shared helper so
|
||||||
|
plugins stop hand-copying resize/cache code:
|
||||||
|
|
||||||
|
- "crop transparent padding, then fill the row height" (football/hockey
|
||||||
|
logo pattern) -> ``crop_to_ink=True, mode="fill_height"``
|
||||||
|
- "crop-to-fill with a top anchor for faces" (masters-tournament headshot
|
||||||
|
pattern) -> ``mode="cover", anchor="top"``
|
||||||
|
- "letterbox to fit, centered on a background" (static-image pattern)
|
||||||
|
-> ``mode="contain"``
|
||||||
|
- NEAREST for pixel art/flags vs LANCZOS for photos (masters flag pattern)
|
||||||
|
-> ``resample=RESAMPLE_NEAREST``
|
||||||
|
|
||||||
|
Unlike PIL's ``thumbnail()`` (downscale-only — the reason plugin imagery
|
||||||
|
stays tiny on big panels), ``fit_image`` upscales by default so content
|
||||||
|
genuinely adapts to larger displays; pass ``upscale=False`` for the old
|
||||||
|
behavior.
|
||||||
|
|
||||||
|
Use via ``LayoutContext.fit_image(...)`` (cached per panel size) or
|
||||||
|
``BasePlugin.draw_image(...)``; the module-level functions are the
|
||||||
|
uncached primitives.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from typing import Any, Optional, Tuple
|
||||||
|
|
||||||
|
from PIL import Image
|
||||||
|
|
||||||
|
# The one Pillow >= 9.1 compat shim (replaces the per-plugin copies).
|
||||||
|
try:
|
||||||
|
RESAMPLE_LANCZOS = Image.Resampling.LANCZOS
|
||||||
|
RESAMPLE_NEAREST = Image.Resampling.NEAREST
|
||||||
|
except AttributeError: # Pillow < 9.1
|
||||||
|
RESAMPLE_LANCZOS = Image.LANCZOS
|
||||||
|
RESAMPLE_NEAREST = Image.NEAREST
|
||||||
|
|
||||||
|
FIT_MODES = ("contain", "cover", "fill_height", "stretch")
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class ImageFitResult:
|
||||||
|
"""A processed RGBA copy of a source image, sized for a target box."""
|
||||||
|
image: Image.Image
|
||||||
|
width: int
|
||||||
|
height: int
|
||||||
|
scale: float # scale applied vs the (possibly ink-cropped) source
|
||||||
|
mode: str
|
||||||
|
source_size: Tuple[int, int]
|
||||||
|
|
||||||
|
@property
|
||||||
|
def is_empty(self) -> bool:
|
||||||
|
return self.width <= 0 or self.height <= 0
|
||||||
|
|
||||||
|
|
||||||
|
_EMPTY_IMAGE = Image.new("RGBA", (1, 1), (0, 0, 0, 0))
|
||||||
|
|
||||||
|
|
||||||
|
def _empty_result(mode: str, source_size: Tuple[int, int]) -> ImageFitResult:
|
||||||
|
return ImageFitResult(_EMPTY_IMAGE, 0, 0, 0.0, mode, source_size)
|
||||||
|
|
||||||
|
|
||||||
|
def _box_dims(box: Any) -> Tuple[int, int]:
|
||||||
|
"""Accept a Region (duck-typed .w/.h) or a (w, h) tuple."""
|
||||||
|
if hasattr(box, "w") and hasattr(box, "h"):
|
||||||
|
return (int(box.w), int(box.h))
|
||||||
|
w, h = box
|
||||||
|
return (int(w), int(h))
|
||||||
|
|
||||||
|
|
||||||
|
def fit_image(img: Image.Image, box: Any, *, mode: str = "contain",
|
||||||
|
crop_to_ink: bool = False, anchor: str = "center",
|
||||||
|
resample: Any = None, upscale: bool = True) -> ImageFitResult:
|
||||||
|
"""Fit an image into a box, preserving crispness policy per content type.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
img: Source PIL image (any mode; output is always RGBA).
|
||||||
|
box: Region or (w, h) target box.
|
||||||
|
mode: "contain" (letterbox), "cover" (crop-to-fill),
|
||||||
|
"fill_height" (height == box height, contain-capped by width),
|
||||||
|
"stretch" (exact resize).
|
||||||
|
crop_to_ink: Trim fully-transparent padding (getbbox) before fitting —
|
||||||
|
logos shipped with generous padding otherwise render small.
|
||||||
|
anchor: For "cover" crops: "center" or "top" (keeps faces/tops).
|
||||||
|
resample: PIL resampling filter; defaults to RESAMPLE_LANCZOS.
|
||||||
|
Use RESAMPLE_NEAREST for pixel art, flags, and sprite icons.
|
||||||
|
upscale: Allow scaling above source size (default True — the adaptive
|
||||||
|
point). False mimics the legacy thumbnail() behavior.
|
||||||
|
"""
|
||||||
|
if mode not in FIT_MODES:
|
||||||
|
raise ValueError(f"Unknown fit mode '{mode}' (expected one of {FIT_MODES})")
|
||||||
|
box_w, box_h = _box_dims(box)
|
||||||
|
if box_w <= 0 or box_h <= 0 or img.width <= 0 or img.height <= 0:
|
||||||
|
return _empty_result(mode, img.size)
|
||||||
|
|
||||||
|
resample = RESAMPLE_LANCZOS if resample is None else resample
|
||||||
|
|
||||||
|
work = img if img.mode == "RGBA" else img.convert("RGBA")
|
||||||
|
if crop_to_ink:
|
||||||
|
bbox = work.getbbox()
|
||||||
|
if bbox is None: # fully transparent
|
||||||
|
return _empty_result(mode, img.size)
|
||||||
|
work = work.crop(bbox)
|
||||||
|
|
||||||
|
src_w, src_h = work.size
|
||||||
|
|
||||||
|
if mode == "stretch":
|
||||||
|
out = work.resize((box_w, box_h), resample)
|
||||||
|
return ImageFitResult(out, box_w, box_h, box_w / src_w, mode, (src_w, src_h))
|
||||||
|
|
||||||
|
if mode == "cover":
|
||||||
|
scale = max(box_w / src_w, box_h / src_h)
|
||||||
|
if not upscale:
|
||||||
|
scale = min(scale, 1.0)
|
||||||
|
scaled_w = max(1, round(src_w * scale))
|
||||||
|
scaled_h = max(1, round(src_h * scale))
|
||||||
|
out = work.resize((scaled_w, scaled_h), resample)
|
||||||
|
# Crop the overhang down to the box (only when the scaled image is
|
||||||
|
# larger; with upscale=False it may be smaller and is left as-is).
|
||||||
|
crop_w, crop_h = min(box_w, scaled_w), min(box_h, scaled_h)
|
||||||
|
left = (scaled_w - crop_w) // 2
|
||||||
|
top = 0 if anchor == "top" else (scaled_h - crop_h) // 2
|
||||||
|
out = out.crop((left, top, left + crop_w, top + crop_h))
|
||||||
|
return ImageFitResult(out, out.width, out.height, scale, mode, (src_w, src_h))
|
||||||
|
|
||||||
|
# contain / fill_height share the "preserve aspect, no crop" path
|
||||||
|
if mode == "fill_height":
|
||||||
|
scale = box_h / src_h
|
||||||
|
# contain-cap: never exceed the box width (football's logo_slot rule)
|
||||||
|
scale = min(scale, box_w / src_w)
|
||||||
|
else: # contain
|
||||||
|
scale = min(box_w / src_w, box_h / src_h)
|
||||||
|
if not upscale:
|
||||||
|
scale = min(scale, 1.0)
|
||||||
|
out_w = max(1, round(src_w * scale))
|
||||||
|
out_h = max(1, round(src_h * scale))
|
||||||
|
if (out_w, out_h) == (src_w, src_h):
|
||||||
|
# No resize needed — but `work` may still BE the caller's original
|
||||||
|
# image (RGBA source, no ink crop). The result must always be an
|
||||||
|
# independent copy: LayoutContext caches ImageFitResults, and an
|
||||||
|
# aliased image would let later mutations of the source corrupt
|
||||||
|
# cached fits (or vice versa).
|
||||||
|
out = work.copy() if work is img else work
|
||||||
|
else:
|
||||||
|
out = work.resize((out_w, out_h), resample)
|
||||||
|
return ImageFitResult(out, out_w, out_h, scale, mode, (src_w, src_h))
|
||||||
|
|
||||||
|
|
||||||
|
def draw_fitted_image(display_manager: Any, ifit: ImageFitResult, box: Any, *,
|
||||||
|
align: str = "center", valign: str = "center",
|
||||||
|
offset: Tuple[int, int] = (0, 0)) -> Optional[Tuple[int, int]]:
|
||||||
|
"""Paste a fitted image aligned within a Region onto the display canvas.
|
||||||
|
|
||||||
|
Pastes with the image's own alpha mask. Returns the (x, y) actually used
|
||||||
|
so callers can position adjacent decorations, or None when nothing was
|
||||||
|
drawn (empty fit / no canvas).
|
||||||
|
"""
|
||||||
|
if ifit is None or ifit.is_empty:
|
||||||
|
return None
|
||||||
|
image = getattr(display_manager, "image", None)
|
||||||
|
if image is None:
|
||||||
|
return None
|
||||||
|
if hasattr(box, "align_xy"):
|
||||||
|
x, y = box.align_xy(ifit.width, ifit.height, align, valign)
|
||||||
|
else:
|
||||||
|
box_w, box_h = _box_dims(box)
|
||||||
|
x = (box_w - ifit.width) // 2
|
||||||
|
y = (box_h - ifit.height) // 2
|
||||||
|
x += int(offset[0])
|
||||||
|
y += int(offset[1])
|
||||||
|
image.paste(ifit.image, (x, y), ifit.image)
|
||||||
|
return (x, y)
|
||||||
@@ -0,0 +1,746 @@
|
|||||||
|
"""
|
||||||
|
Adaptive layout and font scaling helpers for plugins.
|
||||||
|
|
||||||
|
Generalizes the three size-adaptation patterns proven in the plugin
|
||||||
|
ecosystem into small composable core helpers, so plugins render legibly on
|
||||||
|
any panel size (64x32, 128x32, 96x48, 128x64, 256x64, ...) without
|
||||||
|
hand-tuned per-display layouts:
|
||||||
|
|
||||||
|
- Region: integer rect algebra (bands, columns, weighted splits, centering).
|
||||||
|
Regions partition space, so text bands can't overlap by construction —
|
||||||
|
replacing the magic ``y = 1`` / ``y = height - 7`` offsets tuned for 128x32.
|
||||||
|
- Font ladders: ordered (family, size) steps known to render crisply.
|
||||||
|
Pixel fonts (BDF, PressStart2P) only look right at native/integer sizes,
|
||||||
|
so fonts are never scaled continuously — fitting walks a ladder from the
|
||||||
|
largest rung down until the measured text fits the target box. This is
|
||||||
|
baseball-scoreboard's fallback-ladder pattern promoted to core.
|
||||||
|
- LayoutContext: per-(width, height) facts — breakpoint tiers
|
||||||
|
(masters-tournament's pattern), a geometry scale factor vs. a declared
|
||||||
|
design size (f1-scoreboard's pattern), and cached fit-text queries.
|
||||||
|
|
||||||
|
Everything is opt-in: plugins get a context via ``self.layout`` on
|
||||||
|
BasePlugin (or construct one directly) and existing plugins are unaffected.
|
||||||
|
|
||||||
|
Fonts are resolved through FontManager's catalog (family names are
|
||||||
|
lowercased file stems from assets/fonts, e.g. "9x15", "tom-thumb", plus
|
||||||
|
aliases like "press_start"). FitResult.font is a plain PIL font or
|
||||||
|
freetype.Face, so it drops straight into DisplayManager.draw_text().
|
||||||
|
"""
|
||||||
|
|
||||||
|
import logging
|
||||||
|
from collections import OrderedDict
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
|
||||||
|
|
||||||
|
import freetype
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
# Height-based breakpoint tiers, smallest to largest. A 32px-tall panel is
|
||||||
|
# the ecosystem baseline ("sm"); 96x48 lands in "md"; 128x64 in "lg".
|
||||||
|
_HEIGHT_TIERS: Tuple[Tuple[str, int], ...] = (
|
||||||
|
("xs", 16), ("sm", 32), ("md", 48), ("lg", 64), ("xl", 10 ** 9),
|
||||||
|
)
|
||||||
|
TIER_ORDER: Tuple[str, ...] = tuple(name for name, _ in _HEIGHT_TIERS)
|
||||||
|
|
||||||
|
_WIDTH_TIERS: Tuple[Tuple[str, int], ...] = (
|
||||||
|
("narrow", 64), ("normal", 128), ("wide", 256), ("ultrawide", 10 ** 9),
|
||||||
|
)
|
||||||
|
WIDTH_TIER_ORDER: Tuple[str, ...] = tuple(name for name, _ in _WIDTH_TIERS)
|
||||||
|
|
||||||
|
# The panel size most existing plugins were authored against.
|
||||||
|
DEFAULT_DESIGN_SIZE: Tuple[int, int] = (128, 32)
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class Region:
|
||||||
|
"""An integer rectangle. Carving methods return sub-Regions clamped to
|
||||||
|
non-negative dimensions, so degenerate panels never produce negative
|
||||||
|
boxes — a band request larger than the region simply consumes it all."""
|
||||||
|
|
||||||
|
x: int
|
||||||
|
y: int
|
||||||
|
w: int
|
||||||
|
h: int
|
||||||
|
|
||||||
|
def __post_init__(self):
|
||||||
|
object.__setattr__(self, "w", max(0, int(self.w)))
|
||||||
|
object.__setattr__(self, "h", max(0, int(self.h)))
|
||||||
|
object.__setattr__(self, "x", int(self.x))
|
||||||
|
object.__setattr__(self, "y", int(self.y))
|
||||||
|
|
||||||
|
@property
|
||||||
|
def right(self) -> int:
|
||||||
|
return self.x + self.w
|
||||||
|
|
||||||
|
@property
|
||||||
|
def bottom(self) -> int:
|
||||||
|
return self.y + self.h
|
||||||
|
|
||||||
|
@property
|
||||||
|
def center(self) -> Tuple[int, int]:
|
||||||
|
return (self.x + self.w // 2, self.y + self.h // 2)
|
||||||
|
|
||||||
|
# ---- carving -----------------------------------------------------
|
||||||
|
|
||||||
|
def inset(self, dx: int, dy: Optional[int] = None) -> "Region":
|
||||||
|
"""Shrink by dx horizontally and dy (default dx) vertically, each side."""
|
||||||
|
if dy is None:
|
||||||
|
dy = dx
|
||||||
|
return Region(self.x + dx, self.y + dy, self.w - 2 * dx, self.h - 2 * dy)
|
||||||
|
|
||||||
|
def offset(self, dx: int, dy: int) -> "Region":
|
||||||
|
"""Translate without resizing — the hook for user x/y-offset
|
||||||
|
customization: compute regions first, then apply the user's
|
||||||
|
configured offsets as a final translation."""
|
||||||
|
return Region(self.x + dx, self.y + dy, self.w, self.h)
|
||||||
|
|
||||||
|
def top_band(self, h: int) -> "Region":
|
||||||
|
return Region(self.x, self.y, self.w, min(h, self.h))
|
||||||
|
|
||||||
|
def bottom_band(self, h: int) -> "Region":
|
||||||
|
h = min(h, self.h)
|
||||||
|
return Region(self.x, self.bottom - h, self.w, h)
|
||||||
|
|
||||||
|
def middle(self, top_h: int = 0, bottom_h: int = 0) -> "Region":
|
||||||
|
"""What remains between a top band and a bottom band."""
|
||||||
|
return Region(self.x, self.y + top_h, self.w, self.h - top_h - bottom_h)
|
||||||
|
|
||||||
|
def left_col(self, w: int) -> "Region":
|
||||||
|
return Region(self.x, self.y, min(w, self.w), self.h)
|
||||||
|
|
||||||
|
def right_col(self, w: int) -> "Region":
|
||||||
|
w = min(w, self.w)
|
||||||
|
return Region(self.right - w, self.y, w, self.h)
|
||||||
|
|
||||||
|
def split_h(self, *weights: float, gap: int = 0) -> List["Region"]:
|
||||||
|
"""Side-by-side columns sized by weight; gaps between them."""
|
||||||
|
sizes = _weighted_sizes(self.w, weights, gap)
|
||||||
|
cols, cursor = [], self.x
|
||||||
|
for size in sizes:
|
||||||
|
cols.append(Region(cursor, self.y, size, self.h))
|
||||||
|
cursor += size + gap
|
||||||
|
return cols
|
||||||
|
|
||||||
|
def split_v(self, *weights: float, gap: int = 0) -> List["Region"]:
|
||||||
|
"""Stacked rows sized by weight; gaps between them."""
|
||||||
|
sizes = _weighted_sizes(self.h, weights, gap)
|
||||||
|
rows, cursor = [], self.y
|
||||||
|
for size in sizes:
|
||||||
|
rows.append(Region(self.x, cursor, self.w, size))
|
||||||
|
cursor += size + gap
|
||||||
|
return rows
|
||||||
|
|
||||||
|
# ---- placement ---------------------------------------------------
|
||||||
|
|
||||||
|
def align_xy(self, w: int, h: int, align: str = "center",
|
||||||
|
valign: str = "center") -> Tuple[int, int]:
|
||||||
|
"""Top-left position for a w x h box aligned within this region.
|
||||||
|
align: left|center|right; valign: top|center|bottom."""
|
||||||
|
if align == "left":
|
||||||
|
x = self.x
|
||||||
|
elif align == "right":
|
||||||
|
x = self.right - w
|
||||||
|
else:
|
||||||
|
x = self.x + (self.w - w) // 2
|
||||||
|
if valign == "top":
|
||||||
|
y = self.y
|
||||||
|
elif valign == "bottom":
|
||||||
|
y = self.bottom - h
|
||||||
|
else:
|
||||||
|
y = self.y + (self.h - h) // 2
|
||||||
|
return (x, y)
|
||||||
|
|
||||||
|
def center_xy(self, w: int, h: int) -> Tuple[int, int]:
|
||||||
|
return self.align_xy(w, h)
|
||||||
|
|
||||||
|
def contains(self, w: int, h: int) -> bool:
|
||||||
|
return w <= self.w and h <= self.h
|
||||||
|
|
||||||
|
|
||||||
|
def _weighted_sizes(total: int, weights: Sequence[float], gap: int) -> List[int]:
|
||||||
|
"""Integer sizes proportional to weights, remainder spread left-to-right."""
|
||||||
|
if not weights:
|
||||||
|
return []
|
||||||
|
usable = max(0, total - gap * (len(weights) - 1))
|
||||||
|
weight_sum = sum(weights) or 1
|
||||||
|
sizes = [int(usable * w / weight_sum) for w in weights]
|
||||||
|
remainder = usable - sum(sizes)
|
||||||
|
for i in range(remainder):
|
||||||
|
sizes[i % len(sizes)] += 1
|
||||||
|
return sizes
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Font ladders
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class FontStep:
|
||||||
|
"""One rung: a FontManager catalog family at a size it renders crisply."""
|
||||||
|
family: str
|
||||||
|
size_px: int
|
||||||
|
|
||||||
|
|
||||||
|
FontLadder = Tuple[FontStep, ...]
|
||||||
|
|
||||||
|
# X11 BDF bitmap fonts at their native pixel sizes, largest to smallest —
|
||||||
|
# baseball-scoreboard's fallback ladder extended upward. Same-height rungs
|
||||||
|
# are ordered widest first so width-constrained text steps to a narrower
|
||||||
|
# face before dropping a size.
|
||||||
|
LADDER_GRID: FontLadder = (
|
||||||
|
FontStep("10x20", 20),
|
||||||
|
FontStep("9x18", 18),
|
||||||
|
FontStep("9x15", 15),
|
||||||
|
FontStep("8x13", 13),
|
||||||
|
FontStep("7x13", 13),
|
||||||
|
FontStep("6x13", 13),
|
||||||
|
FontStep("6x12", 12),
|
||||||
|
FontStep("6x10", 10),
|
||||||
|
FontStep("6x9", 9),
|
||||||
|
FontStep("5x8", 8),
|
||||||
|
FontStep("5x7", 7),
|
||||||
|
FontStep("4x6", 6),
|
||||||
|
FontStep("tom-thumb", 6),
|
||||||
|
)
|
||||||
|
|
||||||
|
# PressStart2P at integer multiples of its 8px pixel grid only — fractional
|
||||||
|
# sizes blur a pixel font. For headline text (clocks, scores).
|
||||||
|
LADDER_ARCADE: FontLadder = (
|
||||||
|
FontStep("press_start", 32),
|
||||||
|
FontStep("press_start", 24),
|
||||||
|
FontStep("press_start", 16),
|
||||||
|
FontStep("press_start", 8),
|
||||||
|
)
|
||||||
|
|
||||||
|
LADDER_DEFAULT: FontLadder = LADDER_GRID
|
||||||
|
|
||||||
|
ELLIPSIS = "…"
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class FitResult:
|
||||||
|
"""A fitted font plus the ink metrics of the (possibly ellipsized) text.
|
||||||
|
|
||||||
|
``y_offset`` is the gap between the y passed to draw_text() and where
|
||||||
|
ink actually starts; subtract it from the desired ink-top position when
|
||||||
|
drawing (draw_fitted_text does this for you).
|
||||||
|
"""
|
||||||
|
font: Any
|
||||||
|
family: str
|
||||||
|
size_px: int
|
||||||
|
text: str
|
||||||
|
width: int
|
||||||
|
height: int
|
||||||
|
baseline: int
|
||||||
|
y_offset: int
|
||||||
|
fits: bool
|
||||||
|
line_height: int = 0
|
||||||
|
|
||||||
|
|
||||||
|
def measure_ink(text: str, font: Any) -> Tuple[int, int, int, int]:
|
||||||
|
"""Measure the ink box of text: (width, height, baseline, y_offset).
|
||||||
|
|
||||||
|
y_offset is the distance from the y coordinate DisplayManager.draw_text()
|
||||||
|
is given to the top of the actual ink — PIL draws TTF from the em-box
|
||||||
|
top and _draw_bdf_text derives the baseline from y + ascender, so both
|
||||||
|
leave a font-dependent gap that matters when centering in short bands.
|
||||||
|
"""
|
||||||
|
if isinstance(font, freetype.Face):
|
||||||
|
width = 0
|
||||||
|
ascender = font.size.ascender >> 6
|
||||||
|
ink_top, ink_bottom = None, None
|
||||||
|
for char in text:
|
||||||
|
font.load_char(char)
|
||||||
|
width += font.glyph.advance.x >> 6
|
||||||
|
rows = font.glyph.bitmap.rows
|
||||||
|
if rows:
|
||||||
|
top = ascender - font.glyph.bitmap_top
|
||||||
|
ink_top = top if ink_top is None else min(ink_top, top)
|
||||||
|
ink_bottom = top + rows if ink_bottom is None else max(ink_bottom, top + rows)
|
||||||
|
if ink_top is None:
|
||||||
|
ink_top, ink_bottom = 0, 0
|
||||||
|
return (width, ink_bottom - ink_top, ascender, ink_top)
|
||||||
|
bbox = font.getbbox(text)
|
||||||
|
return (bbox[2] - bbox[0], bbox[3] - bbox[1], -bbox[1], bbox[1])
|
||||||
|
|
||||||
|
|
||||||
|
def font_line_height(font: Any) -> int:
|
||||||
|
"""Recommended line spacing for a font (matches DisplayManager.get_font_height)."""
|
||||||
|
if isinstance(font, freetype.Face):
|
||||||
|
return font.size.height >> 6
|
||||||
|
ascent, descent = font.getmetrics()
|
||||||
|
return ascent + descent
|
||||||
|
|
||||||
|
|
||||||
|
def measure_font_crispness(font: Any, sample_text: str = "Ay0",
|
||||||
|
canvas_size: Tuple[int, int] = (250, 60)) -> float:
|
||||||
|
"""Fraction of the rendered sample's ink-bbox pixels that are neither
|
||||||
|
pure black nor pure white — i.e. antialiased.
|
||||||
|
|
||||||
|
BDF (freetype.Face) glyphs are true bitmaps and always render at 0.0.
|
||||||
|
"Pixel-style" TTFs (PressStart2P, and similar fonts bundled for
|
||||||
|
plugins that draw through ImageDraw.text() and so can't take a BDF
|
||||||
|
face) are NOT automatically crisp at arbitrary sizes — PIL antialiases
|
||||||
|
TTF outlines by default, and a pixel-grid font only lands on whole
|
||||||
|
pixels at specific sizes (for PressStart2P: exact multiples of 8).
|
||||||
|
Requesting an unverified size silently produces soft/blurry glyphs on
|
||||||
|
an LED panel, which reads as fuzzy compared to a true BDF rung.
|
||||||
|
|
||||||
|
Use this to vet any custom FontLadder rung that mixes TTF fonts before
|
||||||
|
shipping it — see test_adaptive_layout.py::test_ladder_is_crisp for the
|
||||||
|
pattern. A rung should score 0.0 (or very close, to allow for the odd
|
||||||
|
diagonal stroke) before it belongs in a "crisp" ladder.
|
||||||
|
"""
|
||||||
|
if isinstance(font, freetype.Face):
|
||||||
|
return 0.0
|
||||||
|
from PIL import Image, ImageDraw
|
||||||
|
img = Image.new("L", canvas_size, 0)
|
||||||
|
ImageDraw.Draw(img).text((2, 2), sample_text, font=font, fill=255)
|
||||||
|
bbox = img.getbbox()
|
||||||
|
if bbox is None:
|
||||||
|
return 0.0
|
||||||
|
pixels = img.crop(bbox).tobytes()
|
||||||
|
pure = sum(1 for p in pixels if p == 0 or p == 255)
|
||||||
|
return (len(pixels) - pure) / len(pixels)
|
||||||
|
|
||||||
|
|
||||||
|
class LayoutContext:
|
||||||
|
"""Per-render-size layout facts and fit-text queries for one panel size.
|
||||||
|
|
||||||
|
Construct once per (width, height); BasePlugin.layout does this and
|
||||||
|
rebuilds automatically when the logical display size changes.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, width: int, height: int, font_manager: Any,
|
||||||
|
design_size: Tuple[int, int] = DEFAULT_DESIGN_SIZE):
|
||||||
|
self.width = int(width)
|
||||||
|
self.height = int(height)
|
||||||
|
self.font_manager = font_manager
|
||||||
|
self.design_size = design_size
|
||||||
|
self.bounds = Region(0, 0, self.width, self.height)
|
||||||
|
self.aspect = self.width / max(1, self.height)
|
||||||
|
self.tier = _pick_tier(_HEIGHT_TIERS, self.height)
|
||||||
|
self.width_tier = _pick_tier(_WIDTH_TIERS, self.width)
|
||||||
|
self.is_wide_short = self.aspect >= 2.5 and self.height <= 32
|
||||||
|
design_w, design_h = design_size
|
||||||
|
# Geometry scale only (gaps, icon/logo sizes) — never applied to
|
||||||
|
# fonts, which step between crisp ladder rungs instead.
|
||||||
|
self.scale = min(self.width / max(1, design_w),
|
||||||
|
self.height / max(1, design_h))
|
||||||
|
# LRU-bounded: entries are small, but keys embed the fitted TEXT —
|
||||||
|
# a plugin fitting changing text (a live game clock, a ticker) on a
|
||||||
|
# 24/7 service would otherwise grow this without bound.
|
||||||
|
self._fit_cache: "OrderedDict[Any, FitResult]" = OrderedDict()
|
||||||
|
# LRU-bounded (images are big). Entries hold a strong reference to
|
||||||
|
# the source image when keyed by id() so the id can't be recycled
|
||||||
|
# out from under the cache.
|
||||||
|
self._image_cache: "OrderedDict[Any, Tuple[Any, Any]]" = OrderedDict()
|
||||||
|
|
||||||
|
_IMAGE_CACHE_MAX = 64
|
||||||
|
_FIT_CACHE_MAX = 512
|
||||||
|
|
||||||
|
def _fit_cache_get(self, key: Any) -> Optional["FitResult"]:
|
||||||
|
cached = self._fit_cache.get(key)
|
||||||
|
if cached is not None:
|
||||||
|
self._fit_cache.move_to_end(key)
|
||||||
|
return cached
|
||||||
|
|
||||||
|
def _fit_cache_put(self, key: Any, result: "FitResult") -> None:
|
||||||
|
self._fit_cache[key] = result
|
||||||
|
while len(self._fit_cache) > self._FIT_CACHE_MAX:
|
||||||
|
self._fit_cache.popitem(last=False)
|
||||||
|
|
||||||
|
# ---- the three adaptation patterns --------------------------------
|
||||||
|
|
||||||
|
def px(self, base: int, minimum: int = 1, maximum: Optional[int] = None) -> int:
|
||||||
|
"""Scale a design-size pixel measurement (f1's pattern): gaps,
|
||||||
|
icon sizes, logo slots. Clamped to [minimum, maximum]."""
|
||||||
|
value = max(minimum, round(base * self.scale))
|
||||||
|
if maximum is not None:
|
||||||
|
value = min(value, maximum)
|
||||||
|
return value
|
||||||
|
|
||||||
|
def by_tier(self, mapping: Dict[str, Any], default: Any = None) -> Any:
|
||||||
|
"""Pick the value for the nearest defined tier at-or-below the
|
||||||
|
panel's height tier (masters' pattern). Falls forward to the
|
||||||
|
smallest defined tier above, then to default.
|
||||||
|
|
||||||
|
by_tier({"sm": 10, "lg": 18}) -> 10 on 128x32, 18 on 128x64.
|
||||||
|
Keys may also use width tiers ("narrow", "wide", ...)."""
|
||||||
|
order = TIER_ORDER if any(k in TIER_ORDER for k in mapping) else WIDTH_TIER_ORDER
|
||||||
|
current = self.tier if order is TIER_ORDER else self.width_tier
|
||||||
|
idx = order.index(current)
|
||||||
|
for name in reversed(order[: idx + 1]):
|
||||||
|
if name in mapping:
|
||||||
|
return mapping[name]
|
||||||
|
for name in order[idx + 1:]:
|
||||||
|
if name in mapping:
|
||||||
|
return mapping[name]
|
||||||
|
return default
|
||||||
|
|
||||||
|
def fit_text(self, text: str, box: Union[Region, Tuple[int, int]],
|
||||||
|
ladder: FontLadder = LADDER_DEFAULT,
|
||||||
|
ellipsis: bool = True) -> FitResult:
|
||||||
|
"""Largest ladder rung whose rendered text fits the box (baseball's
|
||||||
|
pattern). If even the smallest rung is too wide, the text is
|
||||||
|
ellipsized to fit (unless ellipsis=False); fits=False only when no
|
||||||
|
acceptable rendering exists."""
|
||||||
|
box_w, box_h = _box_dims(box)
|
||||||
|
key = ("text", text, box_w, box_h, ladder, ellipsis)
|
||||||
|
cached = self._fit_cache_get(key)
|
||||||
|
if cached is not None:
|
||||||
|
return cached
|
||||||
|
result = self._walk_ladder(text, ladder, box_w, box_h, ellipsis)
|
||||||
|
self._fit_cache_put(key, result)
|
||||||
|
return result
|
||||||
|
|
||||||
|
def fit_text_proportional(self, text: str, box: Union[Region, Tuple[int, int]],
|
||||||
|
base_size_px: int, ladder: FontLadder = LADDER_DEFAULT,
|
||||||
|
ellipsis: bool = True,
|
||||||
|
scale: Optional[float] = None) -> FitResult:
|
||||||
|
"""Ladder rung closest to (but not exceeding) ``base_size_px * scale``
|
||||||
|
that still fits the box — proportional sizing instead of ``fit_text``'s
|
||||||
|
"always maximize" behavior.
|
||||||
|
|
||||||
|
Use this when several independently-fitted elements need to stay
|
||||||
|
visually harmonious as the panel grows (e.g. a scoreboard's score,
|
||||||
|
status, and detail text) — ``fit_text`` maximizes each one within
|
||||||
|
its own region, which can make one element balloon out of
|
||||||
|
proportion to its neighbors (a huge score overlapping logos it fit
|
||||||
|
fine at the design size) even though every individual pick is
|
||||||
|
independently "correct". ``base_size_px`` is the size that element
|
||||||
|
renders at on the design size (``design_size``, typically 128x32)
|
||||||
|
— commonly a plugin's existing classic/fixed font size for that
|
||||||
|
element.
|
||||||
|
|
||||||
|
``scale`` defaults to ``self.scale`` (the same conservative
|
||||||
|
min(width_ratio, height_ratio) factor ``px()`` uses — safe for
|
||||||
|
content whose aspect ratio matters). Pass an explicit axis-specific
|
||||||
|
value when the surrounding composition already scales that way —
|
||||||
|
e.g. a scoreboard whose logos scale with height alone
|
||||||
|
(``logo_slot = min(height, width // 2)``) should size its score
|
||||||
|
text by ``height / design_height`` too, or its text will look
|
||||||
|
under-scaled next to bigger logos on a panel that only grew taller.
|
||||||
|
|
||||||
|
Falls back to the smallest rung when even that exceeds the target
|
||||||
|
(a tiny scale factor), and to fit_text's ordinary smaller-rung
|
||||||
|
fallback when the closest-to-target rung doesn't actually fit the
|
||||||
|
box.
|
||||||
|
"""
|
||||||
|
box_w, box_h = _box_dims(box)
|
||||||
|
effective_scale = self.scale if scale is None else scale
|
||||||
|
key = ("text_prop", text, box_w, box_h, ladder, base_size_px, ellipsis, effective_scale)
|
||||||
|
cached = self._fit_cache_get(key)
|
||||||
|
if cached is not None:
|
||||||
|
return cached
|
||||||
|
target = base_size_px * effective_scale
|
||||||
|
eligible = [step for step in ladder if step.size_px <= target]
|
||||||
|
candidates = eligible if eligible else (min(ladder, key=lambda s: s.size_px),)
|
||||||
|
result = self._walk_ladder(text, candidates, box_w, box_h, ellipsis)
|
||||||
|
self._fit_cache_put(key, result)
|
||||||
|
return result
|
||||||
|
|
||||||
|
def _walk_ladder(self, text: str, ladder: Sequence[FontStep],
|
||||||
|
box_w: int, box_h: int, ellipsis: bool) -> FitResult:
|
||||||
|
"""Shared by fit_text/fit_text_proportional: first ladder entry (in
|
||||||
|
the order given) whose rendered text fits, ellipsizing the last one
|
||||||
|
tried if none do."""
|
||||||
|
result = None
|
||||||
|
for step in ladder:
|
||||||
|
font = self.font_manager.get_font(step.family, step.size_px)
|
||||||
|
width, height, baseline, y_offset = measure_ink(text, font)
|
||||||
|
result = FitResult(font, step.family, step.size_px, text,
|
||||||
|
width, height, baseline, y_offset,
|
||||||
|
fits=(width <= box_w and height <= box_h),
|
||||||
|
line_height=font_line_height(font))
|
||||||
|
if result.fits:
|
||||||
|
break
|
||||||
|
|
||||||
|
if result is not None and not result.fits and ellipsis:
|
||||||
|
short = self.ellipsize(text, result.font, box_w)
|
||||||
|
width, height, baseline, y_offset = measure_ink(short, result.font)
|
||||||
|
result = FitResult(result.font, result.family, result.size_px,
|
||||||
|
short, width, height, baseline, y_offset,
|
||||||
|
fits=(width <= box_w and height <= box_h),
|
||||||
|
line_height=result.line_height)
|
||||||
|
return result
|
||||||
|
|
||||||
|
def fit_lines(self, lines: Sequence[str], box: Union[Region, Tuple[int, int]],
|
||||||
|
ladder: FontLadder = LADDER_DEFAULT,
|
||||||
|
spacing: int = 1) -> FitResult:
|
||||||
|
"""Largest rung where every line fits the box width and the stacked
|
||||||
|
lines (line_height + spacing apart) fit the box height. Measures the
|
||||||
|
actual strings, so a long line pushes the ladder down a rung a short
|
||||||
|
one wouldn't (baseball's multiline pattern). Text is the widest line."""
|
||||||
|
box_w, box_h = _box_dims(box)
|
||||||
|
key = ("lines", tuple(lines), box_w, box_h, ladder, spacing)
|
||||||
|
cached = self._fit_cache_get(key)
|
||||||
|
if cached is not None:
|
||||||
|
return cached
|
||||||
|
|
||||||
|
rows = max(1, len(lines))
|
||||||
|
result = None
|
||||||
|
for step in ladder:
|
||||||
|
font = self.font_manager.get_font(step.family, step.size_px)
|
||||||
|
line_h = font_line_height(font)
|
||||||
|
widest, metrics = "", (0, 0, 0, 0)
|
||||||
|
for line in lines:
|
||||||
|
m = measure_ink(line, font)
|
||||||
|
if m[0] >= metrics[0]:
|
||||||
|
widest, metrics = line, m
|
||||||
|
total_h = rows * line_h + (rows - 1) * spacing
|
||||||
|
result = FitResult(font, step.family, step.size_px, widest,
|
||||||
|
metrics[0], metrics[1], metrics[2], metrics[3],
|
||||||
|
fits=(metrics[0] <= box_w and total_h <= box_h),
|
||||||
|
line_height=line_h)
|
||||||
|
if result.fits:
|
||||||
|
break
|
||||||
|
|
||||||
|
self._fit_cache_put(key, result)
|
||||||
|
return result
|
||||||
|
|
||||||
|
def font_for_rows(self, rows: int, box_h: int,
|
||||||
|
ladder: FontLadder = LADDER_GRID) -> FitResult:
|
||||||
|
"""Largest rung whose line height lets `rows` rows fit in box_h
|
||||||
|
(baseball's traditional-scoreboard pattern). Measures a digit/cap
|
||||||
|
sample rather than specific strings."""
|
||||||
|
key = ("rows", rows, box_h, ladder)
|
||||||
|
cached = self._fit_cache_get(key)
|
||||||
|
if cached is not None:
|
||||||
|
return cached
|
||||||
|
|
||||||
|
sample = "0Ay"
|
||||||
|
result = None
|
||||||
|
for step in ladder:
|
||||||
|
font = self.font_manager.get_font(step.family, step.size_px)
|
||||||
|
line_h = font_line_height(font)
|
||||||
|
width, height, baseline, y_offset = measure_ink(sample, font)
|
||||||
|
result = FitResult(font, step.family, step.size_px, sample,
|
||||||
|
width, height, baseline, y_offset,
|
||||||
|
fits=(max(1, rows) * line_h <= box_h),
|
||||||
|
line_height=line_h)
|
||||||
|
if result.fits:
|
||||||
|
break
|
||||||
|
|
||||||
|
self._fit_cache_put(key, result)
|
||||||
|
return result
|
||||||
|
|
||||||
|
# ---- images ---------------------------------------------------------
|
||||||
|
|
||||||
|
def fit_image(self, img: Any, box: Union[Region, Tuple[int, int]], *,
|
||||||
|
mode: str = "contain", crop_to_ink: bool = False,
|
||||||
|
anchor: str = "center", resample: Any = None,
|
||||||
|
upscale: bool = True, cache_key: Any = None) -> Any:
|
||||||
|
"""Fit an image into a box (see src/adaptive_images.py for modes),
|
||||||
|
cached per (image, box size, options) for this panel size.
|
||||||
|
|
||||||
|
Prefer a stable ``cache_key`` (e.g. "logo:KC") for images that get
|
||||||
|
reloaded — the default id()-based key is safe (the entry pins the
|
||||||
|
source image) but misses across reloads of the same content.
|
||||||
|
"""
|
||||||
|
from src.adaptive_images import fit_image as _fit_image
|
||||||
|
|
||||||
|
box_w, box_h = _box_dims(box)
|
||||||
|
resample_name = getattr(resample, "name", repr(resample)) if resample is not None else "default"
|
||||||
|
identity = cache_key if cache_key is not None else ("id", id(img))
|
||||||
|
key = ("image", identity, img.size, box_w, box_h, mode,
|
||||||
|
crop_to_ink, anchor, resample_name, upscale)
|
||||||
|
|
||||||
|
cached = self._image_cache.get(key)
|
||||||
|
if cached is not None:
|
||||||
|
self._image_cache.move_to_end(key)
|
||||||
|
return cached[0]
|
||||||
|
|
||||||
|
result = _fit_image(img, (box_w, box_h), mode=mode,
|
||||||
|
crop_to_ink=crop_to_ink, anchor=anchor,
|
||||||
|
resample=resample, upscale=upscale)
|
||||||
|
# Pin the source only for id()-keyed entries (see docstring).
|
||||||
|
self._image_cache[key] = (result, img if cache_key is None else None)
|
||||||
|
while len(self._image_cache) > self._IMAGE_CACHE_MAX:
|
||||||
|
self._image_cache.popitem(last=False)
|
||||||
|
return result
|
||||||
|
|
||||||
|
# ---- text utilities ------------------------------------------------
|
||||||
|
|
||||||
|
def ellipsize(self, text: str, font: Any, max_w: int) -> str:
|
||||||
|
"""Trim text to fit max_w, appending an ellipsis. Returns '' when
|
||||||
|
not even the ellipsis fits."""
|
||||||
|
if measure_ink(text, font)[0] <= max_w:
|
||||||
|
return text
|
||||||
|
for end in range(len(text) - 1, 0, -1):
|
||||||
|
candidate = text[:end].rstrip() + ELLIPSIS
|
||||||
|
if measure_ink(candidate, font)[0] <= max_w:
|
||||||
|
return candidate
|
||||||
|
return ELLIPSIS if measure_ink(ELLIPSIS, font)[0] <= max_w else ""
|
||||||
|
|
||||||
|
def measure(self, text: str, font: Any) -> Tuple[int, int, int]:
|
||||||
|
"""Ink (width, height, baseline) of text — see measure_ink."""
|
||||||
|
width, height, baseline, _ = measure_ink(text, font)
|
||||||
|
return (width, height, baseline)
|
||||||
|
|
||||||
|
def clear_cache(self) -> None:
|
||||||
|
"""Drop cached fit results (call after fonts are reloaded)."""
|
||||||
|
self._fit_cache.clear()
|
||||||
|
self._image_cache.clear()
|
||||||
|
|
||||||
|
|
||||||
|
def _pick_tier(tiers: Tuple[Tuple[str, int], ...], value: int) -> str:
|
||||||
|
for name, limit in tiers:
|
||||||
|
if value <= limit:
|
||||||
|
return name
|
||||||
|
return tiers[-1][0]
|
||||||
|
|
||||||
|
|
||||||
|
def _box_dims(box: Union[Region, Tuple[int, int]]) -> Tuple[int, int]:
|
||||||
|
if isinstance(box, Region):
|
||||||
|
return (box.w, box.h)
|
||||||
|
w, h = box
|
||||||
|
return (int(w), int(h))
|
||||||
|
|
||||||
|
|
||||||
|
def draw_fitted_text(display_manager: Any, fit: FitResult,
|
||||||
|
box: Union[Region, Tuple[int, int]],
|
||||||
|
color: Tuple[int, int, int] = (255, 255, 255),
|
||||||
|
align: str = "center", valign: str = "center") -> None:
|
||||||
|
"""Draw a FitResult's text aligned within a Region via
|
||||||
|
DisplayManager.draw_text(), compensating for the font's ink offset so
|
||||||
|
the ink (not the em box) is what gets aligned."""
|
||||||
|
region = box if isinstance(box, Region) else Region(0, 0, box[0], box[1])
|
||||||
|
x, y = region.align_xy(fit.width, fit.height, align, valign)
|
||||||
|
display_manager.draw_text(fit.text, x=x, y=y - fit.y_offset,
|
||||||
|
color=color, font=fit.font)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Composite layouts — the region arrangements repeated across plugins,
|
||||||
|
# expressed as Region math so migrated plugins stop hand-copying coordinate
|
||||||
|
# formulas. Deliberately tiny: these return Regions, they don't draw.
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class ScoreboardRegions:
|
||||||
|
"""The two-logos-plus-center-score card shared by the sports plugins."""
|
||||||
|
bounds: Region
|
||||||
|
logo_slot: int # width of each logo slot: min(H, W // 2), center-reserved
|
||||||
|
away_slot: Region # left logo slot
|
||||||
|
home_slot: Region # right logo slot
|
||||||
|
center_col: Region # column between the slots (>= min_center_fraction of width)
|
||||||
|
status_band: Region # top band (replaces the magic y = 1)
|
||||||
|
score_area: Region # center_col's true width, between the bands (replaces y = H//2 - 3)
|
||||||
|
detail_band: Region # bottom band (replaces the magic y = H - 7)
|
||||||
|
bottom_left: Region # bottom corner: away records / timeouts
|
||||||
|
bottom_right: Region # bottom corner: home records / timeouts
|
||||||
|
|
||||||
|
|
||||||
|
def scoreboard_regions(bounds: Region, *, ctx: Optional["LayoutContext"] = None,
|
||||||
|
status_h: Optional[int] = None,
|
||||||
|
detail_h: Optional[int] = None,
|
||||||
|
min_center_fraction: float = 0.15,
|
||||||
|
min_center_design_px: int = 40,
|
||||||
|
score_bleed_fraction: float = 0.5) -> ScoreboardRegions:
|
||||||
|
"""Carve a game-card Region into the standard scoreboard arrangement.
|
||||||
|
|
||||||
|
Encodes the invariant duplicated across the sports plugins:
|
||||||
|
``logo_slot = min(height, width // 2)`` (capped at half the card so the
|
||||||
|
home slot never collapses), away logo centered in the left slot, home in
|
||||||
|
the right.
|
||||||
|
|
||||||
|
That formula alone has a blind spot: at exactly 2:1 aspect ratio
|
||||||
|
(width == 2 * height — a very common shape, e.g. two, four, or more
|
||||||
|
square modules stacked into a taller panel) ``width // 2`` and
|
||||||
|
``height`` are equal, so the two logo slots claim the *entire* width
|
||||||
|
and leave zero pixels for a center column, no matter how large the
|
||||||
|
panel gets. It isn't a "small panel" problem: 96x48, 128x64, and
|
||||||
|
256x128 (all exactly 2:1) hit it identically, while wide panels like
|
||||||
|
the 128x32 design baseline or a 192x48/256x32 panel never do, because
|
||||||
|
height is already the tighter constraint there.
|
||||||
|
|
||||||
|
Two knobs fix it, both defaulted to values verified against the full
|
||||||
|
harness size spread (see test_adaptive_layout.py::TestScoreboardRegions):
|
||||||
|
|
||||||
|
- ``min_center_fraction`` / ``min_center_design_px`` reserve at least
|
||||||
|
``max(width * min_center_fraction, min_center_design_px * ctx.scale)``
|
||||||
|
for the center column, capping ``logo_slot`` further when needed. The
|
||||||
|
design-px term (scaled by the context's geometry factor, so it grows
|
||||||
|
on bigger panels like everything else in ``px()``) matters most on
|
||||||
|
small panels where a flat fraction alone reserves too little absolute
|
||||||
|
space for even a short score string. On wide panels the height
|
||||||
|
constraint already leaves more room than either reserves, so both are
|
||||||
|
a no-op there — 128x32/192x48-style layouts are unaffected.
|
||||||
|
- ``score_bleed_fraction`` extends the score's own *fit box* (not the
|
||||||
|
logo slots themselves) an extra ``logo_slot * score_bleed_fraction``
|
||||||
|
into each side — controlled, intentional overlap with the logo art,
|
||||||
|
the same way real broadcast scoreboards let a big score number's
|
||||||
|
edges cross into the team marks flanking it. Without this, on a
|
||||||
|
square-ish panel the center reserve alone can be too narrow for even
|
||||||
|
a modest score to render without truncating (`"17-21"` -> `"17-2…"`),
|
||||||
|
which is worse than a little overlap.
|
||||||
|
|
||||||
|
status_band and detail_band span the FULL card width and overlay the
|
||||||
|
logo slots — matching the classic layouts, where short outlined status/
|
||||||
|
date text is drawn over the logos without issue; only score_area (the
|
||||||
|
one element whose size actively grows with the panel) uses the
|
||||||
|
narrower, bleed-adjusted box. Band heights default to the classic
|
||||||
|
128x32 values, scaled by the context's geometry factor when one is
|
||||||
|
provided. Works on a full panel or on a scroll-mode card Region.
|
||||||
|
"""
|
||||||
|
if status_h is None:
|
||||||
|
status_h = ctx.px(9, minimum=7) if ctx else 9
|
||||||
|
if detail_h is None:
|
||||||
|
detail_h = ctx.px(8, minimum=7) if ctx else 8
|
||||||
|
|
||||||
|
logo_slot = min(bounds.h, bounds.w // 2)
|
||||||
|
design_reserve = int(min_center_design_px * (ctx.scale if ctx else 1.0))
|
||||||
|
min_center_w = max(1, int(bounds.w * min_center_fraction), design_reserve)
|
||||||
|
max_logo_slot_by_center = max(1, (bounds.w - min_center_w) // 2)
|
||||||
|
logo_slot = min(logo_slot, max_logo_slot_by_center)
|
||||||
|
away_slot = bounds.left_col(logo_slot)
|
||||||
|
home_slot = bounds.right_col(logo_slot)
|
||||||
|
center_col = Region(bounds.x + logo_slot, bounds.y,
|
||||||
|
bounds.w - 2 * logo_slot, bounds.h)
|
||||||
|
status_band = bounds.top_band(status_h)
|
||||||
|
detail_band = bounds.bottom_band(detail_h)
|
||||||
|
middle = bounds.middle(status_band.h, detail_band.h)
|
||||||
|
# score_area is the true center gap's width plus a controlled bleed
|
||||||
|
# into each logo slot (see score_bleed_fraction above) -- narrower than
|
||||||
|
# the full card width status/detail get, since it's the one element
|
||||||
|
# whose size actively grows with the panel and needs its *fit box* to
|
||||||
|
# reflect real available space, but generous enough that a short score
|
||||||
|
# string never has to truncate on a square-ish panel.
|
||||||
|
bleed = int(logo_slot * score_bleed_fraction)
|
||||||
|
score_area = Region(center_col.x - bleed, middle.y,
|
||||||
|
center_col.w + 2 * bleed, middle.h)
|
||||||
|
bottom = bounds.bottom_band(detail_h)
|
||||||
|
return ScoreboardRegions(
|
||||||
|
bounds=bounds, logo_slot=logo_slot,
|
||||||
|
away_slot=away_slot, home_slot=home_slot, center_col=center_col,
|
||||||
|
status_band=status_band, score_area=score_area, detail_band=detail_band,
|
||||||
|
bottom_left=bottom.left_col(logo_slot),
|
||||||
|
bottom_right=bottom.right_col(logo_slot),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class MediaRow:
|
||||||
|
"""Art/icon on the left, text column on the right (music's idiom)."""
|
||||||
|
art: Region
|
||||||
|
body: Region
|
||||||
|
|
||||||
|
|
||||||
|
def media_row(bounds: Region, *, ctx: Optional["LayoutContext"] = None,
|
||||||
|
square: bool = True, gap: Optional[int] = None) -> MediaRow:
|
||||||
|
"""Split a Region into an art slot and a body column.
|
||||||
|
|
||||||
|
With ``square=True`` the art slot is bounds.h wide (album-art style);
|
||||||
|
otherwise it takes the left half. The gap defaults to 2px scaled by the
|
||||||
|
context's geometry factor.
|
||||||
|
"""
|
||||||
|
if gap is None:
|
||||||
|
gap = ctx.px(2, minimum=1) if ctx else 2
|
||||||
|
art_w = bounds.h if square else bounds.w // 2
|
||||||
|
art_w = min(art_w, bounds.w)
|
||||||
|
art = bounds.left_col(art_w)
|
||||||
|
body = Region(bounds.x + art_w + gap, bounds.y,
|
||||||
|
bounds.w - art_w - gap, bounds.h)
|
||||||
|
return MediaRow(art=art, body=body)
|
||||||
Vendored
+48
-9
@@ -10,6 +10,7 @@ import time
|
|||||||
import tempfile
|
import tempfile
|
||||||
import logging
|
import logging
|
||||||
import threading
|
import threading
|
||||||
|
import zlib
|
||||||
from typing import Dict, Any, Optional, Protocol
|
from typing import Dict, Any, Optional, Protocol
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
|
|
||||||
@@ -53,6 +54,11 @@ class DiskCache:
|
|||||||
self.cache_dir = cache_dir
|
self.cache_dir = cache_dir
|
||||||
self.logger = logger or logging.getLogger(__name__)
|
self.logger = logger or logging.getLogger(__name__)
|
||||||
self._lock = threading.Lock()
|
self._lock = threading.Lock()
|
||||||
|
# key -> adler32 of the last payload successfully written to the
|
||||||
|
# primary cache path; lets set() skip rewriting identical data
|
||||||
|
# (per-process only — worst case another process rewrites, never
|
||||||
|
# a missed write). Guarded by _lock.
|
||||||
|
self._write_digests: Dict[str, int] = {}
|
||||||
|
|
||||||
def get_cache_path(self, key: str) -> Optional[str]:
|
def get_cache_path(self, key: str) -> Optional[str]:
|
||||||
"""
|
"""
|
||||||
@@ -155,10 +161,35 @@ class DiskCache:
|
|||||||
cache_path = self.get_cache_path(key)
|
cache_path = self.get_cache_path(key)
|
||||||
if not cache_path:
|
if not cache_path:
|
||||||
return
|
return
|
||||||
|
|
||||||
|
# Serialize once, compact (no indent): the payload is reused by every
|
||||||
|
# write path below, and cache files are machine-read only — indenting
|
||||||
|
# them just multiplied the bytes written to the SD card.
|
||||||
|
try:
|
||||||
|
payload = json.dumps(data, cls=DateTimeEncoder)
|
||||||
|
except (TypeError, ValueError) as e:
|
||||||
|
self.logger.warning("Cache data for key '%s' not serializable: %s", key, e)
|
||||||
|
return
|
||||||
|
|
||||||
|
digest = zlib.adler32(payload.encode('utf-8'))
|
||||||
|
|
||||||
try:
|
try:
|
||||||
# Atomic write to avoid partial/corrupt files
|
# Atomic write to avoid partial/corrupt files
|
||||||
with self._lock:
|
with self._lock:
|
||||||
|
# Skip the disk entirely when this exact payload was already
|
||||||
|
# written for this key (plugins re-save unchanged API data
|
||||||
|
# every update cycle — each write is real SD-card wear).
|
||||||
|
# Refresh the file mtime so records that rely on it for TTL
|
||||||
|
# (no embedded 'timestamp') don't expire early; a metadata
|
||||||
|
# touch is journal-cheap compared to rewriting the data.
|
||||||
|
if self._write_digests.get(key) == digest:
|
||||||
|
try:
|
||||||
|
os.utime(cache_path, None)
|
||||||
|
return
|
||||||
|
except OSError:
|
||||||
|
# File vanished or perms changed — fall through and write
|
||||||
|
self._write_digests.pop(key, None)
|
||||||
|
|
||||||
tmp_dir = os.path.dirname(cache_path)
|
tmp_dir = os.path.dirname(cache_path)
|
||||||
# Try to create temp file in cache directory first
|
# Try to create temp file in cache directory first
|
||||||
# If that fails due to permissions, fall back to direct write
|
# If that fails due to permissions, fall back to direct write
|
||||||
@@ -181,13 +212,17 @@ class DiskCache:
|
|||||||
fd = None
|
fd = None
|
||||||
|
|
||||||
if tmp_path and fd is not None:
|
if tmp_path and fd is not None:
|
||||||
# Use atomic write with temp file
|
# Atomic write with temp file. No fsync: os.replace
|
||||||
|
# already guarantees readers never see a torn file,
|
||||||
|
# and cache data is re-fetchable — forcing a disk
|
||||||
|
# flush per write was the single biggest SD-card
|
||||||
|
# wear source (dozens of fsyncs/min on API-heavy
|
||||||
|
# installs) for data that can be re-downloaded.
|
||||||
try:
|
try:
|
||||||
with os.fdopen(fd, 'w', encoding='utf-8') as tmp_file:
|
with os.fdopen(fd, 'w', encoding='utf-8') as tmp_file:
|
||||||
json.dump(data, tmp_file, indent=4, cls=DateTimeEncoder)
|
tmp_file.write(payload)
|
||||||
tmp_file.flush()
|
|
||||||
os.fsync(tmp_file.fileno())
|
|
||||||
os.replace(tmp_path, cache_path)
|
os.replace(tmp_path, cache_path)
|
||||||
|
self._write_digests[key] = digest
|
||||||
# Set proper permissions: 660 (rw-rw----) for group-readable cache files
|
# Set proper permissions: 660 (rw-rw----) for group-readable cache files
|
||||||
try:
|
try:
|
||||||
os.chmod(cache_path, 0o660) # nosec B103 - intentional; web UI and service share a group
|
os.chmod(cache_path, 0o660) # nosec B103 - intentional; web UI and service share a group
|
||||||
@@ -203,9 +238,8 @@ class DiskCache:
|
|||||||
# Fallback: direct write (not atomic, but better than failing)
|
# Fallback: direct write (not atomic, but better than failing)
|
||||||
try:
|
try:
|
||||||
with open(cache_path, 'w', encoding='utf-8') as cache_file:
|
with open(cache_path, 'w', encoding='utf-8') as cache_file:
|
||||||
json.dump(data, cache_file, indent=4, cls=DateTimeEncoder)
|
cache_file.write(payload)
|
||||||
cache_file.flush()
|
self._write_digests[key] = digest
|
||||||
os.fsync(cache_file.fileno())
|
|
||||||
# Set proper permissions: 660 (rw-rw----) for group-readable cache files
|
# Set proper permissions: 660 (rw-rw----) for group-readable cache files
|
||||||
try:
|
try:
|
||||||
os.chmod(cache_path, 0o660) # nosec B103 - intentional; web UI and service share a group
|
os.chmod(cache_path, 0o660) # nosec B103 - intentional; web UI and service share a group
|
||||||
@@ -229,9 +263,12 @@ class DiskCache:
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
if os.path.isdir(fallback_dir) and os.access(fallback_dir, os.W_OK):
|
if os.path.isdir(fallback_dir) and os.access(fallback_dir, os.W_OK):
|
||||||
|
# NOTE: no digest record here — the fallback file
|
||||||
|
# is a different path, so future sets must keep
|
||||||
|
# retrying the primary location.
|
||||||
fallback_path = os.path.join(fallback_dir, os.path.basename(cache_path))
|
fallback_path = os.path.join(fallback_dir, os.path.basename(cache_path))
|
||||||
with open(fallback_path, 'w', encoding='utf-8') as tmp_file:
|
with open(fallback_path, 'w', encoding='utf-8') as tmp_file:
|
||||||
json.dump(data, tmp_file, indent=4, cls=DateTimeEncoder)
|
tmp_file.write(payload)
|
||||||
# Set proper permissions: 660 (rw-rw----) for group-readable cache files
|
# Set proper permissions: 660 (rw-rw----) for group-readable cache files
|
||||||
try:
|
try:
|
||||||
os.chmod(fallback_path, 0o660) # nosec B103 - intentional; web UI and service share a group
|
os.chmod(fallback_path, 0o660) # nosec B103 - intentional; web UI and service share a group
|
||||||
@@ -272,6 +309,7 @@ class DiskCache:
|
|||||||
|
|
||||||
with self._lock:
|
with self._lock:
|
||||||
if key:
|
if key:
|
||||||
|
self._write_digests.pop(key, None)
|
||||||
cache_path = self.get_cache_path(key)
|
cache_path = self.get_cache_path(key)
|
||||||
if cache_path and os.path.exists(cache_path):
|
if cache_path and os.path.exists(cache_path):
|
||||||
try:
|
try:
|
||||||
@@ -280,6 +318,7 @@ class DiskCache:
|
|||||||
self.logger.warning("Could not remove cache file %s: %s", cache_path, e)
|
self.logger.warning("Could not remove cache file %s: %s", cache_path, e)
|
||||||
else:
|
else:
|
||||||
# Clear all cache files
|
# Clear all cache files
|
||||||
|
self._write_digests.clear()
|
||||||
if os.path.exists(self.cache_dir):
|
if os.path.exists(self.cache_dir):
|
||||||
for filename in os.listdir(self.cache_dir):
|
for filename in os.listdir(self.cache_dir):
|
||||||
if filename.endswith('.json'):
|
if filename.endswith('.json'):
|
||||||
|
|||||||
@@ -2,6 +2,28 @@
|
|||||||
|
|
||||||
This directory contains reusable utilities and helpers for LEDMatrix plugins and core modules.
|
This directory contains reusable utilities and helpers for LEDMatrix plugins and core modules.
|
||||||
|
|
||||||
|
## Adaptive Layout & Images (`src/adaptive_layout.py`, `src/adaptive_images.py`)
|
||||||
|
|
||||||
|
The recommended way to lay out plugins that render legibly on **any** panel
|
||||||
|
size (64x32 through 256x128+) without hand-tuned coordinates. Re-exported
|
||||||
|
from `src.common` for convenience; canonical import paths are
|
||||||
|
`src.adaptive_layout` / `src.adaptive_images`.
|
||||||
|
|
||||||
|
```python
|
||||||
|
# Every BasePlugin already has self.layout and the draw helpers:
|
||||||
|
regs = scoreboard_regions(self.layout.bounds, ctx=self.layout)
|
||||||
|
self.draw_image(away_logo, regs.away_slot, mode="fill_height",
|
||||||
|
crop_to_ink=True, cache_key=f"logo:{abbr}")
|
||||||
|
self.draw_fit(score_text, regs.score_area) # largest crisp font that fits
|
||||||
|
self.draw_fit(status, regs.status_band)
|
||||||
|
```
|
||||||
|
|
||||||
|
Key pieces: `Region` (rect algebra: bands/columns/splits/offset),
|
||||||
|
font ladders (`LADDER_GRID`, `LADDER_ARCADE` — discrete crisp sizes, never
|
||||||
|
fractional scaling), `LayoutContext` (`fit_text`, `fit_image`, `by_tier`,
|
||||||
|
`px`), and composite carvers `scoreboard_regions()` / `media_row()`.
|
||||||
|
Full guide: [docs/ADAPTIVE_LAYOUT.md](../../docs/ADAPTIVE_LAYOUT.md).
|
||||||
|
|
||||||
## Error Handling (`error_handler.py`)
|
## Error Handling (`error_handler.py`)
|
||||||
|
|
||||||
Common error handling patterns and utilities:
|
Common error handling patterns and utilities:
|
||||||
|
|||||||
@@ -26,6 +26,31 @@ from src.common.scroll_helper import ScrollHelper
|
|||||||
from src.common.logo_helper import LogoHelper
|
from src.common.logo_helper import LogoHelper
|
||||||
from src.common.text_helper import TextHelper
|
from src.common.text_helper import TextHelper
|
||||||
|
|
||||||
|
# Adaptive layout & images (canonical homes: src.adaptive_layout /
|
||||||
|
# src.adaptive_images — re-exported here so plugin authors find them in the
|
||||||
|
# blessed-helpers package). See docs/ADAPTIVE_LAYOUT.md.
|
||||||
|
from src.adaptive_layout import (
|
||||||
|
Region,
|
||||||
|
LayoutContext,
|
||||||
|
FontStep,
|
||||||
|
FontLadder,
|
||||||
|
LADDER_GRID,
|
||||||
|
LADDER_ARCADE,
|
||||||
|
FitResult,
|
||||||
|
draw_fitted_text,
|
||||||
|
ScoreboardRegions,
|
||||||
|
scoreboard_regions,
|
||||||
|
MediaRow,
|
||||||
|
media_row,
|
||||||
|
)
|
||||||
|
from src.adaptive_images import (
|
||||||
|
ImageFitResult,
|
||||||
|
fit_image,
|
||||||
|
draw_fitted_image,
|
||||||
|
RESAMPLE_LANCZOS,
|
||||||
|
RESAMPLE_NEAREST,
|
||||||
|
)
|
||||||
|
|
||||||
__all__ = [
|
__all__ = [
|
||||||
'handle_file_operation',
|
'handle_file_operation',
|
||||||
'handle_json_operation',
|
'handle_json_operation',
|
||||||
@@ -37,4 +62,22 @@ __all__ = [
|
|||||||
'ScrollHelper',
|
'ScrollHelper',
|
||||||
'LogoHelper',
|
'LogoHelper',
|
||||||
'TextHelper',
|
'TextHelper',
|
||||||
|
# adaptive layout & images
|
||||||
|
'Region',
|
||||||
|
'LayoutContext',
|
||||||
|
'FontStep',
|
||||||
|
'FontLadder',
|
||||||
|
'LADDER_GRID',
|
||||||
|
'LADDER_ARCADE',
|
||||||
|
'FitResult',
|
||||||
|
'draw_fitted_text',
|
||||||
|
'ScoreboardRegions',
|
||||||
|
'scoreboard_regions',
|
||||||
|
'MediaRow',
|
||||||
|
'media_row',
|
||||||
|
'ImageFitResult',
|
||||||
|
'fit_image',
|
||||||
|
'draw_fitted_image',
|
||||||
|
'RESAMPLE_LANCZOS',
|
||||||
|
'RESAMPLE_NEAREST',
|
||||||
]
|
]
|
||||||
|
|||||||
@@ -72,9 +72,20 @@ class LogoHelper:
|
|||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
PIL Image object or None if loading fails
|
PIL Image object or None if loading fails
|
||||||
|
|
||||||
|
Note: for new adaptive-layout code prefer ``BasePlugin.draw_image``
|
||||||
|
/ ``LayoutContext.fit_image`` (src/adaptive_images.py) for the
|
||||||
|
fitting step — LogoHelper remains useful for its download and
|
||||||
|
placeholder logic.
|
||||||
"""
|
"""
|
||||||
# Check cache first
|
# Resolve the effective target size BEFORE the cache lookup so the
|
||||||
cache_key = f"{team_abbr}_{logo_path}"
|
# key is size-qualified — a panel-size change must not return a
|
||||||
|
# logo resized for the old dimensions.
|
||||||
|
if max_width is None:
|
||||||
|
max_width = int(self.display_width * 1.5)
|
||||||
|
if max_height is None:
|
||||||
|
max_height = int(self.display_height * 1.5)
|
||||||
|
cache_key = f"{team_abbr}_{logo_path}_{max_width}x{max_height}"
|
||||||
if cache_key in self._logo_cache:
|
if cache_key in self._logo_cache:
|
||||||
self.logger.debug(f"Using cached logo for {team_abbr}")
|
self.logger.debug(f"Using cached logo for {team_abbr}")
|
||||||
# Update LRU order (move to end)
|
# Update LRU order (move to end)
|
||||||
|
|||||||
@@ -0,0 +1,68 @@
|
|||||||
|
"""Snapshot write policy for the display preview mirror.
|
||||||
|
|
||||||
|
The display service mirrors frames to /tmp/led_matrix_preview.png, which
|
||||||
|
serves two consumers with different needs:
|
||||||
|
|
||||||
|
- The web UI's live preview (SSE reader in web_interface/app.py) wants
|
||||||
|
fresh frames — but only while a browser is actually watching.
|
||||||
|
- The health check (web_interface/blueprints/api_v3.py, hardware status)
|
||||||
|
uses the file's AGE as a liveness proxy: age >= 60s reads as degraded.
|
||||||
|
|
||||||
|
PNG-encoding every frame at 5 fps forever — identical frames, no viewers —
|
||||||
|
was one of the biggest fixed CPU costs on the Pi. This module is the pure
|
||||||
|
decision logic (extracted so it's unit-testable off-Pi; display_manager
|
||||||
|
imports rgbmatrix unconditionally and can't be):
|
||||||
|
|
||||||
|
WRITE — encode + atomically replace the snapshot file
|
||||||
|
TOUCH — os.utime only: keeps the health-check mtime fresh and lets
|
||||||
|
the SSE reader (mtime-gated) resend at a low rate, without
|
||||||
|
paying for a PNG encode of an unchanged frame
|
||||||
|
SKIP — do nothing
|
||||||
|
|
||||||
|
Policy:
|
||||||
|
- With a fresh viewer marker: changed frames write at up to 1/VIEWER_INTERVAL.
|
||||||
|
- Without viewers: changed frames still write at 1/IDLE_INTERVAL so the
|
||||||
|
preview page shows something recent on open.
|
||||||
|
- Unchanged frames are never re-encoded; the mtime is touched every
|
||||||
|
TOUCH_INTERVAL so the health check (60s threshold) never degrades.
|
||||||
|
|
||||||
|
If any constant here changes, re-check the health threshold in
|
||||||
|
api_v3.py (get_hardware_status) — TOUCH_INTERVAL must stay well under it.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from enum import Enum
|
||||||
|
|
||||||
|
# Snapshot cadence with a browser preview open (seconds).
|
||||||
|
VIEWER_INTERVAL = 0.2
|
||||||
|
# Snapshot cadence with no viewers — cheap freshness for page-open (seconds).
|
||||||
|
IDLE_INTERVAL = 30.0
|
||||||
|
# Max age of the last write/touch before bumping mtime for the health
|
||||||
|
# check. MUST stay well under api_v3's 60s degraded threshold.
|
||||||
|
TOUCH_INTERVAL = 20.0
|
||||||
|
# A viewer marker older than this no longer counts as a live viewer.
|
||||||
|
VIEWER_MARKER_FRESH_SEC = 5.0
|
||||||
|
|
||||||
|
|
||||||
|
class SnapshotAction(Enum):
|
||||||
|
WRITE = "write"
|
||||||
|
TOUCH = "touch"
|
||||||
|
SKIP = "skip"
|
||||||
|
|
||||||
|
|
||||||
|
def decide(now: float, last_write_ts: float, last_touch_ts: float,
|
||||||
|
viewer_fresh: bool, frame_changed: bool) -> SnapshotAction:
|
||||||
|
"""Decide what to do with the current frame.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
now: current monotonic-ish timestamp (same clock as the ts args)
|
||||||
|
last_write_ts: when a frame was last actually encoded+written
|
||||||
|
last_touch_ts: when the file mtime was last bumped (write or touch)
|
||||||
|
viewer_fresh: a browser preview is currently watching
|
||||||
|
frame_changed: the frame differs from the last WRITTEN frame
|
||||||
|
"""
|
||||||
|
interval = VIEWER_INTERVAL if viewer_fresh else IDLE_INTERVAL
|
||||||
|
if frame_changed and (now - last_write_ts) >= interval:
|
||||||
|
return SnapshotAction.WRITE
|
||||||
|
if (now - max(last_write_ts, last_touch_ts)) >= TOUCH_INTERVAL:
|
||||||
|
return SnapshotAction.TOUCH
|
||||||
|
return SnapshotAction.SKIP
|
||||||
@@ -11,6 +11,9 @@ from typing import Dict, List, Optional, Tuple, Union
|
|||||||
|
|
||||||
from PIL import Image, ImageDraw, ImageFont
|
from PIL import Image, ImageDraw, ImageFont
|
||||||
|
|
||||||
|
# Shared throwaway draw surface for measuring text without a target canvas.
|
||||||
|
_measure_draw = ImageDraw.Draw(Image.new("RGB", (1, 1)))
|
||||||
|
|
||||||
|
|
||||||
class TextHelper:
|
class TextHelper:
|
||||||
"""
|
"""
|
||||||
@@ -112,10 +115,10 @@ class TextHelper:
|
|||||||
Width in pixels
|
Width in pixels
|
||||||
"""
|
"""
|
||||||
try:
|
try:
|
||||||
return draw.textlength(text, font=font)
|
return int(_measure_draw.textlength(text, font=font))
|
||||||
except AttributeError:
|
except AttributeError:
|
||||||
# Fallback for older PIL versions
|
# Fallback for older PIL versions
|
||||||
bbox = draw.textbbox((0, 0), text, font=font)
|
bbox = _measure_draw.textbbox((0, 0), text, font=font)
|
||||||
return bbox[2] - bbox[0]
|
return bbox[2] - bbox[0]
|
||||||
|
|
||||||
def get_text_height(self, text: str, font: ImageFont.ImageFont) -> int:
|
def get_text_height(self, text: str, font: ImageFont.ImageFont) -> int:
|
||||||
@@ -129,13 +132,8 @@ class TextHelper:
|
|||||||
Returns:
|
Returns:
|
||||||
Height in pixels
|
Height in pixels
|
||||||
"""
|
"""
|
||||||
try:
|
bbox = _measure_draw.textbbox((0, 0), text, font=font)
|
||||||
bbox = draw.textbbox((0, 0), text, font=font)
|
return bbox[3] - bbox[1]
|
||||||
return bbox[3] - bbox[1]
|
|
||||||
except AttributeError:
|
|
||||||
# Fallback for older PIL versions
|
|
||||||
bbox = draw.textbbox((0, 0), text, font=font)
|
|
||||||
return bbox[3] - bbox[1]
|
|
||||||
|
|
||||||
def get_text_dimensions(self, text: str, font: ImageFont.ImageFont) -> Tuple[int, int]:
|
def get_text_dimensions(self, text: str, font: ImageFont.ImageFont) -> Tuple[int, int]:
|
||||||
"""
|
"""
|
||||||
|
|||||||
+78
-14
@@ -199,6 +199,10 @@ class DisplayController:
|
|||||||
self.wifi_status_file = WIFI_STATUS_FILE
|
self.wifi_status_file = WIFI_STATUS_FILE
|
||||||
self.wifi_status_active = False
|
self.wifi_status_active = False
|
||||||
self.wifi_status_expires_at: Optional[float] = None
|
self.wifi_status_expires_at: Optional[float] = None
|
||||||
|
# _check_wifi_status_message throttle state (checked at frame rate,
|
||||||
|
# stat'd at most once per second)
|
||||||
|
self._wifi_status_check_ts = 0.0
|
||||||
|
self._wifi_status_last_result: Optional[Dict[str, Any]] = None
|
||||||
|
|
||||||
# Plugin display() signature cache — must be initialised before the plugin
|
# Plugin display() signature cache — must be initialised before the plugin
|
||||||
# loading loop below so the .pop() invalidation at load time is always safe.
|
# loading loop below so the .pop() invalidation at load time is always safe.
|
||||||
@@ -502,7 +506,10 @@ class DisplayController:
|
|||||||
|
|
||||||
# Run plugin updates inside the Vegas loop so the inter-iteration
|
# Run plugin updates inside the Vegas loop so the inter-iteration
|
||||||
# gap is <1 ms (nothing left for _tick_plugin_updates() to do).
|
# gap is <1 ms (nothing left for _tick_plugin_updates() to do).
|
||||||
self.vegas_coordinator.set_update_callback(self._tick_plugin_updates)
|
# Use the Vegas-aware variant so plugins that got fresh data are
|
||||||
|
# hot-swapped into the scroll promptly instead of waiting for the
|
||||||
|
# next full cycle.
|
||||||
|
self.vegas_coordinator.set_update_callback(self._tick_plugin_updates_for_vegas)
|
||||||
|
|
||||||
# Wire multi-display sync into Vegas render pipeline
|
# Wire multi-display sync into Vegas render pipeline
|
||||||
follower_pos = self.config.get("sync", {}).get("follower_position", "left")
|
follower_pos = self.config.get("sync", {}).get("follower_position", "left")
|
||||||
@@ -621,18 +628,28 @@ class DisplayController:
|
|||||||
|
|
||||||
current_day = current_time.strftime('%A').lower() # e.g. 'monday'
|
current_day = current_time.strftime('%A').lower() # e.g. 'monday'
|
||||||
current_time_only = current_time.time()
|
current_time_only = current_time.time()
|
||||||
|
|
||||||
# Check if per-day schedule is configured
|
# Check if per-day schedule is configured
|
||||||
days_config = schedule_config.get('days')
|
days_config = schedule_config.get('days')
|
||||||
|
|
||||||
# Determine which schedule to use
|
# Determine which schedule to use. Respect an explicit 'mode' field
|
||||||
|
# (like the dim schedule does) so a stray/legacy 'days' dict left over
|
||||||
|
# from config migration or a prior per-day setup can't silently
|
||||||
|
# override a user's Global schedule selection.
|
||||||
|
mode = schedule_config.get('mode')
|
||||||
|
mode_normalized = mode.replace('_', '-') if mode else None
|
||||||
|
|
||||||
use_per_day = False
|
use_per_day = False
|
||||||
if days_config:
|
if mode_normalized == 'global':
|
||||||
# Check if days dict is not empty and contains current day
|
use_per_day = False
|
||||||
if days_config and current_day in days_config:
|
elif mode_normalized == 'per-day':
|
||||||
|
use_per_day = bool(days_config and current_day in days_config)
|
||||||
|
elif days_config:
|
||||||
|
# No explicit mode recorded (legacy config) - fall back to
|
||||||
|
# inferring from presence of a 'days' dict for the current day.
|
||||||
|
if current_day in days_config:
|
||||||
use_per_day = True
|
use_per_day = True
|
||||||
elif days_config:
|
else:
|
||||||
# Days dict exists but doesn't have current day - fall back to global
|
|
||||||
logger.debug("Per-day schedule exists but %s not configured, using global schedule", current_day)
|
logger.debug("Per-day schedule exists but %s not configured, using global schedule", current_day)
|
||||||
|
|
||||||
if use_per_day:
|
if use_per_day:
|
||||||
@@ -828,6 +845,42 @@ class DisplayController:
|
|||||||
if hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
|
if hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
|
||||||
self.plugin_manager.health_tracker.record_failure(plugin_id, exc)
|
self.plugin_manager.health_tracker.record_failure(plugin_id, exc)
|
||||||
|
|
||||||
|
def _tick_plugin_updates_for_vegas(self) -> None:
|
||||||
|
"""Run scheduled plugin updates and tell Vegas mode which plugins
|
||||||
|
actually got fresh data, so it can hot-swap them into the scroll
|
||||||
|
without waiting for a full cycle to complete.
|
||||||
|
|
||||||
|
Used as the Vegas coordinator's update callback instead of the plain
|
||||||
|
_tick_plugin_updates() so that a live score change is reflected in
|
||||||
|
the ticker within a few seconds rather than at the next cycle
|
||||||
|
boundary (which, depending on min/max_cycle_duration, can be
|
||||||
|
minutes away). Restores wiring that PR #299 added and PR #330's
|
||||||
|
sync-mode refactor inadvertently dropped: coordinator.mark_plugin_updated()
|
||||||
|
has been unreachable dead code since.
|
||||||
|
|
||||||
|
Delegates the before/after plugin_last_update snapshot to
|
||||||
|
PluginManager.run_scheduled_updates_with_changes() so the snapshot,
|
||||||
|
update pass, and diff are lock-protected against this callback's own
|
||||||
|
background update-tick thread racing the main render loop.
|
||||||
|
"""
|
||||||
|
if not self.plugin_manager or not hasattr(self.plugin_manager, "run_scheduled_updates_with_changes"):
|
||||||
|
self._tick_plugin_updates()
|
||||||
|
return
|
||||||
|
|
||||||
|
updated = self.plugin_manager.run_scheduled_updates_with_changes()
|
||||||
|
|
||||||
|
vc = getattr(self, "vegas_coordinator", None)
|
||||||
|
if vc is None:
|
||||||
|
return
|
||||||
|
|
||||||
|
if updated:
|
||||||
|
logger.info("Vegas update tick: %d plugin(s) updated: %s", len(updated), updated)
|
||||||
|
for plugin_id in updated:
|
||||||
|
try:
|
||||||
|
vc.mark_plugin_updated(plugin_id)
|
||||||
|
except Exception: # pylint: disable=broad-except
|
||||||
|
logger.exception("Error marking plugin %s updated for Vegas", plugin_id)
|
||||||
|
|
||||||
def _tick_plugin_updates(self):
|
def _tick_plugin_updates(self):
|
||||||
"""Run scheduled plugin updates if the plugin manager supports them."""
|
"""Run scheduled plugin updates if the plugin manager supports them."""
|
||||||
if not self.plugin_manager:
|
if not self.plugin_manager:
|
||||||
@@ -1635,7 +1688,7 @@ class DisplayController:
|
|||||||
self._sleep_with_plugin_updates(60)
|
self._sleep_with_plugin_updates(60)
|
||||||
continue
|
continue
|
||||||
|
|
||||||
logger.info(f"Display active, processing mode: {self.current_display_mode}")
|
logger.debug("Display active, processing mode: %s", self.current_display_mode)
|
||||||
|
|
||||||
# Plugins update on their own schedules - no forced sync updates needed
|
# Plugins update on their own schedules - no forced sync updates needed
|
||||||
# Each plugin has its own update_interval and background services
|
# Each plugin has its own update_interval and background services
|
||||||
@@ -1803,7 +1856,7 @@ class DisplayController:
|
|||||||
if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
|
if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
|
||||||
should_skip = self.plugin_manager.health_tracker.should_skip_plugin(plugin_id)
|
should_skip = self.plugin_manager.health_tracker.should_skip_plugin(plugin_id)
|
||||||
if should_skip:
|
if should_skip:
|
||||||
logger.info(f"Skipping plugin {plugin_id} due to circuit breaker (mode: {active_mode})")
|
logger.info("Skipping plugin %s due to circuit breaker (mode: %s)", plugin_id, active_mode)
|
||||||
display_result = False
|
display_result = False
|
||||||
# Skip to next mode - let existing logic handle it
|
# Skip to next mode - let existing logic handle it
|
||||||
manager_to_display = None
|
manager_to_display = None
|
||||||
@@ -1861,7 +1914,7 @@ class DisplayController:
|
|||||||
if isinstance(result, bool):
|
if isinstance(result, bool):
|
||||||
display_result = result
|
display_result = result
|
||||||
if not display_result:
|
if not display_result:
|
||||||
logger.info(f"Plugin {plugin_id} display() returned False for mode {active_mode}")
|
logger.info("Plugin %s display() returned False for mode %s", plugin_id, active_mode)
|
||||||
|
|
||||||
# Record success if display completed without exception
|
# Record success if display completed without exception
|
||||||
if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
|
if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
|
||||||
@@ -2354,6 +2407,16 @@ class DisplayController:
|
|||||||
Returns None on any error or if message is expired/invalid.
|
Returns None on any error or if message is expired/invalid.
|
||||||
"""
|
"""
|
||||||
try:
|
try:
|
||||||
|
# Throttle the existence stat to ~1 Hz: this runs on every render
|
||||||
|
# iteration (60+ fps), and the file usually doesn't exist — the
|
||||||
|
# status message's lifetime is measured in seconds anyway.
|
||||||
|
# Both attributes are initialised in __init__.
|
||||||
|
now = time.time()
|
||||||
|
if (now - self._wifi_status_check_ts) < 1.0:
|
||||||
|
return self._wifi_status_last_result
|
||||||
|
self._wifi_status_check_ts = now
|
||||||
|
self._wifi_status_last_result = None
|
||||||
|
|
||||||
# Check if file exists
|
# Check if file exists
|
||||||
if not self.wifi_status_file or not self.wifi_status_file.exists():
|
if not self.wifi_status_file or not self.wifi_status_file.exists():
|
||||||
return None
|
return None
|
||||||
@@ -2404,13 +2467,14 @@ class DisplayController:
|
|||||||
pass
|
pass
|
||||||
return None
|
return None
|
||||||
|
|
||||||
# Message is valid and not expired
|
# Message is valid and not expired — cache for the throttle window
|
||||||
return {
|
self._wifi_status_last_result = {
|
||||||
'message': message,
|
'message': message,
|
||||||
'timestamp': timestamp,
|
'timestamp': timestamp,
|
||||||
'duration': duration,
|
'duration': duration,
|
||||||
'expires_at': expires_at
|
'expires_at': expires_at
|
||||||
}
|
}
|
||||||
|
return self._wifi_status_last_result
|
||||||
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
# Catch-all for any unexpected errors - log but don't break the display
|
# Catch-all for any unexpected errors - log but don't break the display
|
||||||
|
|||||||
+103
-26
@@ -31,13 +31,24 @@ if os.getenv("EMULATOR", "false") == "true":
|
|||||||
else:
|
else:
|
||||||
from rgbmatrix import RGBMatrix, RGBMatrixOptions
|
from rgbmatrix import RGBMatrix, RGBMatrixOptions
|
||||||
from contextlib import contextmanager
|
from contextlib import contextmanager
|
||||||
|
from pathlib import Path
|
||||||
from PIL import Image, ImageDraw, ImageFont
|
from PIL import Image, ImageDraw, ImageFont
|
||||||
import time
|
import time
|
||||||
from typing import Dict, Any, List, Optional
|
from collections import OrderedDict
|
||||||
|
from typing import Dict, Any, List, Optional, Tuple
|
||||||
import logging
|
import logging
|
||||||
import math
|
import math
|
||||||
|
import zlib
|
||||||
import freetype
|
import freetype
|
||||||
|
|
||||||
|
from src.common import snapshot_policy
|
||||||
|
from src.common.permission_utils import (
|
||||||
|
ensure_directory_permissions,
|
||||||
|
ensure_file_permissions,
|
||||||
|
get_assets_dir_mode,
|
||||||
|
get_assets_file_mode,
|
||||||
|
)
|
||||||
|
|
||||||
# Get logger without configuring
|
# Get logger without configuring
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
logger.setLevel(logging.INFO) # Set to INFO level
|
logger.setLevel(logging.INFO) # Set to INFO level
|
||||||
@@ -180,14 +191,34 @@ class DisplayManager:
|
|||||||
# the logical image is blitted to the matrix unchanged.
|
# the logical image is blitted to the matrix unchanged.
|
||||||
self._double_sided = None # dict {copies, axis, logical_width, logical_height} or None
|
self._double_sided = None # dict {copies, axis, logical_width, logical_height} or None
|
||||||
self._physical_image = None # full-chain buffer reused each frame when tiling
|
self._physical_image = None # full-chain buffer reused each frame when tiling
|
||||||
# Text-width measurement cache: (text, id(font)) -> pixel_width
|
# Text-width measurement cache: (text, id(font)) -> (width, font_ref)
|
||||||
# Avoids re-measuring the same string+font on every display() call.
|
# Avoids re-measuring the same string+font on every display() call.
|
||||||
|
# LRU-bounded: keys embed the TEXT, so changing strings (a clock, a
|
||||||
|
# live score) would otherwise grow it forever on a 24/7 service.
|
||||||
|
# Entries hold a strong reference to the font so its id() can't be
|
||||||
|
# recycled by a different font object — an id-keyed cache without
|
||||||
|
# the reference can return the WRONG width after garbage collection.
|
||||||
# Cleared on _load_fonts() so stale entries don't survive a font reload.
|
# Cleared on _load_fonts() so stale entries don't survive a font reload.
|
||||||
self._text_width_cache: Dict[tuple, int] = {}
|
self._text_width_cache: "OrderedDict[tuple, Tuple[int, Any]]" = OrderedDict()
|
||||||
# Snapshot settings for web preview integration (service writes, web reads)
|
self._TEXT_WIDTH_CACHE_MAX = 1024
|
||||||
|
# Snapshot mirror for web preview + health check (service writes, web
|
||||||
|
# reads). Cadence/skip decisions live in src/common/snapshot_policy.py:
|
||||||
|
# full rate only while the web SSE broadcaster keeps the viewer marker
|
||||||
|
# fresh; unchanged frames are never re-encoded, only mtime-touched.
|
||||||
self._snapshot_path = "/tmp/led_matrix_preview.png" # nosec B108 - fixed path intentional; web UI reads same path
|
self._snapshot_path = "/tmp/led_matrix_preview.png" # nosec B108 - fixed path intentional; web UI reads same path
|
||||||
self._snapshot_min_interval_sec = 0.2 # max ~5 fps
|
self._viewer_marker_path = "/tmp/led_matrix_preview_viewer" # nosec B108 - touched by web SSE broadcaster
|
||||||
self._last_snapshot_ts = 0.0
|
self._last_snapshot_ts = 0.0
|
||||||
|
self._last_snapshot_touch_ts = 0.0
|
||||||
|
self._last_snapshot_digest: Optional[int] = None
|
||||||
|
self._snapshot_dir_prepared = False
|
||||||
|
self._viewer_check_ts = 0.0
|
||||||
|
self._viewer_fresh = False
|
||||||
|
self._viewer_was_fresh = False
|
||||||
|
# Snapshot failures are logged as warnings, rate-limited so a
|
||||||
|
# persistent failure (e.g. an unwritable file) can't spam the log —
|
||||||
|
# but is never silent: the snapshot's mtime doubles as the web UI's
|
||||||
|
# hardware-liveness signal, so a quiet failure makes health checks lie.
|
||||||
|
self._snapshot_fail_log_ts = 0.0
|
||||||
|
|
||||||
# Scrolling state tracking for graceful updates
|
# Scrolling state tracking for graceful updates
|
||||||
self._scrolling_state = {
|
self._scrolling_state = {
|
||||||
@@ -699,12 +730,15 @@ class DisplayManager:
|
|||||||
|
|
||||||
Results are cached by (text, font identity) so plugins that measure
|
Results are cached by (text, font identity) so plugins that measure
|
||||||
the same string every frame (e.g. to centre a score) pay only one
|
the same string every frame (e.g. to centre a score) pay only one
|
||||||
measurement per unique (text, font) pair.
|
measurement per unique (text, font) pair. The entry keeps the font
|
||||||
|
alive so its id() can't be recycled, and the cache is LRU-bounded so
|
||||||
|
ever-changing text (clocks, tickers) can't grow it without limit.
|
||||||
"""
|
"""
|
||||||
cache_key = (text, id(font))
|
cache_key = (text, id(font))
|
||||||
cached = self._text_width_cache.get(cache_key)
|
cached = self._text_width_cache.get(cache_key)
|
||||||
if cached is not None:
|
if cached is not None:
|
||||||
return cached
|
self._text_width_cache.move_to_end(cache_key)
|
||||||
|
return cached[0]
|
||||||
|
|
||||||
try:
|
try:
|
||||||
if isinstance(font, freetype.Face):
|
if isinstance(font, freetype.Face):
|
||||||
@@ -719,7 +753,9 @@ class DisplayManager:
|
|||||||
logger.error("Error getting text width: %s", e)
|
logger.error("Error getting text width: %s", e)
|
||||||
return 0
|
return 0
|
||||||
|
|
||||||
self._text_width_cache[cache_key] = width
|
self._text_width_cache[cache_key] = (width, font)
|
||||||
|
while len(self._text_width_cache) > self._TEXT_WIDTH_CACHE_MAX:
|
||||||
|
self._text_width_cache.popitem(last=False)
|
||||||
return width
|
return width
|
||||||
|
|
||||||
def get_font_height(self, font):
|
def get_font_height(self, font):
|
||||||
@@ -1128,27 +1164,56 @@ class DisplayManager:
|
|||||||
'deferred_update_ttl': self._scrolling_state['deferred_update_ttl']
|
'deferred_update_ttl': self._scrolling_state['deferred_update_ttl']
|
||||||
}
|
}
|
||||||
|
|
||||||
|
def _viewer_is_fresh(self, now: float) -> bool:
|
||||||
|
"""True when a browser preview is watching (marker file touched by
|
||||||
|
the web SSE broadcaster). The marker is stat'd at most once per
|
||||||
|
second — at 125 fps loops a per-call stat would be pure overhead."""
|
||||||
|
if (now - self._viewer_check_ts) >= 1.0:
|
||||||
|
self._viewer_check_ts = now
|
||||||
|
try:
|
||||||
|
marker_age = now - os.stat(self._viewer_marker_path).st_mtime
|
||||||
|
self._viewer_fresh = marker_age < snapshot_policy.VIEWER_MARKER_FRESH_SEC
|
||||||
|
except OSError:
|
||||||
|
self._viewer_fresh = False
|
||||||
|
return self._viewer_fresh
|
||||||
|
|
||||||
def _write_snapshot_if_due(self) -> None:
|
def _write_snapshot_if_due(self) -> None:
|
||||||
"""Write the current image to a PNG snapshot file at a limited frequency."""
|
"""Mirror the current frame to the preview snapshot when the policy
|
||||||
|
says it's worth it — see src/common/snapshot_policy.py. Unchanged
|
||||||
|
frames are never re-encoded; without viewers the cadence drops to
|
||||||
|
the idle keepalive."""
|
||||||
try:
|
try:
|
||||||
now = time.time()
|
now = time.time()
|
||||||
if (now - self._last_snapshot_ts) < self._snapshot_min_interval_sec:
|
viewer_fresh = self._viewer_is_fresh(now)
|
||||||
|
if viewer_fresh and not self._viewer_was_fresh:
|
||||||
|
# A preview just opened: let the next changed frame through
|
||||||
|
# immediately instead of waiting out the idle interval.
|
||||||
|
self._last_snapshot_ts = 0.0
|
||||||
|
self._viewer_was_fresh = viewer_fresh
|
||||||
|
|
||||||
|
digest = zlib.adler32(self.image.tobytes())
|
||||||
|
action = snapshot_policy.decide(
|
||||||
|
now, self._last_snapshot_ts, self._last_snapshot_touch_ts,
|
||||||
|
viewer_fresh, digest != self._last_snapshot_digest)
|
||||||
|
if action is snapshot_policy.SnapshotAction.SKIP:
|
||||||
return
|
return
|
||||||
# Ensure directory exists with proper permissions
|
if action is snapshot_policy.SnapshotAction.TOUCH:
|
||||||
from pathlib import Path
|
# mtime bump only: keeps the health check (snapshot age)
|
||||||
from src.common.permission_utils import (
|
# green without paying for a PNG encode of an unchanged frame
|
||||||
ensure_directory_permissions,
|
os.utime(self._snapshot_path, None)
|
||||||
ensure_file_permissions,
|
self._last_snapshot_touch_ts = now
|
||||||
get_assets_dir_mode,
|
return
|
||||||
get_assets_file_mode
|
|
||||||
)
|
# WRITE: ensure directory permissions once, not per frame
|
||||||
snapshot_path_obj = Path(self._snapshot_path)
|
snapshot_path_obj = Path(self._snapshot_path)
|
||||||
# Only ensure permissions on non-system directories
|
if not self._snapshot_dir_prepared:
|
||||||
# Never modify /tmp permissions - it has special system permissions (1777)
|
# Never modify /tmp permissions - it has special system
|
||||||
# that must not be changed or it breaks apt and other system tools
|
# permissions (1777) that must not be changed or it breaks
|
||||||
parent_dir = snapshot_path_obj.parent
|
# apt and other system tools
|
||||||
if parent_dir and str(parent_dir) != '/tmp': # nosec B108 - guard to skip /tmp for permission ops
|
parent_dir = snapshot_path_obj.parent
|
||||||
ensure_directory_permissions(parent_dir, get_assets_dir_mode())
|
if parent_dir and str(parent_dir) != '/tmp': # nosec B108 - guard to skip /tmp for permission ops
|
||||||
|
ensure_directory_permissions(parent_dir, get_assets_dir_mode())
|
||||||
|
self._snapshot_dir_prepared = True
|
||||||
# Write atomically: temp then replace
|
# Write atomically: temp then replace
|
||||||
tmp_path = f"{self._snapshot_path}.tmp"
|
tmp_path = f"{self._snapshot_path}.tmp"
|
||||||
self.image.save(tmp_path, format='PNG')
|
self.image.save(tmp_path, format='PNG')
|
||||||
@@ -1163,6 +1228,18 @@ class DisplayManager:
|
|||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
self._last_snapshot_ts = now
|
self._last_snapshot_ts = now
|
||||||
|
self._last_snapshot_touch_ts = now
|
||||||
|
self._last_snapshot_digest = digest
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
# Snapshot failures should never break display; log at debug to avoid noise
|
# Snapshot failures must never break display — but they must not
|
||||||
logger.debug(f"Snapshot write skipped: {e}")
|
# be silent either: the snapshot's mtime is the web UI's display
|
||||||
|
# mirror AND its hardware-liveness proxy, so a quietly failing
|
||||||
|
# write freezes the mirror and makes health checks lie (seen in
|
||||||
|
# the field: a stale root-owned /tmp file froze it for a day).
|
||||||
|
# Warn at most once per 5 minutes to avoid log spam.
|
||||||
|
if (now - self._snapshot_fail_log_ts) > 300:
|
||||||
|
self._snapshot_fail_log_ts = now
|
||||||
|
logger.warning("Snapshot write failing (web preview/health "
|
||||||
|
"mirror is stale): %s", e)
|
||||||
|
else:
|
||||||
|
logger.debug(f"Snapshot write skipped: {e}")
|
||||||
+66
-5
@@ -35,6 +35,7 @@ import urllib.request
|
|||||||
import zipfile
|
import zipfile
|
||||||
import tempfile
|
import tempfile
|
||||||
import time
|
import time
|
||||||
|
from collections import OrderedDict
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from PIL import ImageFont
|
from PIL import ImageFont
|
||||||
from typing import Dict, Tuple, Optional, Union, Any, List
|
from typing import Dict, Tuple, Optional, Union, Any, List
|
||||||
@@ -58,7 +59,13 @@ class FontManager:
|
|||||||
# Font discovery and catalog
|
# Font discovery and catalog
|
||||||
self.font_catalog: Dict[str, str] = {} # family_name -> file_path
|
self.font_catalog: Dict[str, str] = {} # family_name -> file_path
|
||||||
self.font_cache: Dict[str, Union[ImageFont.FreeTypeFont, freetype.Face]] = {} # (family, size) -> font
|
self.font_cache: Dict[str, Union[ImageFont.FreeTypeFont, freetype.Face]] = {} # (family, size) -> font
|
||||||
self.metrics_cache: Dict[str, Tuple[int, int, int]] = {} # (text, font_id) -> (width, height, baseline)
|
# (text, id(font)) -> ((width, height, baseline), font_ref).
|
||||||
|
# LRU-bounded — keys embed the measured TEXT, so changing strings
|
||||||
|
# (clocks, live scores) would otherwise grow it forever. Entries
|
||||||
|
# keep the font alive so its id() can't be recycled by a different
|
||||||
|
# font object (which would silently return wrong metrics).
|
||||||
|
self.metrics_cache: "OrderedDict[Any, Tuple[Tuple[int, int, int], Any]]" = OrderedDict()
|
||||||
|
self._METRICS_CACHE_MAX = 1024
|
||||||
|
|
||||||
# Plugin font management
|
# Plugin font management
|
||||||
self.plugin_fonts: Dict[str, Dict[str, Any]] = {} # plugin_id -> font_manifest
|
self.plugin_fonts: Dict[str, Dict[str, Any]] = {} # plugin_id -> font_manifest
|
||||||
@@ -103,6 +110,10 @@ class FontManager:
|
|||||||
# Font overrides storage (for manual overrides)
|
# Font overrides storage (for manual overrides)
|
||||||
self.font_overrides_file = "config/font_overrides.json"
|
self.font_overrides_file = "config/font_overrides.json"
|
||||||
self.font_overrides: Dict[str, Dict[str, Any]] = {}
|
self.font_overrides: Dict[str, Dict[str, Any]] = {}
|
||||||
|
|
||||||
|
# Bumped whenever cached font objects are invalidated, so holders of
|
||||||
|
# derived caches (e.g. adaptive-layout fit results) know to rebuild.
|
||||||
|
self.cache_generation = 0
|
||||||
|
|
||||||
self._initialize_fonts()
|
self._initialize_fonts()
|
||||||
|
|
||||||
@@ -112,6 +123,7 @@ class FontManager:
|
|||||||
self.fonts_config = new_config.get("fonts", {})
|
self.fonts_config = new_config.get("fonts", {})
|
||||||
self.font_cache.clear() # Clear cache to force reload
|
self.font_cache.clear() # Clear cache to force reload
|
||||||
self.metrics_cache.clear() # Clear metrics cache
|
self.metrics_cache.clear() # Clear metrics cache
|
||||||
|
self.cache_generation += 1
|
||||||
self._initialize_fonts()
|
self._initialize_fonts()
|
||||||
logger.info("FontManager configuration reloaded successfully")
|
logger.info("FontManager configuration reloaded successfully")
|
||||||
|
|
||||||
@@ -482,6 +494,14 @@ class FontManager:
|
|||||||
def _load_bdf_font(self, font_path: str, size_px: int) -> freetype.Face:
|
def _load_bdf_font(self, font_path: str, size_px: int) -> freetype.Face:
|
||||||
"""Load a BDF font using FreeType."""
|
"""Load a BDF font using FreeType."""
|
||||||
try:
|
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)
|
face = freetype.Face(font_path)
|
||||||
# Set character size (width, height) in 1/64th of points
|
# Set character size (width, height) in 1/64th of points
|
||||||
face.set_char_size(size_px * 64, size_px * 64, 72, 72)
|
face.set_char_size(size_px * 64, size_px * 64, 72, 72)
|
||||||
@@ -490,6 +510,41 @@ class FontManager:
|
|||||||
logger.error(f"Error loading BDF font {font_path}: {e}")
|
logger.error(f"Error loading BDF font {font_path}: {e}")
|
||||||
raise
|
raise
|
||||||
|
|
||||||
|
def get_native_bdf_size(self, family: str) -> Optional[int]:
|
||||||
|
"""The one true pixel size of a BDF family in the catalog, or None
|
||||||
|
for scalable (TTF) families / unknown families."""
|
||||||
|
font_path = self.font_catalog.get(family)
|
||||||
|
if not font_path or not font_path.endswith('.bdf'):
|
||||||
|
return None
|
||||||
|
return self._read_bdf_native_size(font_path)
|
||||||
|
|
||||||
|
@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
|
||||||
|
|
||||||
def _get_fallback_font(self) -> ImageFont.ImageFont:
|
def _get_fallback_font(self) -> ImageFont.ImageFont:
|
||||||
"""Get a fallback font when loading fails."""
|
"""Get a fallback font when loading fails."""
|
||||||
return ImageFont.load_default()
|
return ImageFont.load_default()
|
||||||
@@ -507,10 +562,14 @@ class FontManager:
|
|||||||
Returns:
|
Returns:
|
||||||
Tuple of (width, height, baseline_offset)
|
Tuple of (width, height, baseline_offset)
|
||||||
"""
|
"""
|
||||||
cache_key = f"{hash(text)}_{id(font)}"
|
# Key on the text itself (hash(text) could collide) + font identity;
|
||||||
|
# the entry below keeps the font referenced so the id stays valid.
|
||||||
|
cache_key = (text, id(font))
|
||||||
|
|
||||||
if cache_key in self.metrics_cache:
|
cached = self.metrics_cache.get(cache_key)
|
||||||
return self.metrics_cache[cache_key]
|
if cached is not None:
|
||||||
|
self.metrics_cache.move_to_end(cache_key)
|
||||||
|
return cached[0]
|
||||||
|
|
||||||
try:
|
try:
|
||||||
if isinstance(font, freetype.Face):
|
if isinstance(font, freetype.Face):
|
||||||
@@ -547,7 +606,9 @@ class FontManager:
|
|||||||
baseline = 10
|
baseline = 10
|
||||||
|
|
||||||
result = (width, height, baseline)
|
result = (width, height, baseline)
|
||||||
self.metrics_cache[cache_key] = result
|
self.metrics_cache[cache_key] = (result, font)
|
||||||
|
while len(self.metrics_cache) > self._METRICS_CACHE_MAX:
|
||||||
|
self.metrics_cache.popitem(last=False)
|
||||||
return result
|
return result
|
||||||
|
|
||||||
def get_font_height(self, font: Union[ImageFont.FreeTypeFont, freetype.Face]) -> int:
|
def get_font_height(self, font: Union[ImageFont.FreeTypeFont, freetype.Face]) -> int:
|
||||||
|
|||||||
@@ -1,3 +1,9 @@
|
|||||||
|
"""Deprecated: use src/adaptive_images.py (fit_image) instead.
|
||||||
|
|
||||||
|
This module predates the adaptive image system and has no known callers.
|
||||||
|
It is kept only so any out-of-tree code importing it keeps working.
|
||||||
|
"""
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
from PIL import Image
|
from PIL import Image
|
||||||
|
|
||||||
|
|||||||
@@ -15,6 +15,19 @@ import logging
|
|||||||
from src.logging_config import get_logger
|
from src.logging_config import get_logger
|
||||||
|
|
||||||
|
|
||||||
|
_shared_fallback_font_manager: Optional[Any] = None
|
||||||
|
|
||||||
|
|
||||||
|
def _fallback_font_manager() -> Any:
|
||||||
|
"""Shared FontManager for environments (unit tests, mocks) where the
|
||||||
|
plugin manager doesn't carry one. Scans assets/fonts like the real one."""
|
||||||
|
global _shared_fallback_font_manager
|
||||||
|
if _shared_fallback_font_manager is None:
|
||||||
|
from src.font_manager import FontManager
|
||||||
|
_shared_fallback_font_manager = FontManager({})
|
||||||
|
return _shared_fallback_font_manager
|
||||||
|
|
||||||
|
|
||||||
class VegasDisplayMode(Enum):
|
class VegasDisplayMode(Enum):
|
||||||
"""
|
"""
|
||||||
Display mode for Vegas scroll integration.
|
Display mode for Vegas scroll integration.
|
||||||
@@ -130,6 +143,133 @@ class BasePlugin(ABC):
|
|||||||
"""
|
"""
|
||||||
raise NotImplementedError("Plugins must implement display()")
|
raise NotImplementedError("Plugins must implement display()")
|
||||||
|
|
||||||
|
# -------------------------------------------------------------------------
|
||||||
|
# Adaptive layout support (opt-in)
|
||||||
|
# -------------------------------------------------------------------------
|
||||||
|
@property
|
||||||
|
def layout(self) -> Any:
|
||||||
|
"""
|
||||||
|
LayoutContext for the current logical display size.
|
||||||
|
|
||||||
|
Lazily built and rebuilt automatically when the display size changes
|
||||||
|
(e.g. Vegas segment widths, double-sided logical screens). Provides
|
||||||
|
Region carving (self.layout.bounds), breakpoint tiers, a geometry
|
||||||
|
scale factor vs. the manifest's display.design_size, and fit-text
|
||||||
|
queries against font ladders. See src/adaptive_layout.py.
|
||||||
|
|
||||||
|
Example:
|
||||||
|
rows = self.layout.bounds.inset(1).split_v(3, 1, gap=1)
|
||||||
|
self.draw_fit(big_text, rows[0], ladder=LADDER_ARCADE)
|
||||||
|
self.draw_fit(small_text, rows[1])
|
||||||
|
"""
|
||||||
|
from src.adaptive_layout import LayoutContext
|
||||||
|
|
||||||
|
width = getattr(self.display_manager, "width", None)
|
||||||
|
height = getattr(self.display_manager, "height", None)
|
||||||
|
if not width or not height:
|
||||||
|
matrix = getattr(self.display_manager, "matrix", None)
|
||||||
|
width = getattr(matrix, "width", 128)
|
||||||
|
height = getattr(matrix, "height", 32)
|
||||||
|
|
||||||
|
font_manager = self._get_font_manager()
|
||||||
|
generation = getattr(font_manager, "cache_generation", 0)
|
||||||
|
cached = getattr(self, "_layout_context", None)
|
||||||
|
if (cached is not None
|
||||||
|
and (cached.width, cached.height) == (width, height)
|
||||||
|
and getattr(self, "_layout_font_generation", None) == generation):
|
||||||
|
return cached
|
||||||
|
|
||||||
|
context = LayoutContext(
|
||||||
|
width, height, font_manager,
|
||||||
|
design_size=self._get_design_size(),
|
||||||
|
)
|
||||||
|
self._layout_context = context
|
||||||
|
self._layout_font_generation = generation
|
||||||
|
return context
|
||||||
|
|
||||||
|
def draw_fit(self, text: str, box: Any,
|
||||||
|
color: tuple = (255, 255, 255),
|
||||||
|
ladder: Optional[Any] = None,
|
||||||
|
align: str = "center", valign: str = "center") -> Any:
|
||||||
|
"""
|
||||||
|
Fit text to a Region with the largest crisp font that fits, then draw
|
||||||
|
it aligned within that region via the display manager.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
text: Text to display (ellipsized if even the smallest rung is too wide)
|
||||||
|
box: Region (or (w, h) tuple anchored at 0,0) to fit and align within
|
||||||
|
color: RGB color tuple
|
||||||
|
ladder: FontLadder to walk (default LADDER_GRID; use LADDER_ARCADE
|
||||||
|
for headline text like clocks and scores)
|
||||||
|
align/valign: alignment of the text ink within the box
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
FitResult (font, family, size_px, text, ink metrics, fits flag)
|
||||||
|
"""
|
||||||
|
from src.adaptive_layout import LADDER_DEFAULT, draw_fitted_text
|
||||||
|
|
||||||
|
fit = self.layout.fit_text(text, box, ladder=ladder or LADDER_DEFAULT)
|
||||||
|
draw_fitted_text(self.display_manager, fit, box,
|
||||||
|
color=color, align=align, valign=valign)
|
||||||
|
return fit
|
||||||
|
|
||||||
|
def draw_image(self, img: Any, box: Any, *,
|
||||||
|
mode: str = "contain", align: str = "center",
|
||||||
|
valign: str = "center", crop_to_ink: bool = False,
|
||||||
|
anchor: str = "center", resample: Optional[Any] = None,
|
||||||
|
cache_key: Optional[Any] = None,
|
||||||
|
offset: tuple = (0, 0)) -> Any:
|
||||||
|
"""
|
||||||
|
Fit an image into a Region and paste it aligned within that region
|
||||||
|
onto the display canvas — the image counterpart to draw_fit().
|
||||||
|
|
||||||
|
Args:
|
||||||
|
img: Source PIL image (logos, art, icons)
|
||||||
|
box: Region (or (w, h) tuple) to fit and align within
|
||||||
|
mode: "contain" (letterbox), "cover" (crop-to-fill),
|
||||||
|
"fill_height" (logo-style), "stretch"
|
||||||
|
crop_to_ink: Trim transparent padding before fitting
|
||||||
|
anchor: "center" or "top" for cover crops
|
||||||
|
resample: PIL filter; default LANCZOS. Use RESAMPLE_NEAREST
|
||||||
|
(from src.adaptive_images) for pixel art/flags
|
||||||
|
cache_key: Stable identity (e.g. "logo:KC") for cross-reload
|
||||||
|
caching; defaults to the image object's identity
|
||||||
|
offset: Final (dx, dy) translation — the hook for user
|
||||||
|
x/y-offset customization
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
ImageFitResult (processed image + dimensions + scale)
|
||||||
|
"""
|
||||||
|
from src.adaptive_images import draw_fitted_image
|
||||||
|
|
||||||
|
ifit = self.layout.fit_image(img, box, mode=mode,
|
||||||
|
crop_to_ink=crop_to_ink, anchor=anchor,
|
||||||
|
resample=resample, cache_key=cache_key)
|
||||||
|
draw_fitted_image(self.display_manager, ifit, box,
|
||||||
|
align=align, valign=valign, offset=offset)
|
||||||
|
return ifit
|
||||||
|
|
||||||
|
def _get_font_manager(self) -> Any:
|
||||||
|
"""The shared FontManager, or a module-level fallback when running
|
||||||
|
under mocks/harnesses that don't provide one."""
|
||||||
|
font_manager = getattr(self.plugin_manager, "font_manager", None)
|
||||||
|
if font_manager is not None and hasattr(font_manager, "get_font"):
|
||||||
|
return font_manager
|
||||||
|
return _fallback_font_manager()
|
||||||
|
|
||||||
|
def _get_design_size(self) -> tuple:
|
||||||
|
"""Panel size this plugin's layout was authored against, from the
|
||||||
|
manifest's optional display.design_size (defaults to 128x32)."""
|
||||||
|
from src.adaptive_layout import DEFAULT_DESIGN_SIZE
|
||||||
|
|
||||||
|
if self.plugin_manager and hasattr(self.plugin_manager, "plugin_manifests"):
|
||||||
|
manifest = self.plugin_manager.plugin_manifests.get(self.plugin_id, {})
|
||||||
|
declared = manifest.get("display", {}).get("design_size", {})
|
||||||
|
width, height = declared.get("width"), declared.get("height")
|
||||||
|
if width and height:
|
||||||
|
return (int(width), int(height))
|
||||||
|
return DEFAULT_DESIGN_SIZE
|
||||||
|
|
||||||
def get_display_duration(self) -> float:
|
def get_display_duration(self) -> float:
|
||||||
"""
|
"""
|
||||||
Get the display duration for this plugin instance.
|
Get the display duration for this plugin instance.
|
||||||
|
|||||||
@@ -437,8 +437,7 @@ class PluginLoader:
|
|||||||
if not Path(existing_file).resolve().is_relative_to(resolved_dir):
|
if not Path(existing_file).resolve().is_relative_to(resolved_dir):
|
||||||
evicted[mod_name] = sys.modules.pop(mod_name)
|
evicted[mod_name] = sys.modules.pop(mod_name)
|
||||||
self.logger.debug(
|
self.logger.debug(
|
||||||
"Evicted stale module '%s' (from %s) before loading plugin in %s",
|
"Evicted stale bare-name module '%s' before loading plugin", mod_name,
|
||||||
mod_name, existing_file, plugin_dir,
|
|
||||||
)
|
)
|
||||||
except (ValueError, TypeError):
|
except (ValueError, TypeError):
|
||||||
continue
|
continue
|
||||||
@@ -551,7 +550,7 @@ class PluginLoader:
|
|||||||
plugin_dir_str = str(plugin_dir)
|
plugin_dir_str = str(plugin_dir)
|
||||||
if plugin_dir_str not in sys.path:
|
if plugin_dir_str not in sys.path:
|
||||||
sys.path.insert(0, plugin_dir_str)
|
sys.path.insert(0, plugin_dir_str)
|
||||||
self.logger.debug("Added plugin directory to sys.path: %s", plugin_dir_str)
|
self.logger.debug("Added plugin %s's directory to sys.path", plugin_id)
|
||||||
|
|
||||||
# Import the plugin module
|
# Import the plugin module
|
||||||
module_name = f"plugin_{plugin_id.replace('-', '_')}"
|
module_name = f"plugin_{plugin_id.replace('-', '_')}"
|
||||||
@@ -563,8 +562,8 @@ class PluginLoader:
|
|||||||
|
|
||||||
spec = importlib.util.spec_from_file_location(module_name, entry_file)
|
spec = importlib.util.spec_from_file_location(module_name, entry_file)
|
||||||
if spec is None or spec.loader is None:
|
if spec is None or spec.loader is None:
|
||||||
|
self.logger.error("Could not create module spec for plugin %s", plugin_id)
|
||||||
error_msg = f"Could not create module spec for {entry_file}"
|
error_msg = f"Could not create module spec for {entry_file}"
|
||||||
self.logger.error(error_msg)
|
|
||||||
raise PluginError(error_msg, plugin_id=plugin_id, context={'entry_file': str(entry_file)})
|
raise PluginError(error_msg, plugin_id=plugin_id, context={'entry_file': str(entry_file)})
|
||||||
|
|
||||||
module = importlib.util.module_from_spec(spec)
|
module = importlib.util.module_from_spec(spec)
|
||||||
@@ -683,6 +682,55 @@ class PluginLoader:
|
|||||||
self.logger.error(error_msg, exc_info=True)
|
self.logger.error(error_msg, exc_info=True)
|
||||||
raise PluginError(error_msg, plugin_id=plugin_id) from e
|
raise PluginError(error_msg, plugin_id=plugin_id) from e
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _parse_semver(value: Any) -> Optional[Tuple[int, int, int]]:
|
||||||
|
"""Parse 'X.Y.Z' (extra parts/suffixes ignored) into a comparable
|
||||||
|
3-tuple, or None when unparseable."""
|
||||||
|
if not isinstance(value, str):
|
||||||
|
return None
|
||||||
|
parts = value.strip().lstrip('v').split('.')
|
||||||
|
try:
|
||||||
|
nums = [int(''.join(ch for ch in p if ch.isdigit()) or 0) for p in parts[:3]]
|
||||||
|
except ValueError:
|
||||||
|
return None
|
||||||
|
while len(nums) < 3:
|
||||||
|
nums.append(0)
|
||||||
|
return tuple(nums) # type: ignore[return-value]
|
||||||
|
|
||||||
|
def _warn_if_incompatible(self, plugin_id: str, manifest: Dict[str, Any]) -> None:
|
||||||
|
"""Log one warning when a plugin declares a minimum LEDMatrix version
|
||||||
|
newer than the running core. Advisory only — never raises — so a
|
||||||
|
plugin that guards optional features with try/except keeps working.
|
||||||
|
"""
|
||||||
|
declared = (
|
||||||
|
manifest.get('min_ledmatrix_version')
|
||||||
|
or manifest.get('requires', {}).get('min_ledmatrix_version')
|
||||||
|
)
|
||||||
|
if not declared:
|
||||||
|
versions = manifest.get('versions') or []
|
||||||
|
if versions and isinstance(versions[0], dict):
|
||||||
|
declared = (versions[0].get('ledmatrix_min_version')
|
||||||
|
or versions[0].get('ledmatrix_min'))
|
||||||
|
needed = self._parse_semver(declared)
|
||||||
|
if needed is None:
|
||||||
|
return
|
||||||
|
|
||||||
|
from src import __version__ as core_version
|
||||||
|
current = self._parse_semver(core_version)
|
||||||
|
# Anti-spam guard: if the core's own version number is stale (below
|
||||||
|
# the ecosystem floor every shipped plugin declares), comparing would
|
||||||
|
# warn on nearly everything — skip with a debug note instead.
|
||||||
|
if current is None or current < (2, 0, 0):
|
||||||
|
self.logger.debug(
|
||||||
|
"Skipping version compatibility check for %s: core __version__ "
|
||||||
|
"(%s) is below the ecosystem floor", plugin_id, core_version)
|
||||||
|
return
|
||||||
|
if needed > current:
|
||||||
|
self.logger.warning(
|
||||||
|
"Plugin %s declares min LEDMatrix version %s but this core is %s — "
|
||||||
|
"features it relies on may be missing; update the core or expect "
|
||||||
|
"degraded fallbacks", plugin_id, declared, core_version)
|
||||||
|
|
||||||
def load_plugin(
|
def load_plugin(
|
||||||
self,
|
self,
|
||||||
plugin_id: str,
|
plugin_id: str,
|
||||||
@@ -715,6 +763,8 @@ class PluginLoader:
|
|||||||
Raises:
|
Raises:
|
||||||
PluginError: If loading fails
|
PluginError: If loading fails
|
||||||
"""
|
"""
|
||||||
|
self._warn_if_incompatible(plugin_id, manifest)
|
||||||
|
|
||||||
# Install dependencies if needed
|
# Install dependencies if needed
|
||||||
if install_deps:
|
if install_deps:
|
||||||
if plugins_dir is None:
|
if plugins_dir is None:
|
||||||
|
|||||||
@@ -76,6 +76,12 @@ class PluginManager:
|
|||||||
# concurrent mutation (background reconciliation) and reads (requests).
|
# concurrent mutation (background reconciliation) and reads (requests).
|
||||||
self._discovery_lock = threading.RLock()
|
self._discovery_lock = threading.RLock()
|
||||||
|
|
||||||
|
# Lock protecting plugin_last_update from concurrent mutation/iteration.
|
||||||
|
# It's written from run_scheduled_updates()/update_all_plugins() (main
|
||||||
|
# loop) and read/diffed by run_scheduled_updates_with_changes(), which
|
||||||
|
# Vegas mode calls from its own background update-tick thread.
|
||||||
|
self._plugin_last_update_lock = threading.RLock()
|
||||||
|
|
||||||
# Active plugins
|
# Active plugins
|
||||||
self.plugins: Dict[str, Any] = {}
|
self.plugins: Dict[str, Any] = {}
|
||||||
self.plugin_manifests: Dict[str, Dict[str, Any]] = {}
|
self.plugin_manifests: Dict[str, Dict[str, Any]] = {}
|
||||||
@@ -317,7 +323,8 @@ class PluginManager:
|
|||||||
|
|
||||||
# Store plugin instance
|
# Store plugin instance
|
||||||
self.plugins[plugin_id] = plugin_instance
|
self.plugins[plugin_id] = plugin_instance
|
||||||
self.plugin_last_update[plugin_id] = 0.0
|
with self._plugin_last_update_lock:
|
||||||
|
self.plugin_last_update[plugin_id] = 0.0
|
||||||
# Invalidate cached interval so next tick re-derives it for this plugin
|
# Invalidate cached interval so next tick re-derives it for this plugin
|
||||||
self._update_interval_cache.pop(plugin_id, None)
|
self._update_interval_cache.pop(plugin_id, None)
|
||||||
|
|
||||||
@@ -429,7 +436,8 @@ class PluginManager:
|
|||||||
|
|
||||||
# Remove from active plugins
|
# Remove from active plugins
|
||||||
del self.plugins[plugin_id]
|
del self.plugins[plugin_id]
|
||||||
self.plugin_last_update.pop(plugin_id, None)
|
with self._plugin_last_update_lock:
|
||||||
|
self.plugin_last_update.pop(plugin_id, None)
|
||||||
self._update_interval_cache.pop(plugin_id, None)
|
self._update_interval_cache.pop(plugin_id, None)
|
||||||
|
|
||||||
# Remove main module from sys.modules if present
|
# Remove main module from sys.modules if present
|
||||||
@@ -698,7 +706,8 @@ class PluginManager:
|
|||||||
'recoverable': True,
|
'recoverable': True,
|
||||||
}
|
}
|
||||||
self.logger.warning("Plugin %s update() failed; will retry after interval", plugin_id)
|
self.logger.warning("Plugin %s update() failed; will retry after interval", plugin_id)
|
||||||
self.plugin_last_update[plugin_id] = failure_time
|
with self._plugin_last_update_lock:
|
||||||
|
self.plugin_last_update[plugin_id] = failure_time
|
||||||
self.state_manager.set_state_with_error(plugin_id, PluginState.ENABLED, error_info, error=err)
|
self.state_manager.set_state_with_error(plugin_id, PluginState.ENABLED, error_info, error=err)
|
||||||
if self.health_tracker:
|
if self.health_tracker:
|
||||||
self.health_tracker.record_failure(plugin_id, err)
|
self.health_tracker.record_failure(plugin_id, err)
|
||||||
@@ -731,7 +740,8 @@ class PluginManager:
|
|||||||
if interval is None:
|
if interval is None:
|
||||||
continue
|
continue
|
||||||
|
|
||||||
last_update = self.plugin_last_update.get(plugin_id, 0.0)
|
with self._plugin_last_update_lock:
|
||||||
|
last_update = self.plugin_last_update.get(plugin_id, 0.0)
|
||||||
|
|
||||||
if last_update == 0.0 or (current_time - last_update) >= interval:
|
if last_update == 0.0 or (current_time - last_update) >= interval:
|
||||||
# Update state to RUNNING
|
# Update state to RUNNING
|
||||||
@@ -762,7 +772,8 @@ class PluginManager:
|
|||||||
success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
|
success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
|
||||||
|
|
||||||
if success:
|
if success:
|
||||||
self.plugin_last_update[plugin_id] = current_time
|
with self._plugin_last_update_lock:
|
||||||
|
self.plugin_last_update[plugin_id] = current_time
|
||||||
self.state_manager.record_update(plugin_id)
|
self.state_manager.record_update(plugin_id)
|
||||||
# Update state back to ENABLED
|
# Update state back to ENABLED
|
||||||
self.state_manager.set_state(plugin_id, PluginState.ENABLED)
|
self.state_manager.set_state(plugin_id, PluginState.ENABLED)
|
||||||
@@ -775,6 +786,31 @@ class PluginManager:
|
|||||||
self.logger.exception("Error updating plugin %s: %s", plugin_id, exc)
|
self.logger.exception("Error updating plugin %s: %s", plugin_id, exc)
|
||||||
self._record_update_failure(plugin_id, exc=exc)
|
self._record_update_failure(plugin_id, exc=exc)
|
||||||
|
|
||||||
|
def run_scheduled_updates_with_changes(self, current_time: Optional[float] = None) -> List[str]:
|
||||||
|
"""
|
||||||
|
Like run_scheduled_updates(), but also returns the plugin_ids whose
|
||||||
|
plugin_last_update timestamp actually advanced during this call.
|
||||||
|
|
||||||
|
The before/after snapshots and the update pass itself are each
|
||||||
|
individually lock-protected against concurrent plugin_last_update
|
||||||
|
mutation (Vegas mode calls this from its own background
|
||||||
|
update-tick thread, racing the main render loop's plugin updates),
|
||||||
|
so callers get an atomic "who got fresh data" answer without
|
||||||
|
reaching into plugin_last_update themselves. The lock is not held
|
||||||
|
across the update pass so slow/blocking plugin update() calls don't
|
||||||
|
serialize against other plugin_last_update readers.
|
||||||
|
"""
|
||||||
|
with self._plugin_last_update_lock:
|
||||||
|
old_times = dict(self.plugin_last_update)
|
||||||
|
|
||||||
|
self.run_scheduled_updates(current_time)
|
||||||
|
|
||||||
|
with self._plugin_last_update_lock:
|
||||||
|
return [
|
||||||
|
plugin_id for plugin_id, new_time in self.plugin_last_update.items()
|
||||||
|
if new_time > old_times.get(plugin_id, 0.0)
|
||||||
|
]
|
||||||
|
|
||||||
def update_all_plugins(self) -> None:
|
def update_all_plugins(self) -> None:
|
||||||
"""
|
"""
|
||||||
Update all enabled plugins.
|
Update all enabled plugins.
|
||||||
@@ -797,7 +833,8 @@ class PluginManager:
|
|||||||
try:
|
try:
|
||||||
success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
|
success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
|
||||||
if success:
|
if success:
|
||||||
self.plugin_last_update[plugin_id] = time.time()
|
with self._plugin_last_update_lock:
|
||||||
|
self.plugin_last_update[plugin_id] = time.time()
|
||||||
self.state_manager.record_update(plugin_id)
|
self.state_manager.record_update(plugin_id)
|
||||||
self.state_manager.set_state(plugin_id, PluginState.ENABLED)
|
self.state_manager.set_state(plugin_id, PluginState.ENABLED)
|
||||||
else:
|
else:
|
||||||
|
|||||||
@@ -10,8 +10,11 @@ that don't scale down to a smaller panel.
|
|||||||
|
|
||||||
Limitations (documented on purpose):
|
Limitations (documented on purpose):
|
||||||
- Overflow past the LEFT or TOP edge (negative coordinates) is still clipped by
|
- Overflow past the LEFT or TOP edge (negative coordinates) is still clipped by
|
||||||
PIL and not detected here. The dominant real-world breakage is content that is
|
PIL and not detected pixel-wise here. The dominant real-world breakage is
|
||||||
too wide/tall for a smaller panel, which this catches.
|
content that is too wide/tall for a smaller panel, which this catches.
|
||||||
|
As a partial net, draw_text/draw_image calls made with negative coordinates
|
||||||
|
through this manager are recorded in `negative_coordinate_calls` — but draws
|
||||||
|
made directly on the raw PIL canvas remain uncovered.
|
||||||
- BDF text is clipped to the declared bounds by the parent's bitmap drawer, so
|
- BDF text is clipped to the declared bounds by the parent's bitmap drawer, so
|
||||||
BDF overflow is not flagged. Golden-image regression covers those plugins.
|
BDF overflow is not flagged. Golden-image regression covers those plugins.
|
||||||
- If a plugin replaces the canvas with its own image (display_manager.image = ...),
|
- If a plugin replaces the canvas with its own image (display_manager.image = ...),
|
||||||
@@ -56,6 +59,21 @@ class BoundsCheckingDisplayManager(VisualTestDisplayManager):
|
|||||||
super().__init__(self._canvas_width, self._canvas_height)
|
super().__init__(self._canvas_width, self._canvas_height)
|
||||||
# Plugins must see the DECLARED size, not the padded canvas size.
|
# Plugins must see the DECLARED size, not the padded canvas size.
|
||||||
self.matrix = _MatrixProxy(self._declared_width, self._declared_height)
|
self.matrix = _MatrixProxy(self._declared_width, self._declared_height)
|
||||||
|
# (text-or-'image', x, y) for every mediated draw call given a
|
||||||
|
# negative coordinate — PIL clips these silently, so record them.
|
||||||
|
self.negative_coordinate_calls: list = []
|
||||||
|
|
||||||
|
# -- negative-coordinate (left/top overflow) recording --
|
||||||
|
|
||||||
|
def draw_text(self, text, x=None, y=None, *args, **kwargs):
|
||||||
|
if (x is not None and x < 0) or (y is not None and y < 0):
|
||||||
|
self.negative_coordinate_calls.append((text, x, y))
|
||||||
|
return super().draw_text(text, x, y, *args, **kwargs)
|
||||||
|
|
||||||
|
def draw_image(self, image, x, y, *args, **kwargs):
|
||||||
|
if x < 0 or y < 0:
|
||||||
|
self.negative_coordinate_calls.append(('image', x, y))
|
||||||
|
return super().draw_image(image, x, y, *args, **kwargs)
|
||||||
|
|
||||||
# -- declared dimensions (override parent's image-derived properties) --
|
# -- declared dimensions (override parent's image-derived properties) --
|
||||||
|
|
||||||
|
|||||||
@@ -73,6 +73,10 @@ class RenderResult:
|
|||||||
golden_ok: Optional[bool] = None
|
golden_ok: Optional[bool] = None
|
||||||
golden_diff_pixels: int = 0
|
golden_diff_pixels: int = 0
|
||||||
golden_max_delta: int = 0
|
golden_max_delta: int = 0
|
||||||
|
# fill / scale-up check (populated only for sizes >= 2x the design size)
|
||||||
|
fill_checked: bool = False
|
||||||
|
fill_ok: Optional[bool] = None # False only in strict mode
|
||||||
|
fill_extent: Optional[Tuple[float, float]] = None # (extent_x, extent_y)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def size_label(self) -> str:
|
def size_label(self) -> str:
|
||||||
@@ -86,6 +90,8 @@ class RenderResult:
|
|||||||
return False
|
return False
|
||||||
if self.golden_checked and self.golden_ok is False:
|
if self.golden_checked and self.golden_ok is False:
|
||||||
return False
|
return False
|
||||||
|
if self.fill_ok is False:
|
||||||
|
return False
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
|
||||||
@@ -301,6 +307,74 @@ def compare_to_goldens(results: List[RenderResult], golden_dir: Path,
|
|||||||
return results
|
return results
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Fill / scale-up check
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
#
|
||||||
|
# Overflow catches content that is too BIG for a panel; nothing catches
|
||||||
|
# content that stays tiny on a panel much larger than the plugin's design
|
||||||
|
# size (e.g. 128x32 content in the corner of a 256x128 renders "green").
|
||||||
|
# These helpers measure how much of the panel the lit content spans so the
|
||||||
|
# harness can flag plugins that don't scale up.
|
||||||
|
|
||||||
|
# A pixel counts as "lit" above this luminance — low enough to catch dim
|
||||||
|
# content, high enough to ignore near-black noise.
|
||||||
|
_LIT_THRESHOLD = 16
|
||||||
|
# Content must span at least this fraction of an axis that is >= 2x the
|
||||||
|
# design size. Lenient on purpose: margins are fine, a tiny corner is not.
|
||||||
|
_MIN_FILL_EXTENT = 0.5
|
||||||
|
|
||||||
|
|
||||||
|
def fill_metrics(image: Image.Image) -> Tuple[float, float, float]:
|
||||||
|
"""Measure lit-content coverage: (extent_x, extent_y, ink_ratio).
|
||||||
|
|
||||||
|
extent_* are the lit bounding box's spans as fractions of the panel;
|
||||||
|
ink_ratio is the fraction of pixels lit (reporting only — sparse pixel
|
||||||
|
fonts legitimately have low ink ratios)."""
|
||||||
|
lit = image.convert("L").point(lambda p: 255 if p > _LIT_THRESHOLD else 0)
|
||||||
|
bbox = lit.getbbox()
|
||||||
|
if bbox is None:
|
||||||
|
return (0.0, 0.0, 0.0)
|
||||||
|
extent_x = (bbox[2] - bbox[0]) / image.width
|
||||||
|
extent_y = (bbox[3] - bbox[1]) / image.height
|
||||||
|
ink = sum(1 for p in lit.getdata() if p) / (image.width * image.height)
|
||||||
|
return (extent_x, extent_y, ink)
|
||||||
|
|
||||||
|
|
||||||
|
def check_scale_up(results: List[RenderResult],
|
||||||
|
design_size: Tuple[int, int] = (128, 32),
|
||||||
|
min_extent: float = _MIN_FILL_EXTENT,
|
||||||
|
strict: bool = False) -> List[RenderResult]:
|
||||||
|
"""Flag renders that leave a big panel mostly empty.
|
||||||
|
|
||||||
|
For each result whose panel is at least 2x the design size on an axis,
|
||||||
|
require the lit content to span >= min_extent of that axis. Mutates the
|
||||||
|
results' fill_* fields. In the default warn-only mode fill_ok is left
|
||||||
|
None (reported, never failing); strict=True sets fill_ok=False, which
|
||||||
|
fails RenderResult.ok — opt in per plugin via harness.json
|
||||||
|
{"fill_check": "strict"} once its adaptive layout is in place.
|
||||||
|
"""
|
||||||
|
design_w, design_h = design_size
|
||||||
|
for r in results:
|
||||||
|
if r.image is None or r.error is not None:
|
||||||
|
continue
|
||||||
|
check_x = r.width >= 2 * design_w
|
||||||
|
check_y = r.height >= 2 * design_h
|
||||||
|
if not (check_x or check_y):
|
||||||
|
continue
|
||||||
|
extent_x, extent_y, _ink = fill_metrics(r.image)
|
||||||
|
r.fill_checked = True
|
||||||
|
r.fill_extent = (round(extent_x, 3), round(extent_y, 3))
|
||||||
|
underfilled = ((check_x and extent_x < min_extent)
|
||||||
|
or (check_y and extent_y < min_extent))
|
||||||
|
if underfilled and strict:
|
||||||
|
r.fill_ok = False
|
||||||
|
elif not underfilled:
|
||||||
|
r.fill_ok = True
|
||||||
|
# warn-only underfill: fill_ok stays None; fill_extent tells the story
|
||||||
|
return results
|
||||||
|
|
||||||
|
|
||||||
def write_goldens(results: List[RenderResult], golden_dir: Path) -> int:
|
def write_goldens(results: List[RenderResult], golden_dir: Path) -> int:
|
||||||
"""Write each successfully-rendered result to its golden path. Returns count."""
|
"""Write each successfully-rendered result to its golden path. Returns count."""
|
||||||
written = 0
|
written = 0
|
||||||
|
|||||||
@@ -56,7 +56,15 @@ def load_harness_spec(plugin_dir: Union[str, Path]) -> Dict[str, Any]:
|
|||||||
"config": {...}, # config overrides
|
"config": {...}, # config overrides
|
||||||
"mock_data": "fixtures/mock.json", # path (relative to plugin dir) to cache fixtures
|
"mock_data": "fixtures/mock.json", # path (relative to plugin dir) to cache fixtures
|
||||||
"freeze_time": "2025-08-01 15:25:00",
|
"freeze_time": "2025-08-01 15:25:00",
|
||||||
"skip_update": false
|
"skip_update": false,
|
||||||
|
"fill_check": "warn", # or "strict": underfilled big panels FAIL
|
||||||
|
"variants": [ # extra runs with config overlays and
|
||||||
|
{ # their own golden dirs — e.g. an
|
||||||
|
"name": "adaptive", # opt-in adaptive mode tested beside
|
||||||
|
"config": {"layout_mode": "adaptive"}, # the classic default
|
||||||
|
"golden_dir": "test/golden-adaptive"
|
||||||
|
}
|
||||||
|
]
|
||||||
}
|
}
|
||||||
Returns {} when no harness.json exists.
|
Returns {} when no harness.json exists.
|
||||||
"""
|
"""
|
||||||
|
|||||||
@@ -161,6 +161,13 @@ class MockPluginManager:
|
|||||||
self.plugin_manifests: Dict[str, Dict] = {}
|
self.plugin_manifests: Dict[str, Dict] = {}
|
||||||
self.get_plugin_calls = []
|
self.get_plugin_calls = []
|
||||||
self.get_all_plugins_calls = []
|
self.get_all_plugins_calls = []
|
||||||
|
# Real FontManager so BasePlugin.layout / draw_fit behave identically
|
||||||
|
# under the harness (it only needs assets/fonts on disk).
|
||||||
|
try:
|
||||||
|
from src.font_manager import FontManager
|
||||||
|
self.font_manager: Optional[Any] = FontManager({})
|
||||||
|
except Exception:
|
||||||
|
self.font_manager = None
|
||||||
|
|
||||||
def get_plugin(self, plugin_id: str) -> Optional[Any]:
|
def get_plugin(self, plugin_id: str) -> Optional[Any]:
|
||||||
"""Get a plugin instance."""
|
"""Get a plugin instance."""
|
||||||
|
|||||||
@@ -28,6 +28,7 @@ DEFAULT_TEST_SIZES: List[Tuple[int, int]] = [
|
|||||||
(64, 32), # 1x1 — single panel, the tightest common rectangle
|
(64, 32), # 1x1 — single panel, the tightest common rectangle
|
||||||
(128, 32), # 2x1 — the baseline most plugins are tuned for
|
(128, 32), # 2x1 — the baseline most plugins are tuned for
|
||||||
(64, 64), # 1x2 — stacked, exercises tall-narrow centering
|
(64, 64), # 1x2 — stacked, exercises tall-narrow centering
|
||||||
|
(96, 48), # non-64x32-module panel (e.g. Waveshare), off-grid dims
|
||||||
(128, 64), # 2x2 — block, icon scaling / vertical centering
|
(128, 64), # 2x2 — block, icon scaling / vertical centering
|
||||||
(256, 32), # 4x1 — long strip, wide horizontal layout
|
(256, 32), # 4x1 — long strip, wide horizontal layout
|
||||||
(128, 96), # 2x3 — tall, exercises vertical overflow
|
(128, 96), # 2x3 — tall, exercises vertical overflow
|
||||||
|
|||||||
@@ -66,6 +66,10 @@ class RenderPipeline:
|
|||||||
else display_manager.height
|
else display_manager.height
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# Reusable blank frame for cycle-end pushes (allocated lazily,
|
||||||
|
# re-blacked before each reuse)
|
||||||
|
self._blank_frame = None
|
||||||
|
|
||||||
# ScrollHelper for optimized scrolling
|
# ScrollHelper for optimized scrolling
|
||||||
self.scroll_helper = ScrollHelper(
|
self.scroll_helper = ScrollHelper(
|
||||||
self.display_width,
|
self.display_width,
|
||||||
@@ -234,11 +238,19 @@ class RenderPipeline:
|
|||||||
)
|
)
|
||||||
# Push blank immediately so the hardware never shows any
|
# Push blank immediately so the hardware never shows any
|
||||||
# post-wrap content while the coordinator recomposes the
|
# post-wrap content while the coordinator recomposes the
|
||||||
# next cycle (~100 ms).
|
# next cycle (~100 ms). The blank is allocated once and
|
||||||
|
# reused across cycle wraps (fresh paste each time in case
|
||||||
|
# a consumer drew on the previous one).
|
||||||
try:
|
try:
|
||||||
from PIL import Image as _Image
|
if self._blank_frame is None or self._blank_frame.size != (
|
||||||
blank = _Image.new('RGB', (self.display_width, self.display_height))
|
self.display_width, self.display_height):
|
||||||
self.display_manager.image = blank
|
self._blank_frame = Image.new(
|
||||||
|
'RGB', (self.display_width, self.display_height))
|
||||||
|
else:
|
||||||
|
self._blank_frame.paste(
|
||||||
|
(0, 0, 0),
|
||||||
|
(0, 0, self.display_width, self.display_height))
|
||||||
|
self.display_manager.image = self._blank_frame
|
||||||
self.display_manager.update_display()
|
self.display_manager.update_display()
|
||||||
except Exception:
|
except Exception:
|
||||||
logger.exception("Failed to write blank frame to display at cycle end")
|
logger.exception("Failed to write blank frame to display at cycle end")
|
||||||
@@ -297,6 +309,8 @@ class RenderPipeline:
|
|||||||
Returns True when:
|
Returns True when:
|
||||||
- Cycle is complete and we should start fresh
|
- Cycle is complete and we should start fresh
|
||||||
- Staging buffer has new content
|
- Staging buffer has new content
|
||||||
|
- A plugin currently visible in the scroll has pending updated data
|
||||||
|
(e.g. a live score changed) — standalone (non-sync) mode only
|
||||||
"""
|
"""
|
||||||
if self._cycle_complete:
|
if self._cycle_complete:
|
||||||
return True
|
return True
|
||||||
@@ -314,6 +328,12 @@ class RenderPipeline:
|
|||||||
if buffer_status['staging_count'] > 0:
|
if buffer_status['staging_count'] > 0:
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
# Trigger recompose when pending updates affect visible segments, so
|
||||||
|
# live score/status changes reach the display within a few seconds
|
||||||
|
# instead of waiting for the next full cycle.
|
||||||
|
if self.stream_manager.has_pending_updates_for_visible_segments():
|
||||||
|
return True
|
||||||
|
|
||||||
return False
|
return False
|
||||||
|
|
||||||
def hot_swap_content(self) -> bool:
|
def hot_swap_content(self) -> bool:
|
||||||
|
|||||||
+5
-1
@@ -38,7 +38,11 @@ def mock_cache_manager():
|
|||||||
mock._memory_cache_timestamps = {}
|
mock._memory_cache_timestamps = {}
|
||||||
mock.cache_dir = "/tmp/test_cache"
|
mock.cache_dir = "/tmp/test_cache"
|
||||||
|
|
||||||
def mock_get(key: str, max_age: int = 300) -> Optional[Dict]:
|
def mock_get(key: str, max_age: Optional[int] = 300,
|
||||||
|
memory_ttl: Optional[int] = None) -> Optional[Dict]:
|
||||||
|
# Signature mirrors CacheManager.get — keep in sync or callers
|
||||||
|
# passing keyword args (health tracker, resource monitor) break
|
||||||
|
# only in tests, hiding real-API compatibility.
|
||||||
return mock._memory_cache.get(key)
|
return mock._memory_cache.get(key)
|
||||||
|
|
||||||
def mock_set(key: str, data: Dict, ttl: Optional[int] = None) -> None:
|
def mock_set(key: str, data: Dict, ttl: Optional[int] = None) -> None:
|
||||||
|
|||||||
@@ -0,0 +1,208 @@
|
|||||||
|
"""Tests for adaptive image fitting (src/adaptive_images.py) and the
|
||||||
|
LayoutContext image cache."""
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from PIL import Image
|
||||||
|
|
||||||
|
from src.adaptive_images import (
|
||||||
|
RESAMPLE_LANCZOS,
|
||||||
|
RESAMPLE_NEAREST,
|
||||||
|
ImageFitResult,
|
||||||
|
draw_fitted_image,
|
||||||
|
fit_image,
|
||||||
|
)
|
||||||
|
from src.adaptive_layout import LayoutContext, Region
|
||||||
|
from src.font_manager import FontManager
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(scope="module")
|
||||||
|
def font_manager():
|
||||||
|
return FontManager({})
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def ctx(font_manager):
|
||||||
|
return LayoutContext(128, 32, font_manager)
|
||||||
|
|
||||||
|
|
||||||
|
def _solid(w, h, color=(255, 0, 0, 255)):
|
||||||
|
return Image.new("RGBA", (w, h), color)
|
||||||
|
|
||||||
|
|
||||||
|
def _padded_logo(ink_w=10, ink_h=10, pad=10):
|
||||||
|
"""Transparent canvas with a solid ink block in the middle — models a
|
||||||
|
logo shipped with generous transparent padding."""
|
||||||
|
img = Image.new("RGBA", (ink_w + 2 * pad, ink_h + 2 * pad), (0, 0, 0, 0))
|
||||||
|
img.paste(_solid(ink_w, ink_h), (pad, pad))
|
||||||
|
return img
|
||||||
|
|
||||||
|
|
||||||
|
class TestFitModes:
|
||||||
|
def test_contain_letterboxes_and_upscales(self):
|
||||||
|
fit = fit_image(_solid(10, 5), (40, 40))
|
||||||
|
assert (fit.width, fit.height) == (40, 20) # aspect preserved
|
||||||
|
assert fit.scale == 4.0
|
||||||
|
|
||||||
|
def test_contain_no_upscale(self):
|
||||||
|
fit = fit_image(_solid(10, 5), (40, 40), upscale=False)
|
||||||
|
assert (fit.width, fit.height) == (10, 5)
|
||||||
|
assert fit.scale == 1.0
|
||||||
|
|
||||||
|
def test_cover_fills_and_crops(self):
|
||||||
|
fit = fit_image(_solid(10, 20), (40, 40), mode="cover")
|
||||||
|
assert (fit.width, fit.height) == (40, 40)
|
||||||
|
|
||||||
|
def test_cover_top_anchor(self):
|
||||||
|
# top half red, bottom half blue; cover-crop a wide box with top anchor
|
||||||
|
img = Image.new("RGBA", (20, 40), (0, 0, 255, 255))
|
||||||
|
img.paste(_solid(20, 20, (255, 0, 0, 255)), (0, 0))
|
||||||
|
fit = fit_image(img, (20, 20), mode="cover", anchor="top")
|
||||||
|
assert fit.image.getpixel((10, 5))[:3] == (255, 0, 0) # kept the top
|
||||||
|
|
||||||
|
def test_fill_height_matches_box_height(self):
|
||||||
|
fit = fit_image(_solid(10, 10), (64, 32), mode="fill_height")
|
||||||
|
assert fit.height == 32 and fit.width == 32
|
||||||
|
|
||||||
|
def test_fill_height_capped_by_width(self):
|
||||||
|
# very wide source: height-fill would overflow the box width
|
||||||
|
fit = fit_image(_solid(100, 10), (40, 32), mode="fill_height")
|
||||||
|
assert fit.width <= 40
|
||||||
|
|
||||||
|
def test_stretch_exact(self):
|
||||||
|
fit = fit_image(_solid(3, 7), (25, 13), mode="stretch")
|
||||||
|
assert (fit.width, fit.height) == (25, 13)
|
||||||
|
|
||||||
|
def test_crop_to_ink(self):
|
||||||
|
fit = fit_image(_padded_logo(), (30, 30), crop_to_ink=True)
|
||||||
|
# 10x10 ink upscaled to fill 30x30 (padding would have kept it small)
|
||||||
|
assert (fit.width, fit.height) == (30, 30)
|
||||||
|
no_crop = fit_image(_padded_logo(), (30, 30), crop_to_ink=False)
|
||||||
|
assert no_crop.width == 30 # whole padded canvas scaled instead
|
||||||
|
|
||||||
|
def test_fully_transparent_source(self):
|
||||||
|
img = Image.new("RGBA", (10, 10), (0, 0, 0, 0))
|
||||||
|
fit = fit_image(img, (20, 20), crop_to_ink=True)
|
||||||
|
assert fit.is_empty
|
||||||
|
|
||||||
|
def test_degenerate_box(self):
|
||||||
|
assert fit_image(_solid(10, 10), (0, 20)).is_empty
|
||||||
|
assert fit_image(_solid(10, 10), Region(0, 0, 20, 0)).is_empty
|
||||||
|
|
||||||
|
def test_output_always_rgba(self):
|
||||||
|
rgb = Image.new("RGB", (10, 10), (1, 2, 3))
|
||||||
|
assert fit_image(rgb, (20, 20)).image.mode == "RGBA"
|
||||||
|
|
||||||
|
def test_nearest_keeps_hard_edges(self):
|
||||||
|
# 2x2 checker scaled 8x: NEAREST keeps pure colors, LANCZOS blends
|
||||||
|
img = Image.new("RGBA", (2, 2), (0, 0, 0, 255))
|
||||||
|
img.putpixel((0, 0), (255, 255, 255, 255))
|
||||||
|
near = fit_image(img, (16, 16), mode="stretch", resample=RESAMPLE_NEAREST)
|
||||||
|
colors = {near.image.getpixel((x, y))[:3] for x in range(16) for y in range(16)}
|
||||||
|
assert colors == {(255, 255, 255), (0, 0, 0)}
|
||||||
|
|
||||||
|
def test_unknown_mode_raises(self):
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
fit_image(_solid(4, 4), (8, 8), mode="tile")
|
||||||
|
|
||||||
|
|
||||||
|
class TestDrawFittedImage:
|
||||||
|
class _DM:
|
||||||
|
def __init__(self, w=64, h=32):
|
||||||
|
self.image = Image.new("RGB", (w, h), (0, 0, 0))
|
||||||
|
|
||||||
|
def test_pastes_aligned_in_region(self):
|
||||||
|
dm = self._DM()
|
||||||
|
box = Region(10, 4, 20, 20)
|
||||||
|
fit = fit_image(_solid(10, 10), box)
|
||||||
|
xy = draw_fitted_image(dm, fit, box)
|
||||||
|
assert xy == box.align_xy(fit.width, fit.height)
|
||||||
|
assert dm.image.getpixel((xy[0] + 1, xy[1] + 1)) == (255, 0, 0)
|
||||||
|
|
||||||
|
def test_offset_translates(self):
|
||||||
|
dm = self._DM()
|
||||||
|
box = Region(0, 0, 20, 20)
|
||||||
|
fit = fit_image(_solid(10, 10), box)
|
||||||
|
x, y = draw_fitted_image(dm, fit, box, align="left", valign="top",
|
||||||
|
offset=(3, 5))
|
||||||
|
assert (x, y) == (3, 5)
|
||||||
|
|
||||||
|
def test_empty_fit_noops(self):
|
||||||
|
dm = self._DM()
|
||||||
|
fit = fit_image(_solid(10, 10), (0, 0))
|
||||||
|
assert draw_fitted_image(dm, fit, Region(0, 0, 10, 10)) is None
|
||||||
|
|
||||||
|
|
||||||
|
class TestContextImageCache:
|
||||||
|
def test_size_keyed_hit_and_miss(self, ctx):
|
||||||
|
img = _solid(10, 10)
|
||||||
|
a = ctx.fit_image(img, (20, 20), cache_key="logo:A")
|
||||||
|
assert ctx.fit_image(img, (20, 20), cache_key="logo:A") is a
|
||||||
|
b = ctx.fit_image(img, (30, 30), cache_key="logo:A")
|
||||||
|
assert b is not a and b.width == 30 # different box size = new entry
|
||||||
|
|
||||||
|
def test_id_keyed_default(self, ctx):
|
||||||
|
img = _solid(10, 10)
|
||||||
|
a = ctx.fit_image(img, (20, 20))
|
||||||
|
assert ctx.fit_image(img, (20, 20)) is a
|
||||||
|
|
||||||
|
def test_id_safety_pins_source(self, ctx):
|
||||||
|
# id()-keyed entries must pin the source image so a recycled id
|
||||||
|
# can't alias a dead image's cache entry.
|
||||||
|
img = _solid(10, 10)
|
||||||
|
ctx.fit_image(img, (20, 20))
|
||||||
|
pinned = [entry[1] for entry in ctx._image_cache.values()]
|
||||||
|
assert img in pinned
|
||||||
|
|
||||||
|
def test_cache_key_entries_do_not_pin(self, ctx):
|
||||||
|
img = _solid(10, 10)
|
||||||
|
ctx.fit_image(img, (20, 20), cache_key="logo:X")
|
||||||
|
key = next(k for k in ctx._image_cache if k[1] == "logo:X")
|
||||||
|
assert ctx._image_cache[key][1] is None
|
||||||
|
|
||||||
|
def test_lru_eviction(self, ctx):
|
||||||
|
for i in range(ctx._IMAGE_CACHE_MAX + 5):
|
||||||
|
ctx.fit_image(_solid(4, 4), (8, 8), cache_key=f"k{i}")
|
||||||
|
assert len(ctx._image_cache) == ctx._IMAGE_CACHE_MAX
|
||||||
|
assert not any(k[1] == "k0" for k in ctx._image_cache) # oldest evicted
|
||||||
|
|
||||||
|
def test_clear_cache_clears_images(self, ctx):
|
||||||
|
ctx.fit_image(_solid(4, 4), (8, 8), cache_key="k")
|
||||||
|
ctx.clear_cache()
|
||||||
|
assert len(ctx._image_cache) == 0
|
||||||
|
|
||||||
|
|
||||||
|
class TestBasePluginDrawImage:
|
||||||
|
def test_draw_image_end_to_end(self):
|
||||||
|
from src.plugin_system.base_plugin import BasePlugin
|
||||||
|
from src.plugin_system.testing.mocks import (
|
||||||
|
MockCacheManager, MockDisplayManager, MockPluginManager,
|
||||||
|
)
|
||||||
|
|
||||||
|
class _P(BasePlugin):
|
||||||
|
def update(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
def display(self, force_clear=False):
|
||||||
|
pass
|
||||||
|
|
||||||
|
plugin = _P("t", {}, MockDisplayManager(64, 32),
|
||||||
|
MockCacheManager(), MockPluginManager())
|
||||||
|
logo = _padded_logo()
|
||||||
|
box = plugin.layout.bounds.left_col(32)
|
||||||
|
ifit = plugin.draw_image(logo, box, mode="fill_height",
|
||||||
|
crop_to_ink=True, cache_key="logo:T")
|
||||||
|
assert ifit.height == 32
|
||||||
|
# pasted onto the mock's canvas
|
||||||
|
assert plugin.display_manager.image.getpixel((16, 16)) != (0, 0, 0)
|
||||||
|
|
||||||
|
|
||||||
|
class TestResultIndependence:
|
||||||
|
def test_same_size_fit_never_aliases_the_source(self):
|
||||||
|
"""LayoutContext caches ImageFitResults — an aliased image would let
|
||||||
|
later mutations of the source corrupt cached fits (or vice versa)."""
|
||||||
|
from PIL import ImageDraw
|
||||||
|
src = Image.new("RGBA", (20, 20), (255, 0, 0, 255))
|
||||||
|
fit = fit_image(src, (20, 20))
|
||||||
|
assert fit.image is not src
|
||||||
|
ImageDraw.Draw(src).rectangle([0, 0, 19, 19], fill=(0, 255, 0, 255))
|
||||||
|
assert fit.image.getpixel((5, 5)) == (255, 0, 0, 255)
|
||||||
@@ -0,0 +1,454 @@
|
|||||||
|
"""Tests for the adaptive layout system (src/adaptive_layout.py)."""
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from src.adaptive_layout import (
|
||||||
|
DEFAULT_DESIGN_SIZE,
|
||||||
|
LADDER_ARCADE,
|
||||||
|
LADDER_GRID,
|
||||||
|
LayoutContext,
|
||||||
|
Region,
|
||||||
|
draw_fitted_text,
|
||||||
|
measure_font_crispness,
|
||||||
|
measure_ink,
|
||||||
|
media_row,
|
||||||
|
scoreboard_regions,
|
||||||
|
)
|
||||||
|
from src.plugin_system.testing.sizes import DEFAULT_TEST_SIZES
|
||||||
|
from src.font_manager import FontManager
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(scope="module")
|
||||||
|
def font_manager():
|
||||||
|
"""Real FontManager over assets/fonts — the ladders depend on it."""
|
||||||
|
return FontManager({})
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def ctx(font_manager):
|
||||||
|
return LayoutContext(128, 32, font_manager)
|
||||||
|
|
||||||
|
|
||||||
|
class TestRegion:
|
||||||
|
"""Pure integer rect algebra."""
|
||||||
|
|
||||||
|
def test_bands_partition_without_overlap(self):
|
||||||
|
r = Region(0, 0, 128, 32)
|
||||||
|
top = r.top_band(7)
|
||||||
|
bottom = r.bottom_band(7)
|
||||||
|
middle = r.middle(7, 7)
|
||||||
|
assert top.bottom == middle.y
|
||||||
|
assert middle.bottom == bottom.y
|
||||||
|
assert top.h + middle.h + bottom.h == r.h
|
||||||
|
|
||||||
|
def test_bands_clamp_on_short_panel(self):
|
||||||
|
# The classic failure: y=1 top band and y=height-7 bottom band
|
||||||
|
# overlapping on a short panel. Bands can't exceed the region.
|
||||||
|
r = Region(0, 0, 32, 8)
|
||||||
|
assert r.top_band(16).h == 8
|
||||||
|
assert r.bottom_band(16).h == 8
|
||||||
|
assert r.middle(8, 8).h == 0 # degenerate, never negative
|
||||||
|
|
||||||
|
def test_split_v_weights_sum_to_height(self):
|
||||||
|
r = Region(0, 0, 64, 33)
|
||||||
|
rows = r.split_v(3, 1, 1, gap=1)
|
||||||
|
assert len(rows) == 3
|
||||||
|
assert sum(row.h for row in rows) == 33 - 2 # two 1px gaps
|
||||||
|
assert rows[0].h > rows[1].h
|
||||||
|
assert rows[-1].bottom == r.bottom
|
||||||
|
|
||||||
|
def test_split_h_columns_advance(self):
|
||||||
|
r = Region(0, 0, 100, 32)
|
||||||
|
cols = r.split_h(1, 1, gap=2)
|
||||||
|
assert cols[0].right + 2 == cols[1].x
|
||||||
|
assert cols[1].right == r.right
|
||||||
|
|
||||||
|
def test_degenerate_sizes_never_negative(self):
|
||||||
|
for w, h in [(8, 8), (32, 16), (1, 1)]:
|
||||||
|
r = Region(0, 0, w, h).inset(4)
|
||||||
|
assert r.w >= 0 and r.h >= 0
|
||||||
|
for sub in r.split_v(1, 1) + r.split_h(1, 1, gap=3):
|
||||||
|
assert sub.w >= 0 and sub.h >= 0
|
||||||
|
|
||||||
|
def test_align_xy(self):
|
||||||
|
r = Region(10, 10, 100, 20)
|
||||||
|
assert r.align_xy(20, 10, "left", "top") == (10, 10)
|
||||||
|
assert r.align_xy(20, 10, "right", "bottom") == (90, 20)
|
||||||
|
assert r.align_xy(20, 10) == (50, 15)
|
||||||
|
|
||||||
|
def test_left_right_cols(self):
|
||||||
|
r = Region(0, 0, 128, 32)
|
||||||
|
assert r.left_col(32) == Region(0, 0, 32, 32)
|
||||||
|
assert r.right_col(32) == Region(96, 0, 32, 32)
|
||||||
|
|
||||||
|
def test_offset_translates_without_resizing(self):
|
||||||
|
r = Region(5, 5, 20, 10).offset(3, -2)
|
||||||
|
assert r == Region(8, 3, 20, 10)
|
||||||
|
|
||||||
|
|
||||||
|
class TestScoreboardRegions:
|
||||||
|
@pytest.mark.parametrize("w,h", DEFAULT_TEST_SIZES + [(8, 8)])
|
||||||
|
def test_invariants_at_all_sizes(self, w, h):
|
||||||
|
regs = scoreboard_regions(Region(0, 0, w, h))
|
||||||
|
assert regs.logo_slot <= min(h, w // 2)
|
||||||
|
# slots hug the edges and never overlap the center column
|
||||||
|
assert regs.away_slot.x == 0 and regs.home_slot.right == w
|
||||||
|
assert regs.away_slot.right <= regs.center_col.x or regs.center_col.w == 0
|
||||||
|
assert regs.center_col.right <= regs.home_slot.x or regs.center_col.w == 0
|
||||||
|
# bands stack inside the center column without overlap
|
||||||
|
assert regs.status_band.bottom <= regs.score_area.y or regs.score_area.h == 0
|
||||||
|
assert regs.score_area.bottom <= regs.detail_band.y or regs.score_area.h == 0
|
||||||
|
# everything within bounds, nothing negative
|
||||||
|
for reg in (regs.away_slot, regs.home_slot, regs.center_col,
|
||||||
|
regs.status_band, regs.score_area, regs.detail_band,
|
||||||
|
regs.bottom_left, regs.bottom_right):
|
||||||
|
assert reg.w >= 0 and reg.h >= 0
|
||||||
|
assert reg.x >= 0 and reg.y >= 0
|
||||||
|
assert reg.right <= w and reg.bottom <= h
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("w,h", [(96, 48), (128, 64), (256, 128), (64, 32), (128, 96)])
|
||||||
|
def test_2to1_aspect_gets_a_center_reserve(self, w, h):
|
||||||
|
"""These sizes are all <= 2:1 aspect, where the raw min(h, w//2)
|
||||||
|
formula claims the entire width for logos and leaves zero pixels
|
||||||
|
for a center column — the bug this reserve exists to fix."""
|
||||||
|
regs = scoreboard_regions(Region(0, 0, w, h))
|
||||||
|
assert regs.center_col.w >= int(w * 0.15) - 1 # -1 for int() rounding
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("w,h", [(128, 32), (192, 48), (256, 32)])
|
||||||
|
def test_wide_panels_unaffected_by_center_reserve(self, w, h):
|
||||||
|
"""Wide (>= ~4:1) panels already have height as the tighter
|
||||||
|
constraint, so the center reserve must be a no-op there — the
|
||||||
|
design-size baseline's proportions shouldn't shift."""
|
||||||
|
regs = scoreboard_regions(Region(0, 0, w, h))
|
||||||
|
assert regs.logo_slot == min(h, w // 2)
|
||||||
|
|
||||||
|
def test_center_reserve_fraction_is_configurable(self):
|
||||||
|
# min_center_design_px=0 isolates the fraction term (otherwise the
|
||||||
|
# scaled absolute floor can dominate and mask a fraction change).
|
||||||
|
regs_default = scoreboard_regions(Region(0, 0, 128, 64), min_center_design_px=0)
|
||||||
|
regs_wider = scoreboard_regions(Region(0, 0, 128, 64), min_center_fraction=0.5,
|
||||||
|
min_center_design_px=0)
|
||||||
|
assert regs_wider.center_col.w > regs_default.center_col.w
|
||||||
|
assert regs_wider.logo_slot < regs_default.logo_slot
|
||||||
|
|
||||||
|
def test_score_bleed_extends_past_center_col(self):
|
||||||
|
regs = scoreboard_regions(Region(0, 0, 128, 64), score_bleed_fraction=0.5)
|
||||||
|
assert regs.score_area.w > regs.center_col.w
|
||||||
|
assert regs.score_area.x < regs.center_col.x
|
||||||
|
assert regs.score_area.right > regs.center_col.right
|
||||||
|
|
||||||
|
def test_score_bleed_zero_matches_center_col(self):
|
||||||
|
regs = scoreboard_regions(Region(0, 0, 128, 64), score_bleed_fraction=0.0)
|
||||||
|
assert regs.score_area.w == regs.center_col.w
|
||||||
|
assert regs.score_area.x == regs.center_col.x
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("w,h", [(64, 32), (96, 48), (128, 64), (256, 128), (128, 96)])
|
||||||
|
def test_score_never_needs_ellipsis_for_a_short_score(self, w, h, font_manager):
|
||||||
|
"""The concrete regression this whole reserve/bleed system exists to
|
||||||
|
prevent: a real game score like '17-21' must always render in full,
|
||||||
|
never truncated, at every 2:1-or-tighter aspect ratio in the sample."""
|
||||||
|
ctx = LayoutContext(w, h, font_manager)
|
||||||
|
regs = scoreboard_regions(Region(0, 0, w, h), ctx=ctx)
|
||||||
|
height_scale = h / 32.0
|
||||||
|
fit = ctx.fit_text_proportional("17-21", regs.score_area, base_size_px=10,
|
||||||
|
ladder=LADDER_ARCADE, scale=height_scale)
|
||||||
|
assert fit.text == "17-21"
|
||||||
|
assert fit.fits
|
||||||
|
|
||||||
|
def test_ctx_scales_band_heights(self, font_manager):
|
||||||
|
small = scoreboard_regions(Region(0, 0, 128, 32),
|
||||||
|
ctx=LayoutContext(128, 32, font_manager))
|
||||||
|
big = scoreboard_regions(Region(0, 0, 256, 64),
|
||||||
|
ctx=LayoutContext(256, 64, font_manager))
|
||||||
|
assert big.status_band.h > small.status_band.h
|
||||||
|
|
||||||
|
def test_works_on_offset_card_region(self):
|
||||||
|
card = Region(10, 4, 100, 24)
|
||||||
|
regs = scoreboard_regions(card)
|
||||||
|
assert regs.away_slot.x == 10
|
||||||
|
assert regs.home_slot.right == card.right
|
||||||
|
|
||||||
|
|
||||||
|
class TestMediaRow:
|
||||||
|
def test_square_art_plus_body(self):
|
||||||
|
row = media_row(Region(0, 0, 128, 32))
|
||||||
|
assert row.art == Region(0, 0, 32, 32)
|
||||||
|
assert row.body.x == 32 + 2 and row.body.right == 128
|
||||||
|
|
||||||
|
def test_non_square(self):
|
||||||
|
row = media_row(Region(0, 0, 100, 20), square=False, gap=4)
|
||||||
|
assert row.art.w == 50
|
||||||
|
assert row.body.x == 54
|
||||||
|
|
||||||
|
def test_narrow_panel_clamps(self):
|
||||||
|
row = media_row(Region(0, 0, 16, 32))
|
||||||
|
assert row.art.w == 16 and row.body.w == 0
|
||||||
|
|
||||||
|
|
||||||
|
class TestLayoutContext:
|
||||||
|
def test_tiers(self, font_manager):
|
||||||
|
assert LayoutContext(128, 32, font_manager).tier == "sm"
|
||||||
|
assert LayoutContext(96, 48, font_manager).tier == "md"
|
||||||
|
assert LayoutContext(128, 64, font_manager).tier == "lg"
|
||||||
|
assert LayoutContext(64, 16, font_manager).tier == "xs"
|
||||||
|
assert LayoutContext(256, 128, font_manager).tier == "xl"
|
||||||
|
|
||||||
|
def test_wide_short_flag(self, font_manager):
|
||||||
|
assert LayoutContext(128, 32, font_manager).is_wide_short
|
||||||
|
assert not LayoutContext(128, 64, font_manager).is_wide_short
|
||||||
|
|
||||||
|
def test_scale_against_design_size(self, font_manager):
|
||||||
|
assert LayoutContext(128, 32, font_manager).scale == 1.0
|
||||||
|
assert LayoutContext(256, 64, font_manager).scale == 2.0
|
||||||
|
# min() of the two axes: don't overscale the constrained one
|
||||||
|
assert LayoutContext(256, 32, font_manager).scale == 1.0
|
||||||
|
assert DEFAULT_DESIGN_SIZE == (128, 32)
|
||||||
|
|
||||||
|
def test_px_scales_and_clamps(self, font_manager):
|
||||||
|
big = LayoutContext(256, 64, font_manager)
|
||||||
|
assert big.px(4) == 8
|
||||||
|
assert big.px(4, maximum=6) == 6
|
||||||
|
tiny = LayoutContext(32, 16, font_manager)
|
||||||
|
assert tiny.px(4, minimum=2) == 2
|
||||||
|
|
||||||
|
def test_by_tier_nearest_at_or_below(self, font_manager):
|
||||||
|
mapping = {"sm": 10, "lg": 18}
|
||||||
|
assert LayoutContext(128, 32, font_manager).by_tier(mapping) == 10
|
||||||
|
assert LayoutContext(96, 48, font_manager).by_tier(mapping) == 10 # md -> sm
|
||||||
|
assert LayoutContext(128, 64, font_manager).by_tier(mapping) == 18
|
||||||
|
assert LayoutContext(256, 128, font_manager).by_tier(mapping) == 18 # xl -> lg
|
||||||
|
# nothing at-or-below: fall forward to smallest defined above
|
||||||
|
assert LayoutContext(64, 16, font_manager).by_tier(mapping) == 10
|
||||||
|
|
||||||
|
|
||||||
|
class TestFontFitting:
|
||||||
|
def test_ladder_monotonic(self, font_manager):
|
||||||
|
"""Each ladder rung must render no taller than the one before it."""
|
||||||
|
for ladder in (LADDER_GRID, LADDER_ARCADE):
|
||||||
|
heights = []
|
||||||
|
for step in ladder:
|
||||||
|
font = font_manager.get_font(step.family, step.size_px)
|
||||||
|
heights.append(measure_ink("Ay0", font)[1])
|
||||||
|
assert heights == sorted(heights, reverse=True), (
|
||||||
|
f"ladder not monotonically shrinking: {heights}")
|
||||||
|
|
||||||
|
def test_ladder_grid_is_crisp(self, font_manager):
|
||||||
|
"""LADDER_GRID's BDF fonts are real bitmaps — always 0% antialiased."""
|
||||||
|
for step in LADDER_GRID:
|
||||||
|
font = font_manager.get_font(step.family, step.size_px)
|
||||||
|
assert measure_font_crispness(font, "Ay0") == 0.0
|
||||||
|
|
||||||
|
def test_ladder_arcade_is_crisp(self, font_manager):
|
||||||
|
"""PressStart2P only rasterizes without antialiasing at exact
|
||||||
|
multiples of its 8px design grid — every LADDER_ARCADE rung must
|
||||||
|
land on one."""
|
||||||
|
for step in LADDER_ARCADE:
|
||||||
|
assert step.size_px % 8 == 0, f"{step} is not a multiple of 8"
|
||||||
|
font = font_manager.get_font(step.family, step.size_px)
|
||||||
|
assert measure_font_crispness(font, "17-21") == 0.0
|
||||||
|
|
||||||
|
def test_crispness_catches_a_bad_size(self, font_manager):
|
||||||
|
"""Sanity check the measurement itself: a known-bad size for a
|
||||||
|
pixel-grid font must NOT read as crisp."""
|
||||||
|
font = font_manager.get_font("press_start", 10) # not a multiple of 8
|
||||||
|
assert measure_font_crispness(font, "17-21") > 0.1
|
||||||
|
|
||||||
|
def test_fit_text_grows_on_taller_panel(self, font_manager):
|
||||||
|
small = LayoutContext(64, 32, font_manager)
|
||||||
|
large = LayoutContext(128, 64, font_manager)
|
||||||
|
text = "12:34"
|
||||||
|
fit_small = small.fit_text(text, small.bounds, ladder=LADDER_ARCADE)
|
||||||
|
fit_large = large.fit_text(text, large.bounds, ladder=LADDER_ARCADE)
|
||||||
|
assert fit_small.fits and fit_large.fits
|
||||||
|
assert fit_large.size_px > fit_small.size_px
|
||||||
|
|
||||||
|
def test_fit_text_fits_the_box(self, ctx):
|
||||||
|
box = ctx.bounds.inset(1)
|
||||||
|
fit = ctx.fit_text("HELLO WORLD", box)
|
||||||
|
assert fit.fits
|
||||||
|
assert fit.width <= box.w and fit.height <= box.h
|
||||||
|
|
||||||
|
def test_fit_text_ellipsizes_overlong_text(self, font_manager):
|
||||||
|
tiny = LayoutContext(32, 16, font_manager)
|
||||||
|
fit = tiny.fit_text("SUPERCALIFRAGILISTIC", tiny.bounds)
|
||||||
|
assert fit.text != "SUPERCALIFRAGILISTIC"
|
||||||
|
assert fit.text.endswith("…")
|
||||||
|
assert fit.width <= tiny.bounds.w
|
||||||
|
|
||||||
|
def test_fit_text_cached(self, ctx):
|
||||||
|
first = ctx.fit_text("CACHED", ctx.bounds)
|
||||||
|
second = ctx.fit_text("CACHED", ctx.bounds)
|
||||||
|
assert first is second
|
||||||
|
ctx.clear_cache()
|
||||||
|
assert ctx.fit_text("CACHED", ctx.bounds) is not first
|
||||||
|
|
||||||
|
def test_fit_text_proportional_tracks_design_scale(self, font_manager):
|
||||||
|
# design size 128x32, base_size_px=10 (a typical classic score size):
|
||||||
|
# at 2x scale the target is 20px -> nearest LADDER_ARCADE rung <= 20
|
||||||
|
# is 16px, not the largest that merely fits the box (32).
|
||||||
|
ctx = LayoutContext(256, 64, font_manager) # scale = min(2,2) = 2
|
||||||
|
fit = ctx.fit_text_proportional("17-21", ctx.bounds, base_size_px=10,
|
||||||
|
ladder=LADDER_ARCADE)
|
||||||
|
assert fit.size_px == 16
|
||||||
|
|
||||||
|
def test_fit_text_proportional_does_not_exceed_max_fit(self, ctx):
|
||||||
|
# at scale=1 (128x32, the design size itself) the target equals
|
||||||
|
# base_size_px, so proportional should never pick something LARGER
|
||||||
|
# than plain fit_text would for the same box.
|
||||||
|
prop = ctx.fit_text_proportional("17-21", ctx.bounds, base_size_px=10,
|
||||||
|
ladder=LADDER_ARCADE)
|
||||||
|
maxed = ctx.fit_text("17-21", ctx.bounds, ladder=LADDER_ARCADE)
|
||||||
|
assert prop.size_px <= maxed.size_px
|
||||||
|
|
||||||
|
def test_fit_text_proportional_floors_at_smallest_rung(self, font_manager):
|
||||||
|
# scale so small the target is below every rung -> use the smallest
|
||||||
|
# rung as a floor rather than refusing to render anything.
|
||||||
|
ctx = LayoutContext(32, 8, font_manager) # scale = min(32/128, 8/32) = 0.25
|
||||||
|
fit = ctx.fit_text_proportional("HI", ctx.bounds, base_size_px=10,
|
||||||
|
ladder=LADDER_ARCADE)
|
||||||
|
assert fit.size_px == min(s.size_px for s in LADDER_ARCADE)
|
||||||
|
|
||||||
|
def test_fit_text_proportional_falls_through_when_target_rung_overflows(self, font_manager):
|
||||||
|
# a long string at the target rung might not fit a narrow box even
|
||||||
|
# though the target size is "correct" -- must fall through to a
|
||||||
|
# smaller rung exactly like fit_text does, not just refuse to fit.
|
||||||
|
ctx = LayoutContext(256, 64, font_manager)
|
||||||
|
narrow_box = Region(0, 0, 40, 64)
|
||||||
|
fit = ctx.fit_text_proportional("A REALLY LONG STRING HERE", narrow_box,
|
||||||
|
base_size_px=10, ladder=LADDER_ARCADE)
|
||||||
|
assert fit.fits or fit.text.endswith("…")
|
||||||
|
|
||||||
|
def test_fit_text_proportional_cached(self, ctx):
|
||||||
|
first = ctx.fit_text_proportional("X", ctx.bounds, base_size_px=10)
|
||||||
|
second = ctx.fit_text_proportional("X", ctx.bounds, base_size_px=10)
|
||||||
|
assert first is second
|
||||||
|
|
||||||
|
def test_fit_text_proportional_scale_override(self, font_manager):
|
||||||
|
# 128x64 vs design 128x32: self.scale (min of both axes) is 1.0
|
||||||
|
# since width didn't grow, but a caller whose composition scales by
|
||||||
|
# HEIGHT alone (e.g. logo_slot = min(h, w//2)) should be able to
|
||||||
|
# override the reference scale so text grows with it too.
|
||||||
|
ctx = LayoutContext(128, 64, font_manager)
|
||||||
|
assert ctx.scale == 1.0
|
||||||
|
default_fit = ctx.fit_text_proportional("17-21", ctx.bounds, base_size_px=10,
|
||||||
|
ladder=LADDER_ARCADE)
|
||||||
|
height_scale = 64 / 32 # matches design height
|
||||||
|
scaled_fit = ctx.fit_text_proportional("17-21", ctx.bounds, base_size_px=10,
|
||||||
|
ladder=LADDER_ARCADE, scale=height_scale)
|
||||||
|
assert scaled_fit.size_px > default_fit.size_px
|
||||||
|
|
||||||
|
def test_fit_lines_stacks_within_height(self, ctx):
|
||||||
|
box = ctx.bounds
|
||||||
|
lines = ["LINE ONE", "LINE TWO", "LINE THREE"]
|
||||||
|
fit = ctx.fit_lines(lines, box, spacing=1)
|
||||||
|
assert fit.fits
|
||||||
|
assert 3 * fit.line_height + 2 <= box.h
|
||||||
|
|
||||||
|
def test_font_for_rows(self, ctx):
|
||||||
|
fit = ctx.font_for_rows(4, 32)
|
||||||
|
assert fit.fits
|
||||||
|
assert 4 * fit.line_height <= 32
|
||||||
|
|
||||||
|
def test_ellipsize_returns_original_when_it_fits(self, ctx):
|
||||||
|
font = ctx.font_manager.get_font("4x6", 6)
|
||||||
|
assert ctx.ellipsize("HI", font, 1000) == "HI"
|
||||||
|
|
||||||
|
|
||||||
|
class TestDrawFittedText:
|
||||||
|
def test_draws_within_region(self, ctx):
|
||||||
|
calls = []
|
||||||
|
|
||||||
|
class _DM:
|
||||||
|
def draw_text(self, text, x=None, y=None, color=None, font=None):
|
||||||
|
calls.append((text, x, y))
|
||||||
|
|
||||||
|
box = Region(10, 4, 100, 24)
|
||||||
|
fit = ctx.fit_text("SCORE", box)
|
||||||
|
draw_fitted_text(_DM(), fit, box)
|
||||||
|
text, x, y = calls[0]
|
||||||
|
assert text == "SCORE"
|
||||||
|
assert box.x <= x <= box.right - fit.width
|
||||||
|
# the ink (y + y_offset .. + height) must land inside the box
|
||||||
|
assert box.y <= y + fit.y_offset
|
||||||
|
assert y + fit.y_offset + fit.height <= box.bottom
|
||||||
|
|
||||||
|
|
||||||
|
class TestBasePluginIntegration:
|
||||||
|
def test_layout_property_and_draw_fit(self):
|
||||||
|
from src.plugin_system.base_plugin import BasePlugin
|
||||||
|
from src.plugin_system.testing.mocks import (
|
||||||
|
MockCacheManager, MockDisplayManager, MockPluginManager,
|
||||||
|
)
|
||||||
|
|
||||||
|
class _Plugin(BasePlugin):
|
||||||
|
def update(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
def display(self, force_clear=False):
|
||||||
|
pass
|
||||||
|
|
||||||
|
plugin = _Plugin("test-plugin", {}, MockDisplayManager(96, 48),
|
||||||
|
MockCacheManager(), MockPluginManager())
|
||||||
|
assert (plugin.layout.width, plugin.layout.height) == (96, 48)
|
||||||
|
assert plugin.layout is plugin.layout # cached
|
||||||
|
fit = plugin.draw_fit("HELLO", plugin.layout.bounds.inset(1))
|
||||||
|
assert fit.fits
|
||||||
|
assert plugin.display_manager.draw_calls # actually drew
|
||||||
|
|
||||||
|
def test_layout_rebuilds_on_size_change(self):
|
||||||
|
from src.plugin_system.base_plugin import BasePlugin
|
||||||
|
from src.plugin_system.testing.mocks import (
|
||||||
|
MockCacheManager, MockDisplayManager, MockPluginManager,
|
||||||
|
)
|
||||||
|
|
||||||
|
class _Plugin(BasePlugin):
|
||||||
|
def update(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
def display(self, force_clear=False):
|
||||||
|
pass
|
||||||
|
|
||||||
|
dm = MockDisplayManager(128, 32)
|
||||||
|
plugin = _Plugin("test-plugin", {}, dm,
|
||||||
|
MockCacheManager(), MockPluginManager())
|
||||||
|
assert plugin.layout.tier == "sm"
|
||||||
|
dm.width, dm.height = 128, 64
|
||||||
|
assert plugin.layout.tier == "lg"
|
||||||
|
|
||||||
|
def test_design_size_from_manifest(self):
|
||||||
|
from src.plugin_system.base_plugin import BasePlugin
|
||||||
|
from src.plugin_system.testing.mocks import (
|
||||||
|
MockCacheManager, MockDisplayManager, MockPluginManager,
|
||||||
|
)
|
||||||
|
|
||||||
|
class _Plugin(BasePlugin):
|
||||||
|
def update(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
def display(self, force_clear=False):
|
||||||
|
pass
|
||||||
|
|
||||||
|
pm = MockPluginManager()
|
||||||
|
pm.plugin_manifests["test-plugin"] = {
|
||||||
|
"display": {"design_size": {"width": 64, "height": 32}}
|
||||||
|
}
|
||||||
|
plugin = _Plugin("test-plugin", {}, MockDisplayManager(128, 64),
|
||||||
|
MockCacheManager(), pm)
|
||||||
|
assert plugin.layout.design_size == (64, 32)
|
||||||
|
assert plugin.layout.scale == 2.0
|
||||||
|
|
||||||
|
|
||||||
|
class TestFitCacheBound:
|
||||||
|
def test_fit_cache_is_lru_bounded(self, ctx):
|
||||||
|
"""A plugin fitting changing text (live game clock, ticker) on a
|
||||||
|
24/7 service must not grow the fit cache without bound."""
|
||||||
|
for i in range(ctx._FIT_CACHE_MAX + 100):
|
||||||
|
ctx.fit_text(f"tick {i}", Region(0, 0, 100, 20))
|
||||||
|
assert len(ctx._fit_cache) <= ctx._FIT_CACHE_MAX
|
||||||
|
|
||||||
|
def test_lru_keeps_recent_entries_hot(self, ctx):
|
||||||
|
hot = ctx.fit_text("stay hot", Region(0, 0, 100, 20))
|
||||||
|
for i in range(ctx._FIT_CACHE_MAX - 1):
|
||||||
|
ctx.fit_text(f"cold {i}", Region(0, 0, 100, 20))
|
||||||
|
ctx.fit_text("stay hot", Region(0, 0, 100, 20)) # keep touching it
|
||||||
|
assert ctx.fit_text("stay hot", Region(0, 0, 100, 20)) is hot
|
||||||
@@ -400,3 +400,61 @@ class TestDiskCache:
|
|||||||
assert stats['fetch_count'] == 3
|
assert stats['fetch_count'] == 3
|
||||||
assert stats['total_fetch_time'] == 1.8
|
assert stats['total_fetch_time'] == 1.8
|
||||||
assert stats['average_fetch_time'] == pytest.approx(0.6, abs=0.01)
|
assert stats['average_fetch_time'] == pytest.approx(0.6, abs=0.01)
|
||||||
|
|
||||||
|
|
||||||
|
class TestDiskCacheWriteEconomy:
|
||||||
|
"""SD-card wear guards: identical payloads skip the disk, files are
|
||||||
|
compact, and TTL semantics survive the skip (see PR: fix/diskcache-sd-wear)."""
|
||||||
|
|
||||||
|
def test_identical_set_skips_rewrite(self, tmp_path):
|
||||||
|
import os
|
||||||
|
cache = DiskCache(cache_dir=str(tmp_path))
|
||||||
|
cache.set("k", {"data": "v"})
|
||||||
|
path = cache.get_cache_path("k")
|
||||||
|
first = os.stat(path)
|
||||||
|
os.utime(path, (first.st_atime - 100, first.st_mtime - 100)) # age it
|
||||||
|
aged_mtime = os.stat(path).st_mtime
|
||||||
|
ino_before = os.stat(path).st_ino
|
||||||
|
cache.set("k", {"data": "v"}) # identical payload
|
||||||
|
after = os.stat(path)
|
||||||
|
# mtime refreshed (TTL for mtime-based records preserved)...
|
||||||
|
assert after.st_mtime > aged_mtime
|
||||||
|
# ...but the file was NOT rewritten (same inode: no replace happened)
|
||||||
|
assert after.st_ino == ino_before
|
||||||
|
|
||||||
|
def test_changed_data_rewrites(self, tmp_path):
|
||||||
|
import os
|
||||||
|
cache = DiskCache(cache_dir=str(tmp_path))
|
||||||
|
cache.set("k", {"data": "v1"})
|
||||||
|
cache.set("k", {"data": "v2"})
|
||||||
|
assert cache.get("k") == {"data": "v2"}
|
||||||
|
|
||||||
|
def test_clear_resets_digest(self, tmp_path):
|
||||||
|
import os
|
||||||
|
cache = DiskCache(cache_dir=str(tmp_path))
|
||||||
|
cache.set("k", {"data": "v"})
|
||||||
|
cache.clear("k")
|
||||||
|
assert cache.get("k") is None
|
||||||
|
cache.set("k", {"data": "v"}) # same payload after clear must WRITE
|
||||||
|
assert cache.get("k") == {"data": "v"}
|
||||||
|
|
||||||
|
def test_skip_self_heals_when_file_deleted_externally(self, tmp_path):
|
||||||
|
import os
|
||||||
|
cache = DiskCache(cache_dir=str(tmp_path))
|
||||||
|
cache.set("k", {"data": "v"})
|
||||||
|
os.remove(cache.get_cache_path("k")) # e.g. expiry cleanup
|
||||||
|
cache.set("k", {"data": "v"}) # digest matches but file is gone
|
||||||
|
assert cache.get("k") == {"data": "v"}
|
||||||
|
|
||||||
|
def test_files_are_compact_json(self, tmp_path):
|
||||||
|
cache = DiskCache(cache_dir=str(tmp_path))
|
||||||
|
cache.set("k", {"a": 1, "b": [1, 2, 3]})
|
||||||
|
raw = open(cache.get_cache_path("k")).read()
|
||||||
|
assert "\n" not in raw.strip() # no indent
|
||||||
|
assert cache.get("k") == {"a": 1, "b": [1, 2, 3]}
|
||||||
|
|
||||||
|
def test_datetime_round_trip_still_works(self, tmp_path):
|
||||||
|
from datetime import datetime
|
||||||
|
cache = DiskCache(cache_dir=str(tmp_path))
|
||||||
|
cache.set("k", {"when": datetime(2026, 7, 12, 10, 30)})
|
||||||
|
assert cache.get("k") == {"when": "2026-07-12T10:30:00"}
|
||||||
|
|||||||
@@ -0,0 +1,88 @@
|
|||||||
|
"""
|
||||||
|
Regression tests for DisplayController._tick_plugin_updates_for_vegas().
|
||||||
|
|
||||||
|
PR #299 added logic to detect which plugins actually got fresh data on a
|
||||||
|
scheduled-update tick and notify Vegas mode via
|
||||||
|
vegas_coordinator.mark_plugin_updated(), so a live score change reaches the
|
||||||
|
scroll within seconds instead of waiting for a full cycle. PR #330's
|
||||||
|
multi-display sync refactor deleted this method (folding the callback back
|
||||||
|
to the plain _tick_plugin_updates(), which reports nothing), silently
|
||||||
|
orphaning VegasModeCoordinator.mark_plugin_updated() -- it has had zero
|
||||||
|
callers since.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from typing import Dict, List, Optional
|
||||||
|
from unittest.mock import MagicMock
|
||||||
|
|
||||||
|
from src.display_controller import DisplayController
|
||||||
|
|
||||||
|
|
||||||
|
def _make_controller(updated: Optional[List[str]] = None, vegas_coordinator: Optional[MagicMock] = None) -> DisplayController:
|
||||||
|
dc = object.__new__(DisplayController)
|
||||||
|
dc.plugin_manager = MagicMock()
|
||||||
|
dc.plugin_manager.run_scheduled_updates_with_changes.return_value = list(updated or [])
|
||||||
|
dc.vegas_coordinator = vegas_coordinator
|
||||||
|
return dc
|
||||||
|
|
||||||
|
|
||||||
|
class TestTickPluginUpdatesForVegas:
|
||||||
|
def test_marks_only_plugins_whose_timestamp_advanced(self):
|
||||||
|
vc = MagicMock()
|
||||||
|
dc = _make_controller(updated=["stock-news"], vegas_coordinator=vc)
|
||||||
|
|
||||||
|
dc._tick_plugin_updates_for_vegas()
|
||||||
|
|
||||||
|
vc.mark_plugin_updated.assert_called_once_with("stock-news")
|
||||||
|
|
||||||
|
def test_no_advance_marks_nothing(self):
|
||||||
|
vc = MagicMock()
|
||||||
|
dc = _make_controller(updated=[], vegas_coordinator=vc)
|
||||||
|
|
||||||
|
dc._tick_plugin_updates_for_vegas()
|
||||||
|
|
||||||
|
vc.mark_plugin_updated.assert_not_called()
|
||||||
|
|
||||||
|
def test_no_vegas_coordinator_does_not_raise(self):
|
||||||
|
dc = _make_controller(updated=["stock-news"], vegas_coordinator=None)
|
||||||
|
|
||||||
|
dc._tick_plugin_updates_for_vegas() # must not raise
|
||||||
|
|
||||||
|
def test_mark_plugin_updated_exception_does_not_propagate(self):
|
||||||
|
"""One plugin's mark_plugin_updated failing must not stop the tick
|
||||||
|
or crash the update loop it runs in."""
|
||||||
|
vc = MagicMock()
|
||||||
|
vc.mark_plugin_updated.side_effect = [RuntimeError("boom"), None]
|
||||||
|
dc = _make_controller(updated=["a", "b"], vegas_coordinator=vc)
|
||||||
|
|
||||||
|
dc._tick_plugin_updates_for_vegas() # must not raise
|
||||||
|
|
||||||
|
assert vc.mark_plugin_updated.call_count == 2
|
||||||
|
|
||||||
|
|
||||||
|
class TestVegasCoordinatorCallbackWiring:
|
||||||
|
def test_initialize_wires_vegas_aware_tick_as_update_callback(self):
|
||||||
|
"""The Vegas coordinator must be given the Vegas-aware
|
||||||
|
_tick_plugin_updates_for_vegas as its update callback, not the plain
|
||||||
|
_tick_plugin_updates() -- that's the exact wiring PR #330 dropped."""
|
||||||
|
dc = object.__new__(DisplayController)
|
||||||
|
dc.config = {"display": {"vegas_scroll": {"enabled": True}}, "sync": {}}
|
||||||
|
dc.display_manager = MagicMock()
|
||||||
|
dc.plugin_manager = MagicMock()
|
||||||
|
dc.sync_manager = MagicMock()
|
||||||
|
dc._check_live_priority = MagicMock()
|
||||||
|
dc._check_vegas_interrupt = MagicMock(return_value=False)
|
||||||
|
|
||||||
|
fake_coordinator = MagicMock()
|
||||||
|
|
||||||
|
import src.display_controller as dc_module
|
||||||
|
original_imported = dc_module._vegas_mode_imported
|
||||||
|
original_class = dc_module.VegasModeCoordinator
|
||||||
|
try:
|
||||||
|
dc_module._vegas_mode_imported = True
|
||||||
|
dc_module.VegasModeCoordinator = MagicMock(return_value=fake_coordinator)
|
||||||
|
dc._initialize_vegas_mode()
|
||||||
|
finally:
|
||||||
|
dc_module._vegas_mode_imported = original_imported
|
||||||
|
dc_module.VegasModeCoordinator = original_class
|
||||||
|
|
||||||
|
fake_coordinator.set_update_callback.assert_called_once_with(dc._tick_plugin_updates_for_vegas)
|
||||||
@@ -0,0 +1,98 @@
|
|||||||
|
"""Tests for the harness fill / scale-up check (src/plugin_system/testing/harness.py)."""
|
||||||
|
|
||||||
|
from PIL import Image
|
||||||
|
|
||||||
|
from src.plugin_system.testing.harness import (
|
||||||
|
RenderResult,
|
||||||
|
check_scale_up,
|
||||||
|
fill_metrics,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _canvas(w, h):
|
||||||
|
return Image.new("RGB", (w, h), (0, 0, 0))
|
||||||
|
|
||||||
|
|
||||||
|
def _with_block(w, h, bx, by, bw, bh, color=(255, 255, 255)):
|
||||||
|
img = _canvas(w, h)
|
||||||
|
img.paste(Image.new("RGB", (bw, bh), color), (bx, by))
|
||||||
|
return img
|
||||||
|
|
||||||
|
|
||||||
|
def _result(w, h, image):
|
||||||
|
return RenderResult("p", w, h, "mode", image=image)
|
||||||
|
|
||||||
|
|
||||||
|
class TestFillMetrics:
|
||||||
|
def test_full_white(self):
|
||||||
|
ex, ey, ink = fill_metrics(Image.new("RGB", (64, 32), (255, 255, 255)))
|
||||||
|
assert (ex, ey, ink) == (1.0, 1.0, 1.0)
|
||||||
|
|
||||||
|
def test_black_is_empty(self):
|
||||||
|
assert fill_metrics(_canvas(64, 32)) == (0.0, 0.0, 0.0)
|
||||||
|
|
||||||
|
def test_corner_dot(self):
|
||||||
|
ex, ey, ink = fill_metrics(_with_block(100, 100, 0, 0, 10, 10))
|
||||||
|
assert ex == 0.1 and ey == 0.1
|
||||||
|
assert ink == 0.01
|
||||||
|
|
||||||
|
def test_centered_half(self):
|
||||||
|
ex, ey, _ = fill_metrics(_with_block(100, 100, 25, 25, 50, 50))
|
||||||
|
assert ex == 0.5 and ey == 0.5
|
||||||
|
|
||||||
|
def test_dim_pixels_ignored(self):
|
||||||
|
img = _canvas(10, 10)
|
||||||
|
img.putpixel((5, 5), (10, 10, 10)) # below the lit threshold
|
||||||
|
assert fill_metrics(img) == (0.0, 0.0, 0.0)
|
||||||
|
|
||||||
|
|
||||||
|
class TestCheckScaleUp:
|
||||||
|
def test_not_checked_below_2x(self):
|
||||||
|
# 128x64 vs design 128x32: only height is 2x -> checked on y only;
|
||||||
|
# 128x32 itself: not checked at all
|
||||||
|
r = _result(128, 32, _with_block(128, 32, 0, 0, 10, 10))
|
||||||
|
check_scale_up([r], design_size=(128, 32))
|
||||||
|
assert not r.fill_checked
|
||||||
|
|
||||||
|
def test_warn_mode_records_but_passes(self):
|
||||||
|
# tiny corner content on a 256x128 (2x both axes)
|
||||||
|
r = _result(256, 128, _with_block(256, 128, 0, 0, 20, 20))
|
||||||
|
check_scale_up([r], design_size=(128, 32), strict=False)
|
||||||
|
assert r.fill_checked
|
||||||
|
assert r.fill_ok is None # warn-only: not a failure
|
||||||
|
assert r.ok # still passes
|
||||||
|
assert r.fill_extent[0] < 0.5
|
||||||
|
|
||||||
|
def test_strict_mode_fails_underfill(self):
|
||||||
|
r = _result(256, 128, _with_block(256, 128, 0, 0, 20, 20))
|
||||||
|
check_scale_up([r], design_size=(128, 32), strict=True)
|
||||||
|
assert r.fill_ok is False
|
||||||
|
assert not r.ok
|
||||||
|
|
||||||
|
def test_well_filled_passes_strict(self):
|
||||||
|
r = _result(256, 128, _with_block(256, 128, 10, 10, 200, 100))
|
||||||
|
check_scale_up([r], design_size=(128, 32), strict=True)
|
||||||
|
assert r.fill_ok is True and r.ok
|
||||||
|
|
||||||
|
def test_axis_selection_wide_only(self):
|
||||||
|
# 256x32 vs design 128x32: width is 2x, height is not -> only the
|
||||||
|
# x-extent matters; content spanning full width but few rows passes
|
||||||
|
r = _result(256, 32, _with_block(256, 32, 0, 12, 250, 8))
|
||||||
|
check_scale_up([r], design_size=(128, 32), strict=True)
|
||||||
|
assert r.fill_ok is True
|
||||||
|
|
||||||
|
def test_axis_selection_wide_only_underfill(self):
|
||||||
|
r = _result(256, 32, _with_block(256, 32, 0, 12, 60, 8))
|
||||||
|
check_scale_up([r], design_size=(128, 32), strict=True)
|
||||||
|
assert r.fill_ok is False
|
||||||
|
|
||||||
|
def test_errored_render_skipped(self):
|
||||||
|
r = RenderResult("p", 256, 128, "m", error="boom")
|
||||||
|
check_scale_up([r], design_size=(128, 32), strict=True)
|
||||||
|
assert not r.fill_checked
|
||||||
|
|
||||||
|
def test_custom_design_size(self):
|
||||||
|
# 128x64 with design 64x32 IS 2x both axes
|
||||||
|
r = _result(128, 64, _with_block(128, 64, 0, 0, 10, 10))
|
||||||
|
check_scale_up([r], design_size=(64, 32), strict=False)
|
||||||
|
assert r.fill_checked
|
||||||
@@ -0,0 +1,70 @@
|
|||||||
|
"""Tests for the plugin loader's advisory version-compatibility warning."""
|
||||||
|
|
||||||
|
import logging
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from src.plugin_system.plugin_loader import PluginLoader
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def loader():
|
||||||
|
return PluginLoader(logger=logging.getLogger("test-loader"))
|
||||||
|
|
||||||
|
|
||||||
|
def _warnings(caplog):
|
||||||
|
return [r for r in caplog.records if r.levelno == logging.WARNING]
|
||||||
|
|
||||||
|
|
||||||
|
class TestParseSemver:
|
||||||
|
def test_basic(self, loader):
|
||||||
|
assert loader._parse_semver("3.1.0") == (3, 1, 0)
|
||||||
|
assert loader._parse_semver("v2.0") == (2, 0, 0)
|
||||||
|
assert loader._parse_semver("2.0.0-beta.1") == (2, 0, 0)
|
||||||
|
|
||||||
|
def test_unparseable(self, loader):
|
||||||
|
assert loader._parse_semver(None) is None
|
||||||
|
assert loader._parse_semver(123) is None
|
||||||
|
|
||||||
|
|
||||||
|
class TestWarnIfIncompatible:
|
||||||
|
def test_warns_when_plugin_needs_newer_core(self, loader, caplog, monkeypatch):
|
||||||
|
import src
|
||||||
|
monkeypatch.setattr(src, "__version__", "3.1.0")
|
||||||
|
with caplog.at_level(logging.WARNING, logger="test-loader"):
|
||||||
|
loader._warn_if_incompatible("p", {"min_ledmatrix_version": "9.0.0"})
|
||||||
|
assert len(_warnings(caplog)) == 1
|
||||||
|
assert "9.0.0" in _warnings(caplog)[0].message
|
||||||
|
|
||||||
|
def test_silent_when_compatible(self, loader, caplog, monkeypatch):
|
||||||
|
import src
|
||||||
|
monkeypatch.setattr(src, "__version__", "3.1.0")
|
||||||
|
with caplog.at_level(logging.WARNING, logger="test-loader"):
|
||||||
|
loader._warn_if_incompatible("p", {"min_ledmatrix_version": "2.0.0"})
|
||||||
|
assert not _warnings(caplog)
|
||||||
|
|
||||||
|
def test_silent_when_field_absent(self, loader, caplog):
|
||||||
|
with caplog.at_level(logging.WARNING, logger="test-loader"):
|
||||||
|
loader._warn_if_incompatible("p", {"name": "no version fields"})
|
||||||
|
assert not _warnings(caplog)
|
||||||
|
|
||||||
|
def test_reads_requires_and_versions_spellings(self, loader, caplog, monkeypatch):
|
||||||
|
import src
|
||||||
|
monkeypatch.setattr(src, "__version__", "3.1.0")
|
||||||
|
with caplog.at_level(logging.WARNING, logger="test-loader"):
|
||||||
|
loader._warn_if_incompatible(
|
||||||
|
"a", {"requires": {"min_ledmatrix_version": "9.0.0"}})
|
||||||
|
loader._warn_if_incompatible(
|
||||||
|
"b", {"versions": [{"ledmatrix_min_version": "9.0.0"}]})
|
||||||
|
loader._warn_if_incompatible(
|
||||||
|
"c", {"versions": [{"ledmatrix_min": "9.0.0"}]})
|
||||||
|
assert len(_warnings(caplog)) == 3
|
||||||
|
|
||||||
|
def test_stale_core_version_skips_comparison(self, loader, caplog, monkeypatch):
|
||||||
|
# Anti-spam guard: a core whose __version__ is below the ecosystem
|
||||||
|
# floor must not warn about every plugin.
|
||||||
|
import src
|
||||||
|
monkeypatch.setattr(src, "__version__", "1.0.0")
|
||||||
|
with caplog.at_level(logging.WARNING, logger="test-loader"):
|
||||||
|
loader._warn_if_incompatible("p", {"min_ledmatrix_version": "2.0.0"})
|
||||||
|
assert not _warnings(caplog)
|
||||||
@@ -0,0 +1,93 @@
|
|||||||
|
"""Tests for the snapshot write policy (src/common/snapshot_policy.py).
|
||||||
|
|
||||||
|
The invariants that matter:
|
||||||
|
- unchanged frames are NEVER re-encoded (the old code PNG-encoded identical
|
||||||
|
frames at 5 fps, 24/7)
|
||||||
|
- the file mtime never goes stale enough to trip the health check's 60s
|
||||||
|
degraded threshold (api_v3 get_hardware_status)
|
||||||
|
- a viewer gets full cadence; no viewer drops to the idle keepalive
|
||||||
|
"""
|
||||||
|
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||||||
|
|
||||||
|
from src.common.snapshot_policy import ( # noqa: E402
|
||||||
|
IDLE_INTERVAL,
|
||||||
|
TOUCH_INTERVAL,
|
||||||
|
VIEWER_INTERVAL,
|
||||||
|
SnapshotAction,
|
||||||
|
decide,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class TestViewerCadence:
|
||||||
|
def test_changed_frame_with_viewer_writes_at_full_rate(self):
|
||||||
|
assert decide(now=100.0, last_write_ts=100.0 - VIEWER_INTERVAL,
|
||||||
|
last_touch_ts=0, viewer_fresh=True,
|
||||||
|
frame_changed=True) is SnapshotAction.WRITE
|
||||||
|
|
||||||
|
def test_changed_frame_with_viewer_respects_min_interval(self):
|
||||||
|
assert decide(now=100.0, last_write_ts=100.0 - VIEWER_INTERVAL / 2,
|
||||||
|
last_touch_ts=100.0, viewer_fresh=True,
|
||||||
|
frame_changed=True) is SnapshotAction.SKIP
|
||||||
|
|
||||||
|
def test_unchanged_frame_with_viewer_never_writes(self):
|
||||||
|
"""A static screen with a viewer must not burn PNG encodes."""
|
||||||
|
assert decide(now=100.0, last_write_ts=90.0, last_touch_ts=90.0,
|
||||||
|
viewer_fresh=True,
|
||||||
|
frame_changed=False) is SnapshotAction.SKIP
|
||||||
|
|
||||||
|
|
||||||
|
class TestIdleCadence:
|
||||||
|
def test_changed_frame_without_viewer_waits_for_idle_interval(self):
|
||||||
|
assert decide(now=100.0, last_write_ts=100.0 - IDLE_INTERVAL / 2,
|
||||||
|
last_touch_ts=100.0, viewer_fresh=False,
|
||||||
|
frame_changed=True) is SnapshotAction.SKIP
|
||||||
|
|
||||||
|
def test_changed_frame_without_viewer_writes_at_idle_rate(self):
|
||||||
|
assert decide(now=100.0, last_write_ts=100.0 - IDLE_INTERVAL,
|
||||||
|
last_touch_ts=0, viewer_fresh=False,
|
||||||
|
frame_changed=True) is SnapshotAction.WRITE
|
||||||
|
|
||||||
|
|
||||||
|
class TestHealthKeepalive:
|
||||||
|
def test_stale_mtime_gets_touched(self):
|
||||||
|
"""Whatever else happens, mtime must be bumped within TOUCH_INTERVAL
|
||||||
|
so the health check (60s threshold) never reads the display as dead."""
|
||||||
|
assert decide(now=100.0, last_write_ts=100.0 - TOUCH_INTERVAL,
|
||||||
|
last_touch_ts=100.0 - TOUCH_INTERVAL, viewer_fresh=False,
|
||||||
|
frame_changed=False) is SnapshotAction.TOUCH
|
||||||
|
|
||||||
|
def test_touch_applies_with_viewer_too(self):
|
||||||
|
"""Viewer watching a static screen: no writes, but health stays green."""
|
||||||
|
assert decide(now=100.0, last_write_ts=100.0 - TOUCH_INTERVAL - 1,
|
||||||
|
last_touch_ts=100.0 - TOUCH_INTERVAL - 1, viewer_fresh=True,
|
||||||
|
frame_changed=False) is SnapshotAction.TOUCH
|
||||||
|
|
||||||
|
def test_recent_touch_suppresses_another(self):
|
||||||
|
assert decide(now=100.0, last_write_ts=0.0,
|
||||||
|
last_touch_ts=100.0 - TOUCH_INTERVAL / 2, viewer_fresh=False,
|
||||||
|
frame_changed=False) is SnapshotAction.SKIP
|
||||||
|
|
||||||
|
def test_touch_interval_stays_under_health_threshold(self):
|
||||||
|
"""api_v3's hardware status treats snapshot age >= 60s as degraded.
|
||||||
|
Keep a 2x margin so scheduling jitter can't trip it."""
|
||||||
|
assert TOUCH_INTERVAL <= 30
|
||||||
|
|
||||||
|
def test_worst_case_mtime_age_is_bounded(self):
|
||||||
|
"""Simulate any interleaving: from any state, within one policy call
|
||||||
|
after TOUCH_INTERVAL elapses, mtime gets refreshed (WRITE or TOUCH)."""
|
||||||
|
for viewer in (True, False):
|
||||||
|
for changed in (True, False):
|
||||||
|
action = decide(now=1000.0, last_write_ts=900.0,
|
||||||
|
last_touch_ts=900.0, viewer_fresh=viewer,
|
||||||
|
frame_changed=changed)
|
||||||
|
assert action in (SnapshotAction.WRITE, SnapshotAction.TOUCH)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
sys.exit(pytest.main([__file__, "-v"]))
|
||||||
@@ -0,0 +1,72 @@
|
|||||||
|
"""
|
||||||
|
Regression tests for RenderPipeline.should_recompose()'s pending-updates check.
|
||||||
|
|
||||||
|
PR #299 added a check so a plugin's live score/status change (a "pending
|
||||||
|
update" in StreamManager) triggers a hot-swap within a few seconds instead
|
||||||
|
of waiting for a full scroll cycle to complete. PR #330 (multi-display sync)
|
||||||
|
refactored should_recompose() and dropped that check entirely -- not just
|
||||||
|
gated behind the new sync-mode deferral it added, but removed outright, so
|
||||||
|
even standalone (non-sync) installations silently lost live-refresh and fell
|
||||||
|
back to waiting for full cycle boundaries (which, depending on
|
||||||
|
min/max_cycle_duration, can be minutes).
|
||||||
|
"""
|
||||||
|
|
||||||
|
from unittest.mock import MagicMock
|
||||||
|
|
||||||
|
from src.vegas_mode.config import VegasModeConfig
|
||||||
|
from src.vegas_mode.render_pipeline import RenderPipeline
|
||||||
|
|
||||||
|
|
||||||
|
class FakeDisplayManager:
|
||||||
|
width = 64
|
||||||
|
height = 32
|
||||||
|
|
||||||
|
|
||||||
|
def _make_pipeline(sync_manager=None):
|
||||||
|
stream_manager = MagicMock()
|
||||||
|
stream_manager.get_buffer_status.return_value = {'staging_count': 0}
|
||||||
|
pipeline = RenderPipeline(VegasModeConfig(), FakeDisplayManager(), stream_manager)
|
||||||
|
pipeline.sync_manager = sync_manager
|
||||||
|
return pipeline, stream_manager
|
||||||
|
|
||||||
|
|
||||||
|
class TestShouldRecompose:
|
||||||
|
def test_cycle_complete_always_recomposes(self):
|
||||||
|
pipeline, stream_manager = _make_pipeline()
|
||||||
|
pipeline._cycle_complete = True
|
||||||
|
stream_manager.has_pending_updates_for_visible_segments.return_value = False
|
||||||
|
assert pipeline.should_recompose() is True
|
||||||
|
|
||||||
|
def test_no_pending_updates_no_staging_does_not_recompose(self):
|
||||||
|
pipeline, stream_manager = _make_pipeline()
|
||||||
|
stream_manager.has_pending_updates_for_visible_segments.return_value = False
|
||||||
|
assert pipeline.should_recompose() is False
|
||||||
|
|
||||||
|
def test_pending_updates_on_visible_segment_triggers_recompose(self):
|
||||||
|
"""The actual regression: a live-updated plugin currently in view
|
||||||
|
must trigger a recompose instead of waiting for cycle end."""
|
||||||
|
pipeline, stream_manager = _make_pipeline()
|
||||||
|
stream_manager.has_pending_updates_for_visible_segments.return_value = True
|
||||||
|
assert pipeline.should_recompose() is True
|
||||||
|
|
||||||
|
def test_staging_buffer_content_triggers_recompose(self):
|
||||||
|
pipeline, stream_manager = _make_pipeline()
|
||||||
|
stream_manager.get_buffer_status.return_value = {'staging_count': 1}
|
||||||
|
stream_manager.has_pending_updates_for_visible_segments.return_value = False
|
||||||
|
assert pipeline.should_recompose() is True
|
||||||
|
|
||||||
|
def test_sync_active_defers_pending_updates_to_cycle_boundary(self):
|
||||||
|
"""Sync-mode deferral (PR #330's actual intent) must still hold:
|
||||||
|
pending updates alone must NOT trigger a mid-cycle hot-swap when a
|
||||||
|
follower display is attached, since that causes a visible
|
||||||
|
freeze+jump on the follower. This must keep working after
|
||||||
|
restoring the non-sync pending-updates check above."""
|
||||||
|
pipeline, stream_manager = _make_pipeline(sync_manager=MagicMock())
|
||||||
|
stream_manager.has_pending_updates_for_visible_segments.return_value = True
|
||||||
|
assert pipeline.should_recompose() is False
|
||||||
|
|
||||||
|
def test_sync_active_still_recomposes_on_cycle_complete(self):
|
||||||
|
pipeline, stream_manager = _make_pipeline(sync_manager=MagicMock())
|
||||||
|
pipeline._cycle_complete = True
|
||||||
|
stream_manager.has_pending_updates_for_visible_segments.return_value = True
|
||||||
|
assert pipeline.should_recompose() is True
|
||||||
+16
-1
@@ -608,9 +608,23 @@ def display_preview_generator():
|
|||||||
import base64
|
import base64
|
||||||
from PIL import Image
|
from PIL import Image
|
||||||
import io
|
import io
|
||||||
|
|
||||||
snapshot_path = "/tmp/led_matrix_preview.png" # nosec B108 - fixed path matches display_manager; only read here
|
snapshot_path = "/tmp/led_matrix_preview.png" # nosec B108 - fixed path matches display_manager; only read here
|
||||||
|
# Viewer marker: this generator only runs while the broadcaster has
|
||||||
|
# subscribers (it exits with no clients), so touching the marker each
|
||||||
|
# loop tells the DISPLAY service a browser is actually watching — it
|
||||||
|
# only pays for full-rate PNG snapshot encodes while this stays fresh
|
||||||
|
# (see src/common/snapshot_policy.py).
|
||||||
|
viewer_marker_path = "/tmp/led_matrix_preview_viewer" # nosec B108 - fixed path matches display_manager
|
||||||
last_modified = None
|
last_modified = None
|
||||||
|
|
||||||
|
def _touch_viewer_marker():
|
||||||
|
try:
|
||||||
|
with open(viewer_marker_path, 'a'):
|
||||||
|
pass
|
||||||
|
os.utime(viewer_marker_path, None)
|
||||||
|
except OSError:
|
||||||
|
pass # display side treats a missing marker as "no viewer"
|
||||||
|
|
||||||
# Get display dimensions from config
|
# Get display dimensions from config
|
||||||
try:
|
try:
|
||||||
@@ -627,6 +641,7 @@ def display_preview_generator():
|
|||||||
|
|
||||||
while True:
|
while True:
|
||||||
try:
|
try:
|
||||||
|
_touch_viewer_marker()
|
||||||
# Check if snapshot file exists and has been modified
|
# Check if snapshot file exists and has been modified
|
||||||
if os.path.exists(snapshot_path):
|
if os.path.exists(snapshot_path):
|
||||||
current_modified = os.path.getmtime(snapshot_path)
|
current_modified = os.path.getmtime(snapshot_path)
|
||||||
|
|||||||
@@ -329,6 +329,7 @@ def save_schedule_config():
|
|||||||
}
|
}
|
||||||
|
|
||||||
mode = data.get('mode', 'global')
|
mode = data.get('mode', 'global')
|
||||||
|
schedule_config['mode'] = mode
|
||||||
|
|
||||||
if mode == 'global':
|
if mode == 'global':
|
||||||
# Simple global schedule
|
# Simple global schedule
|
||||||
|
|||||||
Reference in New Issue
Block a user