"""Non-blocking sender for ESP32 light fixtures.
Each fixture has its own worker thread, which owns that fixture's socket, so an
unreachable unit only ever delays itself. Blender's main thread only ever drops the
latest desired state into a per-host mailbox and returns immediately -- it never
calls connect() or sendall(). The original quadpanel.py did its sendall() inline
in the frame_change handler, so a fixture that had gone away stalled the UI for
the socket timeout on every single frame.
Design notes:
* Coalescing, not queueing. A mailbox holds ONE pending frame per host. If the
UI produces frames faster than the wire drains, intermediate frames are
dropped rather than backing up -- for lighting, the newest state is the only
one that matters, and a backlog would play back visibly late.
* Rate limited. The original's comment ("sending commands every 30 times a
second overwhelms the ESP32") is the reason min_interval exists.
* Deduplicated. Identical consecutive frames are not resent.
* Reconnect with backoff, so an unplugged fixture costs one attempt per
backoff window instead of one per frame.
Shared control (firmware proto 3, docs/DESIGN-collab.md): the unit referees
per-slot grants. Against such a unit a Frame is sent per SLOT, only for slots
whose values changed: grab the slot if we don't hold it, send `slot` lines,
let go after RELEASE_AFTER of quiet (inside the unit's 2 s idle). A slot
someone else holds is skipped and reported ("slot 2 held by alice"); when they
let go, their colour stays until Blender's own scene changes -- the latest
person to act wins, Blender included. Pressing Connect (force) pushes every
slot; a silent reconnect pushes nothing, because the lamp is the truth.
"""
import socket
import threading
import time
try:
from . import protocol
except ImportError: # loaded standalone by the tests
import protocol
CONNECT_TIMEOUT = 1.0
SEND_TIMEOUT = 1.0
BACKOFF_START = 1.0
BACKOFF_MAX = 15.0
# The firmware drops a client that has sent nothing for CLIENT_IDLE_TIMEOUT_MS
# (300 s, esp324xlight.ino:95). Keyframe fade mode deliberately stays silent
# between keyframes, so a static look WILL cross that threshold and be dropped.
# Poking the connection well inside the window prevents it.
KEEPALIVE_INTERVAL = 60.0
# "hostname" with no argument only prints the current hostname (cmdHostname's
# else branch) -- it is the one command that produces traffic without touching
# pin state, so it cannot disturb a pwm2 fade in flight. A pwm/pwm2 keepalive
# would restart or cancel the fade.
KEEPALIVE_PAYLOAD = "hostname\n"
# Let go of a slot after this long without a change -- inside the unit's 2 s
# idle expiry, so a quiet Blender hands the slot back itself.
RELEASE_AFTER = 1.5
GRANT_TIMEOUT = 1.0
class _Target:
__slots__ = ("host", "port", "sock", "desired", "dirty", "force",
"last_sent", "last_tx_time", "next_retry", "backoff",
"status", "error", "addr", "wake", "thread", "closed", "once",
"rbuf", "proto", "epoch", "grants", "busy", "slot_sent",
"slot_tx", "replies")
def __init__(self, host, port):
self.host = host
self.port = port
self.wake = threading.Event()
self.thread = None
self.closed = False # set by forget()/stop(); the worker then exits
self.once = False # deliver the pending payload, then let go
self.sock = None
self.desired = None # latest state the fixture SHOULD hold
self.dirty = False # desired differs from what reached the wire
self.force = False # bypass dedup and rate limit once
self.last_sent = None # what actually reached the wire
self.last_tx_time = 0.0
self.next_retry = 0.0
self.backoff = BACKOFF_START
self.status = "idle"
self.error = ""
self.addr = None # resolved ip, filled in by the worker
self._reset_link()
def _reset_link(self):
"""Per-connection state: none of it survives a reconnect."""
self.rbuf = b""
self.proto = 0 # from the unit's `hello`; 0 = proto <= 2
self.epoch = 0
self.grants = {} # slot -> grant number, ours
self.busy = {} # slot -> holder, someone else's
self.slot_sent = {} # slot -> (b, g, r) last sent (proto 3)
self.slot_tx = {} # slot -> time of our last frame
self.replies = [] # parsed replies to `grant`, oldest first
class Sender:
"""One worker thread PER fixture.
A single shared worker used to service every fixture in turn, so one
powered-off unit stalled all the others: create_connection() resolves the
host first, and an unanswered .local lookup blocks getaddrinfo for ~5 s
(nss-mdns timeout). With three units unplugged the shared worker spent
nearly all its time blocked, and the one live fixture saw multi-second lag.
A thread per target confines that block to the fixture that caused it.
"""
def __init__(self):
self._targets = {}
self._lock = threading.Lock()
self._running = False
self.min_interval = 1.0 / 25.0
# Shown by the unit to everyone while Blender holds a slot.
self.holder = ("blender-" + socket.gethostname())[:32].replace(" ", "-")
# ---- main-thread API -------------------------------------------------
def start(self):
with self._lock:
if self._running:
return
self._running = True
for tgt in self._targets.values():
self._spawn(tgt)
def stop(self):
with self._lock:
self._running = False
targets = list(self._targets.values())
self._targets.clear()
for tgt in targets:
tgt.closed = True
tgt.wake.set()
for tgt in targets:
t = tgt.thread
# A worker blocked in a lookup cannot be interrupted; it exits (and
# closes its socket) when the lookup returns. Daemon threads, so a
# timed-out join never holds Blender open.
if t is not None and t.is_alive():
t.join(timeout=2.0)
def send(self, host, port, payload, force=False, once=False):
"""Record the latest desired state for a host. Never blocks.
once=True is for a one-shot command sent while disconnected (Blackout):
the target makes one connection attempt, delivers, and is dropped, so
it does not linger as a connection nobody asked to keep. A once-send to
a fixture that is already connected leaves that connection alone.
"""
if not host or not payload:
return
key = (host, port)
with self._lock:
tgt = self._targets.get(key)
if tgt is None:
tgt = _Target(host, port)
tgt.once = once
self._targets[key] = tgt
if self._running:
self._spawn(tgt)
elif not once:
tgt.once = False
tgt.desired = payload
if force:
tgt.force = True
tgt.dirty = True
elif payload != tgt.last_sent:
tgt.dirty = True
tgt.wake.set()
def forget(self, host, port):
"""Stop driving a fixture. Its worker exits and closes the socket."""
with self._lock:
tgt = self._targets.pop((host, port), None)
if tgt is not None:
tgt.closed = True
tgt.wake.set()
def forget_all(self):
"""Disconnect everything: every worker exits and closes its socket."""
with self._lock:
targets = list(self._targets.values())
self._targets.clear()
for tgt in targets:
tgt.closed = True
tgt.wake.set()
def has_targets(self):
with self._lock:
return bool(self._targets)
def status_of(self, host, port):
with self._lock:
tgt = self._targets.get((host, port))
if tgt is None:
return ("idle", "")
note = tgt.error
if tgt.status == "connected" and tgt.busy:
note = ", ".join("slot %d held by %s" % (s + 1, h) for s, h in sorted(tgt.busy.items()))
return (tgt.status, note)
# ---- worker threads --------------------------------------------------
def _retire(self, tgt):
"""Drop a target from inside its own worker (one-shot finished)."""
with self._lock:
if self._targets.get((tgt.host, tgt.port)) is tgt:
del self._targets[(tgt.host, tgt.port)]
tgt.closed = True
def _spawn(self, tgt):
# Caller holds self._lock.
tgt.thread = threading.Thread(
target=self._run, args=(tgt,),
name="cthuloid-net-%s" % tgt.host, daemon=True)
tgt.thread.start()
def _run(self, tgt):
while self._running and not tgt.closed:
tgt.wake.wait(timeout=0.25)
tgt.wake.clear()
if tgt.closed:
break
try:
self._service(tgt, time.monotonic())
except Exception as exc: # never let the worker die
tgt.status = "error"
tgt.error = repr(exc)
# The worker owns the socket; closing it here rather than from the
# main thread means a close can never race a send in progress.
self._close(tgt)
def _service(self, tgt, now):
if tgt.desired is None:
return # nothing has ever been asked of this fixture
# (1) Reconnect on our own schedule. Previously a reconnect was only
# attempted when a new payload happened to arrive, so a fixture dropped
# while idle stayed dark until the next scene change.
if tgt.sock is None:
if now < tgt.next_retry:
return
if not self._connect(tgt):
if tgt.once:
self._retire(tgt) # one attempt only; never a retry loop
return
# The fixture may have rebooted or been dropped mid-look; it cannot
# be assumed to still hold last_sent. Re-assert the desired state --
# except on a proto-3 unit after a silent reconnect: there the lamp
# is the truth, and others may have changed it meanwhile. Pressing
# Connect (force) still pushes everything.
tgt.last_sent = None
tgt.dirty = True
if tgt.proto >= 3 and not tgt.force and isinstance(tgt.desired, protocol.Frame):
tgt.dirty = False
tgt.last_sent = tgt.desired
for s in range(4):
tgt.slot_sent[s] = tuple(tgt.desired.values[s * 3:s * 3 + 3])
# (2) Read whatever the firmware sent back. This serves two purposes:
# it stops command acks accumulating until a full receive buffer stalls
# writes, and -- more importantly -- recv() returning b"" is the peer
# having closed. That detects a drop with no traffic of our own, on the
# next 0.25 s tick, instead of waiting for the keepalive to fail.
if not self._drain(tgt):
tgt.status = "lost"
tgt.error = "peer closed"
self._close(tgt)
tgt.dirty = True # re-assert state once reconnected
tgt.next_retry = now + tgt.backoff
tgt.backoff = min(tgt.backoff * 2.0, BACKOFF_MAX)
return
# (3) Push state.
if tgt.dirty:
if not tgt.force and now - tgt.last_tx_time < self.min_interval:
return # rate limited; desired is retained, retry next tick
payload = tgt.desired
if isinstance(payload, protocol.Frame) and tgt.proto >= 3:
if self._push_slots(tgt, payload, now, tgt.force):
tgt.last_sent = payload
tgt.dirty = False
tgt.force = False
return
if self._write(tgt, protocol.frame_command(payload) if isinstance(payload, protocol.Frame)
else payload, now):
tgt.last_sent = payload
tgt.dirty = False
tgt.force = False
if tgt.once:
self._retire(tgt)
elif tgt.once:
self._retire(tgt)
return
# (4) Proto 3: hand back slots we've stopped changing.
for s in list(tgt.grants):
if now - tgt.slot_tx.get(s, 0.0) >= RELEASE_AFTER:
tgt.grants.pop(s, None)
if not self._write(tgt, protocol.release_command(s), now):
return
# (5) Keepalive, so the firmware's idle timer never fires on a look we
# are deliberately not resending.
if now - tgt.last_tx_time >= KEEPALIVE_INTERVAL:
self._write(tgt, KEEPALIVE_PAYLOAD, now) # never becomes last_sent
def _write(self, tgt, payload, now):
"""Write bytes; on failure keep the state queued for the retry."""
try:
tgt.sock.sendall(payload.encode("utf-8"))
tgt.last_tx_time = now
tgt.status = "connected"
tgt.error = ""
tgt.backoff = BACKOFF_START
return True
except (socket.timeout, OSError) as exc:
tgt.error = str(exc)
tgt.status = "lost"
self._close(tgt)
# Do NOT drop the payload -- the previous version discarded it here,
# so the first change after a silent drop was swallowed entirely.
tgt.dirty = True
tgt.next_retry = now + tgt.backoff
tgt.backoff = min(tgt.backoff * 2.0, BACKOFF_MAX)
return False
def _push_slots(self, tgt, frame, now, force):
"""Proto 3: send each changed slot under its grant. False if the link failed."""
out = []
for s in range(4):
bgr = tuple(frame.values[s * 3:s * 3 + 3])
if not force and tgt.slot_sent.get(s) == bgr:
continue
# Remembered even when we can't send it: a slot someone else holds
# is not retried until Blender's own scene changes again.
tgt.slot_sent[s] = bgr
if s not in tgt.grants:
got = self._grant(tgt, s, now)
if got is None:
return False
if not got:
continue
out.append(protocol.slot_command(s, tgt.epoch, tgt.grants[s], bgr, frame.fade_ms))
tgt.slot_tx[s] = now
if out:
return self._write(tgt, "".join(out), now)
return True
def _grant(self, tgt, slot, now):
"""Ask the unit for a slot. True granted, False refused, None link failed."""
if not self._write(tgt, protocol.grant_command(slot, self.holder), now):
return None
deadline = time.monotonic() + GRANT_TIMEOUT
while time.monotonic() < deadline:
for i, r in enumerate(tgt.replies):
if r[1] != slot and r[0] != "err":
continue
del tgt.replies[i]
if r[0] == "granted":
tgt.epoch, tgt.grants[slot] = r[2], r[3]
tgt.busy.pop(slot, None)
return True
if r[0] == "busy":
tgt.busy[slot] = r[2]
return False
if not self._read(tgt, timeout=0.05):
return None
return False
def _read(self, tgt, timeout=0.0):
"""Read and handle whatever the unit sent. False means the peer closed."""
alive = True
try:
if timeout:
tgt.sock.settimeout(timeout)
else:
tgt.sock.setblocking(False)
while True:
data = tgt.sock.recv(4096)
if not data:
alive = False # orderly close from the firmware
break
tgt.rbuf += data
while b"\n" in tgt.rbuf:
line, tgt.rbuf = tgt.rbuf.split(b"\n", 1)
self._handle_line(tgt, line.decode("utf-8", "replace").strip())
if timeout:
break # one read is enough when waiting for a reply
except (BlockingIOError, socket.timeout):
pass # nothing more to read; still connected
except OSError:
alive = False
finally:
try:
tgt.sock.setblocking(True)
tgt.sock.settimeout(SEND_TIMEOUT)
except OSError:
pass
if len(tgt.rbuf) > 4096:
tgt.rbuf = b""
return alive
def _drain(self, tgt):
"""Consume pending lines. False means the peer has closed the socket."""
return self._read(tgt)
def _handle_line(self, tgt, line):
p = line.split()
if not p:
return
try:
if p[0] == "hello":
kv = dict(x.split("=", 1) for x in p[1:] if "=" in x)
tgt.proto = int(kv.get("proto", "0"))
tgt.epoch = int(kv.get("epoch", "0"))
elif p[0] == "granted" and len(p) == 4:
tgt.replies.append(("granted", int(p[1]), int(p[2]), int(p[3])))
elif p[0] == "busy" and len(p) == 3:
tgt.replies.append(("busy", int(p[1]), p[2]))
elif p[0] == "ERR" and len(p) >= 2 and p[1] == "grant":
tgt.replies.append(("err", None, line))
elif p[0] == "held" and len(p) == 5:
s, n = int(p[1]), int(p[4])
if tgt.grants.get(s) != n:
tgt.busy[s] = p[2]
elif p[0] == "free" and len(p) >= 4:
s, n = int(p[1]), int(p[3])
tgt.busy.pop(s, None)
if tgt.grants.get(s) == n:
tgt.grants.pop(s, None) # timed out / blackout: grab again next time
elif p[0] == "ERR" and len(p) == 3 and p[1] == "notheld":
s = int(p[2])
tgt.grants.pop(s, None)
tgt.slot_sent.pop(s, None) # resend once we hold it again
except ValueError:
pass
def _duplicate_of(self, tgt):
"""Another target already holding a live socket to the same device.
Two fixtures pointing at the same unit would both drive it, each with
its own lights, so the unit flickers between two different looks. (On
firmware older than e84fa44, which served ONE client, they also evicted
each other in a tight reconnect loop.) The second one is held off and
reported instead.
Targets are keyed by the host string as typed, so "showlight02.local"
and "192.168.50.233" are two entries for one device. Comparing resolved
addresses is what actually catches that.
"""
if tgt.addr is None:
return None
with self._lock:
others = list(self._targets.values())
for other in others:
if other is tgt or other.sock is None:
continue
if other.addr == tgt.addr and other.port == tgt.port:
return other
return None
def _connect(self, tgt):
tgt.status = "connecting"
# Duplicate check uses the address cached from a previous connect. We
# deliberately do NOT resolve here: gethostbyname blocks, and one
# slow-resolving .local name would stall the single worker for every
# other fixture. create_connection resolves anyway, so the address is
# taken from getpeername() below and reused from then on.
dup = self._duplicate_of(tgt)
if dup is not None:
tgt.status = "duplicate"
tgt.error = ("same device as '%s' (%s) -- two fixtures would drive "
"one unit with different looks" % (dup.host, tgt.addr))
tgt.sock = None
tgt.next_retry = time.monotonic() + BACKOFF_MAX
return False
try:
sock = socket.create_connection((tgt.host, tgt.port), timeout=CONNECT_TIMEOUT)
sock.settimeout(SEND_TIMEOUT)
# TCP_NODELAY: these are tiny latency-sensitive frames; Nagle would
# coalesce them into visible stutter.
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
tgt.sock = sock
tgt.status = "connected"
tgt.error = ""
tgt.backoff = BACKOFF_START
try:
# Cache the peer address so the duplicate check above can run
# without a blocking lookup on later reconnects.
tgt.addr = sock.getpeername()[0]
except OSError:
pass
# The firmware greets with "<hostname> ready", and proto 3 follows
# with `hello proto=3 ...`. Read both before the first send, so the
# first frame already goes out in the right protocol.
tgt._reset_link()
deadline = time.monotonic() + 0.4
while tgt.proto == 0 and time.monotonic() < deadline:
if not self._read(tgt, timeout=0.1):
break
return True
except (socket.timeout, OSError) as exc:
tgt.status = "unreachable"
tgt.error = str(exc)
tgt.sock = None
tgt.next_retry = time.monotonic() + tgt.backoff
tgt.backoff = min(tgt.backoff * 2.0, BACKOFF_MAX)
return False
@staticmethod
def _close(tgt):
if tgt.sock is not None:
try:
tgt.sock.close()
except OSError:
pass
tgt.sock = None
# Forget what the wire holds, but keep `desired` -- that is what gets
# re-asserted once the fixture comes back.
tgt.last_sent = None
tgt.grants.clear() # the unit freed them when the socket went
tgt.busy.clear()
SENDER = Sender()