basic info.json handling
This commit is contained in:
parent
1fbbd72c14
commit
ff1bb76537
7 changed files with 210 additions and 82 deletions
|
@ -218,6 +218,24 @@
|
|||
return_type: string
|
||||
return_constant: QSTR(PRODUCT)
|
||||
}
|
||||
0x05: {
|
||||
type: command
|
||||
name: info.json length
|
||||
define: INFO_LEN_QUERY
|
||||
description: Retrieves the length of info.json
|
||||
return_type: u32
|
||||
return_constant: info_json_gz_len
|
||||
}
|
||||
0x06: {
|
||||
type: command
|
||||
name: info.json
|
||||
define: INFO_QUERY
|
||||
description: Retrieves a chunk of info.json
|
||||
request_type: u16
|
||||
request_purpose: offset
|
||||
return_type: u8[32]
|
||||
return_execute: info_json_gz
|
||||
}
|
||||
}
|
||||
},
|
||||
|
||||
|
|
|
@ -1,79 +1 @@
|
|||
"""Interactions with compatible XAP devices
|
||||
"""
|
||||
import json
|
||||
import random
|
||||
|
||||
from milc import cli
|
||||
|
||||
|
||||
def _is_xap_usage(x):
|
||||
return x['usage_page'] == 0xFF51 and x['usage'] == 0x0058
|
||||
|
||||
|
||||
def _is_filtered_device(x):
|
||||
name = "%04x:%04x" % (x['vendor_id'], x['product_id'])
|
||||
return name.lower().startswith(cli.args.device.lower())
|
||||
|
||||
|
||||
def _search():
|
||||
devices = filter(_is_xap_usage, hid.enumerate())
|
||||
if cli.args.device:
|
||||
devices = filter(_is_filtered_device, devices)
|
||||
|
||||
return list(devices)
|
||||
|
||||
|
||||
def _list_devices():
|
||||
"""Dump out available devices
|
||||
"""
|
||||
cli.log.info('Available devices:')
|
||||
devices = _search()
|
||||
for dev in devices:
|
||||
device = hid.Device(path=dev['path'])
|
||||
|
||||
data = _query_device(device)
|
||||
cli.log.info(" %04x:%04x %s %s [API:%s]", dev['vendor_id'], dev['product_id'], dev['manufacturer_string'], dev['product_string'], data['ver'])
|
||||
|
||||
if cli.config.general.verbose:
|
||||
# TODO: better formatting like "lsusb -v"
|
||||
cli.log.info(" " + json.dumps(data))
|
||||
|
||||
|
||||
def _query_device(device):
|
||||
# gen token
|
||||
tok = random.getrandbits(16)
|
||||
temp = tok.to_bytes(2, byteorder='big')
|
||||
|
||||
# send with padding
|
||||
padding = b"\x00" * 59
|
||||
device.write(temp + b'\x02\x00\x00' + padding)
|
||||
|
||||
# get resp
|
||||
array_alpha = device.read(8, 100)
|
||||
# hex_string = " ".join("%02x" % b for b in array_alpha)
|
||||
|
||||
# validate tok sent == resp
|
||||
ver = "UNKNOWN"
|
||||
if str(temp) == str(array_alpha[:2]):
|
||||
# to BCD string
|
||||
a = (array_alpha[7] << 24) + (array_alpha[6] << 16) + (array_alpha[5] << 8) + (array_alpha[4])
|
||||
ver = f'{a>>24}.{a>>16 & 0xFF}.{a & 0xFFFF}'
|
||||
|
||||
return {'ver': ver}
|
||||
|
||||
|
||||
@cli.argument('-d', '--device', help='device to select - uses format <pid>:<vid>.')
|
||||
@cli.argument('-i', '--index', default=0, help='device index to select.')
|
||||
@cli.argument('-l', '--list', arg_only=True, action='store_true', help='List available devices.')
|
||||
@cli.subcommand('Acquire debugging information from usb XAP devices.', hidden=False if cli.config.user.developer else True)
|
||||
def xap(cli):
|
||||
"""Acquire debugging information from XAP devices
|
||||
"""
|
||||
# Lazy load to avoid issues
|
||||
global hid
|
||||
import hid
|
||||
|
||||
if cli.args.list:
|
||||
return _list_devices()
|
||||
|
||||
cli.log.warn("TODO: Device specific stuff")
|
||||
from .xap import xap
|
||||
|
|
149
lib/python/qmk/cli/xap/xap.py
Normal file
149
lib/python/qmk/cli/xap/xap.py
Normal file
|
@ -0,0 +1,149 @@
|
|||
"""Interactions with compatible XAP devices
|
||||
"""
|
||||
import json
|
||||
import random
|
||||
import gzip
|
||||
|
||||
from milc import cli
|
||||
|
||||
|
||||
def _is_xap_usage(x):
|
||||
return x['usage_page'] == 0xFF51 and x['usage'] == 0x0058
|
||||
|
||||
|
||||
def _is_filtered_device(x):
|
||||
name = "%04x:%04x" % (x['vendor_id'], x['product_id'])
|
||||
return name.lower().startswith(cli.args.device.lower())
|
||||
|
||||
|
||||
def _search():
|
||||
devices = filter(_is_xap_usage, hid.enumerate())
|
||||
if cli.args.device:
|
||||
devices = filter(_is_filtered_device, devices)
|
||||
|
||||
return list(devices)
|
||||
|
||||
|
||||
def print_dotted_output(kb_info_json, prefix=''):
|
||||
"""Print the info.json in a plain text format with dot-joined keys.
|
||||
"""
|
||||
for key in sorted(kb_info_json):
|
||||
new_prefix = f'{prefix}.{key}' if prefix else key
|
||||
|
||||
if key in ['parse_errors', 'parse_warnings']:
|
||||
continue
|
||||
elif key == 'layouts' and prefix == '':
|
||||
cli.echo(' {fg_blue}layouts{fg_reset}: %s', ', '.join(sorted(kb_info_json['layouts'].keys())))
|
||||
elif isinstance(kb_info_json[key], dict):
|
||||
print_dotted_output(kb_info_json[key], new_prefix)
|
||||
elif isinstance(kb_info_json[key], list):
|
||||
cli.echo(' {fg_blue}%s{fg_reset}: %s', new_prefix, ', '.join(map(str, sorted(kb_info_json[key]))))
|
||||
else:
|
||||
cli.echo(' {fg_blue}%s{fg_reset}: %s', new_prefix, kb_info_json[key])
|
||||
|
||||
|
||||
def _list_devices():
|
||||
"""Dump out available devices
|
||||
"""
|
||||
cli.log.info('Available devices:')
|
||||
devices = _search()
|
||||
for dev in devices:
|
||||
device = hid.Device(path=dev['path'])
|
||||
|
||||
data = _query_device_version(device)
|
||||
cli.log.info(" %04x:%04x %s %s [API:%s]", dev['vendor_id'], dev['product_id'], dev['manufacturer_string'], dev['product_string'], data['ver'])
|
||||
|
||||
if cli.config.general.verbose:
|
||||
# TODO: better formatting like "lsusb -v"
|
||||
datalen = _query_device_info_len(device)
|
||||
|
||||
data = []
|
||||
offset = 0
|
||||
while offset < datalen:
|
||||
data += _query_device_info(device, offset)
|
||||
offset += 32
|
||||
str_data = gzip.decompress(bytearray(data[:datalen]))
|
||||
print_dotted_output(json.loads(str_data))
|
||||
|
||||
|
||||
def _query_device_version(device):
|
||||
# gen token
|
||||
tok = random.getrandbits(16)
|
||||
temp = tok.to_bytes(2, byteorder='big')
|
||||
|
||||
# send with padding
|
||||
padding = b"\x00" * 59
|
||||
device.write(temp + b'\x02\x00\x00' + padding)
|
||||
|
||||
# get resp
|
||||
array_alpha = device.read(8, 100)
|
||||
# hex_string = " ".join("%02x" % b for b in array_alpha)
|
||||
|
||||
# validate tok sent == resp
|
||||
ver = "UNKNOWN"
|
||||
if str(temp) == str(array_alpha[:2]):
|
||||
# to BCD string
|
||||
a = (array_alpha[7] << 24) + (array_alpha[6] << 16) + (array_alpha[5] << 8) + (array_alpha[4])
|
||||
ver = f'{a>>24}.{a>>16 & 0xFF}.{a & 0xFFFF}'
|
||||
|
||||
return {'ver': ver}
|
||||
|
||||
|
||||
def _query_device_info_len(device):
|
||||
# gen token
|
||||
tok = random.getrandbits(16)
|
||||
temp = tok.to_bytes(2, byteorder='big')
|
||||
|
||||
# send with padding
|
||||
padding = b"\x00" * 59
|
||||
device.write(temp + b'\x02\x01\x05' + padding)
|
||||
|
||||
# get resp
|
||||
array_alpha = device.read(8, 100)
|
||||
# hex_string = " ".join("%02x" % b for b in array_alpha)
|
||||
|
||||
# validate tok sent == resp
|
||||
datalen = "UNKNOWN"
|
||||
if str(temp) == str(array_alpha[:2]):
|
||||
# to BCD string
|
||||
a = (array_alpha[7] << 24) + (array_alpha[6] << 16) + (array_alpha[5] << 8) + (array_alpha[4])
|
||||
datalen = f'{a & 0xFFFF}'
|
||||
|
||||
return int(datalen)
|
||||
|
||||
|
||||
def _query_device_info(device, offset):
|
||||
# gen token
|
||||
tok = random.getrandbits(16)
|
||||
temp = tok.to_bytes(2, byteorder='big')
|
||||
|
||||
# send with padding
|
||||
padding = b"\x00" * 57
|
||||
device.write(temp + b'\x04\x01\x06' + (offset).to_bytes(2, byteorder='big') + padding)
|
||||
|
||||
# get resp
|
||||
array_alpha = device.read(4 + 32, 100)
|
||||
|
||||
# hex_string = " ".join("%02x" % b for b in array_alpha)
|
||||
|
||||
# validate tok sent == resp
|
||||
if str(temp) == str(array_alpha[:2]):
|
||||
return array_alpha[4:]
|
||||
return None
|
||||
|
||||
|
||||
@cli.argument('-d', '--device', help='device to select - uses format <pid>:<vid>.')
|
||||
@cli.argument('-i', '--index', default=0, help='device index to select.')
|
||||
@cli.argument('-l', '--list', arg_only=True, action='store_true', help='List available devices.')
|
||||
@cli.subcommand('Acquire debugging information from usb XAP devices.', hidden=False if cli.config.user.developer else True)
|
||||
def xap(cli):
|
||||
"""Acquire debugging information from XAP devices
|
||||
"""
|
||||
# Lazy load to avoid issues
|
||||
global hid
|
||||
import hid
|
||||
|
||||
if cli.args.list:
|
||||
return _list_devices()
|
||||
|
||||
cli.log.warn("TODO: Device specific stuff")
|
|
@ -14,6 +14,7 @@ def generate_info(output_file, keyboard, keymap):
|
|||
km_info_json = keymap_json(keyboard, keymap)
|
||||
|
||||
# TODO: Munge to XAP requirements
|
||||
del km_info_json['config_h_features']
|
||||
|
||||
# Minify
|
||||
str_data = json.dumps(km_info_json, separators=(',', ':'))
|
||||
|
@ -32,9 +33,9 @@ def generate_info(output_file, keyboard, keymap):
|
|||
lines = [GPL2_HEADER_C_LIKE, GENERATED_HEADER_C_LIKE, '#pragma once', '']
|
||||
|
||||
# Gen output file
|
||||
lines.append('unsigned char info_json_gz[] = {')
|
||||
lines.append('static const unsigned char info_json_gz[] PROGMEM = {')
|
||||
lines.append(data)
|
||||
lines.append('};')
|
||||
lines.append(f'unsigned int info_json_gz_len = {data_len};')
|
||||
lines.append(f'static const unsigned int info_json_gz_len = {data_len};')
|
||||
|
||||
dump_lines(output_file, lines)
|
||||
|
|
|
@ -26,6 +26,8 @@ def _get_c_type(xap_type):
|
|||
def _get_route_type(container):
|
||||
if 'routes' in container:
|
||||
return 'XAP_ROUTE'
|
||||
elif 'return_execute' in container:
|
||||
return 'XAP_EXECUTE'
|
||||
elif 'return_constant' in container:
|
||||
if container['return_type'] == 'u32':
|
||||
return 'XAP_VALUE'
|
||||
|
@ -47,6 +49,11 @@ def _append_routing_table_declaration(lines, container, container_id, route_stac
|
|||
if 'routes' in container:
|
||||
pass
|
||||
|
||||
elif 'return_execute' in container:
|
||||
execute = container['return_execute']
|
||||
lines.append('')
|
||||
lines.append(f'bool xap_respond_{execute}(xap_token_t token, const uint8_t *data, size_t data_len);')
|
||||
|
||||
elif 'return_constant' in container:
|
||||
|
||||
if container['return_type'] == 'u32':
|
||||
|
@ -99,6 +106,11 @@ def _append_routing_table_entry_route(lines, container, container_id, route_stac
|
|||
lines.append(f' .child_routes_len = sizeof({route_name}_table)/sizeof(xap_route_t),')
|
||||
|
||||
|
||||
def _append_routing_table_entry_execute(lines, container, container_id, route_stack):
|
||||
value = container['return_execute']
|
||||
lines.append(f' .handler = xap_respond_{value},')
|
||||
|
||||
|
||||
def _append_routing_table_entry_u32value(lines, container, container_id, route_stack):
|
||||
value = container['return_constant']
|
||||
lines.append(f' .u32value = {value},')
|
||||
|
@ -134,6 +146,8 @@ def _append_routing_table_entry(lines, container, container_id, route_stack):
|
|||
_append_routing_table_entry_flags(lines, container, container_id, route_stack)
|
||||
if 'routes' in container:
|
||||
_append_routing_table_entry_route(lines, container, container_id, route_stack)
|
||||
elif 'return_execute' in container:
|
||||
_append_routing_table_entry_execute(lines, container, container_id, route_stack)
|
||||
elif 'return_constant' in container:
|
||||
if container['return_type'] == 'u32':
|
||||
_append_routing_table_entry_u32value(lines, container, container_id, route_stack)
|
||||
|
|
|
@ -17,6 +17,16 @@
|
|||
#include <quantum.h>
|
||||
#include <xap.h>
|
||||
|
||||
#include "info_json_gz.h"
|
||||
void get_info_json_chunk(uint8_t *data, size_t offset) {
|
||||
uint8_t len = 32;
|
||||
if (offset + len > info_json_gz_len) {
|
||||
len = info_json_gz_len - offset;
|
||||
}
|
||||
|
||||
memcpy_P(data, &info_json_gz[offset], len);
|
||||
}
|
||||
|
||||
#define QSTR2(z) #z
|
||||
#define QSTR(z) QSTR2(z)
|
||||
|
||||
|
|
|
@ -16,7 +16,6 @@
|
|||
|
||||
#include <quantum.h>
|
||||
#include <xap.h>
|
||||
#include <info_json_gz.h>
|
||||
|
||||
void xap_respond_failure(xap_token_t token, xap_response_flags_t response_flags) {
|
||||
xap_send(token, response_flags, NULL, 0);
|
||||
|
@ -41,3 +40,18 @@ bool xap_respond_u32(xap_token_t token, uint32_t value) {
|
|||
uint32_t xap_route_qmk_ffffffffffffffff_getter(void) {
|
||||
return 0x12345678;
|
||||
}
|
||||
|
||||
bool xap_respond_info_json_gz(xap_token_t token, const uint8_t *data, size_t data_len) {
|
||||
if (data_len != 2) {
|
||||
xap_respond_failure(token, 0);
|
||||
return false;
|
||||
}
|
||||
|
||||
uint8_t blob[32] = {0};
|
||||
uint16_t offset = ((uint16_t)data[0]) << 8 | data[1];
|
||||
|
||||
void get_info_json_chunk(uint8_t * data, size_t offset);
|
||||
get_info_json_chunk(blob, offset);
|
||||
|
||||
return xap_respond_data(token, blob, 32);
|
||||
}
|
||||
|
|
Loading…
Reference in a new issue