diff --git a/components/app_update/otatool.py b/components/app_update/otatool.py index 0e122987b57..172b89a661f 100755 --- a/components/app_update/otatool.py +++ b/components/app_update/otatool.py @@ -3,9 +3,8 @@ # otatool is used to perform ota-level operations - flashing ota partition # erasing ota partition and switching ota partition # -# SPDX-FileCopyrightText: 2018-2025 Espressif Systems (Shanghai) CO LTD +# SPDX-FileCopyrightText: 2018-2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 -import argparse import binascii import collections import os @@ -13,6 +12,15 @@ import struct import sys import tempfile +import rich_click as click +from esp_pylib.cli_options import EspRichGroup +from esp_pylib.cli_options import MutuallyExclusiveOption +from esp_pylib.cli_options import OptionEatAll +from esp_pylib.cli_types import AnyIntType +from esp_pylib.logger import Verbosity +from esp_pylib.logger import log +from rich.markup import escape + try: from parttool import PARTITION_TABLE_OFFSET from parttool import PartitionName @@ -31,13 +39,6 @@ __version__ = '2.0' SPI_FLASH_SEC_SIZE = 0x2000 -quiet = False - - -def status(msg): - if not quiet: - print(msg) - class OtatoolTarget: OTADATA_PARTITION = PartitionType('data', 'ota') @@ -226,8 +227,9 @@ def _read_otadata(target): otadata_info = target._get_otadata_info() - print(' {:8s} \t {:8s} | \t {:8s} \t {:8s}'.format('OTA_SEQ', 'CRC', 'OTA_SEQ', 'CRC')) - print( + # Regular tool output (same stream as pre-pylib status()/print). + log.print(' {:8s} \t {:8s} | \t {:8s} \t {:8s}'.format('OTA_SEQ', 'CRC', 'OTA_SEQ', 'CRC')) + log.print( f'Firmware: {otadata_info[0].seq:#08x} \t{otadata_info[0].crc:#08x} | ' f'\t{otadata_info[1].seq:#08x} \t {otadata_info[1].crc:#08x}' ) @@ -235,7 +237,7 @@ def _read_otadata(target): def _erase_otadata(target): target.erase_otadata() - status('Erased ota_data partition contents') + log.print('Erased ota_data partition contents') def _switch_ota_partition(target, ota_id): @@ -244,165 +246,62 @@ def _switch_ota_partition(target, ota_id): def _read_ota_partition(target, ota_id, output): target.read_ota_partition(ota_id, output) - status(f'Read ota partition contents to file {output}') + log.print(f'Read ota partition contents to file {escape(str(output))}') def _write_ota_partition(target, ota_id, input_file): target.write_ota_partition(ota_id, input_file) - status(f'Written contents of file {input_file} to ota partition') + log.print(f'Written contents of file {escape(str(input_file))} to ota partition') def _erase_ota_partition(target, ota_id): target.erase_ota_partition(ota_id) - status('Erased contents of ota partition') + log.print('Erased contents of ota partition') -def main(): - global quiet +def _target_kwargs_from_ctx(ctx_obj, spi_flash_sec_size=None): + kwargs = {} + for key, value in ( + ('port', ctx_obj.get('port')), + ('baud', ctx_obj.get('baud')), + ('partition_table_offset', ctx_obj.get('partition_table_offset')), + ('partition_table_file', ctx_obj.get('partition_table_file')), + ('esptool_args', ctx_obj.get('esptool_args')), + ('esptool_write_args', ctx_obj.get('esptool_write_args')), + ('esptool_read_args', ctx_obj.get('esptool_read_args')), + ('esptool_erase_args', ctx_obj.get('esptool_erase_args')), + ): + if value is not None and value != (): + kwargs[key] = value + if spi_flash_sec_size is not None: + kwargs['spi_flash_sec_size'] = spi_flash_sec_size + return kwargs - parser = argparse.ArgumentParser('ESP-IDF OTA Partitions Tool') - parser.add_argument('--quiet', '-q', help='suppress stderr messages', action='store_true') - parser.add_argument('--esptool-args', help='additional main arguments for esptool', nargs='+') - parser.add_argument( - '--esptool-write-args', help='additional subcommand arguments for esptool write-flash', nargs='+' - ) - parser.add_argument('--esptool-read-args', help='additional subcommand arguments for esptool read-flash', nargs='+') - parser.add_argument( - '--esptool-erase-args', help='additional subcommand arguments for esptool erase-region', nargs='+' - ) +def _resolve_ota_id(slot, name): + if name is not None: + return name + if slot is not None: + return slot + return None - # There are two possible sources for the partition table: a device attached to the host - # or a partition table CSV/binary file. These sources are mutually exclusive. - parser.add_argument('--port', '-p', help='port where the device to read the partition table from is attached') - - parser.add_argument('--baud', '-b', help='baudrate to use', type=int) - - parser.add_argument('--partition-table-offset', '-o', help='offset to read the partition table from', type=str) - - parser.add_argument( - '--partition-table-file', - '-f', - help='file (CSV/binary) to read the partition table from; ' - 'overrides device attached to specified port as the partition table source when defined', - ) - - subparsers = parser.add_subparsers(dest='operation', help='run otatool -h for additional help') - - spi_flash_sec_size = argparse.ArgumentParser(add_help=False) - spi_flash_sec_size.add_argument('--spi-flash-sec-size', help='value of SPI_FLASH_SEC_SIZE macro', type=str) - - # Specify the supported operations - subparsers.add_parser('read_otadata', help='read otadata partition', parents=[spi_flash_sec_size]) - subparsers.add_parser('erase_otadata', help='erase otadata partition') - - slot_or_name_parser = argparse.ArgumentParser(add_help=False) - slot_or_name_parser_args = slot_or_name_parser.add_mutually_exclusive_group() - slot_or_name_parser_args.add_argument('--slot', help='slot number of the ota partition', type=int) - slot_or_name_parser_args.add_argument('--name', help='name of the ota partition') - - subparsers.add_parser( - 'switch_ota_partition', help='switch otadata partition', parents=[slot_or_name_parser, spi_flash_sec_size] - ) - - read_ota_partition_subparser = subparsers.add_parser( - 'read_ota_partition', help='read contents of an ota partition', parents=[slot_or_name_parser] - ) - read_ota_partition_subparser.add_argument( - '--output', help='file to write the contents of the ota partition to', required=True - ) - - write_ota_partition_subparser = subparsers.add_parser( - 'write_ota_partition', help='write contents to an ota partition', parents=[slot_or_name_parser] - ) - write_ota_partition_subparser.add_argument('--input', help='file whose contents to write to the ota partition') - - subparsers.add_parser( - 'erase_ota_partition', help='erase contents of an ota partition', parents=[slot_or_name_parser] - ) - - args = parser.parse_args() - - quiet = args.quiet - - # No operation specified, display help and exit - if args.operation is None: - if not quiet: - parser.print_help() - sys.exit(1) - - target_args = {} - - if args.port: - target_args['port'] = args.port - - if args.partition_table_file: - target_args['partition_table_file'] = args.partition_table_file - - if args.partition_table_offset: - target_args['partition_table_offset'] = int(args.partition_table_offset, 0) - - try: - if args.spi_flash_sec_size: - target_args['spi_flash_sec_size'] = int(args.spi_flash_sec_size, 0) - except AttributeError: - pass - - if args.esptool_args: - target_args['esptool_args'] = args.esptool_args - - if args.esptool_write_args: - target_args['esptool_write_args'] = args.esptool_write_args - - if args.esptool_read_args: - target_args['esptool_read_args'] = args.esptool_read_args - - if args.esptool_erase_args: - target_args['esptool_erase_args'] = args.esptool_erase_args - - if args.baud: - target_args['baud'] = args.baud - - target = OtatoolTarget(**target_args) - - # Create the operation table and execute the operation - common_args = {'target': target} - - ota_id = [] - - try: - if args.name is not None: - ota_id = ['name'] - else: - if args.slot is not None: - ota_id = ['slot'] - except AttributeError: - pass +def _run_operation(operation, target, quiet=False, **op_kwargs): otatool_ops = { 'read_otadata': (_read_otadata, []), 'erase_otadata': (_erase_otadata, []), - 'switch_ota_partition': (_switch_ota_partition, ota_id), - 'read_ota_partition': (_read_ota_partition, ['output'] + ota_id), - 'write_ota_partition': (_write_ota_partition, ['input'] + ota_id), - 'erase_ota_partition': (_erase_ota_partition, ota_id), + 'switch_ota_partition': (_switch_ota_partition, ['ota_id']), + 'read_ota_partition': (_read_ota_partition, ['ota_id', 'output']), + 'write_ota_partition': (_write_ota_partition, ['ota_id', 'input_file']), + 'erase_ota_partition': (_erase_ota_partition, ['ota_id']), } - (op, op_args) = otatool_ops[args.operation] - - for op_arg in op_args: - common_args.update({op_arg: vars(args)[op_arg]}) - - try: - common_args['ota_id'] = common_args.pop('name') - except KeyError: - try: - common_args['ota_id'] = common_args.pop('slot') - except KeyError: - pass + op, op_arg_names = otatool_ops[operation] + common_args = {'target': target} + for op_arg in op_arg_names: + common_args[op_arg] = op_kwargs[op_arg] if quiet: - # If exceptions occur, suppress and exit quietly try: op(**common_args) except Exception: @@ -411,5 +310,187 @@ def main(): op(**common_args) +def _slot_or_name_options(func): + decorators = [ + click.option( + '--slot', + type=int, + cls=MutuallyExclusiveOption, + exclusive_with=['name'], + help='slot number of the ota partition', + ), + click.option( + '--name', + cls=MutuallyExclusiveOption, + exclusive_with=['slot'], + help='name of the ota partition', + ), + ] + for decorator in reversed(decorators): + func = decorator(func) + return func + + +@click.group( + cls=EspRichGroup, + invoke_without_command=True, + context_settings={'help_option_names': ['-h', '--help']}, + help='ESP-IDF OTA Partitions Tool', +) +@click.option('--quiet', '-q', is_flag=True, help='suppress status messages') +@click.option( + '--esptool-args', + multiple=True, + cls=OptionEatAll, + type=str, + help='additional main arguments for esptool', +) +@click.option( + '--esptool-write-args', + multiple=True, + cls=OptionEatAll, + type=str, + help='additional subcommand arguments for esptool write-flash', +) +@click.option( + '--esptool-read-args', + multiple=True, + cls=OptionEatAll, + type=str, + help='additional subcommand arguments for esptool read-flash', +) +@click.option( + '--esptool-erase-args', + multiple=True, + cls=OptionEatAll, + type=str, + help='additional subcommand arguments for esptool erase-region', +) +@click.option('--port', '-p', help='port where the device to read the partition table from is attached') +@click.option('--baud', '-b', type=int, help='baudrate to use') +@click.option('--partition-table-offset', '-o', type=AnyIntType(), help='offset to read the partition table from') +@click.option( + '--partition-table-file', + '-f', + type=click.Path(), + help='file (CSV/binary) to read the partition table from; ' + 'overrides device attached to specified port as the partition table source when defined', +) +@click.pass_context +def cli( + ctx, + quiet, + esptool_args, + esptool_write_args, + esptool_read_args, + esptool_erase_args, + port, + baud, + partition_table_offset, + partition_table_file, +): + if quiet: + log.set_verbosity(Verbosity.SILENT) + + ctx.ensure_object(dict) + ctx.obj.update( + { + 'quiet': quiet, + 'esptool_args': esptool_args, + 'esptool_write_args': esptool_write_args, + 'esptool_read_args': esptool_read_args, + 'esptool_erase_args': esptool_erase_args, + 'port': port, + 'baud': baud, + 'partition_table_offset': partition_table_offset, + 'partition_table_file': partition_table_file, + } + ) + + # Match pre-click argparse: no subcommand → help (unless quiet) and exit 1. + if ctx.invoked_subcommand is None: + if not quiet: + click.echo(ctx.get_help()) + sys.exit(1) + + +@cli.command('read_otadata', help='read otadata partition') +@click.option('--spi-flash-sec-size', type=AnyIntType(), help='value of SPI_FLASH_SEC_SIZE macro') +@click.pass_context +def read_otadata_cmd(ctx, spi_flash_sec_size): + target = OtatoolTarget(**_target_kwargs_from_ctx(ctx.obj, spi_flash_sec_size)) + _run_operation('read_otadata', target, quiet=ctx.obj.get('quiet')) + + +@cli.command('erase_otadata', help='erase otadata partition') +@click.pass_context +def erase_otadata_cmd(ctx): + target = OtatoolTarget(**_target_kwargs_from_ctx(ctx.obj)) + _run_operation('erase_otadata', target, quiet=ctx.obj.get('quiet')) + + +@cli.command('switch_ota_partition', help='switch otadata partition') +@_slot_or_name_options +@click.option('--spi-flash-sec-size', type=AnyIntType(), help='value of SPI_FLASH_SEC_SIZE macro') +@click.pass_context +def switch_ota_partition_cmd(ctx, slot, name, spi_flash_sec_size): + ota_id = _resolve_ota_id(slot, name) + if ota_id is None: + # Under --quiet, match pre-click: silent exit 2 (exception was swallowed). + if ctx.obj.get('quiet'): + sys.exit(2) + log.die('Partition to switch to should be defined using --slot OR --name') + target = OtatoolTarget(**_target_kwargs_from_ctx(ctx.obj, spi_flash_sec_size)) + _run_operation('switch_ota_partition', target, quiet=ctx.obj.get('quiet'), ota_id=ota_id) + + +@cli.command('read_ota_partition', help='read contents of an ota partition') +@_slot_or_name_options +@click.option('--output', help='file to write the contents of the ota partition to', required=True) +@click.pass_context +def read_ota_partition_cmd(ctx, slot, name, output): + ota_id = _resolve_ota_id(slot, name) + if ota_id is None: + if ctx.obj.get('quiet'): + sys.exit(2) + log.die('OTA partition should be defined using --slot OR --name') + target = OtatoolTarget(**_target_kwargs_from_ctx(ctx.obj)) + _run_operation('read_ota_partition', target, quiet=ctx.obj.get('quiet'), ota_id=ota_id, output=output) + + +@cli.command('write_ota_partition', help='write contents to an ota partition') +@_slot_or_name_options +@click.option('--input', 'input_file', help='file whose contents to write to the ota partition') +@click.pass_context +def write_ota_partition_cmd(ctx, slot, name, input_file): + ota_id = _resolve_ota_id(slot, name) + if ota_id is None: + if ctx.obj.get('quiet'): + sys.exit(2) + log.die('OTA partition should be defined using --slot OR --name') + target = OtatoolTarget(**_target_kwargs_from_ctx(ctx.obj)) + _run_operation('write_ota_partition', target, quiet=ctx.obj.get('quiet'), ota_id=ota_id, input_file=input_file) + + +@cli.command('erase_ota_partition', help='erase contents of an ota partition') +@_slot_or_name_options +@click.pass_context +def erase_ota_partition_cmd(ctx, slot, name): + ota_id = _resolve_ota_id(slot, name) + if ota_id is None: + if ctx.obj.get('quiet'): + sys.exit(2) + log.die('OTA partition should be defined using --slot OR --name') + target = OtatoolTarget(**_target_kwargs_from_ctx(ctx.obj)) + _run_operation('erase_ota_partition', target, quiet=ctx.obj.get('quiet'), ota_id=ota_id) + + +def main(): + cli() + + if __name__ == '__main__': + from esp_pylib.excepthook import install_exception_reporting + + install_exception_reporting() main() diff --git a/components/efuse/efuse_table_gen.py b/components/efuse/efuse_table_gen.py index f9f8112d6aa..6bce727403a 100755 --- a/components/efuse/efuse_table_gen.py +++ b/components/efuse/efuse_table_gen.py @@ -14,11 +14,23 @@ import re import sys from datetime import datetime +from esp_pylib.errors import FatalError +from esp_pylib.excepthook import install_exception_reporting +from esp_pylib.logger import Verbosity +from esp_pylib.logger import log +from rich.markup import escape + __version__ = '1.0' -quiet = False max_blk_len = 256 idf_target = 'esp32' +quiet = False + + +def status(msg: str) -> None: + """Print non-critical status to stderr (suppressed by --quiet).""" + if not quiet: + log.print(msg, file=sys.stderr) def get_copyright(): @@ -31,18 +43,6 @@ def get_copyright(): return copyright_str % datetime.today().year -def status(msg): - """Print status message to stderr""" - if not quiet: - critical(msg) - - -def critical(msg): - """Print critical message to stderr""" - sys.stderr.write(msg) - sys.stderr.write('\n') - - class FuseTable(list): def __init__(self): super().__init__() @@ -69,7 +69,7 @@ class FuseTable(list): except InputError as e: raise InputError(f'Error at line {line_no + 1}: {e}') except Exception: - critical(f'Unexpected error parsing line {line_no + 1}: {line}') + log.err(f'Unexpected error parsing line {line_no + 1}: {line}') raise # fix up missing bit_start @@ -126,7 +126,7 @@ class FuseTable(list): field_name = p.field_name + p.group if field_name != '' and len(duplicates.intersection([field_name])) != 0: fl_error = True - print( + log.err( f'Field at {p.field_name}, {p.efuse_block}, {p.bit_start}, {p.bit_count} ' 'have duplicate field_name' ) @@ -477,7 +477,7 @@ class FuseDefinition: def process_input_file(file, type_table): - status('Parsing efuse CSV input file ' + file.name + ' ...') + status('Parsing efuse CSV input file ' + escape(file.name) + ' ...') input_contents = file.read() table = FuseTable.from_csv(input_contents) status('Verifying efuse table...') @@ -509,27 +509,28 @@ def create_output_files(name, output_table, debug): # src files are the same if ckeck_md5_in_file(output_table.md5_digest_table, file_c_path) is False: - status('Creating efuse *.h file ' + file_h_path + ' ...') + status('Creating efuse *.h file ' + escape(file_h_path) + ' ...') output = output_table.to_header(file_name) with open(file_h_path, 'w', encoding='utf-8') as f: f.write(output) - status('Creating efuse *.c file ' + file_c_path + ' ...') + status('Creating efuse *.c file ' + escape(file_c_path) + ' ...') output = output_table.to_c_file(file_name, debug) with open(file_c_path, 'w', encoding='utf-8') as f: f.write(output) else: + # Always visible (same as pre-pylib plain print), even under --quiet. print('Source files do not require updating correspond to csv file.') def main(): - global quiet global max_blk_len global idf_target + global quiet parser = argparse.ArgumentParser(description='ESP32 eFuse Manager') parser.add_argument('--idf_target', '-t', help='Target chip type', default='esp32') - parser.add_argument('--quiet', '-q', help="Don't print non-critical status messages to stderr", action='store_true') + parser.add_argument('--quiet', '-q', help="Don't print non-critical status messages", action='store_true') parser.add_argument('--debug', help='Create header file with debug info', default=False, action='store_false') parser.add_argument('--info', help='Print info about range of used bits', default=False, action='store_true') parser.add_argument('--max_blk_len', help='Max number of bits in BLOCKs', type=int, default=256) @@ -543,11 +544,14 @@ def main(): idf_target = args.idf_target max_blk_len = args.max_blk_len + # Always print Max bits before applying --quiet (pre-pylib behavior). print(f'Max number of bits in BLK {max_blk_len:d}') if max_blk_len not in [256, 192, 128]: raise InputError(f'Unsupported block length = {max_blk_len:d}') quiet = args.quiet + if quiet: + log.set_verbosity(Verbosity.SILENT) debug = args.debug info = args.info @@ -569,7 +573,7 @@ def main(): return 0 -class InputError(RuntimeError): +class InputError(FatalError): def __init__(self, e): super().__init__(e) @@ -580,8 +584,8 @@ class ValidationError(InputError): if __name__ == '__main__': + install_exception_reporting() try: main() except InputError as e: - print(e, file=sys.stderr) - sys.exit(2) + log.die(str(e), exit_code=2) diff --git a/components/esp_system/check_system_init_priorities.py b/components/esp_system/check_system_init_priorities.py index 82fde48f2a8..bfc2ad10acc 100644 --- a/components/esp_system/check_system_init_priorities.py +++ b/components/esp_system/check_system_init_priorities.py @@ -14,6 +14,10 @@ import os import re import sys +from esp_pylib.excepthook import install_exception_reporting +from esp_pylib.logger import log +from rich.markup import escape + COMMENT_REGEX = re.compile(r'//.*?$|/\*.*?\*/', re.DOTALL | re.MULTILINE) ESP_SYSTEM_INIT_FN_REGEX = ( r'{macro}\((?P[a-zA-Z0-9_]+)\s*,\s*' @@ -80,10 +84,11 @@ def strip_comments(contents: str) -> str: def main() -> None: + install_exception_reporting() try: idf_path = os.environ['IDF_PATH'] except KeyError: - raise SystemExit('IDF_PATH must be set before running this script') + log.die('IDF_PATH must be set before running this script') has_errors = False startup_entries: list[StartupEntry] = [] @@ -108,10 +113,9 @@ def main() -> None: count_expected = len(re.findall(rf'\b{macro}\s*\(', file_contents_no_comments)) found = list(pattern.finditer(file_contents_no_comments)) if len(found) != count_expected: - print( - f'error: In {filename}, found {macro} {count_expected} time(s), ' + log.err( + f'In {filename}, found {macro} {count_expected} time(s), ' f'but regular expression matched {len(found)} time(s)', - file=sys.stderr, ) has_errors = True @@ -153,19 +157,16 @@ def main() -> None: # diff_lines = list(difflib.unified_diff(startup_entries_expected_lines, startup_entries_lines, lineterm='')) if len(diff_lines) > 0: - print( - ( - "error: startup order doesn't match the reference file. " - f'please update {STARTUP_ENTRIES_FILE} to match the actual startup order:' - ), - file=sys.stderr, + log.err( + "startup order doesn't match the reference file. " + f'please update {STARTUP_ENTRIES_FILE} to match the actual startup order:' ) for line in diff_lines: - print(f'{line}', file=sys.stderr) + log.print(escape(line), file=sys.stderr) has_errors = True if has_errors: - raise SystemExit(1) + sys.exit(1) if __name__ == '__main__': diff --git a/tools/activate.py b/tools/activate.py index 7c325705fb2..ecb4377ab04 100755 --- a/tools/activate.py +++ b/tools/activate.py @@ -1,16 +1,19 @@ #!/usr/bin/env python -# SPDX-FileCopyrightText: 2023-2024 Espressif Systems (Shanghai) CO LTD +# SPDX-FileCopyrightText: 2023-2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 """ -Ensure that the Python version used to initiate this script is appropriate for -running the ESP-IDF shell activation. The primary goal is to perform the minimum -necessary checks to identify the virtual environment with the default user Python -and then launch activate.py using the ESP-IDF Python virtual environment. +Bootstrap entry point for ESP-IDF shell activation. + +This script runs under the user's system Python (before the IDF virtualenv is +active). It only locates the IDF Python environment via idf_tools and re-invokes +tools/export_utils/activate_venv.py with that interpreter. CLI parsing and +esp-pylib / rich-click usage happen in activate_venv.py inside the venv. """ + import os import sys -from subprocess import run from subprocess import SubprocessError +from subprocess import run def die(msg: str) -> None: @@ -38,12 +41,26 @@ os.environ['IDF_PYTHON_ENV_PATH'] = idf_python_env_path os.environ['ESP_IDF_VERSION'] = idf_version if not os.path.exists(virtualenv_python): - die((f'ESP-IDF Python virtual environment "{virtualenv_python}" ' - f'not found. Please run the install script to set it up before ' - f'proceeding.')) + die( + f'ESP-IDF Python virtual environment "{virtualenv_python}" ' + f'not found. Please run the install script to set it up before ' + f'proceeding.' + ) try: - run([virtualenv_python, os.path.join(idf_path, 'tools', 'export_utils', 'activate_venv.py')] + sys.argv[1:], check=True, env=os.environ.copy()) + # Forward CLI args unchanged; activate_venv.py (rich-click) parses them in the venv. + run( + [virtualenv_python, os.path.join(idf_path, 'tools', 'export_utils', 'activate_venv.py')] + sys.argv[1:], + check=True, + env=os.environ.copy(), + ) except (OSError, SubprocessError) as e: - die('\n'.join(['Activation script failed', str(e), - 'To view detailed debug information, set ESP_IDF_EXPORT_DEBUG=1 and run the export script again.'])) + die( + '\n'.join( + [ + 'Activation script failed', + str(e), + 'To view detailed debug information, set ESP_IDF_EXPORT_DEBUG=1 and run the export script again.', + ] + ) + ) diff --git a/tools/bsasm.py b/tools/bsasm.py index 2eb4a63547d..e842abe6c5b 100755 --- a/tools/bsasm.py +++ b/tools/bsasm.py @@ -1,5 +1,5 @@ #!/usr/bin/env python -# SPDX-FileCopyrightText: 2024 Espressif Systems (Shanghai) CO LTD +# SPDX-FileCopyrightText: 2024-2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 import argparse import copy @@ -9,12 +9,12 @@ import re import struct import sys from typing import Any -from typing import cast -from typing import Dict -from typing import List -from typing import Tuple -from typing import Type from typing import TypedDict +from typing import cast + +from esp_pylib.errors import FatalError +from esp_pylib.excepthook import install_exception_reporting +from esp_pylib.logger import log # Increase this if you change the on-disk binary output format of the BitScrambler so it's # not compatible with previous versions @@ -55,21 +55,21 @@ class Opcode(TypedDict, total=False): ctr_val: int tgt: int h: int - l: int + l: int # noqa: E741 ctr_add: int ctl_cond_src: Input class Inst(TypedDict, total=False): op: Opcode - mux: Dict[int, Input] + mux: dict[int, Input] write: int read: int class Chipcfg(TypedDict, total=False): chipname: str - extra_instruction_groups: List[str] + extra_instruction_groups: list[str] support_all: bool @@ -112,7 +112,7 @@ class Chipcfg(TypedDict, total=False): # rewrite this and assign it to me - Jeroen) -def bsasm_parse(src: str) -> List[Element]: +def bsasm_parse(src: str) -> list[Element]: # Small hack: we trigger processing things on a newline. If a file is read without # a newline at the end of the last instruction, we'd erroneously ignore the last element. # Easiest way to fix it is to make sure the src always ends in a newline. @@ -126,7 +126,7 @@ def bsasm_parse(src: str) -> List[Element]: # We keep track of row/col for error reporting line = 0 column = 0 - elements: List[Element] = [] + elements: list[Element] = [] curr_element: Element = {} in_comment = False # True if we're anywhere between a # and a newline. for ch in src: @@ -169,11 +169,9 @@ def bsasm_parse(src: str) -> List[Element]: finish_element = True state = ST_AFTER_COMMA elif state == ST_AFTER_COMMA: - raise RuntimeError( - f'Line {line} column {column}: Empty subinstruction found' - ) + raise FatalError(f'Line {line} column {column}: Empty subinstruction found') elif state == ST_WH_PRE: - raise RuntimeError(f'Line {line} column {column}: Stray comma found') + raise FatalError(f'Line {line} column {column}: Stray comma found') elif ch == ':': # This indicates the current element is a label; a colon is not used anywhere else. if state == ST_ELEMENT: @@ -184,11 +182,9 @@ def bsasm_parse(src: str) -> List[Element]: finish_element = True state = ST_WH_PRE else: - raise RuntimeError( - f'Line {line} column {column}: Stray semicolon found' - ) + raise FatalError(f'Line {line} column {column}: Stray semicolon found') else: - raise RuntimeError(f'Line {line} column {column}: Stray semicolon found') + raise FatalError(f'Line {line} column {column}: Stray semicolon found') else: # Any other characters. if state == ST_ELEMENT: @@ -203,9 +199,7 @@ def bsasm_parse(src: str) -> List[Element]: # Handle starting and finishing of elements if start_element: if 'line' in curr_element: - raise RuntimeError( - f'Line {line} column {column}: Internal error: Element started twice!' - ) + raise FatalError(f'Line {line} column {column}: Internal error: Element started twice!') curr_element['line'] = line curr_element['column'] = column curr_element['text'] = ch @@ -213,9 +207,7 @@ def bsasm_parse(src: str) -> List[Element]: curr_element['is_label'] = False if finish_element: if 'line' not in curr_element: - raise RuntimeError( - f'Line {line} column {column}: Internal error: Element finished while none started' - ) + raise FatalError(f'Line {line} column {column}: Internal error: Element finished while none started') elements.append(curr_element) curr_element = {} @@ -231,8 +223,9 @@ def bsasm_parse(src: str) -> List[Element]: # Specific syntax error exception. Reports details about the element[s] to make debugging # assembly sources easier. -class bsasm_syntax_error(Exception): - def __new__(cls: Type['bsasm_syntax_error'], *args: str, **kwargs: str) -> 'bsasm_syntax_error': # noqa: F821 + +class bsasm_syntax_error(FatalError): + def __new__(cls: type['bsasm_syntax_error'], *args: str, **kwargs: str) -> 'bsasm_syntax_error': # noqa: F821 return cast(bsasm_syntax_error, super().__new__(cls)) def __init__(self, *args: Any) -> None: # noqa: F821 @@ -243,14 +236,11 @@ class bsasm_syntax_error(Exception): else: ele1 = args[0] ele2 = args[1] - message = args[1] - self.msg = 'Line {} col {}: "{}" and line {} col {}: "{}": {}'.format(ele1['line'], - ele1['column'], - ele1['text'], - ele2['line'], - ele2['column'], - ele2['text'], - message) + message = args[2] + self.msg = 'Line {} col {}: "{}" and line {} col {}: "{}": {}'.format( + ele1['line'], ele1['column'], ele1['text'], ele2['line'], ele2['column'], ele2['text'], message + ) + super().__init__(self.msg) def __str__(self) -> str: # noqa: F821 return self.msg @@ -260,12 +250,12 @@ class bsasm_syntax_error(Exception): class Meta_inst_def(TypedDict, total=False): op: str default: int - enum: Dict[str, int] + enum: dict[str, int] min: int max: int -meta_inst_defs: List[Meta_inst_def] = [ +meta_inst_defs: list[Meta_inst_def] = [ # RX_FETCH_MODE: 0 - on startup fill M0/M1, 1 - don't {'op': 'prefetch', 'default': 1, 'enum': {'true': 1, 'false': 0, '1': 1, '0': 0}}, # Amount of bytes read from input or written to output (depending on eof_on) @@ -288,11 +278,11 @@ def is_meta(ele: Element) -> bool: # Parse a config meta-instruction: check if the values are within range and convert from enums to values -def parse_meta_cfg(ele: Element) -> Tuple[str, int]: +def parse_meta_cfg(ele: Element) -> tuple[str, int]: words = ele['text'].lower().split(' ') meta_key = '' if len(words) != 3: - raise bsasm_syntax_error(ele, f'too many arguments to cfg statement') + raise bsasm_syntax_error(ele, 'too many arguments to cfg statement') for meta_inst_def in meta_inst_defs: if meta_inst_def['op'] == words[1]: if 'enum' in meta_inst_def: @@ -300,17 +290,13 @@ def parse_meta_cfg(ele: Element) -> Tuple[str, int]: meta_key = words[1] meta_value = meta_inst_def['enum'][words[2]] else: - raise bsasm_syntax_error( - ele, f'{words[2]} is not an allowed value for {words[1]}' - ) + raise bsasm_syntax_error(ele, f'{words[2]} is not an allowed value for {words[1]}') else: v = parse_val(ele, words[2], meta_inst_def['min'], meta_inst_def['max']) meta_key = words[1] meta_value = v if meta_key == '': - raise bsasm_syntax_error( - ele, f'{words[1]} is not a recognized meta-instruction' - ) + raise bsasm_syntax_error(ele, f'{words[1]} is not a recognized meta-instruction') return (meta_key, meta_value) @@ -340,8 +326,9 @@ def parse_output_range(ele: Element, text: str) -> range: # Resolve an input to a mux selection number and CTL_LUT_SEL/CTL_SRC_SEL/rel_addr # settings, if those need those to be in a specific state. -def parse_input(ele: Element, text: str, meta: Dict[str, int]) -> Input: +def parse_input(ele: Element, text: str, meta: dict[str, int]) -> Input: # Note that strings in the input def arrays need to be lower case. + # fmt: off inputs = ( # REG_MEM0 '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', '10', '11', '12', '13', '14', '15', @@ -373,6 +360,7 @@ def parse_input(ele: Element, text: str, meta: Dict[str, int]) -> Input: 'l0', 'l1', 'l2', 'l3', 'l4', 'l5', 'l6', 'l7', 'l8', 'l9', 'l10', 'l11', 'l12', 'l13', 'l14', 'l15', 'l16', 'l17', 'l18', 'l19', 'l20', 'l21', 'l22', 'l23', 'l24', 'l25', 'l26', 'l27', 'l28', 'l29', 'l30', 'l31', ) + # fmt: on # Note where in the counter reg / mem1 reg region the LUT starts, if enabled lut_starts = {8: 24, 16: 16, 32: 0} @@ -418,9 +406,7 @@ def parse_input(ele: Element, text: str, meta: Dict[str, int]) -> Input: raise bsasm_syntax_error(ele, f"'Input {text} is not valid.") if ret['input'] >= 64 and rel_addr: - raise bsasm_syntax_error( - ele, f"'LUT input {text} cannot be relatively addressed." - ) + raise bsasm_syntax_error(ele, f"'LUT input {text} cannot be relatively addressed.") if ret['input'] < 64: ret['flags']['rel_addr'] = rel_addr @@ -451,13 +437,11 @@ def check_input_compatible(in1: Input, in2: Input) -> None: # Returns a dictionary with the selected input in 'muxsel' plus a 'flags' dictionary. If a # rel_addr/lutsel/ctrsel key is in the 'flags' field, that bit must be set or cleared in # the instruction; if it's not set, the value of that bit doesn't matter for that input. -def parse_input_range(ele: Element, text: str, meta: Dict[str, int]) -> List[Input]: +def parse_input_range(ele: Element, text: str, meta: dict[str, int]) -> list[Input]: # Validate the range and split into start and optionally end fields b = re.findall(r'^([a-z0-9><=+]+)(?:\.\.([a-z0-9<>=+]+))?$', text) if not b: - raise bsasm_syntax_error( - ele, f'{text} not a valid input selection or range of input selections)' - ) + raise bsasm_syntax_error(ele, f'{text} not a valid input selection or range of input selections)') start = parse_input(ele, b[0][0], meta) if b[0][1] != '': end = parse_input(ele, b[0][1], meta) @@ -485,8 +469,13 @@ def parse_input_range(ele: Element, text: str, meta: Dict[str, int]) -> List[Inp pass elif math.floor(start['input'] / 32) != math.floor(end['input'] / 32): errtxt = f'{text} is not a valid range of input selections. ' - if 'flags' in start and 'lutsel' in start['flags'] and start['flags']['lutsel'] == 1 \ - and ('flags' not in end or 'lutsel' not in end['flags']) and end['input'] < 32: + if ( + 'flags' in start + and 'lutsel' in start['flags'] + and start['flags']['lutsel'] == 1 + and ('flags' not in end or 'lutsel' not in end['flags']) + and end['input'] < 32 + ): errtxt += 'Did you forget an L at the end of the range? (e.g. L0..31 instead of L0..L31)' else: errtxt += 'Try splitting up the range.' @@ -514,7 +503,7 @@ def parse_input_range(ele: Element, text: str, meta: Dict[str, int]) -> List[Inp r = range(start['input'], end['input'] + 1) else: r = range(start['input'], end['input'] - 1, -1) - ret: List[Input] = [] + ret: list[Input] = [] for i in r: n: Input = {'muxsel': i, 'flags': flags, 'ele': ele} ret.append(n) @@ -533,14 +522,12 @@ def parse_val(ele: Element, text: str, minimum: int, maximum: int) -> int: except ValueError: raise bsasm_syntax_error(ele, f"'{text}' is not an integer") if n < minimum or n > maximum: - raise bsasm_syntax_error( - ele, f"'{text}' is out of range [{minimum}..{maximum}]" - ) + raise bsasm_syntax_error(ele, f"'{text}' is out of range [{minimum}..{maximum}]") return n # Return an IP for a label text -def resolve_label(ele: Element, text: str, labels: Dict[str, int]) -> int: +def resolve_label(ele: Element, text: str, labels: dict[str, int]) -> int: if text in labels: return labels[text] # No match. We could technically also see if the label is a direct IP, but I think @@ -565,35 +552,31 @@ def check_chip_supports_inst(chipcfg: Chipcfg, instgroup: str, ele: Element) -> if instgroup not in chipcfg['extra_instruction_groups']: name = chipcfg['chipname'] - raise bsasm_syntax_error( - ele, f'Chip {name} does not support this instruction' - ) + raise bsasm_syntax_error(ele, f'Chip {name} does not support this instruction') def add_op_to_inst(inst: Inst, op: Opcode, ele: Element) -> None: if 'op' in inst: - raise bsasm_syntax_error( - inst['op']['ele'], ele, f'Cannot have multiple opcodes in one instruction' - ) + raise bsasm_syntax_error(inst['op']['ele'], ele, 'Cannot have multiple opcodes in one instruction') op['ele'] = ele inst['op'] = op # Takes the elements generated by the parse routine and converts it to a # representation of the bits in the Bitscrambler program. -def bsasm_assemble(elements: List[Element], chipcfg: Chipcfg) -> Tuple[List[Inst], Dict[str, int], List[int]]: +def bsasm_assemble(elements: list[Element], chipcfg: Chipcfg) -> tuple[list[Inst], dict[str, int], list[int]]: # This assembler uses two passes: the first finds and resolves global # stuff, the second one encodes the actual instructions. # Set the meta-instruction values to their defaults - meta: Dict[str, int] = {} + meta: dict[str, int] = {} for meta_inst_def in meta_inst_defs: meta[meta_inst_def['op']] = meta_inst_def['default'] # Pass 1a: find IPs for labels, mark meta instructions # ToDo: also resolve 'def' symbols here once we implement them ip = 0 - ip_for_label: Dict[str, int] = {} + ip_for_label: dict[str, int] = {} inst_is_meta = False inst_start = True for ele in elements: @@ -627,9 +610,7 @@ def bsasm_assemble(elements: List[Element], chipcfg: Chipcfg) -> Tuple[List[Inst (key, val) = parse_meta_cfg(ele) meta[key] = val if ele['more_in_instruction']: - raise bsasm_syntax_error( - ele, 'garbage after cfg statement detected' - ) + raise bsasm_syntax_error(ele, 'garbage after cfg statement detected') inst_start = not ele['more_in_instruction'] # Pass 1C: parse LUT data instructions. We do this after the meta instructions pass @@ -637,7 +618,7 @@ def bsasm_assemble(elements: List[Element], chipcfg: Chipcfg) -> Tuple[List[Inst # Note a lut can be written both as 'lut 1 2 3' as well as 'lut 1,2,3' so we need # to account for both cases. lut_minmax_vals = { - 8: (-128, 255), + 8: (-128, 255), 16: (-32768, 65537), 32: (-2147483648, 4294967296 - 1), } @@ -660,7 +641,7 @@ def bsasm_assemble(elements: List[Element], chipcfg: Chipcfg) -> Tuple[List[Inst # Pass 2: Parse any instructions valid_read_write = [0, 8, 16, 32] - insts: List[Inst] = [] + insts: list[Inst] = [] def_inst: Inst = {'mux': {}} inst = copy.deepcopy(def_inst) op: Opcode @@ -677,9 +658,7 @@ def bsasm_assemble(elements: List[Element], chipcfg: Chipcfg) -> Tuple[List[Inst i = 0 for out in outs: if out in inst['mux']: - raise bsasm_syntax_error( - ele, f'output {out} already set earlier in instruction' - ) + raise bsasm_syntax_error(ele, f'output {out} already set earlier in instruction') if len(ins) == 1: # set range input inst['mux'][out] = ins[0] @@ -692,25 +671,21 @@ def bsasm_assemble(elements: List[Element], chipcfg: Chipcfg) -> Tuple[List[Inst check_arg_ct(ele, words, 2) no = parse_val(ele, words[1], 0, 32) if no not in valid_read_write: - raise bsasm_syntax_error( - ele, f'{no} is not a valid amount of bits to write' - ) + raise bsasm_syntax_error(ele, f'{no} is not a valid amount of bits to write') inst['write'] = no elif words[0] == 'read': # Read x bits from input fifo check_arg_ct(ele, words, 2) no = parse_val(ele, words[1], 0, 32) if no not in valid_read_write: - raise bsasm_syntax_error( - ele, f'{no} is not a valid amount of bits to write' - ) + raise bsasm_syntax_error(ele, f'{no} is not a valid amount of bits to write') inst['read'] = no elif re.match('loop[ab]', words[0]): # LOOPc end_val ctr_add tgt check_arg_ct(ele, words, 4) op = {'op': OP_LOOP, 'ele': ele} op['c'] = 1 if words[0][4] == 'b' else 0 - op['end_val'] = parse_val(ele, words[1], -32768, 65535) & 0xffff + op['end_val'] = parse_val(ele, words[1], -32768, 65535) & 0xFFFF op['ctr_add'] = parse_val(ele, words[2], -16, 15) & 31 op['tgt'] = resolve_label(ele, words[3], ip_for_label) add_op_to_inst(inst, op, ele) @@ -725,7 +700,7 @@ def bsasm_assemble(elements: List[Element], chipcfg: Chipcfg) -> Tuple[List[Inst else: op['h'] = 1 if words[0][4] == 'h' else 0 op['l'] = 1 if words[0][4] == 'l' else 0 - op['ctr_add'] = parse_val(ele, words[1], -32768, 65535) & 0xffff + op['ctr_add'] = parse_val(ele, words[1], -32768, 65535) & 0xFFFF add_op_to_inst(inst, op, ele) elif re.match('if(n)?', words[0]): # IF[N] ctl_cond_src tgt @@ -745,7 +720,7 @@ def bsasm_assemble(elements: List[Element], chipcfg: Chipcfg) -> Tuple[List[Inst else: op['h'] = 1 if words[0][6] == 'h' else 0 op['l'] = 1 if words[0][6] == 'l' else 0 - op['ctr_add'] = parse_val(ele, words[1], -32768, 65535) & 0xffff + op['ctr_add'] = parse_val(ele, words[1], -32768, 65535) & 0xFFFF add_op_to_inst(inst, op, ele) elif re.match('ldcti[ab]([hl])?', words[0]): # LDCTIc[h|l] @@ -790,11 +765,7 @@ def bsasm_assemble(elements: List[Element], chipcfg: Chipcfg) -> Tuple[List[Inst else: raise bsasm_syntax_error(ele, 'unknown instruction') - if ( - (not ele['more_in_instruction']) - and (not ele['is_label']) - and (not ele['is_meta']) - ): + if (not ele['more_in_instruction']) and (not ele['is_label']) and (not ele['is_meta']): insts.append(inst) inst = copy.deepcopy(def_inst) return (insts, meta, lut) @@ -838,9 +809,9 @@ class bitstream: # This encodes all the instructions into binary. -def insts_to_binary(insts: List[Inst], meta: Dict[str, int], lut: list) -> bytearray: +def insts_to_binary(insts: list[Inst], meta: dict[str, int], lut: list) -> bytearray: if len(insts) > 8: - raise RuntimeError('Program has more than eight instructions.') + raise FatalError('Program has more than eight instructions.') ret = bytearray() # We need to reformat the LUT into 32-bit values, if not already in that format. @@ -955,13 +926,13 @@ def insts_to_binary(insts: List[Inst], meta: Dict[str, int], lut: list) -> bytea bits.add_bits(flags['ctrsel'], 1) bits.add_bits(flags['lutsel'], 1) if bits.size() != 257: - raise RuntimeError(f'Internal error: instruction size is {bits.size()}!') + raise FatalError(f'Internal error: instruction size is {bits.size()}!') # Pad instruction field to 36 bytes = 9 32-bit words bits.add_bits(0, 31) ret += bits.to_bytearray() for i in lut_reformatted: - ret += struct.pack(' bytea # Return the contents of a file def read_file(filename: str) -> str: try: - with open(filename, 'r') as f: - file_content = f.read() - except OSError: - print(f'Error opening {filename}: {sys.exc_info()[0]}') - return file_content + with open(filename) as f: + return f.read() + except OSError as e: + raise FatalError(f'Error opening {filename}: {e}') from e # Write a bytestring to a file @@ -983,29 +953,35 @@ def write_file(filename: str, data: bytearray) -> None: if __name__ == '__main__': - parser = argparse.ArgumentParser( - prog=sys.argv[0], - description='BitScrambler program assembler') - parser.add_argument('infile', help='File name of assembly source to be assembled into a binary') - parser.add_argument('outfile', help='File name of output binary', nargs='?', default=argparse.SUPPRESS) - parser.add_argument('-c', help='Set chip capabilities json file; if set, returns an error when \ - an unsupported instruction is assembled', default=argparse.SUPPRESS) - args = parser.parse_args() + install_exception_reporting() + try: + parser = argparse.ArgumentParser(prog=sys.argv[0], description='BitScrambler program assembler') + parser.add_argument('infile', help='File name of assembly source to be assembled into a binary') + parser.add_argument('outfile', help='File name of output binary', nargs='?', default=argparse.SUPPRESS) + parser.add_argument( + '-c', + help='Set chip capabilities json file; if set, returns an error when \ + an unsupported instruction is assembled', + default=argparse.SUPPRESS, + ) + args = parser.parse_args() - chipcfg = Chipcfg() - if 'c' in args: - with open(args.c) as chipcfg_json: - chipcfg = json.load(chipcfg_json) - else: - chipcfg = {'chipname': 'chip', 'extra_instruction_groups': [], 'support_all': True} + chipcfg = Chipcfg() + if 'c' in args: + with open(args.c) as chipcfg_json: + chipcfg = json.load(chipcfg_json) + else: + chipcfg = {'chipname': 'chip', 'extra_instruction_groups': [], 'support_all': True} - if 'outfile' in args: - outfile = args.outfile - else: - outfile = re.sub('.bsasm', '', args.infile) + '.bsbin' - asm = read_file(args.infile) - tokens = bsasm_parse(asm) - insts, meta, lut = bsasm_assemble(tokens, chipcfg) - out_data = insts_to_binary(insts, meta, lut) - write_file(outfile, out_data) - print(f'Written {len(insts)} instructions and {len(lut)} 32-bit words of LUT.') + if 'outfile' in args: + outfile = args.outfile + else: + outfile = re.sub('.bsasm', '', args.infile) + '.bsbin' + asm = read_file(args.infile) + tokens = bsasm_parse(asm) + insts, meta, lut = bsasm_assemble(tokens, chipcfg) + out_data = insts_to_binary(insts, meta, lut) + write_file(outfile, out_data) + log.print(f'Written {len(insts)} instructions and {len(lut)} 32-bit words of LUT.') + except FatalError as e: + log.die(str(e), exit_code=2) diff --git a/tools/check_python_dependencies.py b/tools/check_python_dependencies.py index 738ad0975a7..21f8511baf3 100755 --- a/tools/check_python_dependencies.py +++ b/tools/check_python_dependencies.py @@ -1,6 +1,6 @@ #!/usr/bin/env python # -# SPDX-FileCopyrightText: 2018-2025 Espressif Systems (Shanghai) CO LTD +# SPDX-FileCopyrightText: 2018-2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 import argparse import os @@ -23,6 +23,10 @@ from importlib.metadata import PackageNotFoundError from importlib.metadata import requires as _requires from importlib.metadata import version as _version +from esp_pylib.excepthook import install_exception_reporting +from esp_pylib.logger import log +from rich.markup import escape + PYTHON_PACKAGE_RE = re.compile(r'[^<>=~]+') @@ -69,6 +73,7 @@ def get_requires(name: str) -> list | None: if __name__ == '__main__': + install_exception_reporting() parser = argparse.ArgumentParser(description='ESP-IDF Python package dependency checker') parser.add_argument( '--requirements', @@ -104,14 +109,12 @@ if __name__ == '__main__': elif con.startswith('-e') and '#egg=' in con: con_m = re.search(r'#egg=([^\s]+)', con) if not con_m: - print(f'Malformed input. Cannot find name in {con}') - sys.exit(1) + log.die(escape(f'Malformed input. Cannot find name in {con}')) con = con_m[1] name_m = PYTHON_PACKAGE_RE.search(con) if not name_m: - print(f'Malformed input. Cannot find name in {con}') - sys.exit(1) + log.die(escape(f'Malformed input. Cannot find name in {con}')) constr_dict[name_m[0]] = con.partition(' #')[0] # remove comments not_satisfied = [] # in string form which will be printed @@ -178,26 +181,30 @@ if __name__ == '__main__': ) if len(not_satisfied) > 0: - print('The following Python requirements are not satisfied:') - print(os.linesep.join(not_satisfied)) + # Header first (same order as before pylib), then details. Escape Rich markup so + # requirement extras like package[extra] print literally. Keep on stdout to match + # pre-migration stream (callers may capture stdout only). + log.print('The following Python requirements are not satisfied:') + log.print(escape(os.linesep.join(not_satisfied))) if 'IDF_PYTHON_ENV_PATH' in os.environ: # We are running inside a private virtual environment under IDF_TOOLS_PATH, # ask the user to run install.bat again. install_script = 'install.bat' if sys.platform == 'win32' else 'install.sh' - print(f'To install the missing packages, please run "{install_script}"') + log.print(f'To install the missing packages, please run "{install_script}"') else: - print( + log.print( 'Please follow the instructions found in the "Set up the tools" section of ' 'ESP-IDF Getting Started Guide.' ) - print('Diagnostic information:') + log.print('Diagnostic information:') idf_python_env_path = os.environ.get('IDF_PYTHON_ENV_PATH') - print(' IDF_PYTHON_ENV_PATH: {}'.format(idf_python_env_path or '(not set)')) - print(f' Python interpreter used: {sys.executable}') + log.print(escape(f' IDF_PYTHON_ENV_PATH: {idf_python_env_path or "(not set)"}')) + log.print(escape(f' Python interpreter used: {sys.executable}')) if not idf_python_env_path or idf_python_env_path not in sys.executable: - print(' Warning: python interpreter not running from IDF_PYTHON_ENV_PATH') - print(' PATH: {}'.format(os.getenv('PATH'))) + # Keep on stdout with the rest of the diagnostic block (pre-pylib stream). + log.print(' Warning: python interpreter not running from IDF_PYTHON_ENV_PATH') + log.print(escape(f' PATH: {os.getenv("PATH")}')) sys.exit(1) - print('Python requirements are satisfied.') + log.print('Python requirements are satisfied.') diff --git a/tools/export_utils/activate_venv.py b/tools/export_utils/activate_venv.py index 56be2ea5dec..a8490729259 100644 --- a/tools/export_utils/activate_venv.py +++ b/tools/export_utils/activate_venv.py @@ -1,53 +1,22 @@ -# SPDX-FileCopyrightText: 2023-2025 Espressif Systems (Shanghai) CO LTD +# SPDX-FileCopyrightText: 2023-2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 -import argparse import os import sys +from types import SimpleNamespace from typing import Any -from typing import Dict -from console_output import CONSOLE_STDERR -from console_output import CONSOLE_STDOUT -from console_output import debug -from console_output import die -from console_output import eprint -from console_output import oprint +import rich_click as click +from console_output import configure_output from console_output import status_message -from console_output import warn +from esp_pylib.excepthook import install_exception_reporting +from esp_pylib.logger import log +from rich.markup import escape from shell_types import SHELL_CLASSES from shell_types import SUPPORTED_SHELLS from utils import conf from utils import run_cmd -def parse_arguments() -> argparse.Namespace: - parser = argparse.ArgumentParser( - prog='activate', - description='Activate ESP-IDF environment', - epilog='On Windows, run `python activate.py` to execute this script in the current terminal window.', - ) - parser.add_argument( - '-s', - '--shell', - metavar='SHELL', - default=os.environ.get('ESP_IDF_SHELL', 'detect'), - help='Explicitly specify shell to start. For example bash, zsh, powershell.exe, cmd.exe', - ) - parser.add_argument('-l', '--list', action='store_true', help=('List supported shells.')) - parser.add_argument('-e', '--export', action='store_true', help=('Generate commands to run in the terminal.')) - parser.add_argument('-n', '--no-color', action='store_true', help=('Disable ANSI color escape sequences.')) - parser.add_argument( - '-d', - '--debug', - action='store_true', - default=bool(os.environ.get('ESP_IDF_EXPORT_DEBUG')), - help=('Enable debug information.'), - ) - parser.add_argument('-q', '--quiet', action='store_true', help=('Suppress all output.')) - - return parser.parse_args() - - @status_message('Checking python version', rv_on_ok=True) def check_python_version() -> str: # Check the Python version within a virtual environment @@ -72,7 +41,7 @@ def get_deactivate_cmd() -> str: @status_message('Establishing a new ESP-IDF environment') -def get_idf_env() -> Dict[str, str]: +def get_idf_env() -> dict[str, str]: # Get ESP-IDF system environment variables extra_paths_list = [ os.path.join('components', 'espcoredump'), @@ -84,7 +53,7 @@ def get_idf_env() -> Dict[str, str]: stdout = run_cmd(cmd) # idf_tools.py might not export certain environment variables if they are already set - idf_env: Dict[str, Any] = { + idf_env: dict[str, Any] = { 'IDF_PATH': os.environ['IDF_PATH'], 'ESP_IDF_VERSION': os.environ['ESP_IDF_VERSION'], 'IDF_PYTHON_ENV_PATH': os.environ['IDF_PYTHON_ENV_PATH'], @@ -95,7 +64,7 @@ def get_idf_env() -> Dict[str, str]: var, val = line.split('=') idf_env[var] = val except ValueError as e: - debug('\n'.join(['Output from `./tools/idf_tools.py export --format key-value`:', f'{stdout}'])) + log.debug('\n'.join(['Output from `./tools/idf_tools.py export --format key-value`:', f'{stdout}'])) raise ValueError( '\n'.join( [ @@ -117,7 +86,7 @@ def detect_shell(args: Any) -> str: import psutil if args.shell != 'detect': - debug(f'Shell explicitly stated: "{args.shell}"') + log.debug(f'Shell explicitly stated: "{args.shell}"') return str(args.shell) current_pid = os.getpid() @@ -128,9 +97,9 @@ def detect_shell(args: Any) -> str: parent_cmdline = parent.cmdline() parent_exe = parent_cmdline[0].lstrip('-') parent_name = os.path.basename(parent_exe) - debug(f'Parent: pid: {parent_pid}, cmdline: {parent_cmdline}, exe: {parent_exe}, name: {parent_name}') + log.debug(f'Parent: pid: {parent_pid}, cmdline: {parent_cmdline}, exe: {parent_exe}, name: {parent_name}') if not parent_name.lower().startswith('python'): - detected_shell_name = parent_name + detected_shell_name = str(parent_name) break current_pid = parent_pid @@ -155,29 +124,56 @@ def print_uninstall_msg() -> Any: return msg -def main() -> None: - args = parse_arguments() +@click.command( + context_settings={'help_option_names': ['-h', '--help']}, + epilog='On Windows, run `python activate.py` to execute this script in the current terminal window.', +) +@click.option( + '-s', + '--shell', + metavar='SHELL', + default=os.environ.get('ESP_IDF_SHELL', 'detect'), + show_default=True, + help='Explicitly specify shell to start. For example bash, zsh, powershell.exe, cmd.exe', +) +@click.option('-l', '--list', 'list_shells', is_flag=True, help='List supported shells.') +@click.option('-e', '--export', is_flag=True, help='Generate commands to run in the terminal.') +@click.option('-n', '--no-color', is_flag=True, help='Disable ANSI color escape sequences.') +@click.option( + '-d', + '--debug', + 'debug_flag', + is_flag=True, + default=bool(os.environ.get('ESP_IDF_EXPORT_DEBUG')), + help='Enable debug information.', +) +@click.option('-q', '--quiet', is_flag=True, help='Suppress all output.') +def main(shell: str, list_shells: bool, export: bool, no_color: bool, debug_flag: bool, quiet: bool) -> None: + install_exception_reporting() - # Setup parsed arguments - CONSOLE_STDERR.no_color = args.no_color - CONSOLE_STDOUT.no_color = args.no_color - CONSOLE_STDERR.quiet = args.quiet - CONSOLE_STDOUT.quiet = args.quiet - # Fill config global holder - conf.ARGS = args + # Fill config global holder before configure_output + conf.ARGS = SimpleNamespace( + shell=shell, + list=list_shells, + export=export, + no_color=no_color, + debug=debug_flag, + quiet=quiet, + ) + configure_output(no_color=no_color, quiet=quiet, debug=debug_flag) - debug(f'command line: {sys.argv}') + log.debug(f'command line: {sys.argv}') if conf.ARGS.list: - oprint(SUPPORTED_SHELLS) + log.print(SUPPORTED_SHELLS) sys.exit() - eprint(f'[dark_orange]Activating ESP-IDF {conf.IDF_VERSION}') + log.print(f'[dark_orange]Activating ESP-IDF {conf.IDF_VERSION}', file=sys.stderr) if conf.IDF_PATH_OLD and conf.IDF_PATH != conf.IDF_PATH_OLD: - warn(f"IDF_PATH is changed from '{conf.IDF_PATH_OLD}' to '{conf.IDF_PATH}'.") + log.warn(f"IDF_PATH is changed from '{escape(conf.IDF_PATH_OLD)}' to '{escape(conf.IDF_PATH)}'.") else: - eprint(f"Setting IDF_PATH to '{conf.IDF_PATH}'.") + log.print(f"Setting IDF_PATH to '{escape(conf.IDF_PATH)}'.", file=sys.stderr) - debug(f'IDF_PYTHON_ENV_PATH {conf.IDF_PYTHON_ENV_PATH}') + log.debug(f'IDF_PYTHON_ENV_PATH {conf.IDF_PYTHON_ENV_PATH}') check_python_version() check_python_dependencies() @@ -188,19 +184,21 @@ def main() -> None: print_uninstall_msg() if detected_shell not in SHELL_CLASSES: - die(f'"{detected_shell}" shell is not among the supported options: "{SUPPORTED_SHELLS}"') + log.die(f'"{escape(str(detected_shell))}" shell is not among the supported options: "{SUPPORTED_SHELLS}"') - shell = SHELL_CLASSES[detected_shell](detected_shell, deactivate_cmd, new_esp_idf_env) + shell_obj = SHELL_CLASSES[detected_shell](detected_shell, deactivate_cmd, new_esp_idf_env) if conf.ARGS.export: - shell.export() + shell_obj.export() sys.exit() - eprint( - f'[dark_orange]Starting new \'{shell.shell}\' shell with ESP-IDF environment... (use "exit" command to quit)' + log.print( + f"[dark_orange]Starting new '{escape(str(shell_obj.shell))}' shell with ESP-IDF environment..." + ' (use "exit" command to quit)', + file=sys.stderr, ) - shell.spawn() - eprint('[dark_orange]ESP-IDF environment exited.') + shell_obj.spawn() + log.print('[dark_orange]ESP-IDF environment exited.', file=sys.stderr) if __name__ == '__main__': diff --git a/tools/export_utils/console_output.py b/tools/export_utils/console_output.py index af06abdc32d..dd9b2c6fd76 100644 --- a/tools/export_utils/console_output.py +++ b/tools/export_utils/console_output.py @@ -1,71 +1,62 @@ -# SPDX-FileCopyrightText: 2024 Espressif Systems (Shanghai) CO LTD +# SPDX-FileCopyrightText: 2024-2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 import sys +from collections.abc import Callable from typing import Any -from typing import Callable +from typing import cast +from rich.markup import escape from utils import conf try: - # The ESP-IDF virtual environment hasn't been verified yet, so see if the rich library + # The ESP-IDF virtual environment hasn't been verified yet, so see if esp-pylib # can be imported to display error and status messages nicely. - from rich.console import Console + from esp_pylib.logger import EspLog + from esp_pylib.logger import Verbosity + from esp_pylib.logger import log except ImportError as e: - sys.exit(f'error: Unable to import the rich module: {e}. Please execute the install script.') - -CONSOLE_STDERR = Console(stderr=True, width=255) -CONSOLE_STDOUT = Console(width=255) + sys.exit(f'error: Unable to import esp-pylib: {e}. Please execute the install script.') -def status_message(msg: str, msg_result: str='', rv_on_ok: bool=False, die_on_err: bool=True) -> Callable: +def configure_output(*, no_color: bool = False, quiet: bool = False, debug: bool = False) -> None: + # log is typed as EspLogBase; console options live on the default EspLog singleton. + # Only pass no_color=True when requested. False would override Rich's NO_COLOR detection. + options: dict[str, Any] = {'quiet': quiet} + if no_color: + options['no_color'] = True + cast(EspLog, log).set_console_options(**options) + if quiet: + log.set_verbosity(Verbosity.SILENT) + elif debug: + log.set_verbosity(Verbosity.VERBOSE) + else: + log.set_verbosity(Verbosity.NORMAL) + + +def status_message(msg: str, msg_result: str = '', rv_on_ok: bool = False, die_on_err: bool = True) -> Callable: def inner(func: Callable) -> Callable: def wrapper(*args: Any, **kwargs: Any) -> Any: - eprint(f'[dark_orange]*[/dark_orange] {msg} ... ', end='') + log.print(f'[dark_orange]*[/dark_orange] {escape(msg)} ... ', file=sys.stderr, end='') try: rv = func(*args, **kwargs) except Exception as e: - eprint('[red]FAILED[/red]') - if conf.ARGS.debug: + log.print('[red]FAILED[/red]', file=sys.stderr) + if conf.ARGS and conf.ARGS.debug: raise if not die_on_err: return None - die(str(e)) + log.die(escape(str(e))) if rv_on_ok: - eprint(f'[green]{rv}[/green]') + log.print(f'[green]{escape(str(rv))}[/green]', file=sys.stderr) elif msg_result: - eprint(f'[green]{msg_result}[/green]') + log.print(f'[green]{escape(msg_result)}[/green]', file=sys.stderr) else: - eprint('[green]OK[/green]') + log.print('[green]OK[/green]', file=sys.stderr) return rv + return wrapper + return inner - - -def err(*args: Any, **kwargs: Any) -> None: - CONSOLE_STDERR.print('[red]error[/red]: ', *args, **kwargs) # type: ignore - - -def warn(*args: Any, **kwargs: Any) -> None: - CONSOLE_STDERR.print('[yellow]warning[/yellow]: ', *args, **kwargs) # type: ignore - - -def debug(*args: Any, **kwargs: Any) -> None: - if not conf.ARGS.debug: - return - CONSOLE_STDERR.print('[green_yellow]debug[/green_yellow]: ', *args, **kwargs) # type: ignore - - -def die(*args: Any, **kwargs: Any) -> None: - err(*args, **kwargs) - sys.exit(1) - - -def eprint(*args: Any, **kwargs: Any) -> None: - CONSOLE_STDERR.print(*args, **kwargs) # type: ignore - - -def oprint(*args: Any, **kwargs: Any) -> None: - CONSOLE_STDOUT.print(*args, **kwargs) # type: ignore diff --git a/tools/export_utils/shell_types.py b/tools/export_utils/shell_types.py index 331c20a1a0c..8afabd56e15 100644 --- a/tools/export_utils/shell_types.py +++ b/tools/export_utils/shell_types.py @@ -1,4 +1,4 @@ -# SPDX-FileCopyrightText: 2024-2025 Espressif Systems (Shanghai) CO LTD +# SPDX-FileCopyrightText: 2024-2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 import base64 import getpass @@ -17,9 +17,8 @@ from tempfile import TemporaryDirectory from tempfile import gettempdir from typing import TextIO -from console_output import debug from console_output import status_message -from console_output import warn +from esp_pylib.logger import log from utils import conf from utils import run_cmd @@ -42,7 +41,7 @@ class Shell: self.tmp_dir_path = Path(gettempdir()) / f'esp_idf_activate_{username_safe}' except Exception as e: self.tmp_dir_path = Path(gettempdir()) / 'esp_idf_activate' - warn(f'Failed to get username with error: {e}. Using default temporary directory {self.tmp_dir_path}.') + log.warn(f'Failed to get username with error: {e}. Using default temporary directory {self.tmp_dir_path}.') if not conf.ARGS.debug and os.path.exists(self.tmp_dir_path): # Do not cleanup temporary directory when debugging @@ -55,7 +54,7 @@ class Shell: if current_time - file_creation_time > timedelta(hours=1): item.unlink() except Exception as e: - warn(f'Failed to clean temp activation directory with file {item}: {e}') + log.warn(f'Failed to clean temp activation directory with file {item}: {e}') self.tmp_dir_path.mkdir(parents=True, exist_ok=True) @@ -90,7 +89,7 @@ class UnixShell(Shell): with NamedTemporaryFile(dir=self.tmp_dir_path, delete=False, prefix='activate_') as fd: self.script_file_path = Path(fd.name) - debug(f'Temporary script file path: {self.script_file_path}') + log.debug(f'Temporary script file path: {self.script_file_path}') self.new_esp_idf_env['IDF_TOOLS_INSTALL_CMD'] = os.path.join(conf.IDF_PATH, 'install.sh') self.new_esp_idf_env['IDF_TOOLS_EXPORT_CMD'] = os.path.join(conf.IDF_PATH, 'export.sh') @@ -189,7 +188,7 @@ class ZshShell(UnixShell): # Create a temporary directory to use as ZDOTDIR tmpdir = TemporaryDirectory() tmpdir_path = Path(tmpdir.name) - debug(f'Temporary ZDOTDIR {tmpdir_path} with .zshrc file') + log.debug(f'Temporary ZDOTDIR {tmpdir_path} with .zshrc file') # Copy init script to the custom ZDOTDIR zshrc_path = tmpdir_path / '.zshrc' @@ -232,7 +231,7 @@ class PowerShell(Shell): with NamedTemporaryFile(dir=self.tmp_dir_path, delete=False, prefix='activate_', suffix='.ps1') as fd: self.script_file_path = Path(fd.name) - debug(f'Temporary script file path: {self.script_file_path}') + log.debug(f'Temporary script file path: {self.script_file_path}') self.new_esp_idf_env['IDF_TOOLS_INSTALL_CMD'] = os.path.join(conf.IDF_PATH, 'install.ps1') self.new_esp_idf_env['IDF_TOOLS_EXPORT_CMD'] = os.path.join(conf.IDF_PATH, 'export.ps1') @@ -284,7 +283,7 @@ class WinCmd(Shell): with NamedTemporaryFile(dir=self.tmp_dir_path, delete=False, prefix='activate_', suffix='.bat') as fd: self.script_file_path = Path(fd.name) - debug(f'Temporary script file path: {self.script_file_path}') + log.debug(f'Temporary script file path: {self.script_file_path}') self.new_esp_idf_env['IDF_TOOLS_INSTALL_CMD'] = os.path.join(conf.IDF_PATH, 'install.bat') self.new_esp_idf_env['IDF_TOOLS_EXPORT_CMD'] = os.path.join(conf.IDF_PATH, 'export.bat') diff --git a/tools/export_utils/utils.py b/tools/export_utils/utils.py index 2f6dd3f25e4..568a2304d2b 100644 --- a/tools/export_utils/utils.py +++ b/tools/export_utils/utils.py @@ -1,13 +1,9 @@ -# SPDX-FileCopyrightText: 2024 Espressif Systems (Shanghai) CO LTD +# SPDX-FileCopyrightText: 2024-2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 -import argparse import os -from subprocess import run from subprocess import SubprocessError +from subprocess import run from typing import Any -from typing import Dict -from typing import List -from typing import Optional class Config: @@ -15,6 +11,7 @@ class Config: Config serves as global hodler for variables used across modules It holds also arguments from command line """ + def __init__(self) -> None: self.IDF_PATH = os.environ['IDF_PATH'] self.IDF_PATH_OLD = os.environ['IDF_PATH_OLD'] @@ -22,14 +19,14 @@ class Config: self.IDF_PYTHON_ENV_PATH = os.environ['IDF_PYTHON_ENV_PATH'] self.IDF_TOOLS_PY = os.path.join(self.IDF_PATH, 'tools', 'idf_tools.py') self.IDF_PY = os.path.join(self.IDF_PATH, 'tools', 'idf.py') - self.ARGS: Optional[argparse.Namespace] = None + self.ARGS: Any | None = None # Global variable instance conf = Config() -def run_cmd(cmd: List[str], env: Optional[Dict[str, Any]]=None) -> str: +def run_cmd(cmd: list[str], env: dict[str, Any] | None = None) -> str: new_env = os.environ.copy() if env is not None: new_env.update(env) diff --git a/tools/gen_soc_caps_kconfig/gen_soc_caps_kconfig.py b/tools/gen_soc_caps_kconfig/gen_soc_caps_kconfig.py index 8d23a63421b..179ccb9bb24 100755 --- a/tools/gen_soc_caps_kconfig/gen_soc_caps_kconfig.py +++ b/tools/gen_soc_caps_kconfig/gen_soc_caps_kconfig.py @@ -1,6 +1,6 @@ #!/usr/bin/env python # -# SPDX-FileCopyrightText: 2021-2025 Espressif Systems (Shanghai) CO LTD +# SPDX-FileCopyrightText: 2021-2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 """ Generate Kconfig.soc_caps.in with defines from soc_caps.h @@ -9,7 +9,6 @@ Generate Kconfig.soc_caps.in with defines from soc_caps.h import argparse import inspect import io -import logging import os import sys from difflib import unified_diff @@ -18,6 +17,9 @@ from pathlib import Path from string import Template import pyparsing +from esp_pylib.excepthook import install_exception_reporting +from esp_pylib.logger import Verbosity +from esp_pylib.logger import log from pyparsing import CaselessLiteral from pyparsing import Char from pyparsing import Combine @@ -77,7 +79,7 @@ class KconfigWriter: def add_entry(self, name, entry_type, value): # type: (str, str, typing.Any) -> None if name in self.entries: - logging.info(f'Duplicate entry: {name}') + log.debug(f'Duplicate entry: {name}') return # Format values for kconfig @@ -206,7 +208,7 @@ def generate_defines(soc_caps_dir, filename, always_write): # type: (Path, str, try: res = parse_define(line) except pyparsing.ParseException: - logging.debug(f'Failed to parse: {line}') + log.debug(f'Failed to parse: {line}') continue if res.ignore_pragma: @@ -233,7 +235,7 @@ def generate_defines(soc_caps_dir, filename, always_write): # type: (Path, str, def get_defines(header_path): # type: (Path) -> list[str] defines = [] - logging.info(f'Reading macros from {header_path}...') + log.debug(f'Reading macros from {header_path}...') with open(header_path, encoding='utf-8') as f: output = f.read() @@ -246,6 +248,7 @@ def get_defines(header_path): # type: (Path) -> list[str] if __name__ == '__main__': + install_exception_reporting() parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('-d', '--dir', help='SoC caps folder paths, support wildcards', nargs='+', default=[]) parser.add_argument('-n', '--filename', nargs='?', default='*caps.h', help='SoC caps filename, support wildcards') @@ -258,14 +261,10 @@ if __name__ == '__main__': parser.add_argument('--always-write', help='Always generate new output files', action='store_true') args = parser.parse_args() - if not args.verbose: - log_level = logging.WARNING - elif args.verbose == 1: - log_level = logging.INFO + if args.verbose: + log.set_verbosity(Verbosity.VERBOSE) else: - log_level = logging.DEBUG - - logging.basicConfig(level=log_level) + log.set_verbosity(Verbosity.NORMAL) files_updated = [] writers = [] # type: typing.List[typing.Optional[KconfigWriter]] @@ -293,22 +292,19 @@ if __name__ == '__main__': configs[config_name] = writer.entries[config_name][0] if configs_with_differing_types: - print( + details = [ 'The following macro constants would be translated to config options with different types' ' for different targets (which may lead to unexpected behavior).' - ) - + ] for config_name in configs_with_differing_types: - print( + details.append( f' {config_name} has types' f'{", ".join(config_type for config_type in configs_with_differing_types[config_name])}' ) - - print('Please ensure all the macro constants will translate to the same config type for all targets.') - - sys.exit(1) + details.append('Please ensure all the macro constants will translate to the same config type for all targets.') + log.die('\n'.join(details)) files_updated = [writer.update_file() for writer in writers if writer is not None] - print(f'Updated {sum(files_updated)} files') + log.print(f'Updated {sum(files_updated)} files') sys.exit(all(files_updated)) diff --git a/tools/idf_size.py b/tools/idf_size.py index 9d68005f3b8..244a0937848 100755 --- a/tools/idf_size.py +++ b/tools/idf_size.py @@ -1,18 +1,22 @@ #!/usr/bin/env python # -# SPDX-FileCopyrightText: 2017-2025 Espressif Systems (Shanghai) CO LTD +# SPDX-FileCopyrightText: 2017-2026 Espressif Systems (Shanghai) CO LTD # # SPDX-License-Identifier: Apache-2.0 # import subprocess import sys +from esp_pylib.excepthook import install_exception_reporting +from esp_pylib.logger import log + if __name__ == '__main__': + install_exception_reporting() try: import esp_idf_size # noqa: F401 except ImportError: - print('WARNING: esp-idf-size not installed, please run the install script to install it', file=sys.stderr) - raise SystemExit(1) + log.warn('esp-idf-size not installed, please run the install script to install it') + sys.exit(1) sys.exit(subprocess.run([sys.executable, '-m', 'esp_idf_size'] + sys.argv[1:]).returncode) diff --git a/tools/idf_tools.py b/tools/idf_tools.py index 40b3cdacc1e..cccd064bcaa 100755 --- a/tools/idf_tools.py +++ b/tools/idf_tools.py @@ -505,8 +505,10 @@ def get_file_size_sha256(filename: str, block_size: int = 65536) -> tuple[int, s def report_progress(count: int, block_size: int, total_size: int) -> None: """ - Prints progress (count * block_size * 100 / total_size) to stdout. + Prints download progress to stdout. """ + if g.quiet or total_size <= 0: + return percent = int(count * block_size * 100 / total_size) percent = min(100, percent) sys.stdout.write(f'\r{percent}%') diff --git a/tools/mkdfu.py b/tools/mkdfu.py index 725ad994afa..6f87ef90cc4 100755 --- a/tools/mkdfu.py +++ b/tools/mkdfu.py @@ -1,6 +1,6 @@ #!/usr/bin/env python # -# SPDX-FileCopyrightText: 2020-2025 Espressif Systems (Shanghai) CO LTD +# SPDX-FileCopyrightText: 2020-2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 # # This program creates archives compatible with ESP32-S* ROM DFU implementation. @@ -24,6 +24,11 @@ except ImportError: # Only used for type annotations pass +from esp_pylib.constants import ESPRESSIF_VID +from esp_pylib.excepthook import install_exception_reporting +from esp_pylib.logger import log +from rich.markup import escape + try: from itertools import izip as zip # type: ignore # noqa: A004 except ImportError: @@ -90,7 +95,6 @@ DFUINFO_FILE = 'dfuinfo0.dat' # Structure which gets added at the end of the entire DFU file DFUSUFFIX_STRUCT = b' None writer = EspDfuWriter(args['output_file'], args['pid'], args['part_size']) - print('Adding flash chip parameters file with flash_size = {}'.format(args['flash_size'])) + log.note(f'Adding flash chip parameters file with flash_size = {args["flash_size"]}') writer.add_flash_params_file(args['flash_size']) for addr, f in args['files']: - print(f'Adding {f} at {addr:#x}') + log.note(f'Adding {escape(str(f))} at {addr:#x}') writer.add_file(addr, f) writer.finish() - print('"{}" has been written. You may proceed with DFU flashing.'.format(args['output_file'].name)) + log.note(f'"{escape(args["output_file"].name)}" has been written. You may proceed with DFU flashing.') if args['part_size'] % (4 * 1024) != 0: - print('WARNING: Partition size of DFU is not multiple of 4k (4096). You might get unexpected behavior.') + log.warn('Partition size of DFU is not multiple of 4k (4096). You might get unexpected behavior.') def main(): # type: () -> None + install_exception_reporting() parser = argparse.ArgumentParser() # Provision to add "info" command diff --git a/tools/mkuf2.py b/tools/mkuf2.py index ff2cc952629..1b96767bc0e 100755 --- a/tools/mkuf2.py +++ b/tools/mkuf2.py @@ -1,6 +1,6 @@ #!/usr/bin/env python # -# SPDX-FileCopyrightText: 2020-2025 Espressif Systems (Shanghai) CO LTD +# SPDX-FileCopyrightText: 2020-2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 # Module was moved to the esptool in ESP-IDF v5.2 and relicensed under GPL v2.0 license. import argparse @@ -9,13 +9,18 @@ import os import subprocess import sys +from esp_pylib.excepthook import install_exception_reporting +from esp_pylib.logger import log +from rich.markup import escape + def main() -> None: + install_exception_reporting() parser = argparse.ArgumentParser() def parse_chip_id(string: str) -> str: # compatibility layer with old script - print("DEPRECATED option '--chip-id'. Please consider using '--chip' instead") + log.warn("DEPRECATED option '--chip-id'. Please consider using '--chip' instead") # DO NOT add new IDs; they are now maintained in esptool. ids = { 0x1C5F21B0: 'esp32', @@ -90,10 +95,11 @@ def main() -> None: bin_selection = [json_content[b] for b in args.bin] flash_dic = dict((x['offset'], x['file']) for x in bin_selection) except KeyError: - print('Invalid binary was selected.') valid = [k if all(x in v for x in ('offset', 'file')) else None for k, v in json_content.items()] - print('Valid ones:', ' '.join(x for x in valid if x)) - exit(1) + # Keep pre-migration order and stdout stream. + log.print('Invalid binary was selected.') + log.print('Valid ones:', ' '.join(escape(x) for x in valid if x)) + sys.exit(1) else: flash_dic = json_content['flash_files'] @@ -124,7 +130,7 @@ def main() -> None: cmd.append('--md5-disable') cmd_str = ' '.join(cmd + files_flatten) - print(f'Executing: {cmd_str}') + log.note(f'Executing: {escape(cmd_str)}') sys.exit(subprocess.run(cmd + files_flatten).returncode) diff --git a/tools/split_paths_by_spaces.py b/tools/split_paths_by_spaces.py index d5e7a81dec8..888b346a7fe 100644 --- a/tools/split_paths_by_spaces.py +++ b/tools/split_paths_by_spaces.py @@ -1,7 +1,6 @@ #!/usr/bin/env python -# coding=utf-8 # -# SPDX-FileCopyrightText: 2021-2022 Espressif Systems (Shanghai) CO LTD +# SPDX-FileCopyrightText: 2021-2026 Espressif Systems (Shanghai) CO LTD # # SPDX-License-Identifier: Apache-2.0 # @@ -35,12 +34,17 @@ import textwrap import typing import unittest +from esp_pylib.excepthook import install_exception_reporting +from esp_pylib.logger import log +from rich.markup import escape + class PathSplitError(RuntimeError): pass def main() -> None: + install_exception_reporting() parser = argparse.ArgumentParser() parser.add_argument('--var-name', required=True, help='Name of CMake variable, for printing errors and warnings') parser.add_argument('in_variable', help='Input variable, may contain a mix of spaces and semicolons as separators') @@ -54,21 +58,37 @@ def main() -> None: ctx = dict(warnings=False) errors = False for part in semicolon_separated_parts: + def warning_cb(warning_str: str) -> None: - print('\n '.join( - textwrap.wrap('Warning: in CMake variable {}: {}'.format(args.var_name, warning_str), width=120, - break_on_hyphens=False)), file=sys.stderr) + log.warn( + '\n '.join( + textwrap.wrap( + f'in CMake variable {escape(args.var_name)}: {escape(warning_str)}', + width=120, + break_on_hyphens=False, + ) + ) + ) ctx['warnings'] = True try: paths += split_paths_by_spaces(part, warning_cb=warning_cb) except PathSplitError as e: - print('\n '.join(textwrap.wrap('Error: in CMake variable {}: {}'.format(args.var_name, str(e)), width=120, - break_on_hyphens=False)), file=sys.stderr) + log.print( + '\n '.join( + textwrap.wrap( + f'Error: in CMake variable {escape(args.var_name)}: {escape(str(e))}', + width=120, + break_on_hyphens=False, + ) + ), + file=sys.stderr, + ) errors = True if errors or ctx['warnings']: - print(textwrap.dedent(""" + log.print( + textwrap.dedent(""" Note: In ESP-IDF v5.0 and later, COMPONENT_DIRS and EXTRA_COMPONENT_DIRS should be defined as CMake lists, not as space separated strings. @@ -104,21 +124,27 @@ def main() -> None: (If you think these variables are defined correctly in your project and this message is not relevant, please report this as an issue.) - """), file=sys.stderr) + """), + file=sys.stderr, + ) - print('Diagnostic info: {} was invoked in {} with arguments: {}'.format( - sys.argv[0], os.getcwd(), sys.argv[1:] - ), file=sys.stderr) + log.print( + escape(f'Diagnostic info: {sys.argv[0]} was invoked in {os.getcwd()} with arguments: {sys.argv[1:]}'), + file=sys.stderr, + ) if errors: - raise SystemExit(1) + sys.exit(1) sys.stdout.write(';'.join(paths)) sys.stdout.flush() -def split_paths_by_spaces(src: str, path_exists_cb: typing.Callable[[str], bool] = os.path.exists, - warning_cb: typing.Optional[typing.Callable[[str], None]] = None) -> typing.List[str]: +def split_paths_by_spaces( + src: str, + path_exists_cb: typing.Callable[[str], bool] = os.path.exists, + warning_cb: typing.Callable[[str], None] | None = None, +) -> list[str]: if ' ' not in src: # no spaces, complete string should be the path return [src] @@ -130,12 +156,12 @@ def split_paths_by_spaces(src: str, path_exists_cb: typing.Callable[[str], bool] delayed_warnings = [] trimmed = src.lstrip(' ') if trimmed != src: - delayed_warnings.append("Path component '{}' contains leading spaces".format(src)) + delayed_warnings.append(f"Path component '{src}' contains leading spaces") src = trimmed trimmed = src.rstrip(' ') if trimmed != src: - delayed_warnings.append("Path component '{}' contains trailing spaces".format(src)) + delayed_warnings.append(f"Path component '{src}' contains trailing spaces") src = trimmed # Enumerate all possible ways to split the string src into paths by spaces. @@ -148,7 +174,7 @@ def split_paths_by_spaces(src: str, path_exists_cb: typing.Callable[[str], bool] parts = src.split(' ') num_spaces = len(parts) - 1 valid_ways_to_split = [] - all_ways_to_split = [selective_join(parts, i) for i in range(2 ** num_spaces)] + all_ways_to_split = [selective_join(parts, i) for i in range(2**num_spaces)] for paths_list in all_ways_to_split: nonempty_paths = list(filter(bool, paths_list)) if all(map(path_exists_or_empty, nonempty_paths)): @@ -162,29 +188,34 @@ def split_paths_by_spaces(src: str, path_exists_cb: typing.Callable[[str], bool] # Report warnings if warning_cb: if len(result) > 1: - warning_cb("Path component '{}' contains a space separator. It was automatically split into {}".format( - src, pprint.pformat(result) - )) + warning_cb( + f"Path component '{src}' contains a space separator. " + f'It was automatically split into {pprint.pformat(result)}' + ) for w in delayed_warnings: warning_cb(w) return result if num_candidates == 0: - raise PathSplitError(("Didn't find a valid way to split path '{}'. " - 'This error may be reported if one or more paths ' - "are separated with spaces, and at least one path doesn't exist.").format(src)) + raise PathSplitError( + f"Didn't find a valid way to split path '{src}'. " + 'This error may be reported if one or more paths ' + "are separated with spaces, and at least one path doesn't exist." + ) # if num_candidates > 1 - raise PathSplitError("Found more than one valid way to split path '{}':{}".format( - src, ''.join('\n\t- ' + pprint.pformat(p) for p in valid_ways_to_split) - )) + raise PathSplitError( + "Found more than one valid way to split path '{}':{}".format( + src, ''.join('\n\t- ' + pprint.pformat(p) for p in valid_ways_to_split) + ) + ) -def selective_join(parts: typing.List[str], n: int) -> typing.List[str]: +def selective_join(parts: list[str], n: int) -> list[str]: """ Given the list of N+1 strings, and an integer n in [0, 2**N - 1] range, - concatenate i-th and (i+1)-th string with space inbetween if bit i is not set in n. + concatenate i-th and (i+1)-th string with space in between if bit i is not set in n. Examples: selective_join(['a', 'b', 'c'], 0b00) == ['a b c'] selective_join(['a', 'b', 'c'], 0b01) == ['a', 'b c'] @@ -228,9 +259,9 @@ class SplitTests(unittest.TestCase): self.check_paths_concatenated('/absolute/path with more spaces') self.check_paths_concatenated('/absolute/path with spaces/one', '/absolute/path with spaces/two') - self.check_paths_concatenated('/absolute/path with spaces/one', - '/absolute/path with spaces/two', - '/absolute/path with spaces/three') + self.check_paths_concatenated( + '/absolute/path with spaces/one', '/absolute/path with spaces/two', '/absolute/path with spaces/three' + ) def test_split_paths_absolute_relative(self) -> None: self.check_paths_concatenated('/absolute/path/one', 'two') @@ -242,11 +273,11 @@ class SplitTests(unittest.TestCase): self.check_paths_concatenated('/absolute/path with spaces/one', 'two') def test_split_paths_ambiguous(self) -> None: - self.check_paths_concatenated_ambiguous('/absolute/path one', 'two', - additional_paths_exist=['/absolute/path', 'one']) + self.check_paths_concatenated_ambiguous( + '/absolute/path one', 'two', additional_paths_exist=['/absolute/path', 'one'] + ) - self.check_paths_concatenated_ambiguous('/path ', '/path', - additional_paths_exist=['/path /path']) + self.check_paths_concatenated_ambiguous('/path ', '/path', additional_paths_exist=['/path /path']) def test_split_paths_nonexistent(self) -> None: self.check_paths_concatenated_nonexistent('one', 'two') @@ -267,25 +298,24 @@ class SplitTests(unittest.TestCase): path_exists = self.path_exists_by_list(paths) - self.assertListEqual(paths, - split_paths_by_spaces(' /path', path_exists_cb=path_exists, warning_cb=add_warning)) + self.assertListEqual(paths, split_paths_by_spaces(' /path', path_exists_cb=path_exists, warning_cb=add_warning)) self.assertEqual(1, len(ctx['warnings'])) self.assertIn('leading', ctx['warnings'][0]) ctx['warnings'] = [] - self.assertListEqual(paths, - split_paths_by_spaces('/path ', path_exists_cb=path_exists, warning_cb=add_warning)) + self.assertListEqual(paths, split_paths_by_spaces('/path ', path_exists_cb=path_exists, warning_cb=add_warning)) self.assertEqual(1, len(ctx['warnings'])) self.assertIn('trailing', ctx['warnings'][0]) ctx['warnings'] = [] - self.assertListEqual(paths + paths, - split_paths_by_spaces('/path /path', path_exists_cb=path_exists, warning_cb=add_warning)) + self.assertListEqual( + paths + paths, split_paths_by_spaces('/path /path', path_exists_cb=path_exists, warning_cb=add_warning) + ) self.assertEqual(1, len(ctx['warnings'])) self.assertIn('contains a space separator', ctx['warnings'][0]) @staticmethod - def path_exists_by_list(paths_which_exist: typing.List[str]) -> typing.Callable[[str], bool]: + def path_exists_by_list(paths_which_exist: list[str]) -> typing.Callable[[str], bool]: """ Returns a function to check whether a path exists, similar to os.path.exists, but instead of checking for files on the real filesystem it considers only the paths provided in 'paths_which_exist' argument. @@ -305,8 +335,7 @@ class SplitTests(unittest.TestCase): return path_exists - def split_paths_concatenated_base(self, paths_to_concatentate: typing.List[str], - paths_existing: typing.List[str]) -> typing.List[str]: + def split_paths_concatenated_base(self, paths_to_concatentate: list[str], paths_existing: list[str]) -> list[str]: concatenated = ' '.join(paths_to_concatentate) path_exists = self.path_exists_by_list(paths_existing) return split_paths_by_spaces(concatenated, path_exists_cb=path_exists) @@ -316,17 +345,23 @@ class SplitTests(unittest.TestCase): paths_split = self.split_paths_concatenated_base(paths_to_concatentate=paths, paths_existing=paths) self.assertListEqual(paths, paths_split) - def check_paths_concatenated_ambiguous(self, *args: str, - additional_paths_exist: typing.Optional[typing.List[str]] = None) -> None: + def check_paths_concatenated_ambiguous(self, *args: str, additional_paths_exist: list[str] | None = None) -> None: paths = [*args] - self.assertRaises(PathSplitError, self.split_paths_concatenated_base, paths_to_concatentate=paths, - paths_existing=paths + (additional_paths_exist or [])) + self.assertRaises( + PathSplitError, + self.split_paths_concatenated_base, + paths_to_concatentate=paths, + paths_existing=paths + (additional_paths_exist or []), + ) - def check_paths_concatenated_nonexistent(self, *args: str, - additional_paths_exist: typing.List[str] = None) -> None: + def check_paths_concatenated_nonexistent(self, *args: str, additional_paths_exist: list[str] | None = None) -> None: paths = [*args] - self.assertRaises(PathSplitError, self.split_paths_concatenated_base, paths_to_concatentate=paths, - paths_existing=additional_paths_exist) + self.assertRaises( + PathSplitError, + self.split_paths_concatenated_base, + paths_to_concatentate=paths, + paths_existing=additional_paths_exist, + ) if __name__ == '__main__': diff --git a/tools/test_bsasm/conftest.py b/tools/test_bsasm/conftest.py new file mode 100644 index 00000000000..e0ea72ca5d9 --- /dev/null +++ b/tools/test_bsasm/conftest.py @@ -0,0 +1,32 @@ +# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD +# SPDX-License-Identifier: Apache-2.0 +import os +import typing + +import pytest + + +@pytest.fixture(autouse=True, scope='session') +def bsasm_terminal_env() -> typing.Generator[None, None, None]: + """Set terminal env so bsasm subprocesses produce consistent output. + + Same pattern as tools/test_mkdfu/conftest.py and tools/test_idf_py/conftest.py: + COLUMNS raises Rich's non-TTY default of 80, preventing most line wrapping. + NO_COLOR=1 strips ANSI escape codes. + + Current tests mainly check return codes and binary content, but bsasm.py uses + esp-pylib logging; keep this fixture so future string asserts stay stable. + """ + keys = ('COLUMNS', 'LINES', 'NO_COLOR', 'FORCE_COLOR', 'PY_COLORS', 'TERM') + saved = {k: os.environ.get(k) for k in keys} + os.environ['COLUMNS'] = '1000' + os.environ['LINES'] = '40' + os.environ['NO_COLOR'] = '1' + for k in ('FORCE_COLOR', 'PY_COLORS'): + os.environ.pop(k, None) + yield + for k, v in saved.items(): + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v diff --git a/tools/test_idf_tools/test_idf_tools_python_env.py b/tools/test_idf_tools/test_idf_tools_python_env.py index 3414a504ae4..179ef1357c5 100644 --- a/tools/test_idf_tools/test_idf_tools_python_env.py +++ b/tools/test_idf_tools/test_idf_tools_python_env.py @@ -1,4 +1,4 @@ -# SPDX-FileCopyrightText: 2022-2025 Espressif Systems (Shanghai) CO LTD +# SPDX-FileCopyrightText: 2022-2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 # NOTE: unittest is by default sorting tests based on their names, # so the order if which the tests are started may be different from @@ -15,7 +15,6 @@ import subprocess import sys import tempfile import unittest -from typing import List # noqa: F401 try: import idf_tools @@ -30,7 +29,7 @@ PYTHON_DIR_BACKUP = tempfile.mkdtemp() PYTHON_BINARY = os.path.join('Scripts', 'python.exe') if sys.platform == 'win32' else os.path.join('bin', 'python') REQ_SATISFIED = 'Python requirements are satisfied' # Python 3.8 and 3.9 has a different error message that does not include the "No package metadata was found for" part -REQ_MISSING = r'Package was not found and is required by the application: (No package metadata was found for )?{}' +REQ_MISSING = r'Package was not found and is\s+required by the application: (No package metadata was found for )?{}' REQ_CORE = '- {}'.format(os.path.join(IDF_PATH, 'tools', 'requirements', 'requirements.core.txt')) REQ_DOCS = '- {}'.format(os.path.join(IDF_PATH, 'tools', 'requirements', 'requirements.docs.txt')) CONSTR = 'Constraint file: {}'.format(os.path.join(TOOLS_DIR, 'espidf.constraints')) @@ -54,8 +53,16 @@ def tearDownModule(): # type: () -> None class BasePythonInstall(unittest.TestCase): - def run_tool(self, cmd): # type: (List[str]) -> str - ret = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=600) + def run_tool(self, cmd): # type: (list[str]) -> str + # Match tools/test_mkdfu/conftest.py: wide COLUMNS avoids Rich line wrapping mid-message, + # NO_COLOR strips ANSI so string asserts stay stable. + env = os.environ.copy() + env['COLUMNS'] = '1000' + env['LINES'] = '40' + env['NO_COLOR'] = '1' + env.pop('FORCE_COLOR', None) + env.pop('PY_COLORS', None) + ret = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=600, env=env) decoded_output = ret.stdout.decode('utf-8', 'ignore') with open(os.path.join(IDF_PATH, 'tools', 'test_idf_tools', 'test_python_env_logs.txt'), 'a+') as w: # stack() returns list of callers frame records. [1] represent caller of this function @@ -63,11 +70,11 @@ class BasePythonInstall(unittest.TestCase): w.write(decoded_output) return decoded_output - def run_idf_tools(self, args): # type: (List[str]) -> str + def run_idf_tools(self, args): # type: (list[str]) -> str cmd = [sys.executable, '../idf_tools.py'] + args return self.run_tool(cmd) - def run_in_venv(self, args): # type: (List[str]) -> str + def run_in_venv(self, args): # type: (list[str]) -> str _, _, python_venv, _ = idf_tools.get_python_env_path() cmd = [python_venv] + args return self.run_tool(cmd) diff --git a/tools/test_mkdfu/conftest.py b/tools/test_mkdfu/conftest.py new file mode 100644 index 00000000000..d7234fb6bcb --- /dev/null +++ b/tools/test_mkdfu/conftest.py @@ -0,0 +1,28 @@ +# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD +# SPDX-License-Identifier: Apache-2.0 +import os +import typing + +import pytest + + +@pytest.fixture(autouse=True, scope='session') +def mkdfu_terminal_env() -> typing.Generator[None, None, None]: + """Set terminal env so mkdfu subprocesses produce consistent output. + + COLUMNS raises Rich's non-TTY default of 80, preventing most line wrapping. + NO_COLOR=1 strips ANSI escape codes. + """ + keys = ('COLUMNS', 'LINES', 'NO_COLOR', 'FORCE_COLOR', 'PY_COLORS', 'TERM') + saved = {k: os.environ.get(k) for k in keys} + os.environ['COLUMNS'] = '1000' + os.environ['LINES'] = '40' + os.environ['NO_COLOR'] = '1' + for k in ('FORCE_COLOR', 'PY_COLORS'): + os.environ.pop(k, None) + yield + for k, v in saved.items(): + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v diff --git a/tools/test_mkdfu/test_mkdfu.py b/tools/test_mkdfu/test_mkdfu.py index 258c99d4804..70d41a2ac80 100755 --- a/tools/test_mkdfu/test_mkdfu.py +++ b/tools/test_mkdfu/test_mkdfu.py @@ -1,10 +1,8 @@ #!/usr/bin/env python -# -*- coding: utf-8 -*- # -# SPDX-FileCopyrightText: 2020-2022 Espressif Systems (Shanghai) CO LTD +# SPDX-FileCopyrightText: 2020-2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 -from __future__ import unicode_literals import collections import filecmp @@ -13,7 +11,6 @@ import os import shutil import sys import tempfile -import time import unittest import pexpect @@ -24,51 +21,46 @@ mkdfu_path = os.path.join(current_dir, '..', 'mkdfu.py') class TestMkDFU(unittest.TestCase): def common_test(self, json_input=None, file_args=[], output_to_compare=None, part_size=None): - ''' + """ - json_input - input JSON file compatible with mkdfu.py - used when not None - file_args - list of (address, path_to_file) tuples - output_to_compare - path to the file containing the expected output - tested when not None - part_size - partition size - used when not None - ''' + """ with tempfile.NamedTemporaryFile(delete=False) as f_out: self.addCleanup(os.unlink, f_out.name) - args = [mkdfu_path, 'write', - '-o', f_out.name, - '--pid', '2', - '--flash-size', '4MB'] + args = [mkdfu_path, 'write', '-o', f_out.name, '--pid', '2', '--flash-size', '4MB'] if part_size: args += ['--part-size', str(part_size)] if json_input: args += ['--json', json_input] for addr, f_path in file_args: args += [str(addr), f_path] - p = pexpect.spawn(sys.executable, args, timeout=10, encoding='utf-8') + env = os.environ.copy() + p = pexpect.spawn(sys.executable, args, timeout=10, encoding='utf-8', env=env, dimensions=(40, 1000)) self.addCleanup(p.terminate, force=True) - p.expect_exact('Adding flash chip parameters file with flash_size = 4MB') + p.expect(pexpect.EOF, timeout=30) + output = p.before or '' + + self.assertIn('Adding flash chip parameters file with flash_size = 4MB', output) for addr, f_path in sorted(file_args, key=lambda e: e[0]): - p.expect_exact('Adding {} at {}'.format(f_path, hex(addr))) + self.assertIn(f'Adding {f_path} at {addr:#x}', output) - p.expect_exact('"{}" has been written. You may proceed with DFU flashing.'.format(f_out.name)) - - # Need to wait for the process to end because the output file is closed when mkdfu exits. - # Do non-blocking wait instead of the blocking p.wait(): - for _ in range(10): - if not p.isalive(): - break - time.sleep(0.5) - else: - p.terminate() + self.assertIn(f'"{f_out.name}" has been written. You may proceed with DFU flashing.', output) if output_to_compare: - self.assertTrue(filecmp.cmp(f_out.name, os.path.join(current_dir, output_to_compare)), 'Output files are different') + self.assertTrue( + filecmp.cmp(f_out.name, os.path.join(current_dir, output_to_compare)), 'Output files are different' + ) class TestHelloWorldExample(TestMkDFU): - ''' + """ tests with images prepared in the "1" subdirectory - ''' + """ + def test_with_json(self): with tempfile.NamedTemporaryFile(mode='w', dir=os.path.join(current_dir, '1'), delete=False) as f: self.addCleanup(os.unlink, f.name) @@ -79,36 +71,39 @@ class TestHelloWorldExample(TestMkDFU): self.common_test(json_input=f.name, output_to_compare='1/dfu.bin') def test_without_json(self): - - self.common_test(file_args=[(0x1000, '1/1.bin'), - (0x8000, '1/2.bin'), - (0x10000, '1/3.bin')], - output_to_compare='1/dfu.bin') + self.common_test( + file_args=[(0x1000, '1/1.bin'), (0x8000, '1/2.bin'), (0x10000, '1/3.bin')], output_to_compare='1/dfu.bin' + ) def test_filenames(self): temp_dir = tempfile.mkdtemp(prefix='very_long_directory_name' * 8) self.addCleanup(shutil.rmtree, temp_dir, ignore_errors=True) - with tempfile.NamedTemporaryFile(prefix='ľščťžýáíéěř\u0420\u043e\u0441\u0441\u0438\u044f', - dir=temp_dir, - delete=False) as f: + with tempfile.NamedTemporaryFile( + prefix='ľščťžýáíéěř\u0420\u043e\u0441\u0441\u0438\u044f', dir=temp_dir, delete=False + ) as f: bootloader = f.name shutil.copyfile(os.path.join(current_dir, '1', '1.bin'), bootloader) - self.common_test(file_args=[(0x1000, bootloader), - (0x8000, os.path.join(current_dir, '1', '2.bin')), - (0x10000, os.path.join(current_dir, '1', '3.bin'))]) + self.common_test( + file_args=[ + (0x1000, bootloader), + (0x8000, os.path.join(current_dir, '1', '2.bin')), + (0x10000, os.path.join(current_dir, '1', '3.bin')), + ] + ) class TestSplit(TestMkDFU): - ''' + """ tests with images prepared in the "2" subdirectory "2/dfu.bin" was prepared with: mkdfu.py write --part-size 5 --pid 2 --flash-size 4MB -o 2/dfu.bin 0 bin where the content of "bin" is b"\xce" * 10 - ''' + """ + def test_split(self): temp_dir = tempfile.mkdtemp(dir=current_dir) self.addCleanup(shutil.rmtree, temp_dir, ignore_errors=True) @@ -117,9 +112,7 @@ class TestSplit(TestMkDFU): self.addCleanup(os.unlink, f.name) f.write(b'\xce' * 10) - self.common_test(file_args=[(0, f.name)], - part_size=5, - output_to_compare='2/dfu.bin') + self.common_test(file_args=[(0, f.name)], part_size=5, output_to_compare='2/dfu.bin') if __name__ == '__main__':