| … | … | @@ -69,11 +69,14 @@ def _keyframe_payload(scene, fixture, slot_map): |
| 69 | 69 | return SKIP |
| 70 | 70 | _last_pwm2_target[fixture.id] = target |
| 71 | 71 | |
| 72 | return protocol.pwm2_command(values, anim.frames_to_ms(scene, nxt - frame)) | |
| 72 | return protocol.Frame(target, anim.frames_to_ms(scene, nxt - frame)) | |
| 73 | 73 | |
| 74 | 74 | |
| 75 | 75 | def build_command(scene, fixture): |
| 76 | """Payload for one fixture, or SKIP to send nothing this tick.""" | |
| 76 | """A protocol.Frame for one fixture, or SKIP to send nothing this tick. | |
| 77 | ||
| 78 | The sender turns the Frame into whatever the unit speaks: a whole-unit | |
| 79 | pwm/pwm2 (proto <= 2) or per-slot grants and `slot` lines (proto 3).""" | |
| 77 | 80 | slot_map = rig.fixture_lights(scene, fixture) |
| 78 | 81 | |
| 79 | 82 | if fixture.fade_mode == 'KEYFRAME' and anim.is_playing(): |
| … | … | @@ -86,12 +89,12 @@ def build_command(scene, fixture): |
| 86 | 89 | slot_map, fixture.red_max_pwm, fixture.use_energy, fixture.energy_ref) |
| 87 | 90 | |
| 88 | 91 | if fixture.fade_mode == 'FIXED': |
| 89 | return protocol.pwm2_command(values, fixture.fade_ms) | |
| 92 | return protocol.Frame(tuple(values), fixture.fade_ms) | |
| 90 | 93 | |
| 91 | 94 | # Plain pwm also cancels any fade still running on the firmware, so drop the |
| 92 | 95 | # remembered target -- the next segment must be re-sent. |
| 93 | 96 | _last_pwm2_target.pop(fixture.id, None) |
| 94 | return protocol.pwm_command(values) | |
| 97 | return protocol.Frame(tuple(values), 0) | |
| 95 | 98 | |
| 96 | 99 | |
| 97 | 100 | def push_scene(scene, force=False): |
| … | … | @@ -17,12 +17,26 @@ Design notes: |
| 17 | 17 | * Deduplicated. Identical consecutive frames are not resent. |
| 18 | 18 | * Reconnect with backoff, so an unplugged fixture costs one attempt per |
| 19 | 19 | backoff window instead of one per frame. |
| 20 | ||
| 21 | Shared control (firmware proto 3, docs/DESIGN-collab.md): the unit referees | |
| 22 | per-slot grants. Against such a unit a Frame is sent per SLOT, only for slots | |
| 23 | whose values changed: grab the slot if we don't hold it, send `slot` lines, | |
| 24 | let go after RELEASE_AFTER of quiet (inside the unit's 2 s idle). A slot | |
| 25 | someone else holds is skipped and reported ("slot 2 held by alice"); when they | |
| 26 | let go, their colour stays until Blender's own scene changes -- the latest | |
| 27 | person to act wins, Blender included. Pressing Connect (force) pushes every | |
| 28 | slot; a silent reconnect pushes nothing, because the lamp is the truth. | |
| 20 | 29 | """ |
| 21 | 30 | |
| 22 | 31 | import socket |
| 23 | 32 | import threading |
| 24 | 33 | import time |
| 25 | 34 | |
| 35 | try: | |
| 36 | from . import protocol | |
| 37 | except ImportError: # loaded standalone by the tests | |
| 38 | import protocol | |
| 39 | ||
| 26 | 40 | CONNECT_TIMEOUT = 1.0 |
| 27 | 41 | SEND_TIMEOUT = 1.0 |
| 28 | 42 | BACKOFF_START = 1.0 |
| … | … | @@ -40,11 +54,18 @@ KEEPALIVE_INTERVAL = 60.0 |
| 40 | 54 | # would restart or cancel the fade. |
| 41 | 55 | KEEPALIVE_PAYLOAD = "hostname\n" |
| 42 | 56 | |
| 57 | # Let go of a slot after this long without a change -- inside the unit's 2 s | |
| 58 | # idle expiry, so a quiet Blender hands the slot back itself. | |
| 59 | RELEASE_AFTER = 1.5 | |
| 60 | GRANT_TIMEOUT = 1.0 | |
| 61 | ||
| 43 | 62 | |
| 44 | 63 | class _Target: |
| 45 | 64 | __slots__ = ("host", "port", "sock", "desired", "dirty", "force", |
| 46 | 65 | "last_sent", "last_tx_time", "next_retry", "backoff", |
| 47 | "status", "error", "addr", "wake", "thread", "closed", "once") | |
| 66 | "status", "error", "addr", "wake", "thread", "closed", "once", | |
| 67 | "rbuf", "proto", "epoch", "grants", "busy", "slot_sent", | |
| 68 | "slot_tx", "replies") | |
| 48 | 69 | |
| 49 | 70 | def __init__(self, host, port): |
| 50 | 71 | self.host = host |
| … | … | @@ -64,6 +85,18 @@ class _Target: |
| 64 | 85 | self.status = "idle" |
| 65 | 86 | self.error = "" |
| 66 | 87 | self.addr = None # resolved ip, filled in by the worker |
| 88 | self._reset_link() | |
| 89 | ||
| 90 | def _reset_link(self): | |
| 91 | """Per-connection state: none of it survives a reconnect.""" | |
| 92 | self.rbuf = b"" | |
| 93 | self.proto = 0 # from the unit's `hello`; 0 = proto <= 2 | |
| 94 | self.epoch = 0 | |
| 95 | self.grants = {} # slot -> grant number, ours | |
| 96 | self.busy = {} # slot -> holder, someone else's | |
| 97 | self.slot_sent = {} # slot -> (b, g, r) last sent (proto 3) | |
| 98 | self.slot_tx = {} # slot -> time of our last frame | |
| 99 | self.replies = [] # parsed replies to `grant`, oldest first | |
| 67 | 100 | |
| 68 | 101 | |
| 69 | 102 | class Sender: |
| … | … | @@ -82,6 +115,8 @@ class Sender: |
| 82 | 115 | self._lock = threading.Lock() |
| 83 | 116 | self._running = False |
| 84 | 117 | self.min_interval = 1.0 / 25.0 |
| 118 | # Shown by the unit to everyone while Blender holds a slot. | |
| 119 | self.holder = ("blender-" + socket.gethostname())[:32].replace(" ", "-") | |
| 85 | 120 | |
| 86 | 121 | # ---- main-thread API ------------------------------------------------- |
| 87 | 122 | |
| … | … | @@ -164,7 +199,10 @@ class Sender: |
| 164 | 199 | tgt = self._targets.get((host, port)) |
| 165 | 200 | if tgt is None: |
| 166 | 201 | return ("idle", "") |
| 167 | return (tgt.status, tgt.error) | |
| 202 | note = tgt.error | |
| 203 | if tgt.status == "connected" and tgt.busy: | |
| 204 | note = ", ".join("slot %d held by %s" % (s + 1, h) for s, h in sorted(tgt.busy.items())) | |
| 205 | return (tgt.status, note) | |
| 168 | 206 | |
| 169 | 207 | # ---- worker threads -------------------------------------------------- |
| 170 | 208 | |
| … | … | @@ -212,9 +250,17 @@ class Sender: |
| 212 | 250 | self._retire(tgt) # one attempt only; never a retry loop |
| 213 | 251 | return |
| 214 | 252 | # The fixture may have rebooted or been dropped mid-look; it cannot |
| 215 | # be assumed to still hold last_sent. Re-assert the desired state. | |
| 253 | # be assumed to still hold last_sent. Re-assert the desired state -- | |
| 254 | # except on a proto-3 unit after a silent reconnect: there the lamp | |
| 255 | # is the truth, and others may have changed it meanwhile. Pressing | |
| 256 | # Connect (force) still pushes everything. | |
| 216 | 257 | tgt.last_sent = None |
| 217 | 258 | tgt.dirty = True |
| 259 | if tgt.proto >= 3 and not tgt.force and isinstance(tgt.desired, protocol.Frame): | |
| 260 | tgt.dirty = False | |
| 261 | tgt.last_sent = tgt.desired | |
| 262 | for s in range(4): | |
| 263 | tgt.slot_sent[s] = tuple(tgt.desired.values[s * 3:s * 3 + 3]) | |
| 218 | 264 | |
| 219 | 265 | # (2) Read whatever the firmware sent back. This serves two purposes: |
| 220 | 266 | # it stops command acks accumulating until a full receive buffer stalls |
| … | … | @@ -235,7 +281,14 @@ class Sender: |
| 235 | 281 | if not tgt.force and now - tgt.last_tx_time < self.min_interval: |
| 236 | 282 | return # rate limited; desired is retained, retry next tick |
| 237 | 283 | payload = tgt.desired |
| 238 | if self._write(tgt, payload, now): | |
| 284 | if isinstance(payload, protocol.Frame) and tgt.proto >= 3: | |
| 285 | if self._push_slots(tgt, payload, now, tgt.force): | |
| 286 | tgt.last_sent = payload | |
| 287 | tgt.dirty = False | |
| 288 | tgt.force = False | |
| 289 | return | |
| 290 | if self._write(tgt, protocol.frame_command(payload) if isinstance(payload, protocol.Frame) | |
| 291 | else payload, now): | |
| 239 | 292 | tgt.last_sent = payload |
| 240 | 293 | tgt.dirty = False |
| 241 | 294 | tgt.force = False |
| … | … | @@ -245,7 +298,14 @@ class Sender: |
| 245 | 298 | self._retire(tgt) |
| 246 | 299 | return |
| 247 | 300 | |
| 248 | # (4) Keepalive, so the firmware's idle timer never fires on a look we | |
| 301 | # (4) Proto 3: hand back slots we've stopped changing. | |
| 302 | for s in list(tgt.grants): | |
| 303 | if now - tgt.slot_tx.get(s, 0.0) >= RELEASE_AFTER: | |
| 304 | tgt.grants.pop(s, None) | |
| 305 | if not self._write(tgt, protocol.release_command(s), now): | |
| 306 | return | |
| 307 | ||
| 308 | # (5) Keepalive, so the firmware's idle timer never fires on a look we | |
| 249 | 309 | # are deliberately not resending. |
| 250 | 310 | if now - tgt.last_tx_time >= KEEPALIVE_INTERVAL: |
| 251 | 311 | self._write(tgt, KEEPALIVE_PAYLOAD, now) # never becomes last_sent |
| … | … | @@ -270,16 +330,68 @@ class Sender: |
| 270 | 330 | tgt.backoff = min(tgt.backoff * 2.0, BACKOFF_MAX) |
| 271 | 331 | return False |
| 272 | 332 | |
| 273 | @staticmethod | |
| 274 | def _drain(tgt): | |
| 275 | """Consume pending acks. False means the peer has closed the socket.""" | |
| 333 | def _push_slots(self, tgt, frame, now, force): | |
| 334 | """Proto 3: send each changed slot under its grant. False if the link failed.""" | |
| 335 | out = [] | |
| 336 | for s in range(4): | |
| 337 | bgr = tuple(frame.values[s * 3:s * 3 + 3]) | |
| 338 | if not force and tgt.slot_sent.get(s) == bgr: | |
| 339 | continue | |
| 340 | # Remembered even when we can't send it: a slot someone else holds | |
| 341 | # is not retried until Blender's own scene changes again. | |
| 342 | tgt.slot_sent[s] = bgr | |
| 343 | if s not in tgt.grants: | |
| 344 | got = self._grant(tgt, s, now) | |
| 345 | if got is None: | |
| 346 | return False | |
| 347 | if not got: | |
| 348 | continue | |
| 349 | out.append(protocol.slot_command(s, tgt.epoch, tgt.grants[s], bgr, frame.fade_ms)) | |
| 350 | tgt.slot_tx[s] = now | |
| 351 | if out: | |
| 352 | return self._write(tgt, "".join(out), now) | |
| 353 | return True | |
| 354 | ||
| 355 | def _grant(self, tgt, slot, now): | |
| 356 | """Ask the unit for a slot. True granted, False refused, None link failed.""" | |
| 357 | if not self._write(tgt, protocol.grant_command(slot, self.holder), now): | |
| 358 | return None | |
| 359 | deadline = time.monotonic() + GRANT_TIMEOUT | |
| 360 | while time.monotonic() < deadline: | |
| 361 | for i, r in enumerate(tgt.replies): | |
| 362 | if r[1] != slot and r[0] != "err": | |
| 363 | continue | |
| 364 | del tgt.replies[i] | |
| 365 | if r[0] == "granted": | |
| 366 | tgt.epoch, tgt.grants[slot] = r[2], r[3] | |
| 367 | tgt.busy.pop(slot, None) | |
| 368 | return True | |
| 369 | if r[0] == "busy": | |
| 370 | tgt.busy[slot] = r[2] | |
| 371 | return False | |
| 372 | if not self._read(tgt, timeout=0.05): | |
| 373 | return None | |
| 374 | return False | |
| 375 | ||
| 376 | def _read(self, tgt, timeout=0.0): | |
| 377 | """Read and handle whatever the unit sent. False means the peer closed.""" | |
| 276 | 378 | alive = True |
| 277 | 379 | try: |
| 278 | tgt.sock.setblocking(False) | |
| 380 | if timeout: | |
| 381 | tgt.sock.settimeout(timeout) | |
| 382 | else: | |
| 383 | tgt.sock.setblocking(False) | |
| 279 | 384 | while True: |
| 280 | if not tgt.sock.recv(4096): | |
| 385 | data = tgt.sock.recv(4096) | |
| 386 | if not data: | |
| 281 | 387 | alive = False # orderly close from the firmware |
| 282 | 388 | break |
| 389 | tgt.rbuf += data | |
| 390 | while b"\n" in tgt.rbuf: | |
| 391 | line, tgt.rbuf = tgt.rbuf.split(b"\n", 1) | |
| 392 | self._handle_line(tgt, line.decode("utf-8", "replace").strip()) | |
| 393 | if timeout: | |
| 394 | break # one read is enough when waiting for a reply | |
| 283 | 395 | except (BlockingIOError, socket.timeout): |
| 284 | 396 | pass # nothing more to read; still connected |
| 285 | 397 | except OSError: |
| … | … | @@ -290,8 +402,45 @@ class Sender: |
| 290 | 402 | tgt.sock.settimeout(SEND_TIMEOUT) |
| 291 | 403 | except OSError: |
| 292 | 404 | pass |
| 405 | if len(tgt.rbuf) > 4096: | |
| 406 | tgt.rbuf = b"" | |
| 293 | 407 | return alive |
| 294 | 408 | |
| 409 | def _drain(self, tgt): | |
| 410 | """Consume pending lines. False means the peer has closed the socket.""" | |
| 411 | return self._read(tgt) | |
| 412 | ||
| 413 | def _handle_line(self, tgt, line): | |
| 414 | p = line.split() | |
| 415 | if not p: | |
| 416 | return | |
| 417 | try: | |
| 418 | if p[0] == "hello": | |
| 419 | kv = dict(x.split("=", 1) for x in p[1:] if "=" in x) | |
| 420 | tgt.proto = int(kv.get("proto", "0")) | |
| 421 | tgt.epoch = int(kv.get("epoch", "0")) | |
| 422 | elif p[0] == "granted" and len(p) == 4: | |
| 423 | tgt.replies.append(("granted", int(p[1]), int(p[2]), int(p[3]))) | |
| 424 | elif p[0] == "busy" and len(p) == 3: | |
| 425 | tgt.replies.append(("busy", int(p[1]), p[2])) | |
| 426 | elif p[0] == "ERR" and len(p) >= 2 and p[1] == "grant": | |
| 427 | tgt.replies.append(("err", None, line)) | |
| 428 | elif p[0] == "held" and len(p) == 5: | |
| 429 | s, n = int(p[1]), int(p[4]) | |
| 430 | if tgt.grants.get(s) != n: | |
| 431 | tgt.busy[s] = p[2] | |
| 432 | elif p[0] == "free" and len(p) >= 4: | |
| 433 | s, n = int(p[1]), int(p[3]) | |
| 434 | tgt.busy.pop(s, None) | |
| 435 | if tgt.grants.get(s) == n: | |
| 436 | tgt.grants.pop(s, None) # timed out / blackout: grab again next time | |
| 437 | elif p[0] == "ERR" and len(p) == 3 and p[1] == "notheld": | |
| 438 | s = int(p[2]) | |
| 439 | tgt.grants.pop(s, None) | |
| 440 | tgt.slot_sent.pop(s, None) # resend once we hold it again | |
| 441 | except ValueError: | |
| 442 | pass | |
| 443 | ||
| 295 | 444 | def _duplicate_of(self, tgt): |
| 296 | 445 | """Another target already holding a live socket to the same device. |
| 297 | 446 | |
| … | … | @@ -347,18 +496,14 @@ class Sender: |
| 347 | 496 | tgt.addr = sock.getpeername()[0] |
| 348 | 497 | except OSError: |
| 349 | 498 | pass |
| 350 | # The firmware greets with "<hostname> ready"; drain it so it does | |
| 351 | # not sit in the buffer. Non-fatal if nothing arrives. | |
| 352 | try: | |
| 353 | sock.settimeout(0.2) | |
| 354 | sock.recv(256) | |
| 355 | except (socket.timeout, OSError): | |
| 356 | pass | |
| 357 | finally: | |
| 358 | try: | |
| 359 | sock.settimeout(SEND_TIMEOUT) | |
| 360 | except OSError: | |
| 361 | pass | |
| 499 | # The firmware greets with "<hostname> ready", and proto 3 follows | |
| 500 | # with `hello proto=3 ...`. Read both before the first send, so the | |
| 501 | # first frame already goes out in the right protocol. | |
| 502 | tgt._reset_link() | |
| 503 | deadline = time.monotonic() + 0.4 | |
| 504 | while tgt.proto == 0 and time.monotonic() < deadline: | |
| 505 | if not self._read(tgt, timeout=0.1): | |
| 506 | break | |
| 362 | 507 | return True |
| 363 | 508 | except (socket.timeout, OSError) as exc: |
| 364 | 509 | tgt.status = "unreachable" |
| … | … | @@ -379,6 +524,8 @@ class Sender: |
| 379 | 524 | # Forget what the wire holds, but keep `desired` -- that is what gets |
| 380 | 525 | # re-asserted once the fixture comes back. |
| 381 | 526 | tgt.last_sent = None |
| 527 | tgt.grants.clear() # the unit freed them when the socket went | |
| 528 | tgt.busy.clear() | |
| 382 | 529 | |
| 383 | 530 | |
| 384 | 531 | SENDER = Sender() |
| … | … | @@ -14,6 +14,8 @@ are padded with `0 0 0`; with INVERT_HIGH_AND_LOW in the firmware, input 0 is |
| 14 | 14 | LED-off, so padding is dark rather than full-on. |
| 15 | 15 | """ |
| 16 | 16 | |
| 17 | from collections import namedtuple | |
| 18 | ||
| 17 | 19 | CHANNELS_PER_SLOT = 3 |
| 18 | 20 | SLOTS_PER_FIXTURE = 4 |
| 19 | 21 | NUM_CHANNELS = CHANNELS_PER_SLOT * SLOTS_PER_FIXTURE |
| … | … | @@ -117,3 +119,34 @@ def freq_command(hz): |
| 117 | 119 | |
| 118 | 120 | def blackout_command(): |
| 119 | 121 | return "stopfade\n" + pwm_command([0] * NUM_CHANNELS) |
| 122 | ||
| 123 | ||
| 124 | # --------------------------------------------------------------------------- | |
| 125 | # What the engine hands the sender: the fixture's 12 values and how long to | |
| 126 | # fade to them (0 = instantly). The sender turns it into whatever the unit | |
| 127 | # speaks -- a whole-unit pwm/pwm2 for proto <= 2, per-slot grants and `slot` | |
| 128 | # lines for proto 3 (docs/DESIGN-collab.md §5). Comparable, so identical | |
| 129 | # consecutive frames are skipped. | |
| 130 | Frame = namedtuple("Frame", "values fade_ms") | |
| 131 | ||
| 132 | ||
| 133 | def frame_command(frame): | |
| 134 | """Proto <= 2: one whole-unit command.""" | |
| 135 | values = list(frame.values) | |
| 136 | if frame.fade_ms > 0: | |
| 137 | return pwm2_command(values, frame.fade_ms) | |
| 138 | return pwm_command(values) | |
| 139 | ||
| 140 | ||
| 141 | # ---- proto 3: shared control ---------------------------------------------- | |
| 142 | ||
| 143 | def grant_command(slot, holder): | |
| 144 | return "grant %s %s\n" % (slot, holder) | |
| 145 | ||
| 146 | ||
| 147 | def slot_command(slot, epoch, n, bgr, fade_ms): | |
| 148 | return "slot %d %d %d %d %d %d %d\n" % (slot, epoch, n, bgr[0], bgr[1], bgr[2], int(fade_ms)) | |
| 149 | ||
| 150 | ||
| 151 | def release_command(slot): | |
| 152 | return "release %s\n" % slot |
| … | … | @@ -24,7 +24,10 @@ def fixture_state(scene, fixture): |
| 24 | 24 | return "off", 'BLANK1' |
| 25 | 25 | if not scene.cthuloid.live: |
| 26 | 26 | return "not connected", 'RADIOBUT_OFF' |
| 27 | status, _err = SENDER.status_of(fixture.host, fixture.port) | |
| 27 | status, note = SENDER.status_of(fixture.host, fixture.port) | |
| 28 | if status == "connected" and "held by" in note: | |
| 29 | # Shared control: someone else holds a slot ("slot 2 held by alice"). | |
| 30 | return note, 'LOCKED' | |
| 28 | 31 | return { |
| 29 | 32 | "connected": ("connected", 'RADIOBUT_ON'), |
| 30 | 33 | "connecting": ("connecting...", 'SORTTIME'), |
| … | … | @@ -0,0 +1,150 @@ |
| 1 | """A proto-3 light unit for tests: referees per-slot grants like the firmware's | |
| 2 | grants.h (first come first served, per socket). Signatures are a dummy.""" | |
| 3 | ||
| 4 | import socket | |
| 5 | import threading | |
| 6 | import time | |
| 7 | ||
| 8 | ||
| 9 | class FakeUnit3: | |
| 10 | def __init__(self, name="fake3", epoch=5, proto=3): | |
| 11 | self.name = name | |
| 12 | self.epoch = epoch | |
| 13 | self.proto = proto | |
| 14 | self.counter = 0 | |
| 15 | self.values = [0] * 12 | |
| 16 | self.holds = [None] * 4 # (conn_id, holder, n) | |
| 17 | self.lines = [] # (conn_id, line) | |
| 18 | self.conns = {} | |
| 19 | self.lock = threading.Lock() | |
| 20 | self.srv = socket.socket() | |
| 21 | self.srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) | |
| 22 | self.srv.bind(("127.0.0.1", 0)) | |
| 23 | self.srv.listen(8) | |
| 24 | self.port = self.srv.getsockname()[1] | |
| 25 | self.next_id = 1 | |
| 26 | threading.Thread(target=self._accept, daemon=True).start() | |
| 27 | ||
| 28 | def received(self, prefix): | |
| 29 | with self.lock: | |
| 30 | return [l for _, l in self.lines if l.startswith(prefix)] | |
| 31 | ||
| 32 | def _bcast(self, line): | |
| 33 | for c in list(self.conns.values()): | |
| 34 | try: | |
| 35 | c.sendall((line + "\r\n").encode()) | |
| 36 | except OSError: | |
| 37 | pass | |
| 38 | ||
| 39 | def _free(self, s, why): | |
| 40 | conn, holder, n = self.holds[s] | |
| 41 | self.holds[s] = None | |
| 42 | v = self.values[s * 3:s * 3 + 3] | |
| 43 | self._bcast("free %d %d %d %d %d %d %s %s %s" % (s, self.epoch, n, v[0], v[1], v[2], why, holder, "ab" * 64)) | |
| 44 | ||
| 45 | def _set(self, start, vals, skip_held): | |
| 46 | changed = False | |
| 47 | for i, v in enumerate(vals): | |
| 48 | c = start + i | |
| 49 | if c >= 12 or (skip_held and self.holds[c // 3]): | |
| 50 | continue | |
| 51 | if self.values[c] != v: | |
| 52 | self.values[c] = v | |
| 53 | changed = True | |
| 54 | if changed: | |
| 55 | self._bcast("values " + " ".join(map(str, self.values))) | |
| 56 | ||
| 57 | def _accept(self): | |
| 58 | while True: | |
| 59 | try: | |
| 60 | c, _ = self.srv.accept() | |
| 61 | except OSError: | |
| 62 | return | |
| 63 | with self.lock: | |
| 64 | cid = self.next_id | |
| 65 | self.next_id += 1 | |
| 66 | self.conns[cid] = c | |
| 67 | c.sendall(("%s ready\r\n" % self.name).encode()) | |
| 68 | if self.proto >= 3: | |
| 69 | c.sendall(("hello proto=3 unit=%s epoch=%d pk=%s\r\n" % (self.name, self.epoch, "cd" * 32)).encode()) | |
| 70 | threading.Thread(target=self._read, args=(cid, c), daemon=True).start() | |
| 71 | ||
| 72 | def _read(self, cid, c): | |
| 73 | buf = b"" | |
| 74 | while True: | |
| 75 | try: | |
| 76 | d = c.recv(4096) | |
| 77 | except OSError: | |
| 78 | d = b"" | |
| 79 | if not d: | |
| 80 | with self.lock: | |
| 81 | self.conns.pop(cid, None) | |
| 82 | for s in range(4): | |
| 83 | if self.holds[s] and self.holds[s][0] == cid: | |
| 84 | self._free(s, "disconnect") | |
| 85 | return | |
| 86 | buf += d | |
| 87 | while b"\n" in buf: | |
| 88 | line, buf = buf.split(b"\n", 1) | |
| 89 | line = line.decode().strip() | |
| 90 | if line: | |
| 91 | with self.lock: | |
| 92 | self._command(cid, c, line) | |
| 93 | ||
| 94 | def _command(self, cid, c, line): | |
| 95 | self.lines.append((cid, line)) | |
| 96 | p = line.split() | |
| 97 | rng = lambda a: [0, 1, 2, 3] if a == "all" else [int(a)] | |
| 98 | send = lambda s: c.sendall((s + "\r\n").encode()) | |
| 99 | if p[0] == "grant" and self.proto >= 3: | |
| 100 | slots = rng(p[1]) | |
| 101 | for s in slots: | |
| 102 | if self.holds[s] and self.holds[s][0] != cid: | |
| 103 | return send("busy %d %s" % (s, self.holds[s][1])) | |
| 104 | for s in slots: | |
| 105 | if not self.holds[s]: | |
| 106 | self.counter += 1 | |
| 107 | self.holds[s] = (cid, p[2], self.counter) | |
| 108 | self._bcast("held %d %s %d %d" % (s, p[2], self.epoch, self.counter)) | |
| 109 | send("granted %d %d %d" % (s, self.epoch, self.holds[s][2])) | |
| 110 | elif p[0] == "slot" and self.proto >= 3: | |
| 111 | s, e, n, b, g, r = map(int, p[1:7]) | |
| 112 | h = self.holds[s] | |
| 113 | if not h or h[0] != cid or e != self.epoch or n != h[2]: | |
| 114 | return send("ERR notheld %d" % s) | |
| 115 | self._set(s * 3, [b, g, r], False) | |
| 116 | elif p[0] == "release": | |
| 117 | for s in rng(p[1]): | |
| 118 | if self.holds[s] and self.holds[s][0] == cid: | |
| 119 | self._free(s, "release") | |
| 120 | elif p[0] == "pwm": | |
| 121 | self._set(0, list(map(int, p[1:])), True) | |
| 122 | elif p[0] == "pwm2": | |
| 123 | self._set(0, list(map(int, p[3:])), True) | |
| 124 | elif p[0] == "stopfade": | |
| 125 | for s in range(4): | |
| 126 | if self.holds[s]: | |
| 127 | self._free(s, "blackout") | |
| 128 | ||
| 129 | def external(self): | |
| 130 | """Another controller's socket (e.g. the web UI).""" | |
| 131 | s = socket.create_connection(("127.0.0.1", self.port)) | |
| 132 | time.sleep(0.1) | |
| 133 | return s | |
| 134 | ||
| 135 | def close(self): | |
| 136 | self.srv.close() | |
| 137 | for c in list(self.conns.values()): | |
| 138 | try: | |
| 139 | c.close() | |
| 140 | except OSError: | |
| 141 | pass | |
| 142 | ||
| 143 | ||
| 144 | def wait(pred, timeout=3.0): | |
| 145 | end = time.monotonic() + timeout | |
| 146 | while time.monotonic() < end: | |
| 147 | if pred(): | |
| 148 | return True | |
| 149 | time.sleep(0.02) | |
| 150 | return pred() |
| … | … | @@ -17,6 +17,7 @@ import time |
| 17 | 17 | |
| 18 | 18 | HERE = os.path.dirname(os.path.abspath(__file__)) |
| 19 | 19 | NET_PY = os.path.join(HERE, "..", "cthuloid_lights", "net.py") |
| 20 | sys.path.insert(0, os.path.dirname(NET_PY)) # net.py imports protocol | |
| 20 | 21 | |
| 21 | 22 | spec = importlib.util.spec_from_file_location("cthuloid_net", NET_PY) |
| 22 | 23 | net = importlib.util.module_from_spec(spec) |
| … | … | @@ -0,0 +1,144 @@ |
| 1 | """The addon's sender against proto-3 units (shared control). | |
| 2 | ||
| 3 | Run: python3 blender-addon/tests/test_net_v3.py | |
| 4 | ||
| 5 | net.py and protocol.py import nothing from Blender, so they are exercised | |
| 6 | directly against a fake unit that referees grants like the firmware. | |
| 7 | """ | |
| 8 | ||
| 9 | import importlib.util | |
| 10 | import os | |
| 11 | import sys | |
| 12 | import time | |
| 13 | ||
| 14 | HERE = os.path.dirname(os.path.abspath(__file__)) | |
| 15 | sys.path.insert(0, HERE) | |
| 16 | from fake_unit3 import FakeUnit3, wait # noqa: E402 | |
| 17 | ||
| 18 | PKG = os.path.join(HERE, "..", "cthuloid_lights") | |
| 19 | ||
| 20 | ||
| 21 | sys.path.insert(0, PKG) # net.py falls back to a top-level `import protocol` | |
| 22 | import protocol # noqa: E402 | |
| 23 | import net # noqa: E402 | |
| 24 | ||
| 25 | failures = [] | |
| 26 | ||
| 27 | ||
| 28 | def check(name, ok, detail=""): | |
| 29 | print(("PASS " if ok else "FAIL ") + name + ("" if ok else " " + str(detail))) | |
| 30 | if not ok: | |
| 31 | failures.append(name) | |
| 32 | ||
| 33 | ||
| 34 | def frame(slots, fade=0): | |
| 35 | """slots: list of 4 (b, g, r) tuples.""" | |
| 36 | vals = [] | |
| 37 | for s in slots: | |
| 38 | vals.extend(s) | |
| 39 | return protocol.Frame(tuple(vals), fade) | |
| 40 | ||
| 41 | ||
| 42 | DARK = [(0, 0, 0)] * 4 | |
| 43 | ||
| 44 | ||
| 45 | def sender(): | |
| 46 | s = net.Sender() | |
| 47 | s.min_interval = 0.0 | |
| 48 | s.holder = "blender-test" | |
| 49 | s.start() | |
| 50 | return s | |
| 51 | ||
| 52 | ||
| 53 | def test_grants_and_slot_frames(): | |
| 54 | fake = FakeUnit3() | |
| 55 | s = sender() | |
| 56 | s.send("127.0.0.1", fake.port, frame([(1, 2, 3), (4, 5, 6), (0, 0, 0), (0, 0, 0)]), force=True) | |
| 57 | check("connect + force: every slot grabbed and set", | |
| 58 | wait(lambda: fake.values[:6] == [1, 2, 3, 4, 5, 6] and len(fake.received("grant ")) == 4)) | |
| 59 | check("frames carry the grant", all(l.split()[2] == "5" for l in fake.received("slot "))) | |
| 60 | n_before = len(fake.received("slot ")) | |
| 61 | s.send("127.0.0.1", fake.port, frame([(9, 9, 9), (4, 5, 6), (0, 0, 0), (0, 0, 0)])) | |
| 62 | check("a later change sends only the slot that changed", | |
| 63 | wait(lambda: fake.values[:3] == [9, 9, 9]) and | |
| 64 | [l.split()[1] for l in fake.received("slot ")[n_before:]] == ["0"], | |
| 65 | fake.received("slot ")[n_before:]) | |
| 66 | check("no whole-unit pwm on a proto-3 unit", not fake.received("pwm")) | |
| 67 | s.stop() | |
| 68 | fake.close() | |
| 69 | ||
| 70 | ||
| 71 | def test_release_after_quiet(): | |
| 72 | fake = FakeUnit3() | |
| 73 | s = sender() | |
| 74 | s.send("127.0.0.1", fake.port, frame([(7, 7, 7)] + DARK[1:]), force=True) | |
| 75 | wait(lambda: fake.values[:3] == [7, 7, 7]) | |
| 76 | t0 = time.monotonic() | |
| 77 | ok = wait(lambda: len(fake.received("release ")) >= 4, timeout=4) | |
| 78 | dt = time.monotonic() - t0 | |
| 79 | check("held slots are let go after ~1.5 s without change (inside the unit's 2 s idle)", | |
| 80 | ok and 1.2 < dt < 2.2, "dt=%.2f %s" % (dt, fake.received("release "))) | |
| 81 | s.stop() | |
| 82 | fake.close() | |
| 83 | ||
| 84 | ||
| 85 | def test_busy_slot_is_skipped_and_reported(): | |
| 86 | fake = FakeUnit3() | |
| 87 | web = fake.external() | |
| 88 | web.sendall(b"grant 1 alice\n") | |
| 89 | wait(lambda: fake.holds[1] is not None) | |
| 90 | s = sender() | |
| 91 | s.send("127.0.0.1", fake.port, frame([(1, 1, 1), (2, 2, 2), (3, 3, 3), (4, 4, 4)]), force=True) | |
| 92 | check("other slots still set", wait(lambda: fake.values[6:9] == [3, 3, 3])) | |
| 93 | check("alice's slot untouched", fake.values[3:6] == [0, 0, 0]) | |
| 94 | status, note = s.status_of("127.0.0.1", fake.port) | |
| 95 | check("fixture reports who holds it", status == "connected" and "held by alice" in note, (status, note)) | |
| 96 | # Alice lets go after setting her own colour: Blender must NOT stomp on it | |
| 97 | # until Blender's own scene changes. | |
| 98 | web.sendall(b"slot 1 5 1 50 60 70 0\nrelease 1\n") | |
| 99 | wait(lambda: fake.holds[1] is None) | |
| 100 | time.sleep(0.6) | |
| 101 | check("after alice lets go, her colour stays (Blender only sends on its own change)", | |
| 102 | fake.values[3:6] == [50, 60, 70], fake.values[3:6]) | |
| 103 | s.send("127.0.0.1", fake.port, frame([(1, 1, 1), (8, 8, 8), (3, 3, 3), (4, 4, 4)])) | |
| 104 | check("Blender's next change to that slot applies", wait(lambda: fake.values[3:6] == [8, 8, 8])) | |
| 105 | check("note cleared once nothing is held by others", "held by" not in s.status_of("127.0.0.1", fake.port)[1]) | |
| 106 | web.close() | |
| 107 | s.stop() | |
| 108 | fake.close() | |
| 109 | ||
| 110 | ||
| 111 | def test_proto2_unchanged(): | |
| 112 | fake = FakeUnit3(proto=2) | |
| 113 | s = sender() | |
| 114 | s.send("127.0.0.1", fake.port, frame([(1, 2, 3)] + DARK[1:]), force=True) | |
| 115 | check("proto-2 unit gets a whole-unit pwm, no grants", | |
| 116 | wait(lambda: fake.values[:3] == [1, 2, 3]) and not fake.received("grant") and fake.received("pwm 1 2 3")) | |
| 117 | s.send("127.0.0.1", fake.port, frame([(4, 5, 6)] + DARK[1:], fade=500)) | |
| 118 | check("fade becomes pwm2", wait(lambda: any(l.startswith("pwm2 1 500 4 5 6") for _, l in fake.lines))) | |
| 119 | s.stop() | |
| 120 | fake.close() | |
| 121 | ||
| 122 | ||
| 123 | def test_blackout_frees_and_regrabs(): | |
| 124 | fake = FakeUnit3() | |
| 125 | s = sender() | |
| 126 | s.send("127.0.0.1", fake.port, frame([(5, 5, 5)] + DARK[1:]), force=True) | |
| 127 | wait(lambda: fake.values[:3] == [5, 5, 5]) | |
| 128 | s.send("127.0.0.1", fake.port, protocol.blackout_command(), force=True) | |
| 129 | check("blackout reaches the unit", wait(lambda: fake.values[:3] == [0, 0, 0] and fake.holds == [None] * 4)) | |
| 130 | grants = len(fake.received("grant ")) | |
| 131 | s.send("127.0.0.1", fake.port, frame([(6, 6, 6)] + DARK[1:])) | |
| 132 | check("next change grabs again and applies", | |
| 133 | wait(lambda: fake.values[:3] == [6, 6, 6]) and len(fake.received("grant ")) > grants and | |
| 134 | not fake.received("ERR"), fake.received("ERR")) | |
| 135 | s.stop() | |
| 136 | fake.close() | |
| 137 | ||
| 138 | ||
| 139 | for t in (test_grants_and_slot_frames, test_release_after_quiet, test_busy_slot_is_skipped_and_reported, | |
| 140 | test_proto2_unchanged, test_blackout_frees_and_regrabs): | |
| 141 | t() | |
| 142 | ||
| 143 | print("\n%d failure(s)" % len(failures) if failures else "\nALL PASS") | |
| 144 | sys.exit(1 if failures else 0) |