hms-mediaengine/hms_app/artnet.py

"""Art-Net-Empfang (ArtDMX): konfigurierbar, mit Sequenz- und Signalprüfung."""

from __future__ import annotations

import socket
import struct
import threading
import time


class ArtNetInput:
    """ArtDMX auf UDP. Mapping, Port und Universes kommen aus den Settings."""

    OP_ARTDMX = 0x5000

    def __init__(self, engine, settings: dict):
        self.engine = engine
        self.settings = settings
        self.sock: socket.socket | None = None
        self.running = False
        self.thread: threading.Thread | None = None
        self.watchdog: threading.Thread | None = None
        self.last_channels = bytearray(512)
        self.stats = {"packets": 0, "accepted": 0, "seq_errors": 0,
                      "last_universe": None, "last_seq": None,
                      "last_time": None}
        self._last_seq: dict[int, int] = {}
        self._wd_blackout = False

    @property
    def cfg(self) -> dict:
        return self.settings["artnet"]

    def start(self) -> bool:
        if not self.cfg.get("enabled", True):
            print("[ArtNet] deaktiviert (Einstellungen)")
            return True
        port = int(self.cfg.get("listen_port", 6454))
        try:
            self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            self.sock.bind(("0.0.0.0", port))
            self.sock.settimeout(0.5)
        except OSError as e:
            print(f"[ArtNet] FEHLER: UDP {port} nicht bindbar: {e}")
            return False
        self.running = True
        self.thread = threading.Thread(target=self._loop, daemon=True)
        self.thread.start()
        self.watchdog = threading.Thread(target=self._watchdog, daemon=True)
        self.watchdog.start()
        print(f"[ArtNet] Empfang aktiv auf UDP {port}, "
              f"Universes: {self.cfg.get('universes')}")
        return True

    def restart(self) -> None:
        self.stop()
        self.start()

    def stop(self) -> None:
        self.running = False
        if self.thread:
            self.thread.join(timeout=1.0)
            self.thread = None
        if self.watchdog:
            self.watchdog.join(timeout=1.5)
            self.watchdog = None
        if self.sock:
            self.sock.close()
            self.sock = None
        if self._wd_blackout:
            self._wd_blackout = False
            self.engine.set_blackout(False)

    def _loop(self) -> None:
        while self.running:
            try:
                data, _addr = self.sock.recvfrom(2048)
            except socket.timeout:
                continue
            except OSError:
                break
            self._handle(data)

    def _handle(self, data: bytes) -> None:
        if len(data) < 18 or data[0:8] != b"Art-Net\x00":
            return
        opcode = struct.unpack("<H", data[8:10])[0]
        if opcode != self.OP_ARTDMX:
            return
        if struct.unpack(">H", data[10:12])[0] < 14:
            return
        seq = data[12]
        universe = struct.unpack("<H", data[14:16])[0]
        length = struct.unpack(">H", data[16:18])[0]
        if length == 0:
            return
        self.stats["packets"] += 1
        if universe not in (self.cfg.get("universes") or [0]):
            return
        last = self._last_seq.get(universe)
        if last is not None:
            if seq == last:
                return  # Duplikat verwerfen
            if seq != (last + 1) % 256:
                self.stats["seq_errors"] += 1
        self._last_seq[universe] = seq
        self.stats["accepted"] += 1
        self.stats["last_universe"] = universe
        self.stats["last_seq"] = seq
        self.stats["last_time"] = time.time()
        dmx = data[18:18 + min(length, 512)]
        self.last_channels[:len(dmx)] = dmx
        if self._wd_blackout:
            self._wd_blackout = False
            self.engine.set_blackout(False)
        self.engine.apply_dmx(bytes(self.last_channels[:len(dmx)]),
                              universe, seq)

    def _watchdog(self) -> None:
        while self.running:
            time.sleep(1.0)
            if self.cfg.get("signal_loss_policy") != "fade_black":
                continue
            if not self.cfg.get("enabled", True):
                continue
            lt = self.stats.get("last_time")
            if lt is None:
                continue
            timeout = float(self.cfg.get("signal_loss_timeout_s", 10))
            if time.time() - lt > timeout and not self._wd_blackout:
                self._wd_blackout = True
                self.engine.set_blackout(True)
                print("[ArtNet] Signalverlust: Fallback-Blackout aktiv")

    def monitor(self) -> dict:
        return {
            "enabled": bool(self.cfg.get("enabled", True)),
            "listen_port": self.cfg.get("listen_port"),
            "universes": self.cfg.get("universes"),
            "stats": dict(self.stats),
            "channels": list(self.last_channels[:32]),
        }