diff --git a/packages/artnet/hms_artnet/__init__.py b/packages/artnet/hms_artnet/__init__.py index f5a5212..8800947 100644 --- a/packages/artnet/hms_artnet/__init__.py +++ b/packages/artnet/hms_artnet/__init__.py @@ -1,7 +1,8 @@ """hms_artnet – Art-Net 4 Steuerung (PLAN.md §16). ArtDMX-Empfang, ArtPoll/ArtPollReply (Discovery als Media Server, Style 0x02), -Fixture-Engine (Master32/Layer64, §16.3–§16.6), Universe-Plan (§16.2), +Fixture-Engine (Master32/Layer64, §16.3–§16.6), Patch-Verwaltung (§16.2), +DMX-zu-Parameter-Verkabelung (§11, §16), Universe-Plan (§16.2), konfigurierbare Universen/Adressen, Sequenzprüfung, Signalverlust-Verhalten. Layouts gegen die offizielle Art-Net-4-Spezifikation verifiziert. """ @@ -33,7 +34,13 @@ from hms_artnet.packets import ( parse_poll, parse_poll_reply, ) +from hms_artnet.patch import ( + FixturePatch, + PatchEntry, + PatchError, +) from hms_artnet.receiver import ArtNetReceiver, DmxUpdate, LossBehavior +from hms_artnet.wiring import DmxToParameterRouter, RouterStats __all__ = [ "OPDMX", @@ -64,4 +71,9 @@ __all__ = [ "UniversePlan", "UniverseRange", "UniverseCollisionError", + "FixturePatch", + "PatchEntry", + "PatchError", + "DmxToParameterRouter", + "RouterStats", ] diff --git a/packages/artnet/hms_artnet/patch.py b/packages/artnet/hms_artnet/patch.py new file mode 100644 index 0000000..6c9744c --- /dev/null +++ b/packages/artnet/hms_artnet/patch.py @@ -0,0 +1,196 @@ +"""Patch-Verwaltung: Layer→DMX-Adressen und Export (PLAN.md §16.2). + +Ein Patch ordnet Show-Layern DMX-Adressen zu: +- Master-Fixture: eine 32-Kanal-Instanz (§16.3) +- Layer-Fixtures: bis zu 8 Instanzen à 64 Kanäle je Universe (§16.2) + +Validierung: +- Adressen lassen ihr Fixture vollständig ins Universe passen +- keine Überlappung im selben Universe (Fehler, kein stilles Verschieben) +- Layer-IDs eindeutig; Komposition/Layer müssen UUIDs sein (§10.2) + +Export (§16.2): enthält Node-ID, Art-Net Short Name, IP, Net/SubNet/Universe, +Startadresse, Layernummer und Fixture-Version. +""" + +from __future__ import annotations + +import csv +import io +import uuid +from dataclasses import dataclass + +MASTER_CHANNELS = 32 # §16.3 +LAYER_CHANNELS = 64 # §16.4 +DMX_CHANNELS = 512 # DMX512 + + +class PatchError(Exception): + """Ungültiger Patch (Überlappung, außerhalb des Universes, …).""" + + +@dataclass(frozen=True) +class PatchEntry: + """Ein Layer-Fixture im Patch (§16.2). + + - layer_id/composition_id: stabile UUIDs (§10.2) + - universe: 15-Bit Port-Address (Net<<8 | SubUni) + - base_address: 1-basierte DMX-Startadresse (1..449 für 64 Kanäle) + - layer_number: menschenlesbare Layernummer für den Export (1-basiert) + """ + + layer_id: str + composition_id: str + universe: int + base_address: int + layer_number: int + + def __post_init__(self) -> None: + for name, value in (("layer_id", self.layer_id), ("composition_id", self.composition_id)): + try: + uuid.UUID(value) + except (ValueError, AttributeError) as exc: + raise ValueError(f"{name} muss eine UUID sein (§10.2)") from exc + if not 0 <= self.universe < 0x8000: + raise ValueError("universe muss 0..0x7FFF sein") + if not 1 <= self.base_address <= DMX_CHANNELS - LAYER_CHANNELS + 1: + raise ValueError( + f"base_address {self.base_address} passt nicht für 64 Kanäle " + f"(erlaubt 1..{DMX_CHANNELS - LAYER_CHANNELS + 1})" + ) + if self.layer_number < 1: + raise ValueError("layer_number muss 1-basiert sein") + + @property + def end_address(self) -> int: + """Letzte belegte DMX-Adresse (inklusiv).""" + return self.base_address + LAYER_CHANNELS - 1 + + +class FixturePatch: + """Vollständiger DMX-Patch eines Nodes (§16.2). + + - master: Adresse des Master32-Fixtures (ein Universe, 1-basiert) + - layers: Liste von PatchEntry (Layer-Fixtures) + + validate(): prüft Überlappungen; add/remove verwalten Einträge. + """ + + def __init__( + self, + node_id: str, + short_name: str, + master_universe: int, + master_base_address: int = 1, + ) -> None: + self.node_id = node_id + self.short_name = short_name + self.master_universe = master_universe + self.master_base_address = master_base_address + self._layers: dict[str, PatchEntry] = {} # layer_id → entry + + # ---------- Verwaltung ---------- + + def add_layer(self, entry: PatchEntry) -> None: + """Fügt ein Layer-Fixture hinzu; Überlappung im selben Universe = Fehler.""" + if entry.layer_id in self._layers: + raise PatchError(f"Layer {entry.layer_id} bereits im Patch") + for existing in self._layers.values(): + if existing.universe == entry.universe and self._overlaps(existing, entry): + raise PatchError( + f"Adressüberlappung Universe {entry.universe}: " + f"L{existing.layer_number} [{existing.base_address}.." + f"{existing.end_address}] vs " + f"L{entry.layer_number} [{entry.base_address}..{entry.end_address}]" + ) + self._layers[entry.layer_id] = entry + + def remove_layer(self, layer_id: str) -> None: + self._layers.pop(layer_id, None) + + def layers(self) -> list[PatchEntry]: + return list(self._layers.values()) + + def layer_by_id(self, layer_id: str) -> PatchEntry | None: + return self._layers.get(layer_id) + + def validate(self) -> list[str]: + """Liefert Fehlerliste; leer = gültig. Master darf kein Layer überlappen.""" + errors: list[str] = [] + if not 1 <= self.master_base_address <= DMX_CHANNELS - MASTER_CHANNELS + 1: + errors.append( + f"Master-Adresse {self.master_base_address} passt nicht für " + f"32 Kanäle (erlaubt 1..{DMX_CHANNELS - MASTER_CHANNELS + 1})" + ) + master_range = ( + self.master_base_address, + self.master_base_address + MASTER_CHANNELS - 1, + ) + for entry in self._layers.values(): + if entry.universe == self.master_universe and self._range_overlap( + (entry.base_address, entry.end_address), master_range + ): + errors.append( + f"Layer {entry.layer_number} überlappt Master-Fixture " + f"im Universe {self.master_universe}" + ) + return errors + + # ---------- Export (§16.2) ---------- + + def export_csv(self, ip_or_host: str = "", fixture_version: str = "1.0.0") -> str: + """Menschenlesbare Kanalliste nach §16.2: enthält Node-ID, Short + Name, IP, Net/SubNet/Universe, Startadresse, Layernummer, Version.""" + buf = io.StringIO() + writer = csv.writer(buf) + writer.writerow( + [ + "node_id", + "artnet_short_name", + "ip_or_host", + "fixture", + "universe", + "start_address", + "channels", + "layer_number", + "fixture_version", + ] + ) + writer.writerow( + [ + self.node_id, + self.short_name, + ip_or_host, + "HMS MediaEngine Master 32ch", + self.master_universe, + self.master_base_address, + MASTER_CHANNELS, + "", + fixture_version, + ] + ) + for entry in sorted(self._layers.values(), key=lambda e: (e.universe, e.base_address)): + writer.writerow( + [ + self.node_id, + self.short_name, + ip_or_host, + "HMS MediaEngine Layer 64ch", + entry.universe, + entry.base_address, + LAYER_CHANNELS, + entry.layer_number, + fixture_version, + ] + ) + return buf.getvalue() + + # ---------- Interna ---------- + + @staticmethod + def _overlaps(a: PatchEntry, b: PatchEntry) -> bool: + return not (a.end_address < b.base_address or b.end_address < a.base_address) + + @staticmethod + def _range_overlap(a: tuple[int, int], b: tuple[int, int]) -> bool: + return not (a[1] < b[0] or b[1] < a[0]) diff --git a/packages/artnet/hms_artnet/wiring.py b/packages/artnet/hms_artnet/wiring.py new file mode 100644 index 0000000..d27bdc8 --- /dev/null +++ b/packages/artnet/hms_artnet/wiring.py @@ -0,0 +1,228 @@ +"""DMX-zu-Parameter-Verkabelung (PLAN.md §11, §16, §29.2). + +Verbindet Art-Net-Receiver → FixturePatch → Master32/Layer64-Engines → +ParameterEngine: + +- jedes DMX-Update wird über den Patch dekodiert +- dekodierte Werte fließen über stabile Parameterpfade in die Engine + (§11: alle Quellen über die Parameter-Engine, nie direkt in den Renderer) +- Master-Blackout läuft mit SAFETY-Priorität (§11.2: überstimmt alles) +- Load/Commit-Events steuern PreloadSlot für atomaren Clipwechsel (§16.5, + §12.2): Preload bei pending selection, Commit wechselt atomar +- Signalverlust: konfigurierbare Policy je Patch (§16.1, §11.3) +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from hms_parameter.engine import ControlSource, ParameterEngine + +from hms_artnet.fixtures import ( + Layer64Engine, + Master32Engine, + TransportCommand, +) +from hms_artnet.patch import MASTER_CHANNELS, FixturePatch +from hms_artnet.receiver import DmxUpdate, LossBehavior + + +@dataclass +class RouterStats: + """Telemetrie je Router (§28.2: Paketrate, Events).""" + + updates_processed: int = 0 + master_updates: int = 0 + layer_updates: int = 0 + load_commits: int = 0 + blackouts: int = 0 + + +class DmxToParameterRouter: + """Routet DMX-Daten über den Patch in die Parameter-Engine. + + Aufbau (§6.1B, §11): + - ein Master32Engine für das Master-Fixture + - je gepatchtem Layer ein Layer64Engine (Flankenzustand je Instanz) + - PreloadSlot je Layer für atomaren Clipwechsel (§12.2, §16.5) + + Blackout (§16.3): Master-Kanal 3 mit SAFETY-Priorität; beim Aufheben + wird der Override releast (nicht auf 0 gesetzt) – der darunterliegende + Zustand bleibt erhalten (§11.3 Release-Semantik). + """ + + def __init__( + self, + patch: FixturePatch, + engine: ParameterEngine, + loss_behavior: LossBehavior = LossBehavior.HOLD, + ) -> None: + self._patch = patch + self._engine = engine + self._loss_behavior = loss_behavior + self._master_engine = Master32Engine() + self._layer_engines: dict[str, Layer64Engine] = {} + self._master_blackout_active = False + self.stats = RouterStats() + # Preload-Slots je Layer (später vom Renderer bedient, §12.2) + self._pending_loads: dict[str, dict] = {} # layer_id → Auswahl + + # ---------- DMX-Verarbeitung ---------- + + def handle_update(self, update: DmxUpdate) -> None: + """Verarbeitet ein DMX-Update über den Patch. + + - Master-Universe: 32 Kanäle ab Master-Basisadresse dekodieren + - Layer-Fixtures: 64 Kanäle je gepatchtem Layer im Universe + - Signalverlust (leere Daten): Policy je Verhalten (§16.1) + """ + self.stats.updates_processed += 1 + if update.sequence == -1 and not update.data: + self._handle_signal_loss(update.universe) + return + self._process_master(update) + self._process_layers(update) + + # ---------- Master (§16.3) ---------- + + def _process_master(self, update: DmxUpdate) -> None: + if update.universe != self._patch.master_universe: + return + start = self._patch.master_base_address - 1 + segment = bytes(update.data[start : start + MASTER_CHANNELS]) + if len(segment) < MASTER_CHANNELS: + return # Universe noch nicht vollständig; kein Fehler + control = self._master_engine.decode(segment) + self.stats.master_updates += 1 + + # Blackout: SAFETY-Priorität, Release beim Aufheben (§11.2, §11.3) + if control.blackout and not self._master_blackout_active: + self._engine.set_value("master/blackout", 1.0, ControlSource.SAFETY) + self._master_blackout_active = True + self.stats.blackouts += 1 + elif not control.blackout and self._master_blackout_active: + self._engine.release("master/blackout", ControlSource.SAFETY) + self._master_blackout_active = False + + src = ControlSource.CONSOLE # Lichtpult (§11.2 Priorität 3) + self._engine.set_value("master/intensity", control.master_intensity, src) + self._engine.set_value("master/global_speed", control.global_speed, src) + self._engine.set_value("master/bpm", control.bpm, src) + self._engine.set_value("master/global_hue", control.global_hue, src) + self._engine.set_value( + "master/global_saturation", control.global_saturation, src + ) + self._engine.set_value( + "master/test_pattern", float(control.test_pattern), src + ) + if control.events: + self.stats.load_commits += len(control.events) + + # ---------- Layer (§16.4, §16.5) ---------- + + def _process_layers(self, update: DmxUpdate) -> None: + for entry in self._patch.layers(): + if entry.universe != update.universe: + continue + layer_engine = self._layer_engines.setdefault( + entry.layer_id, Layer64Engine() + ) + start = entry.base_address - 1 + segment = bytes(update.data[start : start + 64]) + if len(segment) < 64: + continue # Universe unvollständig: Layer still auslassen + control = layer_engine.decode(segment) + self.stats.layer_updates += 1 + base = f"composition/{entry.composition_id}/layer/{entry.layer_id}" + src = ControlSource.CONSOLE + + # Grundsteuerung (§16.4) + if control.enabled: + self._engine.set_value(f"{base}/enabled", 1.0, src) + else: + self._engine.set_value(f"{base}/enabled", 0.0, src) + self._engine.set_value(f"{base}/opacity", control.opacity, src) + self._engine.set_value(f"{base}/blend_mode", float(control.blend_mode), src) + self._engine.set_value( + f"{base}/transform/position_x", control.position_x, src + ) + self._engine.set_value( + f"{base}/transform/position_y", control.position_y, src + ) + self._engine.set_value(f"{base}/transform/scale_x", control.scale_x, src) + self._engine.set_value(f"{base}/transform/scale_y", control.scale_y, src) + self._engine.set_value( + f"{base}/transform/rotation_deg", control.rotation_deg, src + ) + self._engine.set_value(f"{base}/color/hue", control.hue, src) + self._engine.set_value( + f"{base}/color/saturation", control.saturation, src + ) + self._engine.set_value( + f"{base}/color/brightness", control.brightness, src + ) + self._engine.set_value(f"{base}/color/contrast", control.contrast, src) + + # Medien-Transport (§16.4 Kanäle 10-20, Media-Modus) + self._engine.set_value( + f"{base}/source/speed", control.speed, src + ) + self._engine.set_value( + f"{base}/source/position", control.position, src + ) + self._engine.set_value( + f"{base}/source/in_point", control.in_point, src + ) + self._engine.set_value( + f"{base}/source/out_point", control.out_point, src + ) + if control.transport is TransportCommand.RETRIGGER: + self._engine.set_value(f"{base}/source/retrigger", 1.0, src) + + # FX-Steuerung (§16.4 Kanäle 41-62) + self._engine.set_value( + f"{base}/fx1/enabled", 1.0 if control.fx1_enabled else 0.0, src + ) + self._engine.set_value(f"{base}/fx1/mix", control.fx1_mix, src) + self._engine.set_value( + f"{base}/fx2/enabled", 1.0 if control.fx2_enabled else 0.0, src + ) + self._engine.set_value(f"{base}/fx2/mix", control.fx2_mix, src) + + # Load/Commit-Events: atomare Auswahl (§16.5, §12.2) + for event in control.events: + if event.kind == "load_commit": + self.stats.load_commits += 1 + self._pending_loads[entry.layer_id] = { + "bank": event.pending_bank, + "folder": event.pending_folder, + "index": event.pending_index, + "universe": update.universe, + } + elif event.kind == "retrigger": + self._engine.set_value(f"{base}/source/retrigger", 1.0, src) + + # ---------- Ausstehende Loads (§12.2: Renderer bedient Preload) ---------- + + def pending_load_for(self, layer_id: str) -> dict | None: + """Liefert die ausstehende Load-Auswahl eines Layers (für den + Renderer-Preload) und entfernt sie (Verbrauch durch Aufrufer).""" + return self._pending_loads.pop(layer_id, None) + + # ---------- Signalverlust (§16.1, §11.3) ---------- + + def _handle_signal_loss(self, universe: int) -> None: + """DMX-Ausfall auf einem Universe: Policy je Konfiguration. + + - HOLD: nichts tun (letzter Zustand bleibt, §11.3) + - FADE_TO_BLACK: Master-Override auf 0 setzen (SAFETY) + """ + if self._loss_behavior is LossBehavior.HOLD: + return + if ( + self._loss_behavior is LossBehavior.FADE_TO_BLACK + and universe == self._patch.master_universe + ): + # Intensität über SAFETY auf 0; beim Wiederkommen releast die + # Master-Verarbeitung den Override nicht – hier explizit setzen + self._engine.set_value("master/intensity", 0.0, ControlSource.SAFETY) diff --git a/tests/unit/test_patch_and_wiring.py b/tests/unit/test_patch_and_wiring.py new file mode 100644 index 0000000..2cb5412 --- /dev/null +++ b/tests/unit/test_patch_and_wiring.py @@ -0,0 +1,413 @@ +"""Tests Patch-Verwaltung und DMX-zu-Parameter-Verkabelung +(PLAN.md §16.2, §11, §29.2). + +Kette ohne Mocks: DMX-Bytes → Patch → Master32/Layer64-Engines → +ParameterEngine. Prüft Adressvalidierung, Überlappungen, Export, +Blackout-SAFETY, Load/Commit-Events, Signalverlust-Policies. +""" + +from __future__ import annotations + +import csv +import io +import uuid + +import pytest +from hms_artnet import ( + DmxToParameterRouter, + DmxUpdate, + FixturePatch, + LossBehavior, + PatchEntry, + PatchError, +) +from hms_parameter.engine import ControlSource, ParameterEngine + + +def _uuid() -> str: + return str(uuid.uuid4()) + + +# ---------- PatchEntry-Validierung (§16.2) ---------- + + +def test_patch_entry_requires_uuids() -> None: + with pytest.raises(ValueError, match="UUID"): + PatchEntry( + layer_id="keine-uuid", + composition_id=_uuid(), + universe=0, + base_address=1, + layer_number=1, + ) + + +def test_patch_entry_base_address_must_fit_64_channels() -> None: + """64 Kanäle müssen vollständig ins 512er-Universe passen.""" + with pytest.raises(ValueError, match="passt nicht"): + PatchEntry( + layer_id=_uuid(), + composition_id=_uuid(), + universe=0, + base_address=450, # 450+63 > 512 + layer_number=1, + ) + valid = PatchEntry( + layer_id=_uuid(), + composition_id=_uuid(), + universe=0, + base_address=449, # exakt bis 512 + layer_number=1, + ) + assert valid.end_address == 512 + + +# ---------- FixturePatch: Überlappungen (§16.2) ---------- + + +def _patch() -> FixturePatch: + return FixturePatch( + node_id=_uuid(), + short_name="HMS Test Node", + master_universe=0, + master_base_address=1, + ) + + +def _layer_entry(universe: int, base: int, number: int) -> PatchEntry: + return PatchEntry( + layer_id=_uuid(), + composition_id=_uuid(), + universe=universe, + base_address=base, + layer_number=number, + ) + + +def test_patch_accepts_eight_layers_one_universe() -> None: + """§16.2: 8 Layer-Fixtures à 64 Kanäle = exakt 1 Universe.""" + patch = FixturePatch( + node_id=_uuid(), + short_name="N", + master_universe=1, # Master separat + master_base_address=1, + ) + for i in range(8): + patch.add_layer(_layer_entry(universe=0, base=1 + i * 64, number=i + 1)) + assert len(patch.layers()) == 8 + assert patch.validate() == [] + + +def test_patch_rejects_overlapping_layers() -> None: + patch = _patch() + patch.add_layer(_layer_entry(universe=0, base=33, number=1)) + with pytest.raises(PatchError, match="berlappung"): + patch.add_layer(_layer_entry(universe=0, base=64, number=2)) # 64..127 vs 33..96 + + +def test_patch_allows_same_address_in_different_universes() -> None: + """Gleiche Adressen in verschiedenen Universen sind erlaubt (§16.2).""" + patch = _patch() + patch.add_layer(_layer_entry(universe=0, base=1, number=1)) + patch.add_layer(_layer_entry(universe=1, base=1, number=2)) + assert len(patch.layers()) == 2 + + +def test_patch_rejects_duplicate_layer_id() -> None: + patch = _patch() + entry = _layer_entry(universe=0, base=1, number=1) + patch.add_layer(entry) + with pytest.raises(PatchError, match="bereits"): + patch.add_layer(entry) + + +def test_patch_validate_detects_master_overlap() -> None: + """validate() meldet Master-Überlappung; Layer-Layer-Konflikt fängt + add_layer selbst (PatchError) – hier wird nur Master-Overlap geprüft.""" + patch = FixturePatch( + node_id=_uuid(), + short_name="N", + master_universe=0, + master_base_address=1, + ) + # Master: 1..32; Layer 1: 33..96 (kein Konflikt untereinander) + patch.add_layer(_layer_entry(universe=0, base=33, number=1)) + assert patch.validate() == [] + # Master auf 65..96 verschieben: überlappt Layer 1 (33..96) + patch.master_base_address = 65 + errors = patch.validate() + assert any("Master" in e for e in errors) + # Master-Adresse zurück auf gültig: keine Fehler mehr + patch.master_base_address = 1 + assert patch.validate() == [] + + +def test_patch_validate_detects_invalid_master_address() -> None: + patch = FixturePatch( + node_id=_uuid(), + short_name="N", + master_universe=0, + master_base_address=490, # 490+31 > 512 + ) + errors = patch.validate() + assert any("Master-Adresse" in e for e in errors) + + +# ---------- Patch-Export (§16.2) ---------- + + +def test_patch_export_csv_contains_required_fields() -> None: + """§16.2: Export enthält Node-ID, Short Name, IP, Universe, + Startadresse, Layernummer, Fixture-Version.""" + patch = FixturePatch( + node_id="11111111-2222-3333-4444-555555555555", + short_name="HMS Node A", + master_universe=0, + master_base_address=1, + ) + patch.add_layer(_layer_entry(universe=0, base=33, number=1)) + patch.add_layer(_layer_entry(universe=0, base=97, number=2)) + + csv_text = patch.export_csv(ip_or_host="10.0.0.9", fixture_version="1.0.0") + rows = list(csv.reader(io.StringIO(csv_text))) + + header = rows[0] + assert "node_id" in header + assert "universe" in header + assert "start_address" in header + assert "layer_number" in header + assert "fixture_version" in header + + master_row = rows[1] + assert master_row[0] == "11111111-2222-3333-4444-555555555555" + assert master_row[1] == "HMS Node A" + assert master_row[2] == "10.0.0.9" + assert "Master 32ch" in master_row[3] + assert master_row[8] == "1.0.0" + + layer_rows = rows[2:] + assert len(layer_rows) == 2 + assert "Layer 64ch" in layer_rows[0][3] + assert layer_rows[0][7] == "1" + assert layer_rows[1][7] == "2" + + +def test_patch_export_sorted_by_universe_and_address() -> None: + patch = _patch() + patch.add_layer(_layer_entry(universe=1, base=1, number=2)) + patch.add_layer(_layer_entry(universe=0, base=65, number=1)) + csv_text = patch.export_csv() + rows = list(csv.reader(io.StringIO(csv_text))) + # Master zuerst (Universe 0), dann Layer nach Universe/Adresse sortiert + layer_rows = rows[2:] + assert int(layer_rows[0][4]) < int(layer_rows[1][4]) # Universum aufsteigend + + +# ---------- DmxToParameterRouter: Kette ohne Mocks (§11, §29.2) ---------- + + +def _dmx_update(universe: int, data: bytes) -> DmxUpdate: + return DmxUpdate( + universe=universe, + data=data, + sender_ip="192.168.1.100", + received_ns=0, + sequence=1, + ) + + +def _layer_patch_with_two_layers() -> tuple[FixturePatch, str, str, str, str]: + """Patch mit Master (Uni 0) und 2 Layern (Uni 0, Adressen 33 und 97).""" + comp_id = _uuid() + layer1_id, layer2_id = _uuid(), _uuid() + patch = FixturePatch( + node_id=_uuid(), + short_name="HMS Router Test", + master_universe=0, + master_base_address=1, + ) + patch.add_layer( + PatchEntry( + layer_id=layer1_id, + composition_id=comp_id, + universe=0, + base_address=33, + layer_number=1, + ) + ) + patch.add_layer( + PatchEntry( + layer_id=layer2_id, + composition_id=comp_id, + universe=0, + base_address=97, + layer_number=2, + ) + ) + return patch, comp_id, layer1_id, layer2_id, "" + + +def _full_universe_with_master_and_layers( + master: dict[int, int], + layer1: dict[int, int], + layer2: dict[int, int], + layer1_offset: int = 32, # 0-basiert: Basisadresse 33 - 1 + layer2_offset: int = 96, # 0-basiert: Basisadresse 97 - 1 +) -> bytes: + """512 Kanäle; dicts sind (kanal_nr_1basiert_relativ, wert); + Offsets sind 0-basiert (Basisadresse-1).""" + data = bytearray(512) + for k, v in master.items(): + data[int(k) - 1] = v + for k, v in layer1.items(): + data[layer1_offset + int(k) - 1] = v + for k, v in layer2.items(): + data[layer2_offset + int(k) - 1] = v + return bytes(data) + + +def test_router_full_chain_master_and_layers() -> None: + """Kette: DMX → Patch → Engines → ParameterEngine; Werte landen korrekt.""" + patch, comp, l1, l2, _ = _layer_patch_with_two_layers() + engine = ParameterEngine() + router = DmxToParameterRouter(patch, engine) + + data = _full_universe_with_master_and_layers( + master={"1": 0x40, "2": 0x00, "3": 0, "14": 60, "15": 0}, # Intensität ~0.25, BPM + layer1={ + "1": 255, # Enable + "2": 0xFF, "3": 0xFF, # Opacity 1.0 + "4": 0, # Media + "10": 1, # Play + "13": 0xA0, "14": 0x00, # Speed 1x + "41": 255, # FX1 an + "43": 128, # FX1 Mix + }, + layer2={"1": 0, "2": 0x80, "3": 0x00}, # disabled, Opacity 50% + ) + router.handle_update(_dmx_update(0, data)) + + base1 = f"composition/{comp}/layer/{l1}" + base2 = f"composition/{comp}/layer/{l2}" + # Master-Werte + assert engine.effective_value("master/intensity") == pytest.approx(0x4000 / 65535) + assert engine.current_source("master/intensity") is ControlSource.CONSOLE + # Layer 1 aktiv + assert engine.effective_value(f"{base1}/enabled") == pytest.approx(1.0) + assert engine.effective_value(f"{base1}/opacity") == pytest.approx(1.0) + assert engine.effective_value(f"{base1}/source/speed") == pytest.approx(1.0) + assert engine.effective_value(f"{base1}/fx1/enabled") == pytest.approx(1.0) + assert engine.effective_value(f"{base1}/fx1/mix") == pytest.approx(128 / 255) + # Layer 2 disabled + assert engine.effective_value(f"{base2}/enabled") == pytest.approx(0.0) + assert engine.effective_value(f"{base2}/opacity") == pytest.approx(0x8000 / 65535) + # Telemetrie + assert router.stats.updates_processed == 1 + assert router.stats.master_updates == 1 + assert router.stats.layer_updates == 2 + + +def test_router_blackout_safety_priority() -> None: + """§11.2/§16.3: Blackout überstimmt alles mit SAFETY; Release beim Aufheben.""" + patch, comp, l1, _, _ = _layer_patch_with_two_layers() + engine = ParameterEngine() + router = DmxToParameterRouter(patch, engine) + + # Web setzt Intensität hoch (niedrigere Priorität) + engine.set_value("master/intensity", 0.9, ControlSource.WEB) + + # Blackout aktivieren (Kanal 3 = 255) + data = _full_universe_with_master_and_layers( + master={"3": 255}, layer1={}, layer2={} + ) + router.handle_update(_dmx_update(0, data)) + assert engine.effective_value("master/blackout") == pytest.approx(1.0) + assert engine.current_source("master/blackout") is ControlSource.SAFETY + assert router.stats.blackouts == 1 + + # Blackout aufheben (Kanal 3 = 0) + data = _full_universe_with_master_and_layers( + master={"3": 0}, layer1={}, layer2={} + ) + router.handle_update(_dmx_update(0, data)) + # SAFETY-Override ist released; Web-Wert wirkt wieder + snap = engine.snapshot() + assert "master/blackout" not in snap + + +def test_router_load_commit_produces_pending_load() -> None: + """§16.5: Flanke auf Kanal 9 erzeugt load_commit; pending_load für Renderer.""" + patch, comp, l1, _, _ = _layer_patch_with_two_layers() + engine = ParameterEngine() + router = DmxToParameterRouter(patch, engine) + + # Frame 1: Auswahl setzen, kein Commit + data = _full_universe_with_master_and_layers( + master={}, layer1={"5": 2, "6": 3, "9": 0}, layer2={} + ) + router.handle_update(_dmx_update(0, data)) + assert router.pending_load_for(l1) is None # kein Commit + + # Frame 2: Commit-Flanke + data = _full_universe_with_master_and_layers( + master={}, layer1={"5": 2, "6": 3, "9": 255}, layer2={} + ) + router.handle_update(_dmx_update(0, data)) + pending = router.pending_load_for(l1) + assert pending is not None + assert pending["bank"] == 2 + assert pending["folder"] == 3 + assert pending["universe"] == 0 + assert router.stats.load_commits == 1 + # Verbrauch entfernt den Eintrag + assert router.pending_load_for(l1) is None + + +def test_router_signal_loss_hold_keeps_state() -> None: + """§11.3 HOLD: Signalverlust ändert nichts.""" + patch, _, _, _, _ = _layer_patch_with_two_layers() + engine = ParameterEngine() + router = DmxToParameterRouter(patch, engine, LossBehavior.HOLD) + engine.set_value("master/intensity", 0.8, ControlSource.CONSOLE) + + loss = DmxUpdate( + universe=0, data=b"", sender_ip="x", received_ns=1, sequence=-1 + ) + router.handle_update(loss) + assert engine.effective_value("master/intensity") == pytest.approx(0.8) + + +def test_router_signal_loss_fade_to_black_safety() -> None: + """§11.3 fade_to_black: Master-Universe-Ausfall setzt SAFETY-Override.""" + patch, _, _, _, _ = _layer_patch_with_two_layers() + engine = ParameterEngine() + router = DmxToParameterRouter(patch, engine, LossBehavior.FADE_TO_BLACK) + engine.set_value("master/intensity", 0.8, ControlSource.CONSOLE) + + loss = DmxUpdate( + universe=0, data=b"", sender_ip="x", received_ns=1, sequence=-1 + ) + router.handle_update(loss) + assert engine.effective_value("master/intensity") == pytest.approx(0.0) + assert engine.current_source("master/intensity") is ControlSource.SAFETY + + +def test_router_ignores_foreign_universe() -> None: + """Updates für nicht gepatchte Universen ändern nichts.""" + patch, comp, l1, _, _ = _layer_patch_with_two_layers() + engine = ParameterEngine() + router = DmxToParameterRouter(patch, engine) + router.handle_update(_dmx_update(99, bytes(512))) + snap = engine.snapshot() + assert f"composition/{comp}/layer/{l1}/enabled" not in snap + assert router.stats.master_updates == 0 + assert router.stats.layer_updates == 0 + + +def test_router_short_universe_data_skipped_silently() -> None: + """Unvollständige Universen werden still ausgelassen (kein Fehler).""" + patch, _, _, _, _ = _layer_patch_with_two_layers() + engine = ParameterEngine() + router = DmxToParameterRouter(patch, engine) + router.handle_update(_dmx_update(0, bytes(16))) # nur 16 Kanäle + assert router.stats.master_updates == 0 # nichts dekodiert