Tidy up python client

This commit is contained in:
zvecr 2022-07-17 22:02:18 +01:00
parent e9ec212376
commit ca8af27673
4 changed files with 68 additions and 57 deletions

View File

@ -1,9 +1,10 @@
# TODO: assumption of only one level of children
class XAPRoutes():
{%- for id, route in xap.routes | dictsort %} {%- for id, route in xap.routes | dictsort %}
{%- if route.routes %} {%- if route.routes %}
# {{route.define}} # {{route.define}}
{%- for subid, subroute in route.routes | dictsort %} {%- for subid, subroute in route.routes | dictsort %}
{{route.define}}_{{subroute.define}} = b'\x{{ '%02d' % id|int(base=16) }}\x{{ '%02d' % subid|int(base=16) }}' {{route.define}}_{{subroute.define}} = b'\x{{ '%02d' % id|int(base=16) }}\x{{ '%02d' % subid|int(base=16) }}'
{%- endfor %} {%- endfor %}
{%- endif %} {%- endif %}
{%- endfor %} {%- endfor %}

View File

@ -47,7 +47,7 @@ def _list_devices():
device = XAPClient().connect(dev) device = XAPClient().connect(dev)
data = device.info() data = device.info()
cli.log.info(' %04x:%04x %s %s [API:%s]', dev['vendor_id'], dev['product_id'], dev['manufacturer_string'], dev['product_string'], data['xap']) cli.log.info(' %04x:%04x %s %s [API:%s]', dev['vendor_id'], dev['product_id'], dev['manufacturer_string'], dev['product_string'], data['_version']['xap'])
if cli.config.general.verbose: if cli.config.general.verbose:
# TODO: better formatting like 'lsusb -v'? # TODO: better formatting like 'lsusb -v'?
@ -65,10 +65,10 @@ class XAPShell(cmd.Cmd):
self.keycodes = get_xap_keycodes(device.version()['xap']) self.keycodes = get_xap_keycodes(device.version()['xap'])
def do_about(self, arg): def do_about(self, arg):
"""Prints out the current version of QMK with a build date """Prints out the version info of QMK
""" """
# TODO: request stuff? data = self.device.version()
print(self.device.info()['xap']) print_dotted_output(data)
def do_status(self, arg): def do_status(self, arg):
"""Prints out the current device state """Prints out the current device state

View File

@ -2,8 +2,9 @@
""" """
import hid import hid
import json import json
import random import time
import gzip import gzip
import random
import threading import threading
import functools import functools
from struct import Struct, pack, unpack from struct import Struct, pack, unpack
@ -11,7 +12,8 @@ from collections import namedtuple
from platform import platform from platform import platform
from .types import XAPSecureStatus, XAPFlags, XAPRouteError from .types import XAPSecureStatus, XAPFlags, XAPRouteError
from .routes import XAP_VERSION_QUERY from .routes import XAPRoutes
from .util import u32toBCD
RequestPacket = namedtuple('RequestPacket', 'token length data') RequestPacket = namedtuple('RequestPacket', 'token length data')
RequestStruct = Struct('<HB61s') RequestStruct = Struct('<HB61s')
@ -29,27 +31,33 @@ def _gen_token():
return unpack('<H', pack('>H', token))[0] return unpack('<H', pack('>H', token))[0]
def _u32toBCD(val): # noqa: N802
"""Create BCD string
"""
return f'{val>>24}.{val>>16 & 0xFF}.{val & 0xFFFF}'
class XAPDevice: class XAPDevice:
def __init__(self, dev): def __init__(self, dev):
"""Constructor opens hid device and starts dependent services """Constructor opens hid device and starts dependent services
""" """
self.responses = {} self.responses = {}
self.do_read = True
self.dev = hid.Device(path=dev['path']) self.dev = hid.Device(path=dev['path'])
self.bg = threading.Thread(target=self._read_loop, daemon=True) self.bg = threading.Thread(target=self._read_loop, daemon=True)
self.bg.start() self.bg.start()
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, exc_traceback):
self.close()
def close(self):
self.do_read = False
time.sleep(1)
self.dev.close()
def _read_loop(self): def _read_loop(self):
"""Background thread to signal waiting transactions """Background thread to signal waiting transactions
""" """
while 1: while self.do_read:
array_alpha = self.dev.read(ResponseStruct.size, 100) array_alpha = self.dev.read(ResponseStruct.size, 100)
if array_alpha: if array_alpha:
token = int.from_bytes(array_alpha[:2], 'little') token = int.from_bytes(array_alpha[:2], 'little')
@ -59,14 +67,14 @@ class XAPDevice:
event.set() event.set()
def _query_device_info(self): def _query_device_info(self):
datalen = int.from_bytes(self.transaction(b'\x01\x05') or bytes(0), 'little') datalen = int.from_bytes(self.transaction(XAPRoutes.QMK_CONFIG_BLOB_LEN) or bytes(0), 'little')
if not datalen: if not datalen:
return {} return {}
data = [] data = []
offset = 0 offset = 0
while offset < datalen: while offset < datalen:
chunk = self.transaction(b'\x01\x06', offset) chunk = self.transaction(XAPRoutes.QMK_CONFIG_BLOB_CHUNK, offset)
data += chunk data += chunk
offset += len(chunk) offset += len(chunk)
str_data = gzip.decompress(bytearray(data[:datalen])) str_data = gzip.decompress(bytearray(data[:datalen]))
@ -127,13 +135,14 @@ class XAPDevice:
@functools.lru_cache @functools.lru_cache
def subsystem(self): def subsystem(self):
sub = int.from_bytes(self._transaction(b'\x00\x02') or bytes(0), 'little') sub = int.from_bytes(self._transaction(XAPRoutes.XAP_SUBSYSTEM_QUERY) or bytes(0), 'little')
return sub return sub
@functools.lru_cache @functools.lru_cache
def version(self): def version(self):
ver = int.from_bytes(self._transaction(XAP_VERSION_QUERY) or bytes(0), 'little') xap = int.from_bytes(self._transaction(XAPRoutes.XAP_VERSION_QUERY) or bytes(0), 'little')
return {'xap': _u32toBCD(ver)} qmk = int.from_bytes(self._transaction(XAPRoutes.QMK_VERSION_QUERY) or bytes(0), 'little')
return {'xap': u32toBCD(xap), 'qmk': u32toBCD(qmk)}
def _ensure_route(self, route): def _ensure_route(self, route):
(sub, rt) = route (sub, rt) = route
@ -152,23 +161,23 @@ class XAPDevice:
@functools.lru_cache @functools.lru_cache
def info(self): def info(self):
data = self._query_device_info() data = self._query_device_info()
data['_id'] = self.transaction(b'\x01\x08') data['_id'] = self.transaction(XAPRoutes.QMK_HARDWARE_ID)
data['xap'] = self.version()['xap'] data['_version'] = self.version()
return data return data
def status(self): def status(self):
lock = int.from_bytes(self.transaction(b'\x00\x03') or bytes(0), 'little') lock = int.from_bytes(self.transaction(XAPRoutes.XAP_SECURE_STATUS) or bytes(0), 'little')
data = {} data = {}
data['lock'] = XAPSecureStatus(lock).name data['lock'] = XAPSecureStatus(lock).name
return data return data
def unlock(self): def unlock(self):
self.transaction(b'\x00\x04') self.transaction(XAPRoutes.XAP_SECURE_UNLOCK)
def lock(self): def lock(self):
self.transaction(b'\x00\x05') self.transaction(XAPRoutes.XAP_SECURE_LOCK)
def reset(self): def reset(self):
status = int.from_bytes(self.transaction(b'\x01\x07') or bytes(0), 'little') status = int.from_bytes(self.transaction(XAPRoutes.QMK_BOOTLOADER_JUMP) or bytes(0), 'little')
return status == 1 return status == 1

View File

@ -26,33 +26,34 @@
# #
################################################################################ ################################################################################
# TODO: assumption of only one level of children
# XAP class XAPRoutes():
XAP_VERSION_QUERY = b'\x00\x00' # XAP
XAP_CAPABILITIES_QUERY = b'\x00\x01' XAP_VERSION_QUERY = b'\x00\x00'
XAP_SUBSYSTEM_QUERY = b'\x00\x02' XAP_CAPABILITIES_QUERY = b'\x00\x01'
XAP_SECURE_STATUS = b'\x00\x03' XAP_SUBSYSTEM_QUERY = b'\x00\x02'
XAP_SECURE_UNLOCK = b'\x00\x04' XAP_SECURE_STATUS = b'\x00\x03'
XAP_SECURE_LOCK = b'\x00\x05' XAP_SECURE_UNLOCK = b'\x00\x04'
# QMK XAP_SECURE_LOCK = b'\x00\x05'
QMK_VERSION_QUERY = b'\x01\x00' # QMK
QMK_CAPABILITIES_QUERY = b'\x01\x01' QMK_VERSION_QUERY = b'\x01\x00'
QMK_BOARD_IDENTIFIERS = b'\x01\x02' QMK_CAPABILITIES_QUERY = b'\x01\x01'
QMK_BOARD_MANUFACTURER = b'\x01\x03' QMK_BOARD_IDENTIFIERS = b'\x01\x02'
QMK_PRODUCT_NAME = b'\x01\x04' QMK_BOARD_MANUFACTURER = b'\x01\x03'
QMK_CONFIG_BLOB_LEN = b'\x01\x05' QMK_PRODUCT_NAME = b'\x01\x04'
QMK_CONFIG_BLOB_CHUNK = b'\x01\x06' QMK_CONFIG_BLOB_LEN = b'\x01\x05'
QMK_BOOTLOADER_JUMP = b'\x01\x07' QMK_CONFIG_BLOB_CHUNK = b'\x01\x06'
QMK_HARDWARE_ID = b'\x01\x08' QMK_BOOTLOADER_JUMP = b'\x01\x07'
# KEYMAP QMK_HARDWARE_ID = b'\x01\x08'
KEYMAP_CAPABILITIES_QUERY = b'\x04\x01' # KEYMAP
KEYMAP_GET_LAYER_COUNT = b'\x04\x02' KEYMAP_CAPABILITIES_QUERY = b'\x04\x01'
KEYMAP_GET_KEYMAP_KEYCODE = b'\x04\x03' KEYMAP_GET_LAYER_COUNT = b'\x04\x02'
KEYMAP_GET_ENCODER_KEYCODE = b'\x04\x04' KEYMAP_GET_KEYMAP_KEYCODE = b'\x04\x03'
# REMAPPING KEYMAP_GET_ENCODER_KEYCODE = b'\x04\x04'
REMAPPING_CAPABILITIES_QUERY = b'\x05\x01' # REMAPPING
REMAPPING_GET_DYNAMIC_LAYER_COUNT = b'\x05\x02' REMAPPING_CAPABILITIES_QUERY = b'\x05\x01'
REMAPPING_SET_KEYMAP_KEYCODE = b'\x05\x03' REMAPPING_GET_DYNAMIC_LAYER_COUNT = b'\x05\x02'
REMAPPING_SET_ENCODER_KEYCODE = b'\x05\x04' REMAPPING_SET_KEYMAP_KEYCODE = b'\x05\x03'
# LIGHTING REMAPPING_SET_ENCODER_KEYCODE = b'\x05\x04'
LIGHTING_CAPABILITIES_QUERY = b'\x06\x01' # LIGHTING
LIGHTING_CAPABILITIES_QUERY = b'\x06\x01'