#!/usr/bin/env python3
"""Shared control end to end, on REAL hardware: Blender's sender + the web UI
(cthuloid-web) + a proto-3 unit.
python3 blender-addon/tests/e2e_shared_control.py <unit-ip> <web-ui-url>
Uses the addon's real net.py/protocol.py (no Blender needed) and the web UI's
HTTP API. Values stay dim; the unit's colours are restored at the end. Leaves
the web UI connected (Connect switch on).
"""
import json
import os
import socket
import sys
import time
import urllib.request
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.join(HERE, "..", "cthuloid_lights"))
import net # noqa: E402
import protocol # noqa: E402
UNIT = sys.argv[1] if len(sys.argv) > 1 else "192.168.50.89"
WEB = (sys.argv[2] if len(sys.argv) > 2 else "http://stortron.local:4050").rstrip("/")
failures = []
def check(name, ok, detail=""):
print(("PASS " if ok else "FAIL ") + name + ("" if ok else " " + str(detail)))
if not ok:
failures.append(name)
def wait(pred, timeout=4.0):
end = time.monotonic() + timeout
while time.monotonic() < end:
if pred():
return True
time.sleep(0.05)
return pred()
def api(path, body=None):
req = urllib.request.Request(WEB + path, data=None if body is None else json.dumps(body).encode(),
headers={"content-type": "application/json", "x-cthuloid": "1"},
method="GET" if body is None else "POST")
try:
with urllib.request.urlopen(req, timeout=5) as r:
return r.status, json.loads(r.read() or b"null")
except urllib.error.HTTPError as e:
return e.code, json.loads(e.read() or b"null")
def unit_state():
st = api("/api/state")[1]
return next(u for u in st["units"] if u["addr"] == UNIT)
def lamp_values():
s = socket.create_connection((UNIT, 1337), timeout=4)
time.sleep(0.3)
s.sendall(b"state\n")
time.sleep(0.5)
data = s.recv(8192).decode()
s.close()
line = next(l for l in data.splitlines() if l.startswith("state "))
return [int(x) for x in line.split()[6:]]
# --- the web UI on, with this unit included ---------------------------------------
u = unit_state()
UID = u["id"]
api("/api/units/" + UID, {"enabled": True, "fadeMs": 0})
api("/api/connect", {"on": True})
check("web UI connected to the unit", wait(lambda: unit_state()["status"] == "connected", 8))
before = lamp_values()
print("lamp before:", before)
blender = net.Sender()
blender.min_interval = 0.0
blender.start()
HOLDER = blender.holder
def frame(vals):
return protocol.Frame(tuple(vals), 0)
# Connect (force): Blender takes every slot to set the scene -- here the lamp's
# own values, so nothing visibly changes -- then lets go when quiet.
blender.send(UNIT, 1337, frame(before), force=True)
check("Blender connected with proto 3", wait(lambda: blender.status_of(UNIT, 1337)[0] == "connected"))
time.sleep(2.0)
# 1. Blender changes slot 1 only.
scene = list(before)
scene[0:3] = [1500, 0, 0]
blender.send(UNIT, 1337, frame(scene))
check("web UI shows slot 1 held by Blender",
wait(lambda: (unit_state()["holds"][0] or {}).get("holder") == HOLDER), unit_state()["holds"])
check("lamp slot 1 = Blender's value", wait(lambda: lamp_values()[0:3] == [1500, 0, 0]))
check("Blender lets go after ~1.5 s quiet", wait(lambda: unit_state()["holds"][0] is None, 4))
# 2. The web UI grabs slot 2; Blender's change to slot 2 is refused and reported.
code, r = api("/api/units/%s/grab" % UID, {"index": 1, "session": "e2e", "name": "webtester"})
check("web grabs slot 2", code == 200 and r == {"granted": True}, (code, r))
scene[3:6] = [0, 1500, 0]
blender.send(UNIT, 1337, frame(scene))
check("Blender reports slot 2 held by webtester",
wait(lambda: "slot 2 held by webtester" in blender.status_of(UNIT, 1337)[1]), blender.status_of(UNIT, 1337))
check("Blender did not change slot 2", lamp_values()[3:6] == before[3:6])
# 3. The web user sets their colour and lets go; Blender must not stomp on it.
code, _ = api("/api/units/%s/slot" % UID, {"index": 1, "session": "e2e", "r": 0, "g": 0, "b": 0.02, "level": 1})
check("web sets slot 2", code == 200, code)
api("/api/units/%s/release" % UID, {"index": 1, "session": "e2e"})
check("web's release frees the slot", wait(lambda: unit_state()["holds"][1] is None))
want = [int(round(0.02 * 65535)), 0, 0]
check("lamp slot 2 = the web user's colour", wait(lambda: lamp_values()[3:6] == want), lamp_values()[3:6])
time.sleep(1.0)
check("Blender left it alone after the web user let go", lamp_values()[3:6] == want)
check("web UI stored the released value", abs(unit_state()["slots"][1]["level"] - 0.02) < 0.001,
unit_state()["slots"][1])
blender.stop()
# Restore the lamp (legacy pwm only touches unheld slots; red clamped to the cap).
time.sleep(0.5)
s = socket.create_connection((UNIT, 1337), timeout=4)
s.sendall(("pwm " + " ".join(str(min(v, 45000) if i % 3 == 2 else v) for i, v in enumerate(before)) + "\n").encode())
time.sleep(0.5)
s.close()
print("\n%d failure(s)" % len(failures) if failures else "\nALL PASS")
sys.exit(1 if failures else 0)