ahoneybun/qmk_firmware
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Stub out more of broadcast messages

Browse source
This commit is contained in:
zvecr 2022-04-05 18:54:28 +01:00
parent 0d59f8b42d
commit c1b57354f6
8 changed files with 162 additions and 30 deletions
Download patch file Download diff file Expand all files Collapse all files

82
lib/python/qmk/cli/xap/xap.py
View file

@ -48,7 +48,7 @@ def print_dotted_output(kb_info_json, prefix=''):
cli.echo(' {fg_blue}%s{fg_reset}: %s', new_prefix, kb_info_json[key])
def _xap_transaction(device, sub, route, ret_len, *args):
def _xap_transaction(device, sub, route, *args):
# gen token
tok = random.getrandbits(16)
token = tok.to_bytes(2, byteorder='little')
@ -78,16 +78,20 @@ def _xap_transaction(device, sub, route, ret_len, *args):
device.write(buffer)
# get resp
array_alpha = device.read(4 + ret_len, 100)
array_alpha = device.read(64, 100)
# validate tok sent == resp
if str(token) != str(array_alpha[:2]):
return None
return array_alpha[4:]
if int(array_alpha[2]) != 0x01:
return None
payload_len = int(array_alpha[3])
return array_alpha[4:4 + payload_len]
def _query_device(device):
ver_data = _xap_transaction(device, 0x00, 0x00, 4)
ver_data = _xap_transaction(device, 0x00, 0x00)
if not ver_data:
return {'xap': 'UNKNOWN'}
@ -95,14 +99,14 @@ def _query_device(device):
a = (ver_data[3] << 24) + (ver_data[2] << 16) + (ver_data[1] << 8) + (ver_data[0])
ver = f'{a>>24}.{a>>16 & 0xFF}.{a & 0xFFFF}'
secure = int.from_bytes(_xap_transaction(device, 0x00, 0x03, 1), 'little')
secure = int.from_bytes(_xap_transaction(device, 0x00, 0x03), 'little')
secure = 'unlocked' if secure == 2 else 'LOCKED'
return {'xap': ver, 'secure': secure}
def _query_device_info_len(device):
len_data = _xap_transaction(device, 0x01, 0x05, 4)
len_data = _xap_transaction(device, 0x01, 0x05)
if not len_data:
return 0
@ -111,7 +115,7 @@ def _query_device_info_len(device):
def _query_device_info_chunk(device, offset):
return _xap_transaction(device, 0x01, 0x06, 32, offset)
return _xap_transaction(device, 0x01, 0x06, offset)
def _query_device_info(device):
@ -143,7 +147,42 @@ def _list_devices():
# TODO: better formatting like "lsusb -v"?
data = _query_device_info(device)
print_dotted_output(data)
# _xap_transaction(device, 0x01, 0x07, 1)
def xap_doit():
print("xap_doit")
# get layer count
# layers = _xap_transaction(device, 0x04, 0x01)
# layers = int.from_bytes(layers, "little")
# print(f'layers:{layers}')
# get keycode [layer:0, row:0, col:0]
# keycode = _xap_transaction(device, 0x04, 0x02, b"\x00\x00\x00")
# keycode = int.from_bytes(keycode, "little")
# keycode_map = {
# # TODO: this should be data driven...
# 0x04: 'KC_A',
# 0x05: 'KC_B',
# 0x29: 'KC_ESCAPE'
# }
# print('keycode:' + keycode_map.get(keycode, 'unknown'))
# Reboot
# _xap_transaction(device, 0x01, 0x07)
def xap_broadcast_listen(device):
try:
cli.log.info("Listening for XAP broadcasts...")
while 1:
array_alpha = device.read(64, 100)
if str(b"\xFF\xFF") == str(array_alpha[:2]):
if array_alpha[2] == 1:
cli.log.info(" Broadcast: Secure[%02x]", array_alpha[4])
else:
cli.log.info(" Broadcast: type[%02x] data:[%02x]", array_alpha[2], array_alpha[4])
except KeyboardInterrupt:
cli.log.info("Stopping...")
@cli.argument('-d', '--device', help='device to select - uses format <pid>:<vid>.')
@ -161,25 +200,14 @@ def xap(cli):
return _list_devices()
# Connect to first available device
dev = _search()[0]
devices = _search()
if not devices:
cli.log.error("No devices found!")
return False
dev = devices[0]
device = hid.Device(path=dev['path'])
cli.log.info("Connected to:%04x:%04x %s %s", dev['vendor_id'], dev['product_id'], dev['manufacturer_string'], dev['product_string'])
# get layer count
layers = _xap_transaction(device, 0x04, 0x01, 1)
layers = int.from_bytes(layers, "little")
print(f'layers:{layers}')
# get keycode [layer:0, row:0, col:0]
keycode = _xap_transaction(device, 0x04, 0x02, 2, b"\x00\x00\x00")
keycode = int.from_bytes(keycode, "little")
keycode_map = {
# TODO: this should be data driven...
0x04: 'KC_A',
0x05: 'KC_B',
0x29: 'KC_ESCAPE'
}
print('keycode:' + keycode_map.get(keycode, 'unknown'))
# Reboot
# _xap_transaction(device, 0x01, 0x07, 1)
# xap_doit(device)
xap_broadcast_listen(device)

10
lib/python/qmk/xap/gen_firmware/header_generator.py
View file

@ -177,6 +177,16 @@ def _append_internal_types(lines, container):
lines.append(f'#define {prefix}_FAILED 0x00')
lines.append('')
broadcast_messages = container.get('broadcast_messages', {})
broadcast_prefix = broadcast_messages['define_prefix']
for key, value in broadcast_messages['messages'].items():
define = value.get('define')
lines.append(f'#define {broadcast_prefix}_{define} {key}')
# Add special
lines.append(f'#define {broadcast_prefix}_TOKEN 0xFFFF')
lines.append('')
additional_types = {}
types = container.get('type_definitions', {})
for key, value in types.items():

17
lib/python/qmk/xap/gen_firmware/inline_generator.py
View file

@ -245,6 +245,22 @@ def _append_routing_tables(lines, container, container_id=None, route_stack=None
route_stack.pop()
def _append_broadcast_messages(lines, container):
"""TODO:
"""
broadcast_messages = container.get('broadcast_messages', {})
broadcast_prefix = broadcast_messages['define_prefix']
for key, value in broadcast_messages['messages'].items():
define = value.get('define')
name = to_snake(f'{broadcast_prefix}_{define}')
if 'return_type' in value:
ret_type = _get_c_type(value['return_type'])
lines.append(f'void {name}({ret_type} value) {{ xap_broadcast({key}, &value, sizeof(value)); }}')
else:
lines.append(f'void {name}(const void *data, size_t length){{ xap_broadcast({key}, data, length); }}')
def generate_inline(output_file):
"""Generates the XAP protocol header file, generated during normal build.
"""
@ -254,6 +270,7 @@ def generate_inline(output_file):
lines = [GPL2_HEADER_C_LIKE, GENERATED_HEADER_C_LIKE, '']
# Add all the generated code
_append_broadcast_messages(lines, xap_defs)
_append_routing_tables(lines, xap_defs)
dump_lines(output_file, lines)