cmds + lib: Code and log beautification #30

Merged
Jan Lindemann merged 3 commits from jan/feature/20260624-cmds-projects-cmdbuild-annotate-types into master 2026-06-24 21:37:47 +02:00 AGit
6 changed files with 91 additions and 69 deletions

View file

@ -29,7 +29,7 @@ class CmdInstall(Cmd): # export
'-F',
'--fixed-strings',
action = 'store_true',
help = "Don't expand platform.expand_macros macros in <names>",
help = "Don't expand macros in <names>",
)
async def _run(self, args: Namespace) -> None:

View file

@ -30,7 +30,7 @@ class CmdCopy(Cmd): # export
'-F',
'--fixed-strings',
action = 'store_true',
help = "Don't expand platform.expand_macros macros in <src> and <dst>",
help = "Don't expand macros in <src> and <dst>",
)
async def _run(self, args: Namespace) -> None:

View file

@ -1,15 +1,16 @@
from __future__ import annotations
import datetime
import os
import re
from functools import lru_cache
from typing import TYPE_CHECKING
from ...App import Scope
from ...lib.log import DEBUG, ERR, NOTICE, log
from ...lib.util import get_profile_env, pretty_cmd
from .Cmd import Cmd, Parent
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from argparse import ArgumentParser, Namespace
@ -150,7 +151,9 @@ class CmdBuild(Cmd): # export
dep_tree[k].remove(d)
return 1
async def run_make(module, target, cur_project, num_projects) -> None:
async def run_make(
module: str, target: str, cur_project: int, num_projects: int
) -> None:
patt = self.app.is_excluded_from_build(module)
if patt is not None:
@ -194,45 +197,44 @@ class CmdBuild(Cmd): # export
except Exception as e:
log(
ERR,
(
f'Failed to make target "{target}" in module "{module}" '
f'below base {self.app.projs_root}: {str(e)}'
),
f'Failed to make target "{target}" in module "{module}" '
f'below base {self.app.projs_root}: {str(e)}'
)
raise
async def run_make_on_modules(modules, order, target):
async def run_make_on_modules(
modules: set[str], order: list[str], target: str
) -> None:
cur_project = 0
num_projects = len(order)
if target in ['clean', 'distclean']:
for m in reversed(order):
cur_project += 1
await run_make(m, target, cur_project, num_projects)
if m in modules:
modules.remove(m)
if not len(modules):
log(NOTICE, 'All modules cleaned')
return
else:
if target not in ['clean', 'distclean']:
for m in order:
cur_project += 1
await run_make(m, target, cur_project, num_projects)
return
for m in reversed(order):
cur_project += 1
await run_make(m, target, cur_project, num_projects)
if m in modules:
modules.remove(m)
if not len(modules):
log(NOTICE, 'All modules cleaned')
async def run(args):
async def run(args) -> None:
log(DEBUG, f'-------------------------------------- running {pretty_cmd()}')
modules = args.modules
exclude = args.exclude.split()
modules = set(args.modules)
exclude = set(args.exclude.split())
target = args.target
env_exclude = os.getenv('BUILD_EXCLUDE', '')
if len(env_exclude):
log(NOTICE, 'Exluding modules from environment: ' + env_exclude)
exclude += ' ' + env_exclude
if env_exclude is not None:
log(NOTICE, f'Exluding modules from environment: {env_exclude}')
exclude |= set(env_exclude.split())
# -- build
order = []
order: list[str] = []
glob_prereq_types = ['build']
if re.match('pkg-.*', target) is not None:
@ -251,7 +253,7 @@ class CmdBuild(Cmd): # export
exit(0)
cur_project = 0
log(NOTICE, 'Building target %s in %d projects:' % (target, len(order)))
log(NOTICE, f'Building target {target} in {len(order)} projects:')
for m in order:
cur_project += 1
log(NOTICE, ' %3d %s' % (cur_project, m))
@ -263,12 +265,9 @@ class CmdBuild(Cmd): # export
log(
NOTICE,
(
'Build done at %s' %
(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
),
'Build done at %s' %
(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
)
dep_cache: dict[str, dict[str, list[str]]] = {}
await run(args)

View file

@ -9,9 +9,9 @@ from argparse import ArgumentDefaultsHelpFormatter, ArgumentParser, Namespace
from typing import TYPE_CHECKING, Any
from .AsyncRunner import AsyncRunner
from .log import DEBUG, ERR, NOTICE, log, set_log_flags, set_log_level
from .util import pretty_cmd
from .log import DEBUG, ERR, NOTICE, log, log_m, set_log_flags, set_log_level
from .Types import LoadTypes
from .util import pretty_cmd
if TYPE_CHECKING:
from collections.abc import Awaitable
@ -177,7 +177,9 @@ class App: # export
try:
# Import argcomplete only here to not require it to be compatible
# with minimal environments
from argcomplete.completers import BaseCompleter # type: ignore[import-not-found]
from argcomplete.completers import ( # type: ignore[import-not-found]
BaseCompleter
)
class NoopCompleter(BaseCompleter):
@ -208,7 +210,7 @@ class App: # export
if isinstance(ret, int) and ret >= 0 and ret <= 0xFF:
exit_status = ret
except Exception as e:
log(ERR, 'Failed: {}'.format(repr(e) if self.__back_trace else str(e)))
log_m(ERR, 'Failed: {}'.format(repr(e) if self.__back_trace else str(e)))
exit_status = 1
# AssertionErrors are programming errors, hence a programmer should
# get a chance to figure it out

View file

@ -49,7 +49,7 @@ class Result:
ret = ''.join(chr(b) if 32 <= b <= 126 else '.' for b in chunk)
if (not annotate) or truncate is None or len(stdxxx) <= truncate:
return f'{label}"{ret}"' if quote else label + ret
ret = '{labe}"{ret} ..."' if quote else '{labe}{ret} ...'
ret = f'{label}"{ret} ..."' if quote else f'{label}{ret} ...'
ret += f' + {len(stdxxx) - truncate} more bytes'
return ret
@ -59,9 +59,10 @@ class Result:
wd: str | None = None,
verbose = True
) -> str:
if not verbose:
ret = f'{self.__status}: '
else:
def __status_str(cmd: list[str] | None, wd: str | None, verbose: bool) -> str:
if not verbose:
return str(self.__status)
from .util import pretty_cmd
if cmd is None:
cmd = self.__cmd
@ -70,17 +71,46 @@ class Result:
if wd is None:
wd = self.__wd
call = f'"{pretty_cmd(cmd, wd)}" '
ret = (
f'Command {call}has not yet exited' if self.__status is None else
f'Command {call}has exited with status {self.__status}'
)
label, stdxxx, truncate = (
('stdout', self.__stdout, 40) if self.status == 0
else ('stderr', self.__stderr, None)
)
ret += self.__try_decode(
stdxxx, quote = True, truncate = truncate, annotate = True, label = label
)
if self.__status is None:
return f'Command {call}has not yet exited'
return f'Command {call}has exited with status {self.__status}'
def __out(truncate: int) -> list[str]:
ret: list[str] = []
for label, stdxxx, tr in [
('stdout', self.__stdout, truncate),
('stderr', self.__stderr, None),
]:
if stdxxx is None:
continue
ret.append(
self.__try_decode(
stdxxx,
quote = True,
truncate = tr,
annotate = True,
label = label,
)
)
return ret
max_width = 120
# Try a one-liner first
status_str = __status_str(cmd, wd, verbose)
out = __out(40)
ret = f'{status_str}'
if out:
ret += ', ' + ', '.join(out)
if len(ret) > max_width:
ret = f'{status_str}'
if out:
# Now that we're reaking into multiple lines anyway, they may
# just as well be longer
out = __out(max_width)
ret += ':\n' + '\n'.join([' ' + line for line in out])
return ret
def __repr__(self) -> str:

View file

@ -3,13 +3,12 @@ from __future__ import annotations
import abc
import os
import pwd
import sys
from enum import Flag, auto
from typing import TYPE_CHECKING
from ..ExecContext import ExecContext
from ..log import DEBUG, ERR, INFO, NOTICE, WARNING, log
from ..log import DEBUG, ERR, INFO, NOTICE, get_caller_pos, log, log_m
from ..Uri import Uri
if TYPE_CHECKING:
@ -54,24 +53,18 @@ class SSHClient(ExecContext):
log_prefix: str,
) -> Result:
def __log(prio: int, *args):
log(prio, log_prefix, *args)
def __log(prio: int, *args, **kwargs):
caller = kwargs.get('caller')
if caller is None:
kwargs['caller'] = get_caller_pos(1)
log(prio, log_prefix, *args, **kwargs)
def __log_block(prio: int, title: str, block: bytes | str | None):
def __log_block(prio: int, title: str, block: str | None):
if self.__caps & self.Caps.LogOutput:
return
if not block:
if block is None:
return
if isinstance(block, bytes):
encoding = sys.stdout.encoding or 'utf-8'
block = block.decode(encoding).strip()
# Needed to pacify pyright: block can't be anything else at this point
assert isinstance(block, str)
delim = f'---- {title} ----'
__log(prio, f',{delim}')
for line in block.splitlines():
__log(prio, '|', line)
__log(prio, f'`{delim}')
log_m(prio, f'---- {title} ----\n{block}', caller = get_caller_pos(1))
if wd is not None and not self.__caps & self.Caps.Wd:
cmd = ['cd', wd, '&&', *cmd]
@ -97,8 +90,6 @@ class SSHClient(ExecContext):
if verbose:
__log_block(NOTICE, 'stdout', ret.stdout_str_or_none)
__log_block(NOTICE, 'stderr', ret.stderr_str_or_none)
if ret.status != 0:
__log(WARNING, f'Exit code {ret.status}')
return ret