mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-10 09:06:36 +00:00
Compare commits
1
Commits
v3.7.0
...
c5281d9a45
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c5281d9a45 |
@@ -105,6 +105,7 @@ logical image to multiple chained physical panels.
|
||||
| `display_durations` | object, `{}` | Per-plugin display duration in seconds, keyed by plugin id (e.g. `"clock": 15`) | `DisplayController._get_display_duration()` (`src/display_controller.py`) |
|
||||
| `plugin_rotation_order` | array, `[]` | Explicit rotation order of plugin ids; empty = all enabled plugins in discovery order | `DisplayController._apply_plugin_rotation_order()` (`src/display_controller.py`) |
|
||||
| `use_short_date_format` | bool, `true` | Compact date rendering in sports scoreboards | Nothing since `src/base_classes` was removed; scoreboards read `display.use_short_date_format` from their own plugin config |
|
||||
| `scan_order_compensation` | string, `"auto"` | `"auto"` shows one half of each panel a refresh behind while something scrolls at one frame per refresh, which removes the 1px step a 1:N-scan panel shows across its middle; `"off"` disables it. Applies only to layouts whose row order is known: plain or parallel chains, 0 or 180 degree orientation, `multiplexing` 0, `scan_mode` 0, and not in the emulator | `DisplayManager._setup_scan_order_compensation()` (`src/display_manager.py`, `src/scan_order.py`) |
|
||||
| `dynamic_duration.max_duration_seconds` | int, optional | Cap for plugins that request dynamic display time | `DisplayController._get_global_dynamic_cap()` (`src/display_controller.py`) |
|
||||
|
||||
## `display.vegas_scroll` — continuous scroll mode
|
||||
|
||||
+46
-2
@@ -35,6 +35,7 @@ from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
from PIL import Image, ImageDraw, ImageFont
|
||||
from src.common.font_layout import crisp_size, load_truetype, resolve_asset_path
|
||||
from src import scan_order
|
||||
from src.display_geometry import (
|
||||
DEFAULT_CHAIN_LENGTH, DEFAULT_COLS, DEFAULT_PARALLEL, DEFAULT_ROWS,
|
||||
ORIENTATION_ROTATE_DEGREES, compose_pixel_mapper_config, physical_size,
|
||||
@@ -44,7 +45,7 @@ from src.matrix_support import MatrixSettingsRefused, library_refusals, refusal_
|
||||
from src.pi5_matrix_support import is_raspberry_pi_5
|
||||
import threading
|
||||
import time
|
||||
from collections import OrderedDict
|
||||
from collections import OrderedDict, deque
|
||||
from typing import Dict, Any, List, Optional, Tuple
|
||||
import logging
|
||||
import math
|
||||
@@ -243,6 +244,7 @@ class DisplayManager:
|
||||
|
||||
self._setup_matrix()
|
||||
logger.info("Matrix setup completed in %.3f seconds", time.time() - start_time)
|
||||
self._setup_scan_order_compensation()
|
||||
|
||||
font_time = time.time()
|
||||
self._load_fonts()
|
||||
@@ -811,7 +813,7 @@ class DisplayManager:
|
||||
if self._double_sided is not None:
|
||||
self.offscreen_canvas.SetImage(self._composite_double_sided())
|
||||
else:
|
||||
self.offscreen_canvas.SetImage(self.image)
|
||||
self.offscreen_canvas.SetImage(self._scan_compensated(self.image))
|
||||
|
||||
# Swap buffers immediately. framerate_fraction holds the frame
|
||||
# for N refreshes; SwapOnVSync blocks for all of them, which is
|
||||
@@ -828,6 +830,48 @@ class DisplayManager:
|
||||
except Exception as e:
|
||||
logger.error(f"Error updating display: {e}")
|
||||
|
||||
def _setup_scan_order_compensation(self) -> None:
|
||||
"""Work out which rows to show a refresh behind while scrolling.
|
||||
|
||||
See src/scan_order.py. Only on real hardware: the emulator has no scan
|
||||
order, so there the lag would add the very step it removes elsewhere.
|
||||
"""
|
||||
self._scan_lag_bands = None
|
||||
self._scan_history = deque(maxlen=1)
|
||||
if (self.matrix is None or self._double_sided is not None
|
||||
or os.environ.get('EMULATOR', 'false') == 'true'):
|
||||
return
|
||||
display = self.config.get('display') or {}
|
||||
bands = scan_order.scan_lag_bands(
|
||||
display.get('hardware') or {}, self.height,
|
||||
display.get('scan_order_compensation', 'auto'))
|
||||
if not bands:
|
||||
return
|
||||
self._scan_lag_bands = bands
|
||||
self._scan_history = deque(maxlen=max(lag for _, _, lag in bands))
|
||||
logger.info(
|
||||
"Scan-order compensation on: while scrolling, %s",
|
||||
", ".join(f"rows {top}-{bottom - 1} show {lag} refresh(es) behind"
|
||||
for top, bottom, lag in bands))
|
||||
|
||||
def _scan_compensated(self, image: Image.Image) -> Image.Image:
|
||||
"""The frame to present, with lagging rows taken from earlier frames.
|
||||
|
||||
Only mid-scroll at one frame per refresh: that is when consecutive
|
||||
frames are consecutive refreshes. At a longer hold, or on a static
|
||||
screen, the history is dropped and the frame goes out as it is.
|
||||
"""
|
||||
bands = getattr(self, '_scan_lag_bands', None)
|
||||
if not bands:
|
||||
return image
|
||||
if self._frame_hold != 1 or not self.is_currently_scrolling():
|
||||
self._scan_history.clear()
|
||||
return image
|
||||
presented = scan_order.compose(image, self._scan_history, bands)
|
||||
# A copy: plugins draw into the same image object frame after frame.
|
||||
self._scan_history.appendleft(image.copy())
|
||||
return presented
|
||||
|
||||
def clear(self):
|
||||
"""Clear the display completely."""
|
||||
try:
|
||||
|
||||
@@ -0,0 +1,130 @@
|
||||
"""Compensate for the order an LED panel lights its rows.
|
||||
|
||||
A HUB75 panel with 1:N multiplexing lights its rows in pairs: row ``d`` of the
|
||||
top half together with row ``d`` of the bottom half, ``d`` running from 0 to
|
||||
N-1 across each refresh. So the two rows either side of the middle of a panel
|
||||
are lit at opposite ends of every refresh: the last row of one half near the
|
||||
end, the first row of the other at the start. When the picture moves, each
|
||||
refresh shows it one step further on, and those two neighbouring rows -- lit
|
||||
almost a whole refresh apart -- show the text one step apart. On a strip
|
||||
scrolling a whole pixel per refresh that is a crisp 1px step across the middle
|
||||
of every panel, which the eye and a phone camera both see.
|
||||
|
||||
Measured on hdpi (4x128x64 on one chain, rotated 180) on 2026-09-24: switching
|
||||
the panel to interlaced scanning made the step vanish, and halving the scroll
|
||||
speed halved it, so it is the scan order and not a torn frame. Showing one half
|
||||
of the panel a refresh behind the other lines those two rows up again. What is
|
||||
left is a uniform lean of about one step per half from top to bottom, which
|
||||
reads as nothing at all where the step read as a tear.
|
||||
|
||||
Which half lags follows from the geometry. Walking the logical rows top to
|
||||
bottom, wherever the next row is lit near the start of a refresh and the one
|
||||
above it near the end, the section below must show one more refresh of lag to
|
||||
stay continuous with it (and one less where the order jumps the other way).
|
||||
Stacked parallel chains are lit simultaneously, so each further half adds one.
|
||||
|
||||
Only layouts whose physical row order is known are compensated: plain chains,
|
||||
parallel chains, and a 0 or 180 degree rotation. Other pixel mappers
|
||||
(U-mapper, 90/270 rotation, ...), special multiplexing and interlaced scan are
|
||||
left alone, as is the emulator, which has no scan order to compensate for.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from typing import Any, List, Optional, Sequence, Tuple
|
||||
|
||||
from PIL import Image
|
||||
|
||||
from src.display_geometry import compose_pixel_mapper_config
|
||||
|
||||
#: (first row, last row + 1, refreshes of lag), for bands that lag at all.
|
||||
Band = Tuple[int, int, int]
|
||||
|
||||
|
||||
def _int(value: Any, default: int) -> int:
|
||||
try:
|
||||
return int(value)
|
||||
except (TypeError, ValueError):
|
||||
return default
|
||||
|
||||
|
||||
def _rotation(hardware: Mapping[str, Any]) -> Optional[int]:
|
||||
"""The rotation applied by the pixel mappers, or None if they do anything else."""
|
||||
mappers = [m.strip() for m in compose_pixel_mapper_config(hardware).split(';')
|
||||
if m.strip()]
|
||||
if not mappers:
|
||||
return 0
|
||||
if len(mappers) == 1 and mappers[0].replace(' ', '') in ('Rotate:0', 'Rotate:180'):
|
||||
return int(mappers[0].split(':')[1])
|
||||
return None
|
||||
|
||||
|
||||
def scan_lag_bands(hardware: Mapping[str, Any], height: int,
|
||||
setting: str = "auto") -> Optional[List[Band]]:
|
||||
"""Which rows of the logical canvas to show how many refreshes behind.
|
||||
|
||||
Returns None when compensation is off or the layout is not one this
|
||||
understands; see the module docstring.
|
||||
"""
|
||||
if str(setting or "auto").lower() == "off":
|
||||
return None
|
||||
rows = _int(hardware.get('rows'), 0)
|
||||
parallel = max(1, _int(hardware.get('parallel'), 1))
|
||||
if rows < 4 or rows % 2:
|
||||
return None
|
||||
if _int(hardware.get('multiplexing'), 0) != 0 or _int(hardware.get('scan_mode'), 0) != 0:
|
||||
return None
|
||||
rotation = _rotation(hardware)
|
||||
if rotation is None:
|
||||
return None
|
||||
physical_height = rows * parallel
|
||||
if physical_height != height:
|
||||
return None # something remapped the canvas; its row order is unknown
|
||||
|
||||
half = rows // 2
|
||||
|
||||
def phase(y: int) -> int:
|
||||
"""When in the refresh logical row y is lit, as a row-pair index."""
|
||||
physical = y if rotation == 0 else physical_height - 1 - y
|
||||
return (physical % rows) % half
|
||||
|
||||
lags = [0]
|
||||
for y in range(1, physical_height):
|
||||
step = phase(y) - phase(y - 1)
|
||||
if step < -half / 2: # lit near the start after a row lit near the end
|
||||
lags.append(lags[-1] + 1)
|
||||
elif step > half / 2: # the other way round
|
||||
lags.append(lags[-1] - 1)
|
||||
else:
|
||||
lags.append(lags[-1])
|
||||
low = min(lags)
|
||||
bands: List[Band] = []
|
||||
for y, lag in enumerate(lags):
|
||||
lag -= low
|
||||
if bands and bands[-1][2] == lag and bands[-1][1] == y:
|
||||
bands[-1] = (bands[-1][0], y + 1, lag)
|
||||
else:
|
||||
bands.append((y, y + 1, lag))
|
||||
return [band for band in bands if band[2] > 0] or None
|
||||
|
||||
|
||||
def compose(image: Image.Image, history: Sequence[Image.Image],
|
||||
bands: Sequence[Band]) -> Image.Image:
|
||||
"""``image`` with each band taken from the frame ``lag`` refreshes back.
|
||||
|
||||
``history[0]`` is the previous frame. A band whose frame is not available
|
||||
yet (the first frames of a scroll) is left current. Returns ``image`` itself
|
||||
when nothing changes, so the caller pays for a copy only when it must.
|
||||
"""
|
||||
out = image
|
||||
for top, bottom, lag in bands:
|
||||
if lag > len(history):
|
||||
continue
|
||||
source = history[lag - 1]
|
||||
if source.size != image.size:
|
||||
continue
|
||||
if out is image:
|
||||
out = image.copy()
|
||||
out.paste(source.crop((0, top, source.width, bottom)), (0, top))
|
||||
return out
|
||||
@@ -0,0 +1,142 @@
|
||||
"""Scan-order compensation (src/scan_order.py) and its use in update_display.
|
||||
|
||||
Confirmed on hdpi's panel before any of this was written: rotated 180, one
|
||||
chain, 64 rows. Showing the upper half a refresh behind removed the 1px step
|
||||
across the middle of the panel that a whole-pixel-per-refresh scroll showed;
|
||||
see the module docstring for why.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
os.environ.setdefault("EMULATOR", "true")
|
||||
|
||||
import pytest
|
||||
from PIL import Image, ImageDraw
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||||
|
||||
from src import scan_order # noqa: E402
|
||||
|
||||
|
||||
def hw(**overrides):
|
||||
base = {"rows": 64, "cols": 128, "chain_length": 4, "parallel": 1,
|
||||
"multiplexing": 0, "scan_mode": 0, "pixel_mapper_config": "",
|
||||
"orientation": "normal"}
|
||||
base.update(overrides)
|
||||
return base
|
||||
|
||||
|
||||
class TestWhichRowsLag:
|
||||
def test_hdpi_rotated_180_lags_the_upper_half(self):
|
||||
# The configuration the panel test confirmed.
|
||||
assert scan_order.scan_lag_bands(hw(orientation="180"), 64) == [(0, 32, 1)]
|
||||
|
||||
def test_unrotated_lags_the_lower_half(self):
|
||||
assert scan_order.scan_lag_bands(hw(), 64) == [(32, 64, 1)]
|
||||
|
||||
def test_a_32_row_panel_splits_at_16(self):
|
||||
assert scan_order.scan_lag_bands(hw(rows=32), 32) == [(16, 32, 1)]
|
||||
|
||||
def test_a_48_row_panel_splits_at_24(self):
|
||||
assert scan_order.scan_lag_bands(hw(rows=48), 48) == [(24, 48, 1)]
|
||||
|
||||
def test_stacked_parallel_chains_lag_one_more_per_half(self):
|
||||
# Chains are lit together, so every half boundary down the stack is
|
||||
# another start-after-end jump: a continuous lean, never a step.
|
||||
assert scan_order.scan_lag_bands(hw(parallel=2), 128) == [
|
||||
(32, 64, 1), (64, 96, 2), (96, 128, 3)]
|
||||
|
||||
def test_stacked_and_rotated_mirrors_it(self):
|
||||
assert scan_order.scan_lag_bands(hw(parallel=2, orientation="180"), 128) == [
|
||||
(0, 32, 3), (32, 64, 2), (64, 96, 1)]
|
||||
|
||||
@pytest.mark.parametrize("overrides", [
|
||||
{"pixel_mapper_config": "U-mapper"},
|
||||
{"orientation": "90"},
|
||||
{"multiplexing": 3},
|
||||
{"scan_mode": 1},
|
||||
{"rows": 7},
|
||||
])
|
||||
def test_layouts_it_cannot_reason_about_are_left_alone(self, overrides):
|
||||
assert scan_order.scan_lag_bands(hw(**overrides), 64) is None
|
||||
|
||||
def test_a_canvas_of_another_height_is_left_alone(self):
|
||||
# Something remapped it (double-sided, a mapper): row order unknown.
|
||||
assert scan_order.scan_lag_bands(hw(), 32) is None
|
||||
|
||||
def test_it_can_be_switched_off(self):
|
||||
assert scan_order.scan_lag_bands(hw(), 64, setting="off") is None
|
||||
|
||||
|
||||
def _frame(shade):
|
||||
return Image.new("RGB", (8, 64), (shade, shade, shade))
|
||||
|
||||
|
||||
class TestCompose:
|
||||
def test_a_band_comes_from_the_frame_that_many_refreshes_back(self):
|
||||
now, previous = _frame(30), _frame(20)
|
||||
out = scan_order.compose(now, [previous], [(32, 64, 1)])
|
||||
assert out.getpixel((0, 0)) == (30, 30, 30) # current half
|
||||
assert out.getpixel((0, 63)) == (20, 20, 20) # lagging half
|
||||
assert now.getpixel((0, 63)) == (30, 30, 30) # input untouched
|
||||
|
||||
def test_deeper_bands_use_older_frames(self):
|
||||
out = scan_order.compose(_frame(40), [_frame(30), _frame(20)],
|
||||
[(10, 20, 1), (20, 30, 2)])
|
||||
assert [out.getpixel((0, y))[0] for y in (5, 15, 25)] == [40, 30, 20]
|
||||
|
||||
def test_without_enough_history_the_frame_goes_out_as_is(self):
|
||||
now = _frame(30)
|
||||
assert scan_order.compose(now, [], [(32, 64, 1)]) is now
|
||||
|
||||
|
||||
class TestUpdateDisplay:
|
||||
"""The wiring: when the display manager applies it, and when it doesn't."""
|
||||
|
||||
@pytest.fixture
|
||||
def dm(self):
|
||||
from src.display_manager import DisplayManager
|
||||
DisplayManager._instance = None
|
||||
DisplayManager._initialized = False
|
||||
manager = DisplayManager({"display": {
|
||||
"hardware": {"rows": 32, "cols": 64, "chain_length": 1, "parallel": 1},
|
||||
"runtime": {"gpio_slowdown": 0}}}, suppress_test_pattern=True)
|
||||
presented = []
|
||||
|
||||
# update_display alternates between two canvases; watch both.
|
||||
for canvas in (manager.offscreen_canvas, manager.current_canvas):
|
||||
def capture(image, *args, _real=canvas.SetImage, **kwargs):
|
||||
presented.append(image.copy())
|
||||
return _real(image, *args, **kwargs)
|
||||
canvas.SetImage = capture
|
||||
manager._presented = presented
|
||||
yield manager
|
||||
manager.set_scrolling_state(False)
|
||||
DisplayManager._instance = None
|
||||
DisplayManager._initialized = False
|
||||
|
||||
def _push(self, dm, shade):
|
||||
dm.draw.rectangle([0, 0, dm.width - 1, dm.height - 1], fill=(shade, 0, 0))
|
||||
dm.update_display()
|
||||
return dm._presented[-1]
|
||||
|
||||
def test_off_in_the_emulator(self, dm):
|
||||
# No scan order there: lagging a half would add the step it removes.
|
||||
assert dm._scan_lag_bands is None
|
||||
|
||||
def test_lags_mid_scroll_at_one_frame_per_refresh(self, dm):
|
||||
dm._scan_lag_bands = [(16, 32, 1)]
|
||||
dm.set_scrolling_state(True, 1)
|
||||
self._push(dm, 10)
|
||||
shown = self._push(dm, 20)
|
||||
assert shown.getpixel((0, 0)) == (20, 0, 0)
|
||||
assert shown.getpixel((0, 31)) == (10, 0, 0)
|
||||
|
||||
def test_not_on_a_static_screen_or_a_held_frame(self, dm):
|
||||
dm._scan_lag_bands = [(16, 32, 1)]
|
||||
self._push(dm, 10)
|
||||
assert self._push(dm, 20).getpixel((0, 31)) == (20, 0, 0) # not scrolling
|
||||
dm.set_scrolling_state(True, 2)
|
||||
self._push(dm, 30)
|
||||
assert self._push(dm, 40).getpixel((0, 31)) == (40, 0, 0) # hold 2
|
||||
Reference in New Issue
Block a user