Phase 0: Repository-Initialisierung nach Bauplan v1.2

- Struktur gemäß §8 (Eigentumsgrenzen), PLAN.md als normative Basis - Pflichtdokumente: STATUS.md, ERRORS.md, TEST_REPORT.md, CHANGELOG.md, ADRs - ADR-0001 Python 3.13-Pin, ADR-0002 GStreamer 1.28.6-Pin (Windows), ADR-0003 IPC TCP+MessagePack v1 - Kernpakete: hms_protocol, hms_domain, hms_parameter, hms_artnet, hms_adaptive, hms_capabilities, hms_plugin_sdk - Renderer-Spike: D3D11-Primärpfad + Dev-GL-Pfad (§36 Nr. 4-5) - Control Core: FastAPI REST + WebSocket (§36 Nr. 9) - Beispielplugins: Passthrough + Gaussian Blur (3 Adaptive-Quality- Varianten, HLSL/GLSL/GLES) - Tools: Art-Net-Emulator, Fixture-Generator (Master32/Layer64-CSV), Capability-Probe - JSON-Schemas: IPC, Plugin, Projekt, Cluster - 121 Unit-/Integrationstests grün, Ruff grün Gate 0 bleibt offen: Hardwaremessungen nur auf echter Windows-Referenz- hardware gültig (§29.7, §33).
2026-09-11 00:36:59 +02:00
commit 0922cc1d68
133 changed files with 7939 additions and 0 deletions
@@ -0,0 +1,85 @@
+"""Art-Net-Emulator (PLAN.md §16, §29.2, §36 Nr. 7–8).
+
+Simuliert ein Lichtpult für Gate-0-Tests:
+- poll: sendet ArtPoll und zeigt die ArtPollReply des Nodes (Media Server)
+- sweep: fährt einen Fader über Layer-Opacity (16 Bit, Kanäle 2–3) und
+  sendet ArtDMX mit Sequenznummern
+
+Nur für Testnetze bestimmt; kein Ersatz für den Hardwarepulttest (§29.7).
+"""
+
+from __future__ import annotations
+
+import argparse
+import socket
+import time
+
+from hms_artnet.packets import build_dmx, build_poll, parse_poll_reply
+
+
+def cmd_poll(host: str, port: int) -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
+        sock.settimeout(2.0)
+        sock.bind(("0.0.0.0", 0))
+        local_port = sock.getsockname()[1]
+        sock.sendto(build_poll(talk_to_me=0x00), (host, port))
+        try:
+            data, addr = sock.recvfrom(2048)
+        except TimeoutError:
+            print("keine ArtPollReply erhalten (Timeout)")
+            return 1
+        info = parse_poll_reply(data)
+        if info is None:
+            print(f"ungültige Antwort von {addr}")
+            return 1
+        print(f"ArtPollReply von {addr[0]} → lokal gebunden auf Port {local_port}")
+        print(f"  IP:          {info.ip}")
+        print(f"  ShortName:   {info.short_name}")
+        print(f"  LongName:    {info.long_name}")
+        print(f"  NodeReport:  {info.node_report}")
+        print(f"  Style:       0x{info.style:02X} (0x02 = StMedia/Media Server)")
+        print(f"  Ports:       {info.num_ports}")
+        return 0
+
+
+def cmd_sweep(host: str, port: int, universe: int, rate_hz: float, cycles: int) -> int:
+    sequence = 0
+    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
+        for _ in range(cycles):
+            for step in range(0, 256, 8):
+                # Layer64-Layout: Kanal 1 Enable, Kanal 2-3 Opacity 16 Bit (MSB zuerst)
+                data = bytearray(16)
+                data[0] = 255
+                data[1] = step
+                data[2] = 0
+                sequence = (sequence % 255) + 1
+                sock.sendto(build_dmx(universe, bytes(data), sequence=sequence), (host, port))
+                time.sleep(1.0 / rate_hz)
+            for step in range(255, -1, -8):
+                data = bytearray(16)
+                data[0] = 255
+                data[1] = step
+                data[2] = 0
+                sequence = (sequence % 255) + 1
+                sock.sendto(build_dmx(universe, bytes(data), sequence=sequence), (host, port))
+                time.sleep(1.0 / rate_hz)
+    print(f"sweep abgeschlossen: {cycles} Zyklen auf Universe {universe}")
+    return 0
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(prog="artnet_emulator")
+    parser.add_argument("--host", default="127.0.0.1")
+    parser.add_argument("--port", type=int, default=6454)
+    parser.add_argument("--mode", choices=["poll", "sweep"], required=True)
+    parser.add_argument("--universe", type=int, default=0)
+    parser.add_argument("--rate", type=float, default=40.0, help="Pakete pro Sekunde")
+    parser.add_argument("--cycles", type=int, default=1)
+    args = parser.parse_args(argv)
+    if args.mode == "poll":
+        return cmd_poll(args.host, args.port)
+    return cmd_sweep(args.host, args.port, args.universe, args.rate, args.cycles)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())