Files
2026-04-22 23:13:12 +01:00

384 lines
14 KiB
Python

"""Acquire debugging information from usb hid devices
cli implementation of https://www.pjrc.com/teensy/hid_listen.html
"""
from functools import lru_cache
from pathlib import Path
from platform import platform
from threading import Thread
from time import sleep, strftime
from milc import cli
from milc.questions import yesno
LOG_COLOR = {
'next': 0,
'colors': [
'{fg_blue}',
'{fg_cyan}',
'{fg_green}',
'{fg_magenta}',
'{fg_red}',
'{fg_yellow}',
],
}
KNOWN_BOOTLOADERS = {
# VID , PID
('03EB', '2045'): 'lufa-ms: LUFA Mass Storage Bootloader',
('03EB', '2067'): 'qmk-hid: HID Bootloader',
('03EB', '2FEF'): 'atmel-dfu: ATmega16U2',
('03EB', '2FF0'): 'atmel-dfu: ATmega32U2',
('03EB', '2FF3'): 'atmel-dfu: ATmega16U4',
('03EB', '2FF4'): 'atmel-dfu: ATmega32U4',
('03EB', '2FF9'): 'atmel-dfu: AT90USB64',
('03EB', '2FFA'): 'atmel-dfu: AT90USB162',
('03EB', '2FFB'): 'atmel-dfu: AT90USB128',
('03EB', '6124'): 'Microchip SAM-BA',
('0483', 'DF11'): 'stm32-dfu: STM32 BOOTLOADER',
('16C0', '05DC'): 'usbasploader: USBaspLoader',
('16C0', '05DF'): 'bootloadhid: HIDBoot',
('16C0', '0478'): 'halfkay: Teensy Halfkay',
('1B4F', '9203'): 'caterina: Pro Micro 3.3V',
('1B4F', '9205'): 'caterina: Pro Micro 5V',
('1B4F', '9207'): 'caterina: LilyPadUSB',
('1C11', 'B007'): 'kiibohd: Kiibohd DFU Bootloader',
('1EAF', '0003'): 'stm32duino: Maple 003',
('1FFB', '0101'): 'caterina: Pololu A-Star 32U4 Bootloader',
('2341', '0036'): 'caterina: Arduino Leonardo',
('2341', '0037'): 'caterina: Arduino Micro',
('239A', '000C'): 'caterina: Adafruit Feather 32U4',
('239A', '000D'): 'caterina: Adafruit ItsyBitsy 32U4 3v',
('239A', '000E'): 'caterina: Adafruit ItsyBitsy 32U4 5v',
('2A03', '0036'): 'caterina: Arduino Leonardo',
('2A03', '0037'): 'caterina: Arduino Micro',
('314B', '0106'): 'apm32-dfu: APM32 DFU ISP Mode',
('342D', 'DFA0'): 'wb32-dfu: WB32 Device in DFU Mode'
}
def _get_next_color():
"""Cycles to the next available log color.
"""
color = LOG_COLOR['colors'][LOG_COLOR['next']]
LOG_COLOR['next'] = (LOG_COLOR['next'] + 1) % len(LOG_COLOR['colors'])
return color
def _gen_header(dev):
"""Generate a consistent logging segment.
"""
return f'{dev["color"]}{dev["manufacturer_string"]} {dev["product_string"]}{{style_reset_all}} ({dev["color"]}{int2hex(dev["vendor_id"])}:{int2hex(dev["product_id"])}:{dev["index"]}{{style_reset_all}})'
def int2hex(number):
"""Returns a string representation of the number as hex.
"""
return f'{number:04X}'
def install_deps():
"""Install the necessary dependencies for qmk console.
"""
this_platform = platform().lower()
if 'darwin' in this_platform or 'macos' in this_platform:
command = ['brew', 'install', 'hidapi']
elif 'linux' in this_platform:
command = ['sudo', 'apt', 'install', '-y', 'libhidapi-hidraw0', 'libusb-dev']
elif 'windows' in this_platform:
command = ['pacboy', 'sync', '--needed', '--noconfirm', '--disable-download-timeout', 'hidapi:x']
else:
cli.log.error(f'Unsupported platform: {this_platform}')
cmd_str = ' '.join(command)
if yesno(f'Would you like to run "{cmd_str}" to install the necessary package?'):
cli.run(command, capture_output=False)
return True
@lru_cache(maxsize=0)
def import_usb_core():
"""Attempts to import the usb.core module.
"""
try:
import usb.core
return usb.core
except ImportError as e:
cli.log.error(f'Could not import usb.core: {e}')
if install_deps():
return import_usb_core()
raise
def import_hid():
"""Attempts to import the hid module.
"""
import os
old_cwd = os.getcwd()
try:
# macOS library search paths don't include homebrew by default when launched via a `uv`-based deployment.
# The `hid` python module does a search and attempts to load a set of known library names, but doesn't qualify the path, nor gives the option to do so.
# To work around this, we chdir to the homebrew lib directory before importing hid, then chdir back to where we were.
if 'darwin' in platform().lower() or 'macos' in platform().lower():
for path in ('/opt/homebrew/lib', '/usr/local/lib'):
lib_search = Path(path) / 'libhidapi.dylib'
if lib_search.exists():
os.chdir(path)
break
import hid
return hid
except ImportError as e:
cli.log.error(f'Could not import hid: {e}')
if install_deps():
return import_hid()
raise
finally:
os.chdir(old_cwd)
class MonitorDevice(object):
def __init__(self, hid_device, numeric):
self.hid = import_hid()
self.hid_device = hid_device
self.numeric = numeric
self.device = self.hid.Device(path=hid_device['path'])
self.current_line = ''
cli.log.info(f'Console Connected: {_gen_header(hid_device)}')
def read(self, size, encoding='ascii', timeout=1):
"""Read size bytes from the device.
"""
return self.device.read(size, timeout).decode(encoding)
def read_line(self):
"""Read from the device's console until we get a \n.
"""
while '\n' not in self.current_line:
self.current_line += self.read(32).replace('\x00', '')
lines = self.current_line.split('\n', 1)
self.current_line = lines[1]
return lines[0]
def run_forever(self):
while True:
try:
text = self.read_line()
ts = f'{{style_dim}}{{fg_green}}{strftime(cli.config.general.datetime_fmt)}{{style_reset_all}}' if cli.args.timestamp else ''
cli.echo(f'{ts}{_gen_header(self.hid_device)}: {text}')
except self.hid.HIDException:
break
class FindDevices(object):
def __init__(self, vid, pid, index, numeric):
self.hid = import_hid()
self.vid = vid
self.pid = pid
self.index = index
self.numeric = numeric
def run_forever(self):
"""Process messages from our queue in a loop.
"""
live_devices = {}
live_bootloaders = {}
while True:
try:
for device in list(live_devices):
if not live_devices[device]['thread'].is_alive():
cli.log.info(f'Console Disconnected: {_gen_header(live_devices[device])}')
del live_devices[device]
for device in self.find_devices():
if device['path'] not in live_devices:
device['color'] = _get_next_color()
live_devices[device['path']] = device
try:
monitor = MonitorDevice(device, self.numeric)
device['thread'] = Thread(target=monitor.run_forever, daemon=True)
device['thread'].start()
except Exception as e:
cli.log.error(f'Could not connect to {_gen_header(device)}: {e.__class__.__name__}: {e}')
if cli.config.general.verbose:
cli.log.exception(e)
del live_devices[device['path']]
if cli.args.bootloaders:
for device in self.find_bootloaders():
if device.address in live_bootloaders:
live_bootloaders[device.address]._qmk_found = True
else:
name = KNOWN_BOOTLOADERS[(int2hex(device.idVendor), int2hex(device.idProduct))]
cli.log.info(f'Bootloader Connected: {{style_bright}}{{fg_magenta}}{name}')
device._qmk_found = True
live_bootloaders[device.address] = device
for device in list(live_bootloaders):
if live_bootloaders[device]._qmk_found:
live_bootloaders[device]._qmk_found = False
else:
name = KNOWN_BOOTLOADERS[(int2hex(live_bootloaders[device].idVendor), int2hex(live_bootloaders[device].idProduct))]
cli.log.info(f'Bootloader Disconnected: {{style_bright}}{{fg_magenta}}{name}')
del live_bootloaders[device]
sleep(.1)
except KeyboardInterrupt:
break
def is_bootloader(self, hid_device):
"""Returns true if the device in question matches a known bootloader vid/pid.
"""
return (int2hex(hid_device.idVendor), int2hex(hid_device.idProduct)) in KNOWN_BOOTLOADERS
def is_console_hid(self, hid_device):
"""Returns true when the usage page indicates it's a teensy-style console.
"""
return hid_device['usage_page'] == 0xFF31 and hid_device['usage'] == 0x0074
def is_filtered_device(self, hid_device):
"""Returns True if the device should be included in the list of available consoles.
"""
return int2hex(hid_device['vendor_id']) == self.vid and int2hex(hid_device['product_id']) == self.pid
def find_devices_by_report(self, hid_devices):
"""Returns a list of available teensy-style consoles by doing a brute-force search.
Some versions of linux don't report usage and usage_page. In that case we fallback to reading the report (possibly inaccurately) ourselves.
"""
devices = []
for device in hid_devices:
path = device['path'].decode('utf-8')
if path.startswith('/dev/hidraw'):
number = path[11:]
report = Path(f'/sys/class/hidraw/hidraw{number}/device/report_descriptor')
if report.exists():
rp = report.read_bytes()
if rp[1] == 0x31 and rp[3] == 0x09:
devices.append(device)
return devices
def find_bootloaders(self):
"""Returns a list of available bootloader devices.
"""
import usb.core
return list(filter(self.is_bootloader, usb.core.find(find_all=True)))
def find_devices(self):
"""Returns a list of available teensy-style consoles.
"""
hid_devices = self.hid.enumerate()
devices = list(filter(self.is_console_hid, hid_devices))
if not devices:
devices = self.find_devices_by_report(hid_devices)
if self.vid and self.pid:
devices = list(filter(self.is_filtered_device, devices))
# Add index numbers
device_index = {}
for device in devices:
ident = ':'.join((int2hex(device['vendor_id']), int2hex(device['product_id'])))
if ident not in device_index:
device_index[ident] = 0
device_index[ident] += 1
device['index'] = device_index[ident]
return devices
def list_devices(device_finder):
"""Show the user a nicely formatted list of devices.
"""
devices = device_finder.find_devices()
if devices:
cli.log.info('Available devices:')
for dev in devices:
color = _get_next_color()
cli.log.info(f'\t{color}{int2hex(dev["vendor_id"])}:{int2hex(dev["product_id"])}:{dev["index"]}{{style_reset_all}}\t{dev["manufacturer_string"]} {dev["product_string"]}')
if cli.args.bootloaders:
bootloaders = device_finder.find_bootloaders()
if bootloaders:
cli.log.info('Available Bootloaders:')
for dev in bootloaders:
color = _get_next_color()
cli.log.info(f'\t{color}{int2hex(dev.idVendor)}:{int2hex(dev.idProduct)}{{style_reset_all}}\t{KNOWN_BOOTLOADERS[(int2hex(dev.idVendor), int2hex(dev.idProduct))]}')
@cli.argument('--bootloaders', arg_only=True, default=True, action='store_boolean', help='displaying bootloaders.')
@cli.argument('-d', '--device', help='Device to select - uses format <pid>:<vid>[:<index>].')
@cli.argument('-l', '--list', arg_only=True, action='store_true', help='List available hid_listen devices.')
@cli.argument('-n', '--numeric', arg_only=True, action='store_true', help='Show VID/PID instead of names.')
@cli.argument('-t', '--timestamp', arg_only=True, action='store_true', help='Print the timestamp for received messages as well.')
@cli.argument('-w', '--wait', type=int, default=1, help="How many seconds to wait between checks (Default: 1)")
@cli.subcommand('Acquire debugging information from usb hid devices.')
def console(cli):
"""Acquire debugging information from usb hid devices
"""
vid = None
pid = None
index = 1
if cli.config.console.device:
device = cli.config.console.device.split(':')
if len(device) == 2:
vid, pid = device
elif len(device) == 3:
vid, pid, index = device
if not index.isdigit():
cli.log.error(f'Device index must be a number! Got "{index}" instead.')
exit(1)
index = int(index)
if index < 1:
cli.log.error(f'Device index must be greater than 0! Got "{index}".')
exit(1)
else:
cli.log.error(f'Invalid format for device, expected "<pid>:<vid>[:<index>]" but got "{cli.config.console.device}".')
cli.print_help()
exit(1)
vid = vid.upper()
pid = pid.upper()
device_finder = FindDevices(vid, pid, index, cli.args.numeric)
if cli.args.list:
return list_devices(device_finder)
print('Looking for devices...', flush=True)
device_finder.run_forever()