mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-13 02:26:37 +00:00
fix(scripts): scroll_speeds.py opens the panel with the service's options
--measure and --demo built RGBMatrixOptions from a private copy of the display service's builder that had drifted: gpio_slowdown came from display.hardware (default 2) instead of display.runtime (default 3), and rp1_rio, panel_type, disable_hardware_pulsing, inverse_colors, pixel_mapper_config and orientation were skipped, with different defaults (hardware_mapping "regular", pwm_bits 11). A panel needing a high slowdown was measured -- or garbled -- in a setup the service never drives. The option filling in DisplayManager._setup_matrix moves, unchanged, into DisplayManager.apply_matrix_options(options, config), which _setup_matrix calls and the script reuses (overriding only limit_refresh_rate_hz for --measure). The script now loads the whole config rather than the hardware block. Tests pin the script's options to the service's attribute for attribute. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Chuck
+37
-42
@@ -47,52 +47,47 @@ from src.common import scroll_config # noqa: E402
|
||||
CONFIG = Path(__file__).resolve().parent.parent / "config" / "config.json"
|
||||
|
||||
|
||||
def load_hardware():
|
||||
def load_config():
|
||||
"""The whole config.json, or {} when it is missing or unreadable."""
|
||||
try:
|
||||
with open(CONFIG, encoding="utf-8") as handle:
|
||||
return (json.load(handle).get("display") or {}).get("hardware") or {}
|
||||
config = json.load(handle)
|
||||
except (OSError, ValueError):
|
||||
return {}
|
||||
return config if isinstance(config, dict) else {}
|
||||
|
||||
|
||||
def build_options(hardware, refresh_override=None):
|
||||
from rgbmatrix import RGBMatrixOptions
|
||||
|
||||
o = RGBMatrixOptions()
|
||||
from src.display_geometry import (
|
||||
DEFAULT_CHAIN_LENGTH, DEFAULT_COLS, DEFAULT_PARALLEL, DEFAULT_ROWS,
|
||||
)
|
||||
o.rows = int(hardware.get("rows", DEFAULT_ROWS))
|
||||
o.cols = int(hardware.get("cols", DEFAULT_COLS))
|
||||
o.chain_length = int(hardware.get("chain_length", DEFAULT_CHAIN_LENGTH))
|
||||
o.parallel = int(hardware.get("parallel", DEFAULT_PARALLEL))
|
||||
o.brightness = int(hardware.get("brightness", 80))
|
||||
o.hardware_mapping = hardware.get("hardware_mapping", "regular")
|
||||
o.pwm_bits = int(hardware.get("pwm_bits", 11))
|
||||
o.pwm_dither_bits = int(hardware.get("pwm_dither_bits", 0))
|
||||
o.pwm_lsb_nanoseconds = int(hardware.get("pwm_lsb_nanoseconds", 130))
|
||||
o.led_rgb_sequence = hardware.get("led_rgb_sequence", "RGB")
|
||||
o.scan_mode = int(hardware.get("scan_mode", 0))
|
||||
o.row_address_type = int(hardware.get("row_address_type", 0))
|
||||
o.multiplexing = int(hardware.get("multiplexing", 0))
|
||||
o.gpio_slowdown = int(hardware.get("gpio_slowdown", 2))
|
||||
o.limit_refresh_rate_hz = (
|
||||
int(refresh_override) if refresh_override is not None
|
||||
else int(hardware.get("limit_refresh_rate_hz", 0))
|
||||
)
|
||||
return o
|
||||
def hardware_of(config):
|
||||
return (config.get("display") or {}).get("hardware") or {}
|
||||
|
||||
|
||||
def open_matrix(hardware, refresh_override=None):
|
||||
def build_options(config, refresh_override=None):
|
||||
"""The matrix options the display service would use for this config.
|
||||
|
||||
Built by DisplayManager.apply_matrix_options, not a copy of it, so the
|
||||
measurement and the demo drive the panel exactly as the service does
|
||||
(runtime gpio_slowdown, rp1_rio, panel_type, orientation, defaults).
|
||||
``refresh_override`` replaces limit_refresh_rate_hz; 0 means uncapped.
|
||||
"""
|
||||
from src.display_manager import DisplayManager, RGBMatrixOptions
|
||||
|
||||
options = DisplayManager.apply_matrix_options(RGBMatrixOptions(), config)
|
||||
if refresh_override is not None:
|
||||
options.limit_refresh_rate_hz = int(refresh_override)
|
||||
return options
|
||||
|
||||
|
||||
def open_matrix(config, refresh_override=None):
|
||||
"""Construct the matrix, or explain why it will not open."""
|
||||
if os.geteuid() != 0:
|
||||
sys.exit("this needs root for GPIO access - rerun with sudo")
|
||||
try:
|
||||
from rgbmatrix import RGBMatrix
|
||||
except ImportError:
|
||||
sys.exit("rgbmatrix is not installed on this machine")
|
||||
from src.display_manager import RGBMatrix
|
||||
except ImportError as exc:
|
||||
sys.exit("could not load the display stack ({}); is rgbmatrix "
|
||||
"installed on this machine?".format(exc))
|
||||
try:
|
||||
return RGBMatrix(options=build_options(hardware, refresh_override))
|
||||
return RGBMatrix(options=build_options(config, refresh_override))
|
||||
except Exception as exc: # pragma: no cover - hardware dependent
|
||||
sys.exit(
|
||||
"could not open the panel ({}).\n"
|
||||
@@ -101,7 +96,7 @@ def open_matrix(hardware, refresh_override=None):
|
||||
)
|
||||
|
||||
|
||||
def measure_refresh(hardware, seconds=6.0):
|
||||
def measure_refresh(config, seconds=6.0):
|
||||
"""Actual refresh rate, by running uncapped and timing the swaps.
|
||||
|
||||
SwapOnVSync blocks until the panel's next refresh, so an unthrottled loop
|
||||
@@ -109,7 +104,7 @@ def measure_refresh(hardware, seconds=6.0):
|
||||
chain will really give you, as opposed to whatever limit_refresh_rate_hz
|
||||
optimistically asks for.
|
||||
"""
|
||||
matrix = open_matrix(hardware, refresh_override=0)
|
||||
matrix = open_matrix(config, refresh_override=0)
|
||||
canvas = matrix.CreateFrameCanvas()
|
||||
canvas = matrix.SwapOnVSync(canvas) # discard the first, it includes setup
|
||||
frames = 0
|
||||
@@ -122,17 +117,17 @@ def measure_refresh(hardware, seconds=6.0):
|
||||
return measured
|
||||
|
||||
|
||||
def demo(hardware, target, seconds):
|
||||
def demo(config, target, seconds):
|
||||
"""Scroll text at the crisp speed nearest `target`."""
|
||||
from PIL import Image, ImageDraw, ImageFont
|
||||
|
||||
from src.common.font_layout import load_truetype
|
||||
|
||||
hz = float(hardware.get("limit_refresh_rate_hz") or scroll_config.DEFAULT_REFRESH_HZ)
|
||||
hz = scroll_config.refresh_hz_from_config(config)
|
||||
choice = scroll_config.solve_crisp(target, hz)
|
||||
print("asked for {:.0f} px/s -> {}".format(target, choice.describe()))
|
||||
|
||||
matrix = open_matrix(hardware)
|
||||
matrix = open_matrix(config)
|
||||
canvas = matrix.CreateFrameCanvas()
|
||||
W, H = canvas.width, canvas.height
|
||||
|
||||
@@ -214,15 +209,15 @@ def main():
|
||||
help="highlight the entry nearest this speed")
|
||||
args = ap.parse_args()
|
||||
|
||||
hardware = load_hardware()
|
||||
configured = float(hardware.get("limit_refresh_rate_hz") or 0)
|
||||
config = load_config()
|
||||
configured = float(hardware_of(config).get("limit_refresh_rate_hz") or 0)
|
||||
|
||||
if args.demo is not None:
|
||||
demo(hardware, args.demo, args.seconds)
|
||||
demo(config, args.demo, args.seconds)
|
||||
return
|
||||
|
||||
if args.measure:
|
||||
measured = measure_refresh(hardware)
|
||||
measured = measure_refresh(config)
|
||||
print("measured panel refresh: {:.1f}Hz".format(measured))
|
||||
if configured:
|
||||
print("configured limit_refresh_rate_hz: {:.0f}".format(configured))
|
||||
|
||||
+65
-44
@@ -279,50 +279,9 @@ class DisplayManager:
|
||||
hardware_config = self.config.get('display', {}).get('hardware', {})
|
||||
runtime_config = self.config.get('display', {}).get('runtime', {})
|
||||
|
||||
# Basic hardware settings
|
||||
options.rows = hardware_config.get('rows', DEFAULT_ROWS)
|
||||
options.cols = hardware_config.get('cols', DEFAULT_COLS)
|
||||
options.chain_length = hardware_config.get('chain_length', DEFAULT_CHAIN_LENGTH)
|
||||
options.parallel = hardware_config.get('parallel', DEFAULT_PARALLEL)
|
||||
options.hardware_mapping = hardware_config.get('hardware_mapping', 'adafruit-hat-pwm')
|
||||
|
||||
# Performance and stability settings
|
||||
options.brightness = hardware_config.get('brightness', 90)
|
||||
options.pwm_bits = hardware_config.get('pwm_bits', 10)
|
||||
options.pwm_lsb_nanoseconds = hardware_config.get('pwm_lsb_nanoseconds', 150)
|
||||
options.led_rgb_sequence = hardware_config.get('led_rgb_sequence', 'RGB')
|
||||
options.pixel_mapper_config = self._build_pixel_mapper_config(hardware_config)
|
||||
options.row_address_type = hardware_config.get('row_address_type', 0)
|
||||
options.multiplexing = hardware_config.get('multiplexing', 0)
|
||||
options.panel_type = hardware_config.get('panel_type', '')
|
||||
options.disable_hardware_pulsing = hardware_config.get('disable_hardware_pulsing', False)
|
||||
options.show_refresh_rate = hardware_config.get('show_refresh_rate', False)
|
||||
options.limit_refresh_rate_hz = hardware_config.get('limit_refresh_rate_hz', 90)
|
||||
options.gpio_slowdown = runtime_config.get('gpio_slowdown', 3)
|
||||
|
||||
# Disable internal privilege dropping - we manage this via systemd or remain root
|
||||
# This prevents the library from dropping to 'daemon' user which breaks file permissions
|
||||
options.drop_privileges = False
|
||||
|
||||
# Additional settings from config
|
||||
if 'scan_mode' in hardware_config:
|
||||
options.scan_mode = hardware_config.get('scan_mode')
|
||||
if 'pwm_dither_bits' in hardware_config:
|
||||
options.pwm_dither_bits = hardware_config.get('pwm_dither_bits')
|
||||
if 'inverse_colors' in hardware_config:
|
||||
options.inverse_colors = hardware_config.get('inverse_colors')
|
||||
# Pi 5 only: 0=PIO/RP1 coprocessor (default, less CPU),
|
||||
# 1=RIO/Registered IO (faster; gpio_slowdown effect is inverted in this mode)
|
||||
if 'rp1_rio' in runtime_config:
|
||||
if hasattr(options, 'rp1_rio'):
|
||||
options.rp1_rio = runtime_config.get('rp1_rio')
|
||||
else:
|
||||
logger.warning(
|
||||
"rp1_rio is set in config but the installed rgbmatrix library does "
|
||||
"not support it — the library was likely built without Pi 5 RP1 "
|
||||
"support (mmap to 0x3f000000 instead of RP1 chip). "
|
||||
"Fix: sudo RPI_RGB_FORCE_REBUILD=1 ./first_time_install.sh"
|
||||
)
|
||||
# Every option comes from display.hardware / display.runtime, in
|
||||
# one place that scripts/scroll_speeds.py shares.
|
||||
self.apply_matrix_options(options, self.config)
|
||||
|
||||
logger.info(f"Initializing RGB Matrix with settings: rows={options.rows}, cols={options.cols}, chain_length={options.chain_length}, parallel={options.parallel}, hardware_mapping={options.hardware_mapping}")
|
||||
|
||||
@@ -1370,6 +1329,68 @@ class DisplayManager:
|
||||
|
||||
return dt.strftime(f"%b %-d{suffix}")
|
||||
|
||||
@classmethod
|
||||
def apply_matrix_options(cls, options, config: Dict[str, Any]):
|
||||
"""Fill ``options`` (an ``RGBMatrixOptions``) from the LEDMatrix config.
|
||||
|
||||
This is exactly what the display service drives the panel with, so a
|
||||
tool that opens the matrix itself (``scripts/scroll_speeds.py``) gets
|
||||
the same panel -- same runtime ``gpio_slowdown``, ``rp1_rio``,
|
||||
``panel_type``, orientation and defaults -- rather than a private copy
|
||||
that drifts. Does not open the matrix. Returns ``options``.
|
||||
"""
|
||||
display = config.get('display', {}) if isinstance(config, dict) else {}
|
||||
hardware_config = display.get('hardware', {})
|
||||
runtime_config = display.get('runtime', {})
|
||||
|
||||
# Basic hardware settings
|
||||
options.rows = hardware_config.get('rows', DEFAULT_ROWS)
|
||||
options.cols = hardware_config.get('cols', DEFAULT_COLS)
|
||||
options.chain_length = hardware_config.get('chain_length', DEFAULT_CHAIN_LENGTH)
|
||||
options.parallel = hardware_config.get('parallel', DEFAULT_PARALLEL)
|
||||
options.hardware_mapping = hardware_config.get('hardware_mapping', 'adafruit-hat-pwm')
|
||||
|
||||
# Performance and stability settings
|
||||
options.brightness = hardware_config.get('brightness', 90)
|
||||
options.pwm_bits = hardware_config.get('pwm_bits', 10)
|
||||
options.pwm_lsb_nanoseconds = hardware_config.get('pwm_lsb_nanoseconds', 150)
|
||||
options.led_rgb_sequence = hardware_config.get('led_rgb_sequence', 'RGB')
|
||||
# _build_pixel_mapper_config reads only class attributes, so the class
|
||||
# stands in for an instance here.
|
||||
options.pixel_mapper_config = cls._build_pixel_mapper_config(cls, hardware_config)
|
||||
options.row_address_type = hardware_config.get('row_address_type', 0)
|
||||
options.multiplexing = hardware_config.get('multiplexing', 0)
|
||||
options.panel_type = hardware_config.get('panel_type', '')
|
||||
options.disable_hardware_pulsing = hardware_config.get('disable_hardware_pulsing', False)
|
||||
options.show_refresh_rate = hardware_config.get('show_refresh_rate', False)
|
||||
options.limit_refresh_rate_hz = hardware_config.get('limit_refresh_rate_hz', 90)
|
||||
options.gpio_slowdown = runtime_config.get('gpio_slowdown', 3)
|
||||
|
||||
# Disable internal privilege dropping - we manage this via systemd or remain root
|
||||
# This prevents the library from dropping to 'daemon' user which breaks file permissions
|
||||
options.drop_privileges = False
|
||||
|
||||
# Additional settings from config
|
||||
if 'scan_mode' in hardware_config:
|
||||
options.scan_mode = hardware_config.get('scan_mode')
|
||||
if 'pwm_dither_bits' in hardware_config:
|
||||
options.pwm_dither_bits = hardware_config.get('pwm_dither_bits')
|
||||
if 'inverse_colors' in hardware_config:
|
||||
options.inverse_colors = hardware_config.get('inverse_colors')
|
||||
# Pi 5 only: 0=PIO/RP1 coprocessor (default, less CPU),
|
||||
# 1=RIO/Registered IO (faster; gpio_slowdown effect is inverted in this mode)
|
||||
if 'rp1_rio' in runtime_config:
|
||||
if hasattr(options, 'rp1_rio'):
|
||||
options.rp1_rio = runtime_config.get('rp1_rio')
|
||||
else:
|
||||
logger.warning(
|
||||
"rp1_rio is set in config but the installed rgbmatrix library does "
|
||||
"not support it — the library was likely built without Pi 5 RP1 "
|
||||
"support (mmap to 0x3f000000 instead of RP1 chip). "
|
||||
"Fix: sudo RPI_RGB_FORCE_REBUILD=1 ./first_time_install.sh"
|
||||
)
|
||||
return options
|
||||
|
||||
@property
|
||||
def refresh_hz(self) -> float:
|
||||
"""The panel's refresh rate in Hz, from the hardware config.
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
"""scripts/scroll_speeds.py must drive the panel the way the service does.
|
||||
|
||||
--measure and --demo open the matrix themselves. They used to build the
|
||||
options from a private copy of DisplayManager's builder that had drifted: it
|
||||
read gpio_slowdown from display.hardware (the service reads display.runtime),
|
||||
and skipped rp1_rio, panel_type, orientation and more. On a panel that needs a
|
||||
high slowdown that measured -- or garbled -- a panel the service never drives.
|
||||
"""
|
||||
import importlib.util
|
||||
import os
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
os.environ.setdefault("EMULATOR", "true")
|
||||
|
||||
import src.display_manager as display_manager # noqa: E402
|
||||
from src.display_manager import DisplayManager # noqa: E402
|
||||
|
||||
SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "scroll_speeds.py"
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def script():
|
||||
spec = importlib.util.spec_from_file_location("scroll_speeds_script", SCRIPT)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
class _Options:
|
||||
"""Records what is set; declares rp1_rio like a Pi 5-capable binding."""
|
||||
|
||||
rp1_rio = None
|
||||
|
||||
|
||||
CONFIG = {
|
||||
"display": {
|
||||
"hardware": {
|
||||
"rows": 64, "cols": 64, "chain_length": 2, "parallel": 1,
|
||||
"hardware_mapping": "adafruit-hat-pwm",
|
||||
"brightness": 70, "pwm_bits": 9, "pwm_lsb_nanoseconds": 130,
|
||||
"row_address_type": 5, "panel_type": "FM6126A",
|
||||
"limit_refresh_rate_hz": 120, "orientation": "180",
|
||||
"pixel_mapper_config": "U-mapper",
|
||||
"gpio_slowdown": 2, # a stale key in the wrong block
|
||||
},
|
||||
"runtime": {"gpio_slowdown": 7, "rp1_rio": 1},
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
def _script_options(script, config, **kwargs):
|
||||
with patch.object(display_manager, "RGBMatrixOptions", _Options):
|
||||
return script.build_options(config, **kwargs)
|
||||
|
||||
|
||||
def test_runtime_settings_are_used(script):
|
||||
options = _script_options(script, CONFIG)
|
||||
assert options.gpio_slowdown == 7
|
||||
assert options.rp1_rio == 1
|
||||
assert options.panel_type == "FM6126A"
|
||||
assert options.row_address_type == 5
|
||||
assert options.pixel_mapper_config == "U-mapper;Rotate:180"
|
||||
assert options.limit_refresh_rate_hz == 120
|
||||
|
||||
|
||||
def test_the_options_match_the_display_service(script):
|
||||
"""Same config in, same options out, attribute for attribute."""
|
||||
service = DisplayManager.apply_matrix_options(_Options(), CONFIG)
|
||||
assert vars(_script_options(script, CONFIG)) == vars(service)
|
||||
|
||||
|
||||
def test_measure_runs_uncapped(script):
|
||||
assert _script_options(script, CONFIG, refresh_override=0).limit_refresh_rate_hz == 0
|
||||
|
||||
|
||||
def test_an_empty_config_gets_the_service_defaults(script):
|
||||
options = _script_options(script, {})
|
||||
assert vars(options) == vars(DisplayManager.apply_matrix_options(_Options(), {}))
|
||||
assert options.gpio_slowdown == 3
|
||||
|
||||
|
||||
def test_the_service_uses_the_shared_builder(test_config):
|
||||
"""DisplayManager._setup_matrix fills its options through the same call."""
|
||||
with patch.object(display_manager, "RGBMatrix"), \
|
||||
patch.object(display_manager, "RGBMatrixOptions", _Options), \
|
||||
patch.object(display_manager, "freetype"), \
|
||||
patch.object(DisplayManager, "apply_matrix_options",
|
||||
wraps=DisplayManager.apply_matrix_options) as shared, \
|
||||
patch.dict(os.environ, {"EMULATOR": "false"}):
|
||||
DisplayManager._instance = None
|
||||
try:
|
||||
DisplayManager(test_config)
|
||||
finally:
|
||||
DisplayManager._instance = None
|
||||
shared.assert_called_once()
|
||||
Reference in New Issue
Block a user