Initial commit: VersaPad viewer + programming mode

versapad_serial.py

@@ -0,0 +1,219 @@
+"""
+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)