alexei.dolgolyov/ledgrab
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 14 Wiki Activity

refactor(processing): replace inline effect dispatch with @_effect_renderer registry

Browse Source 
EffectColorStripStream._animate_loop used to rebuild a 12-entry dict
``renderers = {"fire": self._render_fire, ...}`` on every frame, then
look up ``renderers.get(self._effect_type, self._render_fire)``. Two
audit smells (H1) at once: per-frame dict-rebuild churn and a silent
fallback to fire whenever ``self._effect_type`` was a typo or any
``_render_*`` method got renamed without updating the dict.

Fix:

  * ``@_effect_renderer("fire")`` stamps an attribute on the unbound
    method.
  * ``@_collect_effect_renderers`` (applied to the class) walks
    members at class-creation, gathers the marked ones into
    ``cls._RENDERERS``, and raises ``RuntimeError`` on duplicate
    registration.

The loop now reads ``type(self)._RENDERERS`` once and calls the
unbound method with explicit ``self``. An unknown ``_effect_type``
logs a warning and skips the frame (sleep one frame_time) instead of
silently rendering fire — louder failure mode without crashing the
animation thread.

Tests: 5 new tests cover the 12-effect coverage set, callable shape,
class-level (not per-instance) dict identity, duplicate-name
rejection, and the marker stamp contract.

343 existing processing / storage / API tests stay green; ruff clean.
This commit is contained in:
Admin
2026-05-22 23:00:00 +03:00
parent 29bdacf69a
commit 97dae2cd62
2 changed files with 162 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/ledgrab/core/processing/effect_stream.py
+80 -19
View File
@@ -11,7 +11,7 @@ no external dependencies are required.
import math
import threading
import time
from typing import Dict, Optional
from typing import Callable, Dict, Optional
import numpy as np
@@ -22,6 +22,54 @@ from ledgrab.utils.timer import high_resolution_timer
logger = get_logger(__name__)
# -- Effect renderer registry --
# A small attribute-marker + class-decorator pair: the per-method
# ``@_effect_renderer("fire")`` decorator stamps a name onto the unbound
# method; ``_collect_effect_renderers`` (applied to the class) scans those
# stamps and builds a ``cls._RENDERERS`` dict. This replaces the inline
# ``renderers = {"fire": self._render_fire, ...}`` dict that the animation
# loop used to rebuild every frame, and the silent ``.get(..., self._render_fire)``
# fallback that turned a typo in ``_effect_type`` into a hidden fire-renderer.
def _effect_renderer(name: str) -> Callable:
"""Mark a method as the renderer for the given effect type.
The actual collection happens via ``@_collect_effect_renderers`` on the
enclosing class — decorating the method alone is harmless if the class
decorator is omitted.
"""
def _decorator(method):
method._effect_renderer_name = name
return method
return _decorator
def _collect_effect_renderers(cls):
"""Class decorator: gather methods tagged with ``@_effect_renderer``
into ``cls._RENDERERS``.
Runs once at class-creation. Raises ``RuntimeError`` if two methods
register under the same name (catches copy-paste typos).
"""
renderers: Dict[str, Callable] = {}
for attr_name in dir(cls):
member = getattr(cls, attr_name, None)
effect_name = getattr(member, "_effect_renderer_name", None)
if effect_name is None:
continue
if effect_name in renderers:
raise RuntimeError(
f"Duplicate @_effect_renderer({effect_name!r}) registration on {cls.__name__}"
)
renderers[effect_name] = member
cls._RENDERERS = renderers
return cls
# ── Palette LUT system ──────────────────────────────────────────────────
# Each palette is a list of (position, R, G, B) control points.
@@ -211,13 +259,14 @@ _EFFECT_DEFAULT_PALETTE = {
}
@_collect_effect_renderers
class EffectColorStripStream(ColorStripStream):
"""Color strip stream that runs a procedural LED effect.
Dispatches to one of five render methods based on effect_type:
fire, meteor, plasma, noise, aurora.
Uses the same lifecycle pattern as StaticColorStripStream:
Uses the same lifecycle pattern as SingleColorStripStream:
background thread, double-buffered output, configure() for auto-sizing.
"""
@@ -367,21 +416,10 @@ class EffectColorStripStream(ColorStripStream):
_buf_a = _buf_b = None
_use_a = True
# Dispatch table
renderers = {
"fire": self._render_fire,
"meteor": self._render_meteor,
"plasma": self._render_plasma,
"noise": self._render_noise,
"aurora": self._render_aurora,
"rain": self._render_rain,
"comet": self._render_comet,
"bouncing_ball": self._render_bouncing_ball,
"fireworks": self._render_fireworks,
"sparkle_rain": self._render_sparkle_rain,
"lava_lamp": self._render_lava_lamp,
"wave_interference": self._render_wave_interference,
}
# Dispatch via the class-level registry built by
# ``@_collect_effect_renderers`` at class-creation time. Renderers
# are unbound methods, so each call passes ``self`` explicitly.
renderers = type(self)._RENDERERS
limiter = FrameLimiter(self._fps)
@@ -424,8 +462,19 @@ class EffectColorStripStream(ColorStripStream):
buf = _buf_a if _use_a else _buf_b
_use_a = not _use_a
render_fn = renderers.get(self._effect_type, self._render_fire)
render_fn(buf, n, anim_time)
render_fn = renderers.get(self._effect_type)
if render_fn is None:
# Unknown effect type — log once per loop pass and
# skip rendering rather than silently falling back
# to fire (the previous behaviour, which hid typos
# in ``_effect_type``).
logger.warning(
"EffectColorStripStream: unknown effect_type %r — skipping frame",
self._effect_type,
)
time.sleep(frame_time)
continue
render_fn(self, buf, n, anim_time)
with self._colors_lock:
self._colors = buf
@@ -440,6 +489,7 @@ class EffectColorStripStream(ColorStripStream):
# ── Fire ─────────────────────────────────────────────────────────
@_effect_renderer("fire")
def _render_fire(self, buf: np.ndarray, n: int, t: float) -> None:
"""Heat-propagation fire simulation.
@@ -490,6 +540,7 @@ class EffectColorStripStream(ColorStripStream):
# ── Meteor ───────────────────────────────────────────────────────
@_effect_renderer("meteor")
def _render_meteor(self, buf: np.ndarray, n: int, t: float) -> None:
"""Bright meteor head with exponential-decay trail."""
speed = self._effective_speed
@@ -560,6 +611,7 @@ class EffectColorStripStream(ColorStripStream):
# ── Plasma ───────────────────────────────────────────────────────
@_effect_renderer("plasma")
def _render_plasma(self, buf: np.ndarray, n: int, t: float) -> None:
"""Overlapping sine waves creating colorful plasma patterns."""
speed = self._effective_speed
@@ -587,6 +639,7 @@ class EffectColorStripStream(ColorStripStream):
# ── Perlin Noise ─────────────────────────────────────────────────
@_effect_renderer("noise")
def _render_noise(self, buf: np.ndarray, n: int, t: float) -> None:
"""Smooth scrolling fractal noise mapped to a color palette."""
speed = self._effective_speed
@@ -605,6 +658,7 @@ class EffectColorStripStream(ColorStripStream):
# ── Aurora ───────────────────────────────────────────────────────
@_effect_renderer("aurora")
def _render_aurora(self, buf: np.ndarray, n: int, t: float) -> None:
"""Layered noise bands simulating aurora borealis."""
speed = self._effective_speed
@@ -651,6 +705,7 @@ class EffectColorStripStream(ColorStripStream):
# ── Rain ──────────────────────────────────────────────────────────
@_effect_renderer("rain")
def _render_rain(self, buf: np.ndarray, n: int, t: float) -> None:
"""Raindrops falling down the strip with trailing tails."""
speed = self._effective_speed
@@ -686,6 +741,7 @@ class EffectColorStripStream(ColorStripStream):
# ── Comet ─────────────────────────────────────────────────────────
@_effect_renderer("comet")
def _render_comet(self, buf: np.ndarray, n: int, t: float) -> None:
"""Multiple comets with curved, pulsing tails."""
speed = self._effective_speed
@@ -732,6 +788,7 @@ class EffectColorStripStream(ColorStripStream):
# ── Bouncing Ball ─────────────────────────────────────────────────
@_effect_renderer("bouncing_ball")
def _render_bouncing_ball(self, buf: np.ndarray, n: int, t: float) -> None:
"""Physics-simulated bouncing balls with gravity."""
speed = self._effective_speed
@@ -795,6 +852,7 @@ class EffectColorStripStream(ColorStripStream):
# ── Fireworks ─────────────────────────────────────────────────────
@_effect_renderer("fireworks")
def _render_fireworks(self, buf: np.ndarray, n: int, t: float) -> None:
"""Rockets launch and explode into colorful particle bursts."""
speed = self._effective_speed
@@ -868,6 +926,7 @@ class EffectColorStripStream(ColorStripStream):
# ── Sparkle Rain ──────────────────────────────────────────────────
@_effect_renderer("sparkle_rain")
def _render_sparkle_rain(self, buf: np.ndarray, n: int, t: float) -> None:
"""Twinkling star field with smooth fade-in/fade-out."""
speed = self._effective_speed
@@ -904,6 +963,7 @@ class EffectColorStripStream(ColorStripStream):
# ── Lava Lamp ─────────────────────────────────────────────────────
@_effect_renderer("lava_lamp")
def _render_lava_lamp(self, buf: np.ndarray, n: int, t: float) -> None:
"""Slow-moving colored blobs that merge and separate."""
speed = self._effective_speed
@@ -945,6 +1005,7 @@ class EffectColorStripStream(ColorStripStream):
# ── Wave Interference ─────────────────────────────────────────────
@_effect_renderer("wave_interference")
def _render_wave_interference(self, buf: np.ndarray, n: int, t: float) -> None:
"""Two counter-propagating sine waves creating interference patterns."""
speed = self._effective_speed
server/tests/core/processing/test_effect_registry.py
+82
View File
@@ -0,0 +1,82 @@
"""Tests for the EffectColorStripStream renderer registry.
Lock in the per-class decorator-built _RENDERERS dict so:
* every render method is registered exactly once,
* the registry is built at class-creation (not per-frame), and
* duplicate registration is caught early.
"""
from __future__ import annotations
import pytest
from ledgrab.core.processing.effect_stream import (
EffectColorStripStream,
_collect_effect_renderers,
_effect_renderer,
)
EXPECTED_EFFECTS = frozenset(
{
"fire",
"meteor",
"plasma",
"noise",
"aurora",
"rain",
"comet",
"bouncing_ball",
"fireworks",
"sparkle_rain",
"lava_lamp",
"wave_interference",
}
)
def test_registry_contains_every_known_effect():
assert set(EffectColorStripStream._RENDERERS.keys()) == EXPECTED_EFFECTS
def test_each_registered_method_is_callable_and_unbound():
for name, method in EffectColorStripStream._RENDERERS.items():
assert callable(method), f"renderer for {name!r} is not callable"
# Unbound: bind requires self argument, name should start with _render_
assert method.__name__.startswith(
"_render_"
), f"renderer for {name!r} has unexpected name {method.__name__}"
def test_registry_is_one_per_class_not_per_instance():
"""The _RENDERERS dict lives on the CLASS, not on each instance."""
assert isinstance(EffectColorStripStream._RENDERERS, dict)
# Should be the exact same object across attribute access
assert EffectColorStripStream._RENDERERS is EffectColorStripStream._RENDERERS
def test_duplicate_registration_is_rejected():
"""A class that registers two methods under the same name fails at decoration."""
with pytest.raises(RuntimeError, match="Duplicate"):
@_collect_effect_renderers
class _Bad: # noqa: D401
@_effect_renderer("fire")
def a(self):
pass
@_effect_renderer("fire")
def b(self):
pass
def test_renderer_decorator_stamps_name_attribute():
"""The marker attribute is what the class decorator reads."""
@_effect_renderer("my_effect")
def fn(self):
pass
assert fn._effect_renderer_name == "my_effect"