hms-mediaengine/packages/artnet/hms_artnet/patch.py

"""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])