""" Serial-Client fuer das VersaPad-Board -- liest/schreibt die komplette Config (740B, alle 3 Profile) und die Makro-Tabelle (512B, 32 Slots) per CDC-USB-Protokoll. Protokoll 1:1 aus VersaGUI/src/Protocol.cs und VersaMCU/doc/07_serial_protocol.md uebernommen, nicht neu geraten: - 8-Byte-Pakete: [0]=Cmd/Evt, [1]=key_id/chunk-index/-count, [2..7]=Payload - Config: CmdConfigRead(0x13) -> EvtConfigBegin/Data/End (0x92/93/94) CmdConfigBegin/Data/Commit(0x10/11/12) -> EvtConfigAck/Nack (0x90/91) - Makros: CmdMacroRead(0x23) -> EvtMacroBegin/Data/End (0x96/97/98) CmdMacroBegin/Data/Commit(0x20/21/22) -> EvtMacroAck/Nack (0x95/99) Board-Identifikation per VID/PID (VID_239A&PID_0042, wie in SerialManager.cs). Wichtig: Der COM-Port ist exklusiv -- wenn VersaGUI/Tray gerade verbunden ist, schlaegt das Oeffnen hier fehl (kein Konflikt, einfach kein Erfolg). Schreiben ist sicher im Sinne von "kann das Board nicht zerlegen": die Firmware prueft Magic/Version/CRC (Config) bzw. Keycode-Bereich (Makros) VOR jedem NVM-Write und antwortet sonst nur mit NACK (siehe CMainController.cpp, USB_CMD_CONFIG_COMMIT / USB_CMD_MACRO_COMMIT). """ import time try: import serial import serial.tools.list_ports SERIAL_AVAILABLE = True except ImportError: SERIAL_AVAILABLE = False BAUD = 115200 PACKET_SIZE = 8 PAYLOAD_SIZE = 6 CONFIG_SIZE = 740 MACRO_SIZE = 512 ACTIVE_PROFILE_OFFSET = 7 CMD_CONFIG_BEGIN = 0x10 CMD_CONFIG_DATA = 0x11 CMD_CONFIG_COMMIT = 0x12 CMD_CONFIG_READ = 0x13 CMD_MACRO_BEGIN = 0x20 CMD_MACRO_DATA = 0x21 CMD_MACRO_COMMIT = 0x22 CMD_MACRO_READ = 0x23 EVT_CONFIG_ACK = 0x90 EVT_CONFIG_NACK = 0x91 EVT_CONFIG_BEGIN = 0x92 EVT_CONFIG_DATA = 0x93 EVT_CONFIG_END = 0x94 EVT_MACRO_ACK = 0x95 EVT_MACRO_BEGIN = 0x96 EVT_MACRO_DATA = 0x97 EVT_MACRO_END = 0x98 EVT_MACRO_NACK = 0x99 VID_PID_TOKEN = "239A:0042" def find_port(): if not SERIAL_AVAILABLE: return None for p in serial.tools.list_ports.comports(): if p.hwid and VID_PID_TOKEN in p.hwid.upper(): return p.device return None class VersaPadLink: """Haelt optional eine offene Verbindung; verbindet bei Bedarf neu.""" def __init__(self): self.ser = None self.last_error = None # None | "no_pyserial" | "not_found" | "busy" | "timeout" | "nack" @property def connected(self): return self.ser is not None and self.ser.is_open def _ensure_open(self): if self.connected: return True if not SERIAL_AVAILABLE: self.last_error = "no_pyserial" return False port = find_port() if not port: self.last_error = "not_found" return False try: self.ser = serial.Serial(port, BAUD, timeout=0.5) self.ser.dtr = True time.sleep(0.2) self.ser.reset_input_buffer() self.last_error = None return True except serial.SerialException: self.ser = None self.last_error = "busy" return False def close(self): if self.ser is not None: try: self.ser.close() except Exception: pass self.ser = None # ── Generischer Dump-Empfang (CONFIG_READ / MACRO_READ) ───────────── def _read_dump(self, request_cmd, begin_evt, data_evt, end_evt, size, deadline_s=3.0): if not self._ensure_open(): return None try: self.ser.reset_input_buffer() self.ser.write(bytes([request_cmd, 0, 0, 0, 0, 0, 0, 0])) buf = bytearray(size) began = False deadline = time.time() + deadline_s while time.time() < deadline: raw = self.ser.read(PACKET_SIZE) if len(raw) < PACKET_SIZE: continue evt, idx = raw[0], raw[1] if evt == begin_evt: began = True elif evt == data_evt: offset = idx * PAYLOAD_SIZE if 0 <= offset < size: n = min(PAYLOAD_SIZE, size - offset) buf[offset:offset + n] = raw[2:2 + n] elif evt == end_evt: if not began: self.last_error = "timeout" return None self.last_error = None return bytes(buf) self.last_error = "timeout" return None except serial.SerialException: self.close() self.last_error = "busy" return None # ── Generisches Schreiben (CONFIG_BEGIN/DATA/COMMIT / MACRO_*) ──────── def _write_blob(self, begin_cmd, data_cmd, commit_cmd, ack_evt, nack_evt, blob, deadline_s=5.0): if not self._ensure_open(): return False chunks = -(-len(blob) // PAYLOAD_SIZE) if chunks > 255: self.last_error = "too_large" return False try: self.ser.reset_input_buffer() self.ser.write(bytes([begin_cmd, chunks, 0, 0, 0, 0, 0, 0])) for i in range(chunks): offset = i * PAYLOAD_SIZE chunk = blob[offset:offset + PAYLOAD_SIZE] chunk = chunk + bytes(PAYLOAD_SIZE - len(chunk)) self.ser.write(bytes([data_cmd, i]) + chunk) self.ser.write(bytes([commit_cmd, 0, 0, 0, 0, 0, 0, 0])) deadline = time.time() + deadline_s while time.time() < deadline: raw = self.ser.read(PACKET_SIZE) if len(raw) < PACKET_SIZE: continue evt = raw[0] if evt == ack_evt: self.last_error = None return True if evt == nack_evt: self.last_error = "nack" return False self.last_error = "timeout" return False except serial.SerialException: self.close() self.last_error = "busy" return False # ── Oeffentliche API ──────────────────────────────────────── def read_active_profile(self): """0-2 bei Erfolg, None bei Timeout/Fehler/kein Board.""" buf = self._read_dump(CMD_CONFIG_READ, EVT_CONFIG_BEGIN, EVT_CONFIG_DATA, EVT_CONFIG_END, CONFIG_SIZE) if buf is None: return None profile = buf[ACTIVE_PROFILE_OFFSET] if profile not in (0, 1, 2): self.last_error = "timeout" return None return profile def read_full_config(self): """740B Rohdaten (alle 3 Profile) oder None.""" return self._read_dump(CMD_CONFIG_READ, EVT_CONFIG_BEGIN, EVT_CONFIG_DATA, EVT_CONFIG_END, CONFIG_SIZE) def write_full_config(self, blob): """blob: 740B (siehe versapad_protocol.pack_config). True = Board hat ACK geschickt.""" assert len(blob) == CONFIG_SIZE return self._write_blob(CMD_CONFIG_BEGIN, CMD_CONFIG_DATA, CMD_CONFIG_COMMIT, EVT_CONFIG_ACK, EVT_CONFIG_NACK, blob) def read_macros(self): """512B Rohdaten (32 Slots) oder None.""" return self._read_dump(CMD_MACRO_READ, EVT_MACRO_BEGIN, EVT_MACRO_DATA, EVT_MACRO_END, MACRO_SIZE) def write_macros(self, blob): """blob: 512B (siehe versapad_protocol.pack_macros). True = Board hat ACK geschickt.""" assert len(blob) == MACRO_SIZE return self._write_blob(CMD_MACRO_BEGIN, CMD_MACRO_DATA, CMD_MACRO_COMMIT, EVT_MACRO_ACK, EVT_MACRO_NACK, blob)