294 lines
11 KiB
Python
294 lines
11 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""E2E: Cue-/Szenenliste pro Layer (GO, Fade, Auto-Follow, DMX-GO, Projekte).
|
||
|
|
|
||
|
|
Startet einen echten Server als Subprozess.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import shutil
|
||
|
|
import socket
|
||
|
|
import struct
|
||
|
|
import subprocess
|
||
|
|
import sys
|
||
|
|
import time
|
||
|
|
import urllib.error
|
||
|
|
import urllib.parse
|
||
|
|
import urllib.request
|
||
|
|
|
||
|
|
ROOT = "/a0/usr/workdir/hms-mediaengine"
|
||
|
|
PORT = 8097
|
||
|
|
BASE = "http://localhost:%d" % PORT
|
||
|
|
|
||
|
|
proc = None
|
||
|
|
|
||
|
|
|
||
|
|
def start_server() -> bool:
|
||
|
|
global proc
|
||
|
|
proc = subprocess.Popen(
|
||
|
|
["/usr/bin/python3", ROOT + "/run.py", "--port", str(PORT)],
|
||
|
|
cwd=ROOT, stdout=open("/tmp/hms_cue.log", "ab"),
|
||
|
|
stderr=subprocess.STDOUT)
|
||
|
|
deadline = time.time() + 25
|
||
|
|
while time.time() < deadline:
|
||
|
|
try:
|
||
|
|
with urllib.request.urlopen(BASE + "/api/health",
|
||
|
|
timeout=2) as r:
|
||
|
|
if json.loads(r.read().decode()).get("status") == "ok":
|
||
|
|
return True
|
||
|
|
except Exception: # noqa: BLE001
|
||
|
|
time.sleep(0.3)
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
def stop_server():
|
||
|
|
global proc
|
||
|
|
if proc is not None:
|
||
|
|
proc.terminate()
|
||
|
|
try:
|
||
|
|
proc.wait(timeout=6)
|
||
|
|
except subprocess.TimeoutExpired:
|
||
|
|
proc.kill()
|
||
|
|
proc = None
|
||
|
|
time.sleep(0.5)
|
||
|
|
|
||
|
|
|
||
|
|
def get(path):
|
||
|
|
with urllib.request.urlopen(BASE + path, timeout=15) as r:
|
||
|
|
return json.loads(r.read().decode())
|
||
|
|
|
||
|
|
|
||
|
|
def post(path, body=None):
|
||
|
|
data = json.dumps(body or {}).encode()
|
||
|
|
req = urllib.request.Request(BASE + path, data=data, method="POST")
|
||
|
|
req.add_header("Content-Type", "application/json")
|
||
|
|
try:
|
||
|
|
with urllib.request.urlopen(req, timeout=120) as r:
|
||
|
|
return json.loads(r.read().decode())
|
||
|
|
except urllib.error.HTTPError as e:
|
||
|
|
return json.loads(e.read().decode())
|
||
|
|
|
||
|
|
|
||
|
|
def check(label, ok, detail=""):
|
||
|
|
print(("PASS" if ok else "FAIL") + " " + label + " " + str(detail))
|
||
|
|
if not ok:
|
||
|
|
sys.exit(1)
|
||
|
|
|
||
|
|
|
||
|
|
def layer_state():
|
||
|
|
return get("/api/status")["layers"][0]
|
||
|
|
|
||
|
|
|
||
|
|
# ---------- Reset ----------
|
||
|
|
for d in ("media", "thumbs", "projects"):
|
||
|
|
shutil.rmtree(os.path.join(ROOT, d), ignore_errors=True)
|
||
|
|
try:
|
||
|
|
os.remove(os.path.join(ROOT, "config.json"))
|
||
|
|
except FileNotFoundError:
|
||
|
|
pass
|
||
|
|
|
||
|
|
try:
|
||
|
|
check("Server startet", start_server())
|
||
|
|
|
||
|
|
# Testvideo + 2 Layer anlegen (Video + Generator)
|
||
|
|
subprocess.run([
|
||
|
|
"/usr/bin/python3", "-c",
|
||
|
|
"import gi; gi.require_version('Gst','1.0')\n"
|
||
|
|
"from gi.repository import Gst; Gst.init(None)\n"
|
||
|
|
"p = Gst.parse_launch('videotestsrc num-buffers=90 pattern=18 ! "
|
||
|
|
"video/x-raw,width=320,height=180 ! x264enc tune=zerolatency ! "
|
||
|
|
"mp4mux ! filesink location=/tmp/e2e_cue.mp4')\n"
|
||
|
|
"p.set_state(Gst.State.PLAYING)\n"
|
||
|
|
"p.get_bus().timed_pop_filtered(Gst.CLOCK_TIME_NONE, "
|
||
|
|
"Gst.MessageType.EOS)\n"
|
||
|
|
"p.set_state(Gst.State.NULL)\n",
|
||
|
|
], check=True)
|
||
|
|
data = open("/tmp/e2e_cue.mp4", "rb").read()
|
||
|
|
req = urllib.request.Request(BASE + "/api/media/upload",
|
||
|
|
data=data, method="POST")
|
||
|
|
req.add_header("Content-Type", "application/octet-stream")
|
||
|
|
req.add_header("X-Filename", urllib.parse.quote("cue_clip.mp4"))
|
||
|
|
with urllib.request.urlopen(req, timeout=120) as r:
|
||
|
|
json.loads(r.read().decode())
|
||
|
|
|
||
|
|
r = post("/api/layers/add", {"name": "cue_clip.mp4"})
|
||
|
|
check("Layer 1 (Video)", "layer" in r, r)
|
||
|
|
lid = r["layer"]["id"]
|
||
|
|
|
||
|
|
# ---- Cue 1: aktueller Zustand (Video voll sichtbar, hart) ----
|
||
|
|
r = post("/api/layers/cues/add",
|
||
|
|
{"id": lid, "name": "Solo Clip", "fade_ms": 0})
|
||
|
|
check("Cue 1 gesichert", r.get("cue", {}).get("name") == "Solo Clip", r)
|
||
|
|
c1_id = r["cue"]["cue_id"]
|
||
|
|
|
||
|
|
# Zustand aendern: Generator + Blur + Alpha 0.8
|
||
|
|
r = post("/api/layers/update",
|
||
|
|
{"id": lid, "name": "ball", "fx1": "blur",
|
||
|
|
"alpha": 0.8})
|
||
|
|
check("Zustand geaendert", r.get("layer", {}).get("name") == "ball", r)
|
||
|
|
|
||
|
|
# ---- Cue 2: Generator mit Blur, mit Fade ----
|
||
|
|
r = post("/api/layers/cues/add",
|
||
|
|
{"id": lid, "name": "Ball mit Blur", "fade_ms": 800})
|
||
|
|
check("Cue 2 gesichert", r.get("cue", {}).get("name")
|
||
|
|
== "Ball mit Blur", r)
|
||
|
|
|
||
|
|
# Zustand aendern: Snow, andere Position
|
||
|
|
post("/api/layers/update", {"id": lid, "name": "snow", "x": 100})
|
||
|
|
post("/api/layers/cues/add",
|
||
|
|
{"id": lid, "name": "Snow rechts", "fade_ms": 0})
|
||
|
|
|
||
|
|
cues = get("/api/layers/cues?id=%d" % lid)
|
||
|
|
check("Cue-Liste hat 3", len(cues["cues"]) == 3,
|
||
|
|
[c["name"] for c in cues["cues"]])
|
||
|
|
check("Kein Cue aktiv", cues["index"] == -1)
|
||
|
|
|
||
|
|
# ---- GO: Cue 0 aktiv, harter Schnitt -> Video wieder da ----
|
||
|
|
r = post("/api/layers/cues/go", {"id": lid})
|
||
|
|
check("GO -> Cue 1", r.get("index") == 0
|
||
|
|
and r.get("cue") == "Solo Clip", r)
|
||
|
|
st = layer_state()
|
||
|
|
check("Cue 1 angewendet", st["name"] == "cue_clip.mp4"
|
||
|
|
and st["fx1"] is None and st["alpha"] == 1.0,
|
||
|
|
"name=%s fx1=%s" % (st["name"], st["fx1"]))
|
||
|
|
|
||
|
|
# ---- GO: Cue 2 mit Fade (Quelle wechselt -> dip-to-black) ----
|
||
|
|
r = post("/api/layers/cues/go", {"id": lid})
|
||
|
|
check("GO -> Cue 2", r.get("index") == 1
|
||
|
|
and r.get("cue") == "Ball mit Blur", r)
|
||
|
|
# waehrend des Fades: dip-to-black = Phase 1 (0-400ms) runter,
|
||
|
|
# Quelle wechselt, Phase 2 (400-800ms) hoch zum Ziel
|
||
|
|
time.sleep(0.2)
|
||
|
|
mid1 = layer_state()
|
||
|
|
check("Fade Phase 1 (abwaerts)", 0.2 < mid1["alpha"] < 0.95,
|
||
|
|
"alpha=%s" % mid1["alpha"])
|
||
|
|
time.sleep(0.4) # jetzt in Phase 2 (hoch zum Ziel 0.8)
|
||
|
|
mid2 = layer_state()
|
||
|
|
check("Fade Phase 2 (aufwaerts)", 0.05 < mid2["alpha"] < 0.79,
|
||
|
|
"alpha=%s" % mid2["alpha"])
|
||
|
|
time.sleep(0.6) # Fade 800ms komplett abwarten
|
||
|
|
st = layer_state()
|
||
|
|
check("Cue 2 angewendet (Generator+Blur, alpha 0.8)",
|
||
|
|
st["name"] == "ball" and st["fx1"] == "blur"
|
||
|
|
and abs(st["alpha"] - 0.8) < 0.06,
|
||
|
|
"name=%s fx1=%s alpha=%s" % (st["name"], st["fx1"], st["alpha"]))
|
||
|
|
check("Engine laeuft nach Fade", get("/api/status")["error"] is None)
|
||
|
|
|
||
|
|
# ---- Goto: direkt Cue 3 ----
|
||
|
|
r = post("/api/layers/cues/goto", {"id": lid, "index": 2})
|
||
|
|
check("Goto Cue 3", r.get("cue") == "Snow rechts", r)
|
||
|
|
st = layer_state()
|
||
|
|
check("Cue 3 angewendet", st["name"] == "snow" and st["x"] == 100,
|
||
|
|
"name=%s x=%s" % (st["name"], st["x"]))
|
||
|
|
|
||
|
|
# ---- GO vom letzten Cue: Wrap-around zu Cue 1 ----
|
||
|
|
r = post("/api/layers/cues/go", {"id": lid})
|
||
|
|
check("Wrap-around zu Cue 1", r.get("index") == 0, r)
|
||
|
|
|
||
|
|
# ---- GO zurueck (-1) ----
|
||
|
|
r = post("/api/layers/cues/go", {"id": lid, "steps": -1})
|
||
|
|
check("GO zurueck -> letzter Cue", r.get("index") == 2, r)
|
||
|
|
|
||
|
|
# ---- Cue entfernen: Indexkorrektur ----
|
||
|
|
r = post("/api/layers/cues/remove", {"id": lid, "cue_id": c1_id})
|
||
|
|
check("Cue entfernt", r.get("removed") is True, r)
|
||
|
|
cues = get("/api/layers/cues?id=%d" % lid)
|
||
|
|
check("Liste hat noch 2", len(cues["cues"]) == 2,
|
||
|
|
[c["name"] for c in cues["cues"]])
|
||
|
|
|
||
|
|
# ---- Auto-Follow: Cue mit follow_ms schaltet automatisch weiter ----
|
||
|
|
post("/api/layers/cues/goto", {"id": lid, "index": 0}) # Ball mit Blur
|
||
|
|
time.sleep(1.2) # Cue 0 hat 800ms Fade -> erst abwarten, dann pruefen
|
||
|
|
st = layer_state()
|
||
|
|
check("Vor Auto-Follow: Ball", st["name"] == "ball", st["name"])
|
||
|
|
# follow_ms fuer Cue 0 per Neu-Anlegen nicht moeglich -> pruefen ueber
|
||
|
|
# internen Zustand via Projektdatei reicht; wir nutzen cue 0 (800ms fade)
|
||
|
|
# und legen danach direkt Auto-Follow-Cue an: Snow mit follow 600ms
|
||
|
|
post("/api/layers/update", {"id": lid, "name": "snow"})
|
||
|
|
r = post("/api/layers/cues/add",
|
||
|
|
{"id": lid, "name": "Snow kurz", "fade_ms": 0,
|
||
|
|
"follow_ms": 600})
|
||
|
|
check("Auto-Follow-Cue angelegt", r.get("cue", {}).get("follow_ms")
|
||
|
|
== 600, r)
|
||
|
|
cues = get("/api/layers/cues?id=%d" % lid)
|
||
|
|
check("3 Cues fuer Follow-Test", len(cues["cues"]) == 3)
|
||
|
|
# Liste: [Ball mit Blur(0), Snow rechts(1), Snow kurz(2, follow 600ms)]
|
||
|
|
# Gehe zum LETZTEN Cue (Snow kurz) -> Follow schaltet auf Index 0 (Wrap)
|
||
|
|
post("/api/layers/cues/goto", {"id": lid, "index": 2})
|
||
|
|
st = layer_state()
|
||
|
|
check("Follow-Cue aktiv (Snow)", st["name"] == "snow")
|
||
|
|
time.sleep(1.2) # follow 600ms + Puffer
|
||
|
|
cues = get("/api/layers/cues?id=%d" % lid)
|
||
|
|
check("Auto-Follow geschaltet (Wrap zu 0)", cues["index"] == 0,
|
||
|
|
"index=%s" % cues["index"])
|
||
|
|
st = layer_state()
|
||
|
|
check("Nach Follow: Cue 0 aktiv (Ball, evtl. mitten im Fade)",
|
||
|
|
st["name"] == "ball" or st["name"] == "snow", st["name"])
|
||
|
|
|
||
|
|
# ---- DMX Cue-GO: Flanke schaltet weiter ----
|
||
|
|
r = post("/api/settings", {"artnet": {
|
||
|
|
"enabled": True, "listen_port": 6455, "universes": [0],
|
||
|
|
"master_channel": 10, "layer_start_channel": 2,
|
||
|
|
"playback_channel": 6, "retrigger_channel": 7,
|
||
|
|
"blackout_channel": 8, "cue_go_channel": 30,
|
||
|
|
"signal_loss_policy": "hold", "signal_loss_timeout_s": 10}})
|
||
|
|
check("cue_go_channel konfiguriert", bool(r.get("saved")), r)
|
||
|
|
time.sleep(2)
|
||
|
|
cues_before = get("/api/layers/cues?id=%d" % lid)["index"]
|
||
|
|
|
||
|
|
seq_counter = [40]
|
||
|
|
|
||
|
|
def send_dmx(val):
|
||
|
|
seq_counter[0] = (seq_counter[0] + 1) % 256 # Sequenz muss wachsen
|
||
|
|
dmx = bytearray(512)
|
||
|
|
dmx[29] = val # Kanal 30
|
||
|
|
pkt = (b"Art-Net\x00" + struct.pack("<H", 0x5000)
|
||
|
|
+ struct.pack(">H", 14) + bytes([seq_counter[0]]) + b"\x00"
|
||
|
|
+ struct.pack("<H", 0) + struct.pack(">H", 512)
|
||
|
|
+ bytes(dmx))
|
||
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||
|
|
s.sendto(pkt, ("127.0.0.1", 6455))
|
||
|
|
s.close()
|
||
|
|
|
||
|
|
send_dmx(255) # steigende Flanke -> GO
|
||
|
|
time.sleep(0.8)
|
||
|
|
cues_after = get("/api/layers/cues?id=%d" % lid)["index"]
|
||
|
|
check("DMX-Flanke: Cue weitergeschaltet",
|
||
|
|
cues_after != cues_before,
|
||
|
|
"%s -> %s" % (cues_before, cues_after))
|
||
|
|
send_dmx(255) # KEINE neue Flanke (bleibt high) -> kein GO
|
||
|
|
time.sleep(0.8)
|
||
|
|
cues_same = get("/api/layers/cues?id=%d" % lid)["index"]
|
||
|
|
check("Kein GO ohne Flanke", cues_same == cues_after,
|
||
|
|
"%s == %s" % (cues_same, cues_after))
|
||
|
|
send_dmx(0) # fallende Flanke
|
||
|
|
send_dmx(255) # neue steigende Flanke -> GO
|
||
|
|
time.sleep(0.8)
|
||
|
|
cues_next = get("/api/layers/cues?id=%d" % lid)["index"]
|
||
|
|
check("Neue Flanke: wieder GO", cues_next != cues_same,
|
||
|
|
"%s -> %s" % (cues_same, cues_next))
|
||
|
|
|
||
|
|
# ---- Cues in Projekt speichern + laden ----
|
||
|
|
r = post("/api/projects/save", {"name": "Cue Show"})
|
||
|
|
check("Projekt mit Cues gespeichert", r.get("saved") == "Cue Show", r)
|
||
|
|
post("/api/layers/remove", {"id": lid})
|
||
|
|
check("Layer entfernt", len(get("/api/status")["layers"]) == 0)
|
||
|
|
r = post("/api/projects/load", {"name": "Cue Show"})
|
||
|
|
check("Projekt geladen", r.get("loaded") is True, r)
|
||
|
|
cues = get("/api/layers/cues?id=%d" % get("/api/status")["layers"][0]["id"])
|
||
|
|
check("Cues nach Projekt-Load erhalten", len(cues["cues"]) == 3,
|
||
|
|
"%d Cues" % len(cues["cues"]))
|
||
|
|
|
||
|
|
# ---- Fehlerfaelle ----
|
||
|
|
r = post("/api/layers/cues/go", {"id": 999})
|
||
|
|
check("GO auf unbekanntem Layer 400", "error" in r, r)
|
||
|
|
r = post("/api/layers/cues/add",
|
||
|
|
{"id": get("/api/status")["layers"][0]["id"],
|
||
|
|
"state": {"name": "fehlt.mp4", "kind": "video"}})
|
||
|
|
check("Cue mit fehlendem Medium 400", "error" in r, r)
|
||
|
|
|
||
|
|
print("")
|
||
|
|
print("ALLE CUE-E2E-TESTS BESTANDEN")
|
||
|
|
finally:
|
||
|
|
stop_server()
|