219 lines
7.7 KiB
Python
219 lines
7.7 KiB
Python
"""
|
|
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)
|