refactor(processing): hot-path magic numbers -> named module constants

processed_stream lifts the 30-iteration filter recheck cadence to
_FILTER_RECHECK_EVERY_N_FRAMES with a comment explaining the 30 fps
trade-off. wled_target_processor lifts SKIP_REPOLL, the diagnostics
interval, and the CSPT recheck cadence to module-level
_SKIP_REPOLL_SLEEP_SECONDS, _DIAGNOSTICS_REPORT_INTERVAL_SECONDS, and
_CSPT_RECHECK_EVERY_N_ITERATIONS. Tests can monkeypatch them now.
This commit is contained in:
2026-05-23 01:14:31 +03:00
parent 507e1385a6
commit d38021f061
2 changed files with 29 additions and 5 deletions
@@ -13,6 +13,12 @@ from ledgrab.utils import get_logger
logger = get_logger(__name__) logger = get_logger(__name__)
# How often the processing loop re-resolves the CSPT filter chain. Set so a
# template edit lands inside ~1 s at the default 30 fps without bothering
# the per-frame hot path: at 30 fps, 30 iterations ≈ 1 second.
_FILTER_RECHECK_EVERY_N_FRAMES = 30
class ProcessedColorStripStream(ColorStripStream): class ProcessedColorStripStream(ColorStripStream):
"""Color strip stream that wraps an input CSS and applies CSPT filters. """Color strip stream that wraps an input CSS and applies CSPT filters.
@@ -154,9 +160,10 @@ class ProcessedColorStripStream(ColorStripStream):
while self._running: while self._running:
t0 = time.monotonic() t0 = time.monotonic()
# Periodically re-resolve filters (every 30 iterations) # Periodically re-resolve filters so CSPT edits land within
# ~_FILTER_RECHECK_EVERY_N_FRAMES / fps seconds.
self._resolve_count += 1 self._resolve_count += 1
if self._resolve_count >= 30: if self._resolve_count >= _FILTER_RECHECK_EVERY_N_FRAMES:
self._resolve_count = 0 self._resolve_count = 0
self._resolve_filters() self._resolve_filters()
@@ -29,6 +29,23 @@ from ledgrab.utils.timer import high_resolution_timer
logger = get_logger(__name__) logger = get_logger(__name__)
# How long the inner loop waits before re-polling for a fresh frame when the
# upstream colour buffer has not advanced. Kept small so deadlines are not
# missed on slow devices but large enough that we are not spinning the CPU.
_SKIP_REPOLL_SLEEP_SECONDS = 0.005
# Cadence for the timing-diagnostics dump (5 s) — long enough to amortise
# the per-iteration cost across hundreds of frames at 30 fps so the log
# is informative without spamming.
_DIAGNOSTICS_REPORT_INTERVAL_SECONDS = 5.0
# How often the per-target processing loop re-resolves the CSPT (filter
# chain) cache. 30 iterations ≈ 1 second at 30 fps; quick enough that a
# template edit reaches the wire fast, cheap enough that no per-frame
# work is wasted resolving an unchanged chain.
_CSPT_RECHECK_EVERY_N_ITERATIONS = 30
class WledTargetProcessor(TargetProcessor): class WledTargetProcessor(TargetProcessor):
"""Streams LED colors from a single ColorStripStream to a WLED/LED device.""" """Streams LED colors from a single ColorStripStream to a WLED/LED device."""
@@ -887,10 +904,10 @@ class WledTargetProcessor(TargetProcessor):
return max(0, min(255, int(base * static_val))) return max(0, min(255, int(base * static_val)))
return base return base
SKIP_REPOLL = 0.005 # 5 ms SKIP_REPOLL = _SKIP_REPOLL_SLEEP_SECONDS
# --- Timing diagnostics --- # --- Timing diagnostics ---
_diag_interval = 5.0 _diag_interval = _DIAGNOSTICS_REPORT_INTERVAL_SECONDS
_diag_next_report = time.perf_counter() + _diag_interval _diag_next_report = time.perf_counter() + _diag_interval
_diag_sleep_jitters: collections.deque = collections.deque(maxlen=300) _diag_sleep_jitters: collections.deque = collections.deque(maxlen=300)
_diag_slow_iters: collections.deque = collections.deque(maxlen=50) _diag_slow_iters: collections.deque = collections.deque(maxlen=50)
@@ -912,7 +929,7 @@ class WledTargetProcessor(TargetProcessor):
# --- CSPT (Color Strip Processing Template) filter cache --- # --- CSPT (Color Strip Processing Template) filter cache ---
_cspt_cached_template_id: Optional[str] = None # last resolved template ID _cspt_cached_template_id: Optional[str] = None # last resolved template ID
_cspt_filters: list = [] # list of PostprocessingFilter instances _cspt_filters: list = [] # list of PostprocessingFilter instances
_cspt_check_interval = 30 # re-check device template ID every N iterations _cspt_check_interval = _CSPT_RECHECK_EVERY_N_ITERATIONS
_cspt_check_counter = 0 _cspt_check_counter = 0
logger.info( logger.info(