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).
86 lines
3.3 KiB
Python
86 lines
3.3 KiB
Python
"""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())
|