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', '-F',
'--fixed-strings', '--fixed-strings',
action = 'store_true', 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: async def _run(self, args: Namespace) -> None:

View file

@ -30,7 +30,7 @@ class CmdCopy(Cmd): # export
'-F', '-F',
'--fixed-strings', '--fixed-strings',
action = 'store_true', 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: async def _run(self, args: Namespace) -> None:

View file

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

View file

@ -9,9 +9,9 @@ from argparse import ArgumentDefaultsHelpFormatter, ArgumentParser, Namespace
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, Any
from .AsyncRunner import AsyncRunner from .AsyncRunner import AsyncRunner
from .log import DEBUG, ERR, NOTICE, log, set_log_flags, set_log_level from .log import DEBUG, ERR, NOTICE, log, log_m, set_log_flags, set_log_level
from .util import pretty_cmd
from .Types import LoadTypes from .Types import LoadTypes
from .util import pretty_cmd
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Awaitable from collections.abc import Awaitable
@ -177,7 +177,9 @@ class App: # export
try: try:
# Import argcomplete only here to not require it to be compatible # Import argcomplete only here to not require it to be compatible
# with minimal environments # 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): class NoopCompleter(BaseCompleter):
@ -208,7 +210,7 @@ class App: # export
if isinstance(ret, int) and ret >= 0 and ret <= 0xFF: if isinstance(ret, int) and ret >= 0 and ret <= 0xFF:
exit_status = ret exit_status = ret
except Exception as e: 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 exit_status = 1
# AssertionErrors are programming errors, hence a programmer should # AssertionErrors are programming errors, hence a programmer should
# get a chance to figure it out # 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) ret = ''.join(chr(b) if 32 <= b <= 126 else '.' for b in chunk)
if (not annotate) or truncate is None or len(stdxxx) <= truncate: if (not annotate) or truncate is None or len(stdxxx) <= truncate:
return f'{label}"{ret}"' if quote else label + ret 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' ret += f' + {len(stdxxx) - truncate} more bytes'
return ret return ret
@ -59,9 +59,10 @@ class Result:
wd: str | None = None, wd: str | None = None,
verbose = True verbose = True
) -> str: ) -> str:
if not verbose:
ret = f'{self.__status}: ' def __status_str(cmd: list[str] | None, wd: str | None, verbose: bool) -> str:
else: if not verbose:
return str(self.__status)
from .util import pretty_cmd from .util import pretty_cmd
if cmd is None: if cmd is None:
cmd = self.__cmd cmd = self.__cmd
@ -70,17 +71,46 @@ class Result:
if wd is None: if wd is None:
wd = self.__wd wd = self.__wd
call = f'"{pretty_cmd(cmd, wd)}" ' call = f'"{pretty_cmd(cmd, wd)}" '
ret = ( if self.__status is None:
f'Command {call}has not yet exited' if self.__status is None else return f'Command {call}has not yet exited'
f'Command {call}has exited with status {self.__status}' return f'Command {call}has exited with status {self.__status}'
)
label, stdxxx, truncate = ( def __out(truncate: int) -> list[str]:
('stdout', self.__stdout, 40) if self.status == 0 ret: list[str] = []
else ('stderr', self.__stderr, None) for label, stdxxx, tr in [
) ('stdout', self.__stdout, truncate),
ret += self.__try_decode( ('stderr', self.__stderr, None),
stdxxx, quote = True, truncate = truncate, annotate = True, label = label ]:
) 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 return ret
def __repr__(self) -> str: def __repr__(self) -> str:

View file

@ -3,13 +3,12 @@ from __future__ import annotations
import abc import abc
import os import os
import pwd import pwd
import sys
from enum import Flag, auto from enum import Flag, auto
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from ..ExecContext import ExecContext 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 from ..Uri import Uri
if TYPE_CHECKING: if TYPE_CHECKING:
@ -54,24 +53,18 @@ class SSHClient(ExecContext):
log_prefix: str, log_prefix: str,
) -> Result: ) -> Result:
def __log(prio: int, *args): def __log(prio: int, *args, **kwargs):
log(prio, log_prefix, *args) 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: if self.__caps & self.Caps.LogOutput:
return return
if not block: if block is None:
return return
if isinstance(block, bytes): log_m(prio, f'---- {title} ----\n{block}', caller = get_caller_pos(1))
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}')
if wd is not None and not self.__caps & self.Caps.Wd: if wd is not None and not self.__caps & self.Caps.Wd:
cmd = ['cd', wd, '&&', *cmd] cmd = ['cd', wd, '&&', *cmd]
@ -97,8 +90,6 @@ class SSHClient(ExecContext):
if verbose: if verbose:
__log_block(NOTICE, 'stdout', ret.stdout_str_or_none) __log_block(NOTICE, 'stdout', ret.stdout_str_or_none)
__log_block(NOTICE, 'stderr', ret.stderr_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 return ret