Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 7 additions & 12 deletions qmk_cli/git.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,6 @@

from milc import cli

default_repo = 'qmk_firmware'
default_fork = 'qmk/' + default_repo
default_branch = 'master'


def git_clone(url, destination, branch):
git_clone = [
Expand All @@ -18,7 +14,7 @@ def git_clone(url, destination, branch):
url,
str(destination),
]
cli.log.debug('Git clone command: %s', git_clone)
cli.log.debug(f'Git clone command: {git_clone}', )

try:
with subprocess.Popen(git_clone, stderr=subprocess.STDOUT, stdout=subprocess.PIPE, bufsize=1, universal_newlines=True, encoding='utf-8') as p:
Expand All @@ -28,13 +24,12 @@ def git_clone(url, destination, branch):
except Exception as e:
git_cmd = ' '.join([s.replace(' ', r'\ ') for s in git_clone])

cli.log.error("Could not run '%s': %s: %s", git_cmd, e.__class__.__name__, e)
cli.log.error(f'Could not run "{git_cmd}": {e.__class__}: {e}')
return False

if p.returncode == 0:
cli.log.info('Successfully cloned %s to %s!', url, destination)
return True

else:
cli.log.error('git clone exited %d', p.returncode)
if p.returncode != 0:
cli.log.error(f'git clone exited {p.returncode}')
return False

cli.log.info(f'Successfully cloned {url} to {destination}!')
return True
6 changes: 3 additions & 3 deletions qmk_cli/script_qmk.py
Original file line number Diff line number Diff line change
Expand Up @@ -96,9 +96,9 @@ def main():

except ImportError as e:
if qmk_firmware.name != 'qmk_firmware':
print('Warning: %s does not end in "qmk_firmware". Do you need to set QMK_HOME to "%s/qmk_firmware"?' % (qmk_firmware, qmk_firmware))
print(f'Warning: {qmk_firmware} does not end in "qmk_firmware". Do you need to set QMK_HOME to "{qmk_firmware}/qmk_firmware"?')

print('Error: %s: %s', (e.__class__.__name__, e))
print(f'Error: {e.__class__}: {e}')
print_exc()
sys.exit(1)

Expand All @@ -110,7 +110,7 @@ def main():

elif return_code is not True and isinstance(return_code, int):
if return_code < 0 or return_code > 255:
milc.cli.log.error('Invalid return_code: %d', return_code)
milc.cli.log.error(f'Invalid return_code: {return_code}')
exit(255)

exit(return_code)
Expand Down
6 changes: 3 additions & 3 deletions qmk_cli/subcommands/clone.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,16 +13,16 @@


@cli.argument('--baseurl', arg_only=True, default='https://github.com', help='The URL all git operations start from (Default: https://github.com)')
@cli.argument('-b', '--branch', arg_only=True, default=default_branch, help='The branch to clone. Default: %s' % default_branch)
@cli.argument('-b', '--branch', arg_only=True, default=default_branch, help=f'The branch to clone. Default: {default_branch}')
@cli.argument('destination', arg_only=True, default=Path(os.environ['ORIG_CWD']) / default_repo, type=AbsPath, nargs='?', help='The directory to clone to. Default: (current directory)')
@cli.argument('fork', arg_only=True, default=default_fork, nargs='?', help='The qmk_firmware fork to clone. Default: %s' % default_fork)
@cli.argument('fork', arg_only=True, default=default_fork, nargs='?', help=f'The qmk_firmware fork to clone. Default: {default_fork}')
@cli.subcommand('Clone a qmk_firmware fork.')
def clone(cli):
git_url = '/'.join((cli.args.baseurl, cli.args.fork))

# Exists (but not an empty dir)
if cli.args.destination.exists() and any(cli.args.destination.iterdir()):
cli.log.error('Destination already exists: %s', cli.args.destination)
cli.log.error(f'Destination already exists: {cli.args.destination}')
exit(1)

return git_clone(git_url, cli.args.destination, cli.args.branch)
83 changes: 46 additions & 37 deletions qmk_cli/subcommands/console.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,26 @@
}


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.
"""
Expand All @@ -69,9 +89,10 @@ def install_deps():
elif 'windows' in this_platform:
command = ['pacboy', 'sync', '--needed', '--noconfirm', '--disable-download-timeout', 'hidapi:x']
else:
cli.log.error('Unsupported platform: %s', this_platform)
cli.log.error(f'Unsupported platform: {this_platform}')

if yesno("Would you like to run `%s` to install the necessary package?", ' '.join(command)):
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

Expand All @@ -85,7 +106,7 @@ def import_usb_core():
return usb.core

except ImportError as e:
cli.log.error('Could not import usb.core: %s', e)
cli.log.error(f'Could not import usb.core: {e}')

if install_deps():
return import_usb_core()
Expand Down Expand Up @@ -113,7 +134,7 @@ def import_hid():
return hid

except ImportError as e:
cli.log.error('Could not import hid: %s', e)
cli.log.error(f'Could not import hid: {e}')

if install_deps():
return import_hid()
Expand All @@ -131,8 +152,7 @@ def __init__(self, hid_device, numeric):
self.device = self.hid.Device(path=hid_device['path'])
self.current_line = ''

cli.log.info('Console Connected: %(color)s%(manufacturer_string)s %(product_string)s{style_reset_all} (%(color)s%(vendor_id)04X:%(product_id)04X:%(index)d{style_reset_all})', hid_device)

cli.log.info(f'Console Connected: {_gen_header(hid_device)}')
def read(self, size, encoding='ascii', timeout=1):
"""Read size bytes from the device.
"""
Expand All @@ -152,12 +172,10 @@ def read_line(self):
def run_forever(self):
while True:
try:
message = {**self.hid_device, 'text': self.read_line()}
identifier = (int2hex(message['vendor_id']), int2hex(message['product_id'])) if self.numeric else (message['manufacturer_string'], message['product_string'])
message['identifier'] = ':'.join(identifier)
message['ts'] = '{style_dim}{fg_green}%s{style_reset_all} ' % (strftime(cli.config.general.datetime_fmt),) if cli.args.timestamp else ''
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('%s', '%(ts)s%(color)s%(identifier)s:%(index)d{style_reset_all}: %(text)s' % message)
cli.echo(f'{ts}{_gen_header(self.hid_device)}: {text}')

except self.hid.HIDException:
break
Expand All @@ -181,13 +199,12 @@ def run_forever(self):
try:
for device in list(live_devices):
if not live_devices[device]['thread'].is_alive():
cli.log.info('Console Disconnected: %(color)s%(manufacturer_string)s %(product_string)s{style_reset_all} (%(color)s%(vendor_id)04X:%(product_id)04X:%(index)d{style_reset_all})', live_devices[device])
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'] = LOG_COLOR['colors'][LOG_COLOR['next']]
LOG_COLOR['next'] = (LOG_COLOR['next'] + 1) % len(LOG_COLOR['colors'])
device['color'] = _get_next_color()
live_devices[device['path']] = device

try:
Expand All @@ -196,9 +213,7 @@ def run_forever(self):

device['thread'].start()
except Exception as e:
device['e'] = e
device['e_name'] = e.__class__.__name__
cli.log.error("Could not connect to %(color)s%(manufacturer_string)s %(product_string)s{style_reset_all} (%(color)s%(vendor_id)04X:%(product_id)04X:%(index)d{style_reset_all}): %(e_name)s: %(e)s", device)
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']]
Expand All @@ -209,7 +224,7 @@ def run_forever(self):
live_bootloaders[device.address]._qmk_found = True
else:
name = KNOWN_BOOTLOADERS[(int2hex(device.idVendor), int2hex(device.idProduct))]
cli.log.info('Bootloader Connected: {style_bright}{fg_magenta}%s', name)
cli.log.info(f'Bootloader Connected: {{style_bright}}{{fg_magenta}}{name}')
device._qmk_found = True
live_bootloaders[device.address] = device

Expand All @@ -218,7 +233,7 @@ def run_forever(self):
live_bootloaders[device]._qmk_found = False
else:
name = KNOWN_BOOTLOADERS[(int2hex(live_bootloaders[device].idVendor), int2hex(live_bootloaders[device].idProduct))]
cli.log.info('Bootloader Disconnected: {style_bright}{fg_magenta}%s', name)
cli.log.info(f'Bootloader Disconnected: {{style_bright}}{{fg_magenta}}{name}')
del live_bootloaders[device]

sleep(.1)
Expand Down Expand Up @@ -285,23 +300,17 @@ def find_devices(self):
# Add index numbers
device_index = {}
for device in devices:
id = ':'.join((int2hex(device['vendor_id']), int2hex(device['product_id'])))
ident = ':'.join((int2hex(device['vendor_id']), int2hex(device['product_id'])))

if id not in device_index:
device_index[id] = 0
if ident not in device_index:
device_index[ident] = 0

device_index[id] += 1
device['index'] = device_index[id]
device_index[ident] += 1
device['index'] = device_index[ident]

return devices


def int2hex(number):
"""Returns a string representation of the number as hex.
"""
return "%04X" % number


def list_devices(device_finder):
"""Show the user a nicely formatted list of devices.
"""
Expand All @@ -310,9 +319,8 @@ def list_devices(device_finder):
if devices:
cli.log.info('Available devices:')
for dev in devices:
color = LOG_COLOR['colors'][LOG_COLOR['next']]
LOG_COLOR['next'] = (LOG_COLOR['next'] + 1) % len(LOG_COLOR['colors'])
cli.log.info("\t%s%s:%s:%d{style_reset_all}\t%s %s", color, int2hex(dev['vendor_id']), int2hex(dev['product_id']), dev['index'], dev['manufacturer_string'], dev['product_string'])
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()
Expand All @@ -321,7 +329,8 @@ def list_devices(device_finder):
cli.log.info('Available Bootloaders:')

for dev in bootloaders:
cli.log.info("\t%s:%s\t%s", int2hex(dev.idVendor), int2hex(dev.idProduct), KNOWN_BOOTLOADERS[(int2hex(dev.idVendor), int2hex(dev.idProduct))])
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.')
Expand All @@ -348,17 +357,17 @@ def console(cli):
vid, pid, index = device

if not index.isdigit():
cli.log.error('Device index must be a number! Got "%s" instead.', index)
cli.log.error(f'Device index must be a number! Got "{index}" instead.')
exit(1)

index = int(index)

if index < 1:
cli.log.error('Device index must be greater than 0! Got %s', index)
cli.log.error(f'Device index must be greater than 0! Got "{index}".')
exit(1)

else:
cli.log.error('Invalid format for device, expected "<pid>:<vid>[:<index>]" but got "%s".', cli.config.console.device)
cli.log.error(f'Invalid format for device, expected "<pid>:<vid>[:<index>]" but got "{cli.config.console.device}".')
cli.print_help()
exit(1)

Expand Down
12 changes: 6 additions & 6 deletions qmk_cli/subcommands/env.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,14 +10,14 @@
@cli.argument('var', arg_only=True, default=None, nargs='?', help='Optional variable to query')
@cli.subcommand('Prints environment information.')
def env(cli):
home = os.environ.get('QMK_HOME', "")
userspace = os.environ.get('QMK_USERSPACE', "")
home = os.environ.get('QMK_HOME', '')
userspace = os.environ.get('QMK_USERSPACE', '')
data = {
'QMK_HOME': home,
'QMK_FIRMWARE': home if is_qmk_firmware(Path(home)) else "",
'QMK_USERSPACE': userspace if is_qmk_userspace(Path(userspace)) else "",
'QMK_DISTRIB_DIR': os.environ.get('QMK_DISTRIB_DIR', ""),
'QMK_PATH_PREFIX': os.environ.get('QMK_PATH_PREFIX', "")
'QMK_FIRMWARE': home if is_qmk_firmware(Path(home)) else '',
'QMK_USERSPACE': userspace if is_qmk_userspace(Path(userspace)) else '',
'QMK_DISTRIB_DIR': os.environ.get('QMK_DISTRIB_DIR', ''),
'QMK_PATH_PREFIX': os.environ.get('QMK_PATH_PREFIX', '')
}

# Now munge the current cli config
Expand Down
Loading
Loading