change(tools): Unified outputs of various standalone scripts (esp-pylib)

2026-10-01 18:50:34 +03:00 · 2026-09-09 09:43:59 +02:00
parent 22ade8b843
commit d6771a0608
20 changed files with 780 additions and 601 deletions
@@ -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'))
+    # Regular tool output (same stream as pre-pylib status()/print).
-    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():
+def _target_kwargs_from_ctx(ctx_obj, spi_flash_sec_size=None):
-    global quiet
+    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')
+def _resolve_ota_id(slot, name):
-    parser.add_argument('--esptool-args', help='additional main arguments for esptool', nargs='+')
+    if name is not None:
-    parser.add_argument(
+        return name
-        '--esptool-write-args', help='additional subcommand arguments for esptool write-flash', nargs='+'
+    if slot is not None:
-    )
+        return slot
-    parser.add_argument('--esptool-read-args', help='additional subcommand arguments for esptool read-flash', nargs='+')
+    return None
    parser.add_argument(
        '--esptool-erase-args', help='additional subcommand arguments for esptool erase-region', nargs='+'
    )
    # 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),
+        'switch_ota_partition': (_switch_ota_partition, ['ota_id']),
-        'read_ota_partition': (_read_ota_partition, ['output'] + ota_id),
+        'read_ota_partition': (_read_ota_partition, ['ota_id', 'output']),
-        'write_ota_partition': (_write_ota_partition, ['input'] + ota_id),
+        'write_ota_partition': (_write_ota_partition, ['ota_id', 'input_file']),
-        'erase_ota_partition': (_erase_ota_partition, ota_id),
+        'erase_ota_partition': (_erase_ota_partition, ['ota_id']),
    }
-    (op, op_args) = otatool_ops[args.operation]
+    op, op_arg_names = otatool_ops[operation]
-
+    common_args = {'target': target}
-    for op_arg in op_args:
+    for op_arg in op_arg_names:
-        common_args.update({op_arg: vars(args)[op_arg]})
+        common_args[op_arg] = op_kwargs[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
    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()
@@ -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)
+        log.die(str(e), exit_code=2)
        sys.exit(2)
@@ -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<func>[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(
+                log.err(
-                    f'error: In {filename}, found {macro} {count_expected} time(s), '
+                    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(
+        log.err(
-            (
+            "startup order doesn't match the reference file. "
-                "error: startup order doesn't match the reference file. "
+            f'please update {STARTUP_ENTRIES_FILE} to match the actual startup order:'
                f'please update {STARTUP_ENTRIES_FILE} to match the actual startup order:'
            ),
            file=sys.stderr,
        )
        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__':
@@ -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
+Bootstrap entry point for ESP-IDF shell activation.
-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
+This script runs under the user's system Python (before the IDF virtualenv is
-and then launch activate.py using the ESP-IDF Python virtual environment.
+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}" '
+    die(
-         f'not found. Please run the install script to set it up before '
+        f'ESP-IDF Python virtual environment "{virtualenv_python}" '
-         f'proceeding.'))
+        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),
+    die(
-                   'To view detailed debug information, set ESP_IDF_EXPORT_DEBUG=1 and run the export script again.']))
+        '\n'.join(
            [
                'Activation script failed',
                str(e),
                'To view detailed debug information, set ESP_IDF_EXPORT_DEBUG=1 and run the export script again.',
            ]
        )
    )
@@ -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(
+                raise FatalError(f'Line {line} column {column}: Empty subinstruction found')
                    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(
+                    raise FatalError(f'Line {line} column {column}: Stray semicolon found')
                        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(
+                raise FatalError(f'Line {line} column {column}: Internal error: Element started twice!')
                    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(
+                raise FatalError(f'Line {line} column {column}: Internal error: Element finished while none started')
                    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]
+            message = args[2]
-            self.msg = 'Line {} col {}: "{}" and line {} col {}: "{}": {}'.format(ele1['line'],
+            self.msg = 'Line {} col {}: "{}" and line {} col {}: "{}": {}'.format(
-                                                                                  ele1['column'],
+                ele1['line'], ele1['column'], ele1['text'], ele2['line'], ele2['column'], ele2['text'], message
-                                                                                  ele1['text'],
+            )
-                                                                                  ele2['line'],
+        super().__init__(self.msg)
                                                                                  ele2['column'],
                                                                                  ele2['text'],
                                                                                  message)
    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(
+                    raise bsasm_syntax_error(ele, f'{words[2]} is not an allowed value for {words[1]}')
                        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(
+        raise bsasm_syntax_error(ele, f'{words[1]} is not a recognized meta-instruction')
            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(
+        raise bsasm_syntax_error(ele, f"'LUT input {text} cannot be relatively addressed.")
            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(
+        raise bsasm_syntax_error(ele, f'{text} not a valid input selection or range of input selections)')
            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 \
+        if (
-                and ('flags' not in end or 'lutsel' not in end['flags']) and end['input'] < 32:
+            '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(
+        raise bsasm_syntax_error(ele, f"'{text}' is out of range [{minimum}..{maximum}]")
            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(
+        raise bsasm_syntax_error(ele, f'Chip {name} does not support this instruction')
            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(
+        raise bsasm_syntax_error(inst['op']['ele'], ele, 'Cannot have multiple opcodes in one instruction')
            inst['op']['ele'], ele, f'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(
+                    raise bsasm_syntax_error(ele, 'garbage after cfg statement detected')
                        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(
+                        raise bsasm_syntax_error(ele, f'output {out} already set earlier in instruction')
                            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(
+                    raise bsasm_syntax_error(ele, f'{no} is not a valid amount of bits to write')
                        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(
+                    raise bsasm_syntax_error(ele, f'{no} is not a valid amount of bits to write')
                        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 (
+            if (not ele['more_in_instruction']) and (not ele['is_label']) and (not ele['is_meta']):
                (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('<I', i & 0xffffffff)
+        ret += struct.pack('<I', i & 0xFFFFFFFF)
    return ret
@@ -969,11 +940,10 @@ def insts_to_binary(insts: List[Inst], meta: Dict[str, int], lut: list) -> bytea
 # Return the contents of a file
 def read_file(filename: str) -> str:
    try:
-        with open(filename, 'r') as f:
+        with open(filename) as f:
-            file_content = f.read()
+            return f.read()
-    except OSError:
+    except OSError as e:
-        print(f'Error opening {filename}: {sys.exc_info()[0]}')
+        raise FatalError(f'Error opening {filename}: {e}') from e
    return file_content
 # Write a bytestring to a file
@@ -983,29 +953,35 @@ def write_file(filename: str, data: bytearray) -> None:
 if __name__ == '__main__':
-    parser = argparse.ArgumentParser(
+    install_exception_reporting()
-        prog=sys.argv[0],
+    try:
-        description='BitScrambler program assembler')
+        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('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('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 \
+        parser.add_argument(
-            an unsupported instruction is assembled',  default=argparse.SUPPRESS)
+            '-c',
-    args = parser.parse_args()
+            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()
+        chipcfg = Chipcfg()
-    if 'c' in args:
+        if 'c' in args:
-        with open(args.c) as chipcfg_json:
+            with open(args.c) as chipcfg_json:
-            chipcfg = json.load(chipcfg_json)
+                chipcfg = json.load(chipcfg_json)
-    else:
+        else:
-        chipcfg = {'chipname': 'chip', 'extra_instruction_groups': [], 'support_all': True}
+            chipcfg = {'chipname': 'chip', 'extra_instruction_groups': [], 'support_all': True}
-    if 'outfile' in args:
+        if 'outfile' in args:
-        outfile = args.outfile
+            outfile = args.outfile
-    else:
+        else:
-        outfile = re.sub('.bsasm', '', args.infile) + '.bsbin'
+            outfile = re.sub('.bsasm', '', args.infile) + '.bsbin'
-    asm = read_file(args.infile)
+        asm = read_file(args.infile)
-    tokens = bsasm_parse(asm)
+        tokens = bsasm_parse(asm)
-    insts, meta, lut = bsasm_assemble(tokens, chipcfg)
+        insts, meta, lut = bsasm_assemble(tokens, chipcfg)
-    out_data = insts_to_binary(insts, meta, lut)
+        out_data = insts_to_binary(insts, meta, lut)
-    write_file(outfile, out_data)
+        write_file(outfile, out_data)
-    print(f'Written {len(insts)} instructions and {len(lut)} 32-bit words of LUT.')
+        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)
@@ -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}')
+                        log.die(escape(f'Malformed input. Cannot find name in {con}'))
                        sys.exit(1)
                    con = con_m[1]
                name_m = PYTHON_PACKAGE_RE.search(con)
                if not name_m:
-                    print(f'Malformed input. Cannot find name in {con}')
+                    log.die(escape(f'Malformed input. Cannot find name in {con}'))
                    sys.exit(1)
                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:')
+        # Header first (same order as before pylib), then details. Escape Rich markup so
-        print(os.linesep.join(not_satisfied))
+        # 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)'))
+        log.print(escape(f'    IDF_PYTHON_ENV_PATH: {idf_python_env_path or "(not set)"}'))
-        print(f'    Python interpreter used: {sys.executable}')
+        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')
+            # Keep on stdout with the rest of the diagnostic block (pre-pylib stream).
-            print('    PATH: {}'.format(os.getenv('PATH')))
+            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.')
@@ -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
+import rich_click as click
-from console_output import CONSOLE_STDOUT
+from console_output import configure_output
 from console_output import debug
 from console_output import die
 from console_output import eprint
 from console_output import oprint
 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:
+@click.command(
-    args = parse_arguments()
+    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
+    # Fill config global holder before configure_output
-    CONSOLE_STDERR.no_color = args.no_color
+    conf.ARGS = SimpleNamespace(
-    CONSOLE_STDOUT.no_color = args.no_color
+        shell=shell,
-    CONSOLE_STDERR.quiet = args.quiet
+        list=list_shells,
-    CONSOLE_STDOUT.quiet = args.quiet
+        export=export,
-    # Fill config global holder
+        no_color=no_color,
-    conf.ARGS = args
+        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(
+    log.print(
-        f'[dark_orange]Starting new \'{shell.shell}\' shell with ESP-IDF environment... (use "exit" command to quit)'
+        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()
+    shell_obj.spawn()
-    eprint('[dark_orange]ESP-IDF environment exited.')
+    log.print('[dark_orange]ESP-IDF environment exited.', file=sys.stderr)
 if __name__ == '__main__':
@@ -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.')
+    sys.exit(f'error: Unable to import esp-pylib: {e}. Please execute the install script.')
 CONSOLE_STDERR = Console(stderr=True, width=255)
 CONSOLE_STDOUT = Console(width=255)
-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]')
+                log.print('[red]FAILED[/red]', file=sys.stderr)
-                if conf.ARGS.debug:
+                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
@@ -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')
@@ -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)
@@ -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:
+    if args.verbose:
-        log_level = logging.WARNING
+        log.set_verbosity(Verbosity.VERBOSE)
    elif args.verbose == 1:
        log_level = logging.INFO
    else:
-        log_level = logging.DEBUG
+        log.set_verbosity(Verbosity.NORMAL)
    logging.basicConfig(level=log_level)
    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])}'
            )
-
+        details.append('Please ensure all the macro constants will translate to the same config type for all targets.')
-        print('Please ensure all the macro constants will translate to the same config type for all targets.')
+        log.die('\n'.join(details))
        sys.exit(1)
    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))
@@ -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)
+        log.warn('esp-idf-size not installed, please run the install script to install it')
-        raise SystemExit(1)
+        sys.exit(1)
    sys.exit(subprocess.run([sys.executable, '-m', 'esp_idf_size'] + sys.argv[1:]).returncode)
@@ -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}%')
@@ -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'<H H H H 3s B'
 DFUSuffix = namedtuple('DFUSuffix', ['bcd_device', 'pid', 'vid', 'bcd_dfu', 'sig', 'len'])
 ESPRESSIF_VID = 12346
 # This CRC32 gets added after DFUSUFFIX_STRUCT
 DFUCRC_STRUCT = b'<I'
@@ -235,18 +239,19 @@ class EspDfuWriter:
 def action_write(args):  # type: (typing.Mapping[str, typing.Any]) -> 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
@@ -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))
+                # Keep pre-migration order and stdout stream.
-                exit(1)
+                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)
@@ -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(
+            log.warn(
-                textwrap.wrap('Warning: in CMake variable {}: {}'.format(args.var_name, warning_str), width=120,
+                '\n  '.join(
-                              break_on_hyphens=False)), file=sys.stderr)
+                    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,
+            log.print(
-                                            break_on_hyphens=False)), file=sys.stderr)
+                '\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(
+        log.print(
-            sys.argv[0], os.getcwd(), sys.argv[1:]
+            escape(f'Diagnostic info: {sys.argv[0]} was invoked in {os.getcwd()} with arguments: {sys.argv[1:]}'),
-        ), file=sys.stderr)
+            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,
+def split_paths_by_spaces(
-                          warning_cb: typing.Optional[typing.Callable[[str], None]] = None) -> typing.List[str]:
+    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(
+                warning_cb(
-                    src, pprint.pformat(result)
+                    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 '{}'. "
+        raise PathSplitError(
-                              'This error may be reported if one or more paths '
+            f"Didn't find a valid way to split path '{src}'. "
-                              "are separated with spaces, and at least one path doesn't exist.").format(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(
+    raise PathSplitError(
-        src, ''.join('\n\t- ' + pprint.pformat(p) for p in valid_ways_to_split)
+        "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',
+        self.check_paths_concatenated(
-                                      '/absolute/path with spaces/two',
+            '/absolute/path with spaces/one', '/absolute/path with spaces/two', '/absolute/path with spaces/three'
-                                      '/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',
+        self.check_paths_concatenated_ambiguous(
-                                                additional_paths_exist=['/absolute/path', 'one'])
+            '/absolute/path one', 'two', additional_paths_exist=['/absolute/path', 'one']
        )
-        self.check_paths_concatenated_ambiguous('/path    ', '/path',
+        self.check_paths_concatenated_ambiguous('/path    ', '/path', additional_paths_exist=['/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,
+        self.assertListEqual(paths, split_paths_by_spaces(' /path', path_exists_cb=path_exists, warning_cb=add_warning))
                             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,
+        self.assertListEqual(paths, split_paths_by_spaces('/path ', path_exists_cb=path_exists, warning_cb=add_warning))
                             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,
+        self.assertListEqual(
-                             split_paths_by_spaces('/path /path', path_exists_cb=path_exists, warning_cb=add_warning))
+            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],
+    def split_paths_concatenated_base(self, paths_to_concatentate: list[str], paths_existing: list[str]) -> list[str]:
                                      paths_existing: typing.List[str]) -> typing.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,
+    def check_paths_concatenated_ambiguous(self, *args: str, additional_paths_exist: list[str] | None = None) -> None:
                                           additional_paths_exist: typing.Optional[typing.List[str]] = None) -> None:
        paths = [*args]
-        self.assertRaises(PathSplitError, self.split_paths_concatenated_base, paths_to_concatentate=paths,
+        self.assertRaises(
-                          paths_existing=paths + (additional_paths_exist or []))
+            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,
+    def check_paths_concatenated_nonexistent(self, *args: str, additional_paths_exist: list[str] | None = None) -> None:
                                             additional_paths_exist: typing.List[str] = None) -> None:
        paths = [*args]
-        self.assertRaises(PathSplitError, self.split_paths_concatenated_base, paths_to_concatentate=paths,
+        self.assertRaises(
-                          paths_existing=additional_paths_exist)
+            PathSplitError,
            self.split_paths_concatenated_base,
            paths_to_concatentate=paths,
            paths_existing=additional_paths_exist,
        )
 if __name__ == '__main__':
@@ -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
@@ -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
+    def run_tool(self, cmd):  # type: (list[str]) -> str
-        ret = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=600)
+        # 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)
@@ -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
@@ -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',
+        args = [mkdfu_path, 'write', '-o', f_out.name, '--pid', '2', '--flash-size', '4MB']
                '-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))
+        self.assertIn(f'"{f_out.name}" has been written. You may proceed with DFU flashing.', output)
        # 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()
        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(
-        self.common_test(file_args=[(0x1000, '1/1.bin'),
+            file_args=[(0x1000, '1/1.bin'), (0x8000, '1/2.bin'), (0x10000, '1/3.bin')], output_to_compare='1/dfu.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',
+        with tempfile.NamedTemporaryFile(
-                                         dir=temp_dir,
+            prefix='ľščťžýáíéěř\u0420\u043e\u0441\u0441\u0438\u044f', dir=temp_dir, delete=False
-                                         delete=False) as f:
+        ) as f:
            bootloader = f.name
        shutil.copyfile(os.path.join(current_dir, '1', '1.bin'), bootloader)
-        self.common_test(file_args=[(0x1000, bootloader),
+        self.common_test(
-                                    (0x8000, os.path.join(current_dir, '1', '2.bin')),
+            file_args=[
-                                    (0x10000, os.path.join(current_dir, '1', '3.bin'))])
+                (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)],
+        self.common_test(file_args=[(0, f.name)], part_size=5, output_to_compare='2/dfu.bin')
                         part_size=5,
                         output_to_compare='2/dfu.bin')
 if __name__ == '__main__':