"""Smatfarm GD32F130 fixture driver (MicroPython, ESP32). Protocol, as implemented by the v2 firmware patch: write : 0xA0..0xA5 (select channel 0..5), then one value byte 0..100 release : 0xB2 -> hand control back to the fixture's own schedule timeout : 0xB3 lo hi -> set the firmware failsafe, in ticks (~seconds) readback: 0xB0 -> [0xB0][len][payload x (len-1)][XOR of payload] payload[0] schedule_mode (1 = our manual override is in effect; 0/2/3 = the fixture's own scheduler is running, i.e. it power-cycled on us) payload[1] fan state, 0/1 payload[2:4] heatsink temperature, degC x10, little-endian payload[4:8] firmware poll counter, little-endian payload[8:14] six setpoints (what was commanded) payload[14:20] six live values (after the fixture's fade ramp) The length byte is what makes this frame extensible: fields are only ever appended, so older offsets stay valid. Two hard-won constraints baked in below: * There is no ack on writes and the USART has a one-byte receive buffer. Commands sent less than ~0.5 s apart drop bytes; 2 s was reliable every time. MIN_GAP_MS enforces this. * MicroPython's UART timeout parameter does not reliably bound read(); every read here polls uart.any() against its own deadline instead. """ from machine import UART, Pin import time CMD_READBACK = 0xB0 CMD_RELEASE = 0xB2 CMD_SET_TIMEOUT = 0xB3 MIN_PAYLOAD = 20 # everything this driver knows how to parse NUM_CH = 6 # The patched firmware polls the UART from the fixture's 1 Hz task, and the # USART has a ONE-BYTE receive buffer. Two consequences, both measured on # hardware (USART_STAT showed ORERR, and no setpoint ever changed): # * at most one byte may be in flight per poll period, so every single # byte must be spaced > 1000 ms -- including the two bytes of a channel # command, which must NOT be written in one uart.write(); # * a reply can take up to a full poll period to even begin. MIN_GAP_MS = 1150 # > the fixture's 1000 ms poll period REPLY_TIMEOUT_MS = 1600 # up to 1 poll period of latency, plus the frame class SmatfarmError(Exception): pass class Smatfarm: def __init__(self, uart_id=1, tx=17, rx=16, baud=115200): self.uart = UART(uart_id, baudrate=baud, tx=Pin(tx), rx=Pin(rx)) self._last_tx = time.ticks_ms() # Diagnostics: distinguishes "replied but we mis-parsed it" from # "nothing arrives on RX at all", which look identical otherwise. self.rx_bytes = 0 self.rx_last = b"" self.fail = "" # which reject path fired, for /status self.raw = b"" # the actual frame bytes we rejected # ---- low level ---------------------------------------------------- def _pace(self): """Never send faster than the fixture can keep up with.""" waited = time.ticks_diff(time.ticks_ms(), self._last_tx) if waited < MIN_GAP_MS: time.sleep_ms(MIN_GAP_MS - waited) def _write_byte(self, b): self._pace() self.uart.write(bytes((b,))) self._last_tx = time.ticks_ms() def _drain(self): while self.uart.any(): self.uart.read(self.uart.any()) def _read_exact(self, n, timeout_ms): """Deadline-bounded read; do not trust the UART timeout kwarg.""" buf = b"" deadline = time.ticks_add(time.ticks_ms(), timeout_ms) while len(buf) < n: if time.ticks_diff(deadline, time.ticks_ms()) <= 0: return buf if self.uart.any(): buf += self.uart.read(n - len(buf)) or b"" else: time.sleep_ms(2) return buf # ---- commands ---------------------------------------------------- def set_channel(self, ch, value): if not 0 <= ch < NUM_CH: raise ValueError("channel out of range") value = max(0, min(100, int(value))) # One byte per poll period: these must be two separately paced # writes, never a single two-byte write. self._write_byte(0xA0 + ch) self._write_byte(value) return value def set_all(self, values): if len(values) != NUM_CH: raise ValueError("need %d values" % NUM_CH) return [self.set_channel(i, v) for i, v in enumerate(values)] def readback(self): """Return a dict of fixture state, or None if no valid reply. The reply header 0xB0 is non-ASCII, so it cannot collide with the fixture's own IR debug logging on the same TX line -- we resync on it rather than assuming the buffer starts clean. """ self._pace() self._drain() self.uart.write(bytes((CMD_READBACK,))) self._last_tx = time.ticks_ms() # Scan for the header rather than assuming a clean buffer: the # fixture logs "IR RAW[%x] KEY[%x]" on this same TX line, so an # arbitrary amount of ASCII can precede the reply. Read greedily # against one deadline and resync on the 0xB0. deadline = time.ticks_add(time.ticks_ms(), REPLY_TIMEOUT_MS) frame = b"" while True: idx = frame.find(bytes((CMD_READBACK,))) if idx >= 0: frame = frame[idx:] break frame = frame[-1:] if frame else b"" # keep nothing useful if time.ticks_diff(deadline, time.ticks_ms()) <= 0: self.fail = "no-header-deadline"; self.raw = frame; return None more = self._read_exact(MIN_PAYLOAD + 2, REPLY_TIMEOUT_MS) if not more: self.fail = "no-header-empty"; self.raw = frame; return None self.rx_bytes += len(more) self.rx_last = (self.rx_last + more)[-16:] frame += more n = 0 while True: if len(frame) >= 2: n = frame[1] # payload bytes + checksum if n < MIN_PAYLOAD + 1: self.fail = "short-len=%d" % n; self.raw = frame; return None if len(frame) >= 2 + n: break if time.ticks_diff(deadline, time.ticks_ms()) <= 0: self.fail = "body-deadline"; self.raw = frame; return None more = self._read_exact((2 + n - len(frame)) or 1, REPLY_TIMEOUT_MS) if not more: self.fail = "body-empty"; self.raw = frame; return None self.rx_bytes += len(more) self.rx_last = (self.rx_last + more)[-16:] frame += more body = frame[2:2 + n - 1] ck = 0 for byte in body: ck ^= byte if ck != frame[2 + n - 1]: self.fail = "checksum calc=%02x got=%02x" % (ck, frame[2 + n - 1]) self.raw = frame return None self.fail = "" self.raw = frame return { "mode": body[0], "override": body[0] == 1, "fan": body[1], "temp_c": (body[2] | (body[3] << 8)) / 10.0, "ticks": body[4] | (body[5] << 8) | (body[6] << 16) | (body[7] << 24), "setpoints": list(body[8:14]), "live": list(body[14:20]), } def release(self): """Hand control back to the fixture's internal sunrise/sunset schedule.""" self._pace() self.uart.write(bytes((CMD_RELEASE,))) self._last_tx = time.ticks_ms() def set_failsafe(self, units): """Set the firmware's dead-controller timeout, in ticks. The firmware hands control back to its own sunrise/sunset schedule if it hears nothing from us within this window, so the lights do not freeze at whatever was last commanded when this controller dies. Both readback and channel writes count as contact. The fixture's main loop calls the patched poll behind a 1000 ms gate, so one tick is one second and the range is ~18 hours. The firmware default is 3600. Confirm the rate on real hardware by comparing the `ticks` field of two readbacks a known time apart. """ if not 1 <= units <= 0xFFFF: raise ValueError("units out of range") self._pace() self.uart.write(bytes((CMD_SET_TIMEOUT, units & 0xFF, units >> 8))) self._last_tx = time.ticks_ms() # ---- state keeping ----------------------------------------------- def override_active(self): """False once the fixture has power-cycled back to its own schedule.""" r = self.readback() return None if r is None else r["override"] def ensure(self, desired, retries=2): """Re-assert `desired` if the fixture drifted or reverted. Needed because a fixture power cycle resets the override flag and the fixture resumes its internal sunrise/sunset schedule. """ for _ in range(retries + 1): r = self.readback() if r is not None and r["override"] and r["setpoints"] == list(desired): return True self.set_all(desired) r = self.readback() return r is not None and r["override"] and r["setpoints"] == list(desired)