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

This commit is contained in:
Jakub Kocka
2026-09-09 09:43:59 +02:00
parent 22ade8b843
commit d6771a0608
20 changed files with 780 additions and 601 deletions
+231 -150
View File
@@ -3,9 +3,8 @@
# otatool is used to perform ota-level operations - flashing ota partition
# erasing ota partition and switching ota partition
#
# SPDX-FileCopyrightText: 2018-2025 Espressif Systems (Shanghai) CO LTD
# SPDX-FileCopyrightText: 2018-2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
import argparse
import binascii
import collections
import os
@@ -13,6 +12,15 @@ import struct
import sys
import tempfile
import rich_click as click
from esp_pylib.cli_options import EspRichGroup
from esp_pylib.cli_options import MutuallyExclusiveOption
from esp_pylib.cli_options import OptionEatAll
from esp_pylib.cli_types import AnyIntType
from esp_pylib.logger import Verbosity
from esp_pylib.logger import log
from rich.markup import escape
try:
from parttool import PARTITION_TABLE_OFFSET
from parttool import PartitionName
@@ -31,13 +39,6 @@ __version__ = '2.0'
SPI_FLASH_SEC_SIZE = 0x2000
quiet = False
def status(msg):
if not quiet:
print(msg)
class OtatoolTarget:
OTADATA_PARTITION = PartitionType('data', 'ota')
@@ -226,8 +227,9 @@ def _read_otadata(target):
otadata_info = target._get_otadata_info()
print(' {:8s} \t {:8s} | \t {:8s} \t {:8s}'.format('OTA_SEQ', 'CRC', 'OTA_SEQ', 'CRC'))
print(
# Regular tool output (same stream as pre-pylib status()/print).
log.print(' {:8s} \t {:8s} | \t {:8s} \t {:8s}'.format('OTA_SEQ', 'CRC', 'OTA_SEQ', 'CRC'))
log.print(
f'Firmware: {otadata_info[0].seq:#08x} \t{otadata_info[0].crc:#08x} | '
f'\t{otadata_info[1].seq:#08x} \t {otadata_info[1].crc:#08x}'
)
@@ -235,7 +237,7 @@ def _read_otadata(target):
def _erase_otadata(target):
target.erase_otadata()
status('Erased ota_data partition contents')
log.print('Erased ota_data partition contents')
def _switch_ota_partition(target, ota_id):
@@ -244,165 +246,62 @@ def _switch_ota_partition(target, ota_id):
def _read_ota_partition(target, ota_id, output):
target.read_ota_partition(ota_id, output)
status(f'Read ota partition contents to file {output}')
log.print(f'Read ota partition contents to file {escape(str(output))}')
def _write_ota_partition(target, ota_id, input_file):
target.write_ota_partition(ota_id, input_file)
status(f'Written contents of file {input_file} to ota partition')
log.print(f'Written contents of file {escape(str(input_file))} to ota partition')
def _erase_ota_partition(target, ota_id):
target.erase_ota_partition(ota_id)
status('Erased contents of ota partition')
log.print('Erased contents of ota partition')
def main():
global quiet
def _target_kwargs_from_ctx(ctx_obj, spi_flash_sec_size=None):
kwargs = {}
for key, value in (
('port', ctx_obj.get('port')),
('baud', ctx_obj.get('baud')),
('partition_table_offset', ctx_obj.get('partition_table_offset')),
('partition_table_file', ctx_obj.get('partition_table_file')),
('esptool_args', ctx_obj.get('esptool_args')),
('esptool_write_args', ctx_obj.get('esptool_write_args')),
('esptool_read_args', ctx_obj.get('esptool_read_args')),
('esptool_erase_args', ctx_obj.get('esptool_erase_args')),
):
if value is not None and value != ():
kwargs[key] = value
if spi_flash_sec_size is not None:
kwargs['spi_flash_sec_size'] = spi_flash_sec_size
return kwargs
parser = argparse.ArgumentParser('ESP-IDF OTA Partitions Tool')
parser.add_argument('--quiet', '-q', help='suppress stderr messages', action='store_true')
parser.add_argument('--esptool-args', help='additional main arguments for esptool', nargs='+')
parser.add_argument(
'--esptool-write-args', help='additional subcommand arguments for esptool write-flash', nargs='+'
)
parser.add_argument('--esptool-read-args', help='additional subcommand arguments for esptool read-flash', nargs='+')
parser.add_argument(
'--esptool-erase-args', help='additional subcommand arguments for esptool erase-region', nargs='+'
)
def _resolve_ota_id(slot, name):
if name is not None:
return name
if slot is not None:
return slot
return None
# There are two possible sources for the partition table: a device attached to the host
# or a partition table CSV/binary file. These sources are mutually exclusive.
parser.add_argument('--port', '-p', help='port where the device to read the partition table from is attached')
parser.add_argument('--baud', '-b', help='baudrate to use', type=int)
parser.add_argument('--partition-table-offset', '-o', help='offset to read the partition table from', type=str)
parser.add_argument(
'--partition-table-file',
'-f',
help='file (CSV/binary) to read the partition table from; '
'overrides device attached to specified port as the partition table source when defined',
)
subparsers = parser.add_subparsers(dest='operation', help='run otatool -h for additional help')
spi_flash_sec_size = argparse.ArgumentParser(add_help=False)
spi_flash_sec_size.add_argument('--spi-flash-sec-size', help='value of SPI_FLASH_SEC_SIZE macro', type=str)
# Specify the supported operations
subparsers.add_parser('read_otadata', help='read otadata partition', parents=[spi_flash_sec_size])
subparsers.add_parser('erase_otadata', help='erase otadata partition')
slot_or_name_parser = argparse.ArgumentParser(add_help=False)
slot_or_name_parser_args = slot_or_name_parser.add_mutually_exclusive_group()
slot_or_name_parser_args.add_argument('--slot', help='slot number of the ota partition', type=int)
slot_or_name_parser_args.add_argument('--name', help='name of the ota partition')
subparsers.add_parser(
'switch_ota_partition', help='switch otadata partition', parents=[slot_or_name_parser, spi_flash_sec_size]
)
read_ota_partition_subparser = subparsers.add_parser(
'read_ota_partition', help='read contents of an ota partition', parents=[slot_or_name_parser]
)
read_ota_partition_subparser.add_argument(
'--output', help='file to write the contents of the ota partition to', required=True
)
write_ota_partition_subparser = subparsers.add_parser(
'write_ota_partition', help='write contents to an ota partition', parents=[slot_or_name_parser]
)
write_ota_partition_subparser.add_argument('--input', help='file whose contents to write to the ota partition')
subparsers.add_parser(
'erase_ota_partition', help='erase contents of an ota partition', parents=[slot_or_name_parser]
)
args = parser.parse_args()
quiet = args.quiet
# No operation specified, display help and exit
if args.operation is None:
if not quiet:
parser.print_help()
sys.exit(1)
target_args = {}
if args.port:
target_args['port'] = args.port
if args.partition_table_file:
target_args['partition_table_file'] = args.partition_table_file
if args.partition_table_offset:
target_args['partition_table_offset'] = int(args.partition_table_offset, 0)
try:
if args.spi_flash_sec_size:
target_args['spi_flash_sec_size'] = int(args.spi_flash_sec_size, 0)
except AttributeError:
pass
if args.esptool_args:
target_args['esptool_args'] = args.esptool_args
if args.esptool_write_args:
target_args['esptool_write_args'] = args.esptool_write_args
if args.esptool_read_args:
target_args['esptool_read_args'] = args.esptool_read_args
if args.esptool_erase_args:
target_args['esptool_erase_args'] = args.esptool_erase_args
if args.baud:
target_args['baud'] = args.baud
target = OtatoolTarget(**target_args)
# Create the operation table and execute the operation
common_args = {'target': target}
ota_id = []
try:
if args.name is not None:
ota_id = ['name']
else:
if args.slot is not None:
ota_id = ['slot']
except AttributeError:
pass
def _run_operation(operation, target, quiet=False, **op_kwargs):
otatool_ops = {
'read_otadata': (_read_otadata, []),
'erase_otadata': (_erase_otadata, []),
'switch_ota_partition': (_switch_ota_partition, ota_id),
'read_ota_partition': (_read_ota_partition, ['output'] + ota_id),
'write_ota_partition': (_write_ota_partition, ['input'] + ota_id),
'erase_ota_partition': (_erase_ota_partition, ota_id),
'switch_ota_partition': (_switch_ota_partition, ['ota_id']),
'read_ota_partition': (_read_ota_partition, ['ota_id', 'output']),
'write_ota_partition': (_write_ota_partition, ['ota_id', 'input_file']),
'erase_ota_partition': (_erase_ota_partition, ['ota_id']),
}
(op, op_args) = otatool_ops[args.operation]
for op_arg in op_args:
common_args.update({op_arg: vars(args)[op_arg]})
try:
common_args['ota_id'] = common_args.pop('name')
except KeyError:
try:
common_args['ota_id'] = common_args.pop('slot')
except KeyError:
pass
op, op_arg_names = otatool_ops[operation]
common_args = {'target': target}
for op_arg in op_arg_names:
common_args[op_arg] = op_kwargs[op_arg]
if quiet:
# If exceptions occur, suppress and exit quietly
try:
op(**common_args)
except Exception:
@@ -411,5 +310,187 @@ def main():
op(**common_args)
def _slot_or_name_options(func):
decorators = [
click.option(
'--slot',
type=int,
cls=MutuallyExclusiveOption,
exclusive_with=['name'],
help='slot number of the ota partition',
),
click.option(
'--name',
cls=MutuallyExclusiveOption,
exclusive_with=['slot'],
help='name of the ota partition',
),
]
for decorator in reversed(decorators):
func = decorator(func)
return func
@click.group(
cls=EspRichGroup,
invoke_without_command=True,
context_settings={'help_option_names': ['-h', '--help']},
help='ESP-IDF OTA Partitions Tool',
)
@click.option('--quiet', '-q', is_flag=True, help='suppress status messages')
@click.option(
'--esptool-args',
multiple=True,
cls=OptionEatAll,
type=str,
help='additional main arguments for esptool',
)
@click.option(
'--esptool-write-args',
multiple=True,
cls=OptionEatAll,
type=str,
help='additional subcommand arguments for esptool write-flash',
)
@click.option(
'--esptool-read-args',
multiple=True,
cls=OptionEatAll,
type=str,
help='additional subcommand arguments for esptool read-flash',
)
@click.option(
'--esptool-erase-args',
multiple=True,
cls=OptionEatAll,
type=str,
help='additional subcommand arguments for esptool erase-region',
)
@click.option('--port', '-p', help='port where the device to read the partition table from is attached')
@click.option('--baud', '-b', type=int, help='baudrate to use')
@click.option('--partition-table-offset', '-o', type=AnyIntType(), help='offset to read the partition table from')
@click.option(
'--partition-table-file',
'-f',
type=click.Path(),
help='file (CSV/binary) to read the partition table from; '
'overrides device attached to specified port as the partition table source when defined',
)
@click.pass_context
def cli(
ctx,
quiet,
esptool_args,
esptool_write_args,
esptool_read_args,
esptool_erase_args,
port,
baud,
partition_table_offset,
partition_table_file,
):
if quiet:
log.set_verbosity(Verbosity.SILENT)
ctx.ensure_object(dict)
ctx.obj.update(
{
'quiet': quiet,
'esptool_args': esptool_args,
'esptool_write_args': esptool_write_args,
'esptool_read_args': esptool_read_args,
'esptool_erase_args': esptool_erase_args,
'port': port,
'baud': baud,
'partition_table_offset': partition_table_offset,
'partition_table_file': partition_table_file,
}
)
# Match pre-click argparse: no subcommand → help (unless quiet) and exit 1.
if ctx.invoked_subcommand is None:
if not quiet:
click.echo(ctx.get_help())
sys.exit(1)
@cli.command('read_otadata', help='read otadata partition')
@click.option('--spi-flash-sec-size', type=AnyIntType(), help='value of SPI_FLASH_SEC_SIZE macro')
@click.pass_context
def read_otadata_cmd(ctx, spi_flash_sec_size):
target = OtatoolTarget(**_target_kwargs_from_ctx(ctx.obj, spi_flash_sec_size))
_run_operation('read_otadata', target, quiet=ctx.obj.get('quiet'))
@cli.command('erase_otadata', help='erase otadata partition')
@click.pass_context
def erase_otadata_cmd(ctx):
target = OtatoolTarget(**_target_kwargs_from_ctx(ctx.obj))
_run_operation('erase_otadata', target, quiet=ctx.obj.get('quiet'))
@cli.command('switch_ota_partition', help='switch otadata partition')
@_slot_or_name_options
@click.option('--spi-flash-sec-size', type=AnyIntType(), help='value of SPI_FLASH_SEC_SIZE macro')
@click.pass_context
def switch_ota_partition_cmd(ctx, slot, name, spi_flash_sec_size):
ota_id = _resolve_ota_id(slot, name)
if ota_id is None:
# Under --quiet, match pre-click: silent exit 2 (exception was swallowed).
if ctx.obj.get('quiet'):
sys.exit(2)
log.die('Partition to switch to should be defined using --slot OR --name')
target = OtatoolTarget(**_target_kwargs_from_ctx(ctx.obj, spi_flash_sec_size))
_run_operation('switch_ota_partition', target, quiet=ctx.obj.get('quiet'), ota_id=ota_id)
@cli.command('read_ota_partition', help='read contents of an ota partition')
@_slot_or_name_options
@click.option('--output', help='file to write the contents of the ota partition to', required=True)
@click.pass_context
def read_ota_partition_cmd(ctx, slot, name, output):
ota_id = _resolve_ota_id(slot, name)
if ota_id is None:
if ctx.obj.get('quiet'):
sys.exit(2)
log.die('OTA partition should be defined using --slot OR --name')
target = OtatoolTarget(**_target_kwargs_from_ctx(ctx.obj))
_run_operation('read_ota_partition', target, quiet=ctx.obj.get('quiet'), ota_id=ota_id, output=output)
@cli.command('write_ota_partition', help='write contents to an ota partition')
@_slot_or_name_options
@click.option('--input', 'input_file', help='file whose contents to write to the ota partition')
@click.pass_context
def write_ota_partition_cmd(ctx, slot, name, input_file):
ota_id = _resolve_ota_id(slot, name)
if ota_id is None:
if ctx.obj.get('quiet'):
sys.exit(2)
log.die('OTA partition should be defined using --slot OR --name')
target = OtatoolTarget(**_target_kwargs_from_ctx(ctx.obj))
_run_operation('write_ota_partition', target, quiet=ctx.obj.get('quiet'), ota_id=ota_id, input_file=input_file)
@cli.command('erase_ota_partition', help='erase contents of an ota partition')
@_slot_or_name_options
@click.pass_context
def erase_ota_partition_cmd(ctx, slot, name):
ota_id = _resolve_ota_id(slot, name)
if ota_id is None:
if ctx.obj.get('quiet'):
sys.exit(2)
log.die('OTA partition should be defined using --slot OR --name')
target = OtatoolTarget(**_target_kwargs_from_ctx(ctx.obj))
_run_operation('erase_ota_partition', target, quiet=ctx.obj.get('quiet'), ota_id=ota_id)
def main():
cli()
if __name__ == '__main__':
from esp_pylib.excepthook import install_exception_reporting
install_exception_reporting()
main()
+27 -23
View File
@@ -14,11 +14,23 @@ import re
import sys
from datetime import datetime
from esp_pylib.errors import FatalError
from esp_pylib.excepthook import install_exception_reporting
from esp_pylib.logger import Verbosity
from esp_pylib.logger import log
from rich.markup import escape
__version__ = '1.0'
quiet = False
max_blk_len = 256
idf_target = 'esp32'
quiet = False
def status(msg: str) -> None:
"""Print non-critical status to stderr (suppressed by --quiet)."""
if not quiet:
log.print(msg, file=sys.stderr)
def get_copyright():
@@ -31,18 +43,6 @@ def get_copyright():
return copyright_str % datetime.today().year
def status(msg):
"""Print status message to stderr"""
if not quiet:
critical(msg)
def critical(msg):
"""Print critical message to stderr"""
sys.stderr.write(msg)
sys.stderr.write('\n')
class FuseTable(list):
def __init__(self):
super().__init__()
@@ -69,7 +69,7 @@ class FuseTable(list):
except InputError as e:
raise InputError(f'Error at line {line_no + 1}: {e}')
except Exception:
critical(f'Unexpected error parsing line {line_no + 1}: {line}')
log.err(f'Unexpected error parsing line {line_no + 1}: {line}')
raise
# fix up missing bit_start
@@ -126,7 +126,7 @@ class FuseTable(list):
field_name = p.field_name + p.group
if field_name != '' and len(duplicates.intersection([field_name])) != 0:
fl_error = True
print(
log.err(
f'Field at {p.field_name}, {p.efuse_block}, {p.bit_start}, {p.bit_count} '
'have duplicate field_name'
)
@@ -477,7 +477,7 @@ class FuseDefinition:
def process_input_file(file, type_table):
status('Parsing efuse CSV input file ' + file.name + ' ...')
status('Parsing efuse CSV input file ' + escape(file.name) + ' ...')
input_contents = file.read()
table = FuseTable.from_csv(input_contents)
status('Verifying efuse table...')
@@ -509,27 +509,28 @@ def create_output_files(name, output_table, debug):
# src files are the same
if ckeck_md5_in_file(output_table.md5_digest_table, file_c_path) is False:
status('Creating efuse *.h file ' + file_h_path + ' ...')
status('Creating efuse *.h file ' + escape(file_h_path) + ' ...')
output = output_table.to_header(file_name)
with open(file_h_path, 'w', encoding='utf-8') as f:
f.write(output)
status('Creating efuse *.c file ' + file_c_path + ' ...')
status('Creating efuse *.c file ' + escape(file_c_path) + ' ...')
output = output_table.to_c_file(file_name, debug)
with open(file_c_path, 'w', encoding='utf-8') as f:
f.write(output)
else:
# Always visible (same as pre-pylib plain print), even under --quiet.
print('Source files do not require updating correspond to csv file.')
def main():
global quiet
global max_blk_len
global idf_target
global quiet
parser = argparse.ArgumentParser(description='ESP32 eFuse Manager')
parser.add_argument('--idf_target', '-t', help='Target chip type', default='esp32')
parser.add_argument('--quiet', '-q', help="Don't print non-critical status messages to stderr", action='store_true')
parser.add_argument('--quiet', '-q', help="Don't print non-critical status messages", action='store_true')
parser.add_argument('--debug', help='Create header file with debug info', default=False, action='store_false')
parser.add_argument('--info', help='Print info about range of used bits', default=False, action='store_true')
parser.add_argument('--max_blk_len', help='Max number of bits in BLOCKs', type=int, default=256)
@@ -543,11 +544,14 @@ def main():
idf_target = args.idf_target
max_blk_len = args.max_blk_len
# Always print Max bits before applying --quiet (pre-pylib behavior).
print(f'Max number of bits in BLK {max_blk_len:d}')
if max_blk_len not in [256, 192, 128]:
raise InputError(f'Unsupported block length = {max_blk_len:d}')
quiet = args.quiet
if quiet:
log.set_verbosity(Verbosity.SILENT)
debug = args.debug
info = args.info
@@ -569,7 +573,7 @@ def main():
return 0
class InputError(RuntimeError):
class InputError(FatalError):
def __init__(self, e):
super().__init__(e)
@@ -580,8 +584,8 @@ class ValidationError(InputError):
if __name__ == '__main__':
install_exception_reporting()
try:
main()
except InputError as e:
print(e, file=sys.stderr)
sys.exit(2)
log.die(str(e), exit_code=2)
@@ -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(
f'error: In {filename}, found {macro} {count_expected} time(s), '
log.err(
f'In {filename}, found {macro} {count_expected} time(s), '
f'but regular expression matched {len(found)} time(s)',
file=sys.stderr,
)
has_errors = True
@@ -153,19 +157,16 @@ def main() -> None:
#
diff_lines = list(difflib.unified_diff(startup_entries_expected_lines, startup_entries_lines, lineterm=''))
if len(diff_lines) > 0:
print(
(
"error: startup order doesn't match the reference file. "
f'please update {STARTUP_ENTRIES_FILE} to match the actual startup order:'
),
file=sys.stderr,
log.err(
"startup order doesn't match the reference file. "
f'please update {STARTUP_ENTRIES_FILE} to match the actual startup order:'
)
for line in diff_lines:
print(f'{line}', file=sys.stderr)
log.print(escape(line), file=sys.stderr)
has_errors = True
if has_errors:
raise SystemExit(1)
sys.exit(1)
if __name__ == '__main__':