"""A proto-3 light unit for tests: referees per-slot grants like the firmware's
grants.h (first come first served, per socket). Signatures are a dummy."""
import socket
import threading
import time
class FakeUnit3:
def __init__(self, name="fake3", epoch=5, proto=3):
self.name = name
self.epoch = epoch
self.proto = proto
self.counter = 0
self.values = [0] * 12
self.holds = [None] * 4 # (conn_id, holder, n)
self.lines = [] # (conn_id, line)
self.conns = {}
self.lock = threading.Lock()
self.srv = socket.socket()
self.srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.srv.bind(("127.0.0.1", 0))
self.srv.listen(8)
self.port = self.srv.getsockname()[1]
self.next_id = 1
threading.Thread(target=self._accept, daemon=True).start()
def received(self, prefix):
with self.lock:
return [l for _, l in self.lines if l.startswith(prefix)]
def _bcast(self, line):
for c in list(self.conns.values()):
try:
c.sendall((line + "\r\n").encode())
except OSError:
pass
def _free(self, s, why):
conn, holder, n = self.holds[s]
self.holds[s] = None
v = self.values[s * 3:s * 3 + 3]
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))
def _set(self, start, vals, skip_held):
changed = False
for i, v in enumerate(vals):
c = start + i
if c >= 12 or (skip_held and self.holds[c // 3]):
continue
if self.values[c] != v:
self.values[c] = v
changed = True
if changed:
self._bcast("values " + " ".join(map(str, self.values)))
def _accept(self):
while True:
try:
c, _ = self.srv.accept()
except OSError:
return
with self.lock:
cid = self.next_id
self.next_id += 1
self.conns[cid] = c
c.sendall(("%s ready\r\n" % self.name).encode())
if self.proto >= 3:
c.sendall(("hello proto=3 unit=%s epoch=%d pk=%s\r\n" % (self.name, self.epoch, "cd" * 32)).encode())
threading.Thread(target=self._read, args=(cid, c), daemon=True).start()
def _read(self, cid, c):
buf = b""
while True:
try:
d = c.recv(4096)
except OSError:
d = b""
if not d:
with self.lock:
self.conns.pop(cid, None)
for s in range(4):
if self.holds[s] and self.holds[s][0] == cid:
self._free(s, "disconnect")
return
buf += d
while b"\n" in buf:
line, buf = buf.split(b"\n", 1)
line = line.decode().strip()
if line:
with self.lock:
self._command(cid, c, line)
def _command(self, cid, c, line):
self.lines.append((cid, line))
p = line.split()
rng = lambda a: [0, 1, 2, 3] if a == "all" else [int(a)]
send = lambda s: c.sendall((s + "\r\n").encode())
if p[0] == "grant" and self.proto >= 3:
slots = rng(p[1])
for s in slots:
if self.holds[s] and self.holds[s][0] != cid:
return send("busy %d %s" % (s, self.holds[s][1]))
for s in slots:
if not self.holds[s]:
self.counter += 1
self.holds[s] = (cid, p[2], self.counter)
self._bcast("held %d %s %d %d" % (s, p[2], self.epoch, self.counter))
send("granted %d %d %d" % (s, self.epoch, self.holds[s][2]))
elif p[0] == "slot" and self.proto >= 3:
s, e, n, b, g, r = map(int, p[1:7])
h = self.holds[s]
if not h or h[0] != cid or e != self.epoch or n != h[2]:
return send("ERR notheld %d" % s)
self._set(s * 3, [b, g, r], False)
elif p[0] == "release":
for s in rng(p[1]):
if self.holds[s] and self.holds[s][0] == cid:
self._free(s, "release")
elif p[0] == "pwm":
self._set(0, list(map(int, p[1:])), True)
elif p[0] == "pwm2":
self._set(0, list(map(int, p[3:])), True)
elif p[0] == "stopfade":
for s in range(4):
if self.holds[s]:
self._free(s, "blackout")
def external(self):
"""Another controller's socket (e.g. the web UI)."""
s = socket.create_connection(("127.0.0.1", self.port))
time.sleep(0.1)
return s
def close(self):
self.srv.close()
for c in list(self.conns.values()):
try:
c.close()
except OSError:
pass
def wait(pred, timeout=3.0):
end = time.monotonic() + timeout
while time.monotonic() < end:
if pred():
return True
time.sleep(0.02)
return pred()