HMS MediaEngine Agent
362e089be0
Der Nutzer hat recht: Der Ordner war voller Entwicklungs-Muell. Jetzt ist sauber getrennt: ROOT (was der Nutzer sieht und braucht): - run.py = das Programm - hms_app/ = der Anwendungscode - HMS MediaEngine.app = macOS Doppelklick-Starter - HMS-Start.vbs = Windows Doppelklick-Starter - HMS-Install.vbs = Windows Erst-Installation - HMS-Mac-Install.command = macOS Homebrew-Installation - HMS-Portable-Install.command = macOS Portable-Installation (16GB-Fix) - installer_gui.py = grafischer Installer - launcher.pyw + launcher_core.py = interne Start-Logik - LIESMICH.txt = 10-Zeilen-Kurzanleitung - .gitignore _entwicklung/ (alles andere, NICHT benoetigt): - packages/ apps/ native/ plugins/ tools/ schemas/ tests/ docs/ build/ fixture_profiles/ - PLAN.md STATUS.md ERRORS.md TEST_REPORT.md CHANGELOG.md README.md - pyproject.toml uv.lock setup_*.sh/ps1 make_mac_app.py Diese Trennung gilt ab sofort fuer alle Commits. Der Nutzer kann _entwicklung/ loeschen wenn er Platz braucht - die App laeuft ohne. Verifiziert: App startet nach Aufraeumen unveraendert (Health 200).
414 lines
14 KiB
Python
414 lines
14 KiB
Python
"""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
|