diff --git a/components/nvs_flash/nvs_partition_tool/nvs_check.py b/components/nvs_flash/nvs_partition_tool/nvs_check.py index b93dec809bc..bbaf01c16d4 100644 --- a/components/nvs_flash/nvs_partition_tool/nvs_check.py +++ b/components/nvs_flash/nvs_partition_tool/nvs_check.py @@ -1,62 +1,41 @@ #!/usr/bin/env python3 -# SPDX-FileCopyrightText: 2023-2025 Espressif Systems (Shanghai) CO LTD +# SPDX-FileCopyrightText: 2023-2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 -from typing import Dict -from typing import List -from typing import Optional -from typing import Set from nvs_logger import NVS_Logger -from nvs_parser import nvs_const from nvs_parser import NVS_Entry from nvs_parser import NVS_Page from nvs_parser import NVS_Partition - +from nvs_parser import nvs_const EMPTY_ENTRY = NVS_Entry(-1, bytearray(32), 'Erased') -used_namespaces: Dict[int, Optional[str]] = {} -found_namespaces: Dict[int, str] = {} -blobs: Dict = {} -blob_chunks: List[NVS_Entry] = [] +used_namespaces: dict[int, str | None] = {} +found_namespaces: dict[int, str] = {} +blobs: dict = {} +blob_chunks: list[NVS_Entry] = [] -def check_partition_size(nvs_partition: NVS_Partition, nvs_log: NVS_Logger, read_only: bool=False) -> bool: - """ Checks if the partition is large enough and has enough pages - """ +def check_partition_size(nvs_partition: NVS_Partition, nvs_log: NVS_Logger, read_only: bool = False) -> bool: + """Checks if the partition is large enough and has enough pages""" if len(nvs_partition.raw_data) // 0x1000 < 3 and not read_only: - nvs_log.info( - nvs_log.yellow( - 'NVS Partition size must be at least 0x3000 (4kiB * 3 pages == 12kiB)!' - ) - ) + nvs_log.warn('NVS Partition size must be at least 0x3000 (4kiB * 3 pages == 12kiB)!') return False if len(nvs_partition.raw_data) % 0x1000 != 0: - nvs_log.info( - nvs_log.yellow( - 'NVS Partition size must be a multiple of 0x1000 (4kiB)!' - ) - ) + nvs_log.warn('NVS Partition size must be a multiple of 0x1000 (4kiB)!') return False if len(nvs_partition.pages) < 3 and not read_only: - nvs_log.info( - nvs_log.yellow( - 'NVS Partition must contain 3 pages (sectors) at least to function properly!' - ) - ) + nvs_log.warn('NVS Partition must contain 3 pages (sectors) at least to function properly!') return False return True def check_empty_page_present(nvs_partition: NVS_Partition, nvs_log: NVS_Logger) -> bool: if not any(page.header['status'] == 'Empty' for page in nvs_partition.pages): - nvs_log.info( - nvs_log.red( - '''No free (empty) page found in the NVS partition, -at least one free page is required for proper function!''' - ) + nvs_log.err( + 'No free (empty) page found in the NVS partition, at least one free page is required for proper function!' ) - nvs_log.info(nvs_log.red('NVS partition possibly truncated?\n')) + nvs_log.err('NVS partition possibly truncated?') return False return True @@ -67,39 +46,31 @@ def check_empty_page_content(nvs_page: NVS_Page, nvs_log: NVS_Logger) -> bool: if nvs_page.raw_entry_state_bitmap != bytearray({0xFF}) * nvs_const.entry_size: result = False - nvs_log.info( - nvs_log.red( - 'The page is reported as Empty but its entry state bitmap is not empty!' - ) - ) + nvs_log.err('The page is reported as Empty but its entry state bitmap is not empty!') if any([not e.is_empty for e in nvs_page.entries]): result = False - nvs_log.info( - nvs_log.red('The page is reported as Empty but there are data written!') - ) + nvs_log.err('The page is reported as Empty but there are data written!') return result def check_page_crc(nvs_page: NVS_Page, nvs_log: NVS_Logger) -> bool: if nvs_page.header['crc']['original'] == nvs_page.header['crc']['computed']: - nvs_log.info( - nvs_log.cyan(f'Page no. {nvs_page.header["page_index"]}'), '\tCRC32: OK' - ) + nvs_log.info(nvs_log.cyan(f'Page no. {nvs_page.header["page_index"]}'), '\tCRC32: OK') return True else: nvs_log.info( nvs_log.cyan(f'Page no. {nvs_page.header["page_index"]}'), - f'Original CRC32:', + 'Original CRC32:', nvs_log.red(f'{nvs_page.header["crc"]["original"]:x}'), - f'Generated CRC32:', + 'Generated CRC32:', nvs_log.green(f'{nvs_page.header["crc"]["computed"]:x}'), ) return False -def identify_entry_duplicates(entry: NVS_Entry, entry_dict: Dict[str, List[NVS_Entry]]) -> Dict[str, List[NVS_Entry]]: +def identify_entry_duplicates(entry: NVS_Entry, entry_dict: dict[str, list[NVS_Entry]]) -> dict[str, list[NVS_Entry]]: """Identifies and logs written entries Part 1 of duplicate entry check mechanism @@ -112,10 +83,9 @@ def identify_entry_duplicates(entry: NVS_Entry, entry_dict: Dict[str, List[NVS_E return entry_dict -def check_page_entries(nvs_page: NVS_Page, nvs_log: NVS_Logger) -> Dict[str, List[NVS_Entry]]: - """Checks entries in the given page (entry state, children CRC32, entry type, span and gathers blobs and namespaces) - """ - seen_written_entires: Dict[str, List[NVS_Entry]] = {} +def check_page_entries(nvs_page: NVS_Page, nvs_log: NVS_Logger) -> dict[str, list[NVS_Entry]]: + """Checks entries in the given page (state, CRC32, type, span; gathers blobs and namespaces).""" + seen_written_entires: dict[str, list[NVS_Entry]] = {} for entry in nvs_page.entries: # entry: NVS_Entry @@ -129,90 +99,65 @@ def check_page_entries(nvs_page: NVS_Page, nvs_log: NVS_Logger) -> Dict[str, Lis # Duplicate entry check (1) - same key, different index - find duplicates seen_written_entires = identify_entry_duplicates(entry, seen_written_entires) - # Entry state check - doesn't check variable length values (metadata such as state are meaningless as all 32 bytes are pure data) + # Entry state check - skip variable-length payload bytes (metadata not meaningful there) if entry.is_empty: if entry.state == 'Written': - nvs_log.info( - nvs_log.red( - f' Entry #{entry.index:03d} is reported as Written but it is empty!' - ) - ) + nvs_log.err(f' Entry #{entry.index:03d} is reported as Written but it is empty!') continue elif entry.state == 'Erased': - nvs_log.info( - nvs_log.yellow( - f' Entry #{entry.index:03d} is reported as Erased but it is empty! (Only entries reported as Empty should be empty)' - ) + nvs_log.warn( + f' Entry #{entry.index:03d} is reported as Erased but it is empty! ' + '(Only entries reported as Empty should be empty)' ) if entry.state == 'Written': # Entry CRC32 check - if ( - entry.metadata['crc']['original'] - != entry.metadata['crc']['computed'] - ): + if entry.metadata['crc']['original'] != entry.metadata['crc']['computed']: nvs_log.info( - nvs_log.red( - f' Entry #{entry.index:03d} {entry.key} has wrong CRC32!{"": <5}' - ), - f'Written:', + nvs_log.red(f' Entry #{entry.index:03d} {entry.key} has wrong CRC32!'), + 'Written:', nvs_log.red(f'{entry.metadata["crc"]["original"]:x}'), - f'Generated:', + 'Generated:', nvs_log.green(f'{entry.metadata["crc"]["computed"]:x}'), ) # Entry children CRC32 check - if ( - entry.metadata['span'] > 1 - and (entry.metadata['crc']['data_original'] != entry.metadata['crc']['data_computed']) + if entry.metadata['span'] > 1 and ( + entry.metadata['crc']['data_original'] != entry.metadata['crc']['data_computed'] ): nvs_log.info( - nvs_log.red( - f' Entry #{entry.index:03d} {entry.key} data (string, blob) has wrong CRC32!' - ), - f'Written:', + nvs_log.red(f' Entry #{entry.index:03d} {entry.key} data (string, blob) has wrong CRC32!'), + 'Written:', nvs_log.red(f'{entry.metadata["crc"]["data_original"]:x}'), - f'Generated:', + 'Generated:', nvs_log.green(f'{entry.metadata["crc"]["data_computed"]:x}'), ) # Entry type check - if entry.metadata['type'] not in [ - nvs_const.item_type[key] for key in nvs_const.item_type - ]: - nvs_log.info( - nvs_log.yellow( - f' Type of entry #{entry.index:03d} {entry.key} is unrecognized!' - ), - f'Type: {entry.metadata["type"]}', + if entry.metadata['type'] not in [nvs_const.item_type[key] for key in nvs_const.item_type]: + nvs_log.warn( + f' Type of entry #{entry.index:03d} {entry.key} is unrecognized! Type: {entry.metadata["type"]}' ) # Span check - if ( - entry.index + entry.metadata['span'] - 1 - >= int(nvs_const.page_size / nvs_const.entry_size) - 2 - ): - nvs_log.info( - nvs_log.red( - f' Variable length entry #{entry.index:03d} {entry.key} is out of bounds!' - ) - ) + if entry.index + entry.metadata['span'] - 1 >= int(nvs_const.page_size / nvs_const.entry_size) - 2: + nvs_log.err(f' Variable length entry #{entry.index:03d} {entry.key} is out of bounds!') # Spanned entry state checks elif entry.metadata['span'] > 1: parent_state = entry.state for kid in entry.children: if parent_state != kid.state: - nvs_log.info( - nvs_log.yellow(' Inconsistent data state!'), - f'Entry #{entry.index:03d} {entry.key} state: {parent_state},', - f'Data entry #{kid.index:03d} {entry.key} state: {kid.state}', + nvs_log.warn( + f' Inconsistent data state! Entry #{entry.index:03d} {entry.key} ' + f'state: {parent_state}, Data entry #{kid.index:03d} {entry.key} ' + f'state: {kid.state}' ) # Gather blobs & namespaces if entry.metadata['type'] == 'blob_index': - blobs[f'{entry.metadata["namespace"]:03d}{entry.key}'] = [entry] + [ - EMPTY_ENTRY - ] * entry.data['chunk_count'] + blobs[f'{entry.metadata["namespace"]:03d}{entry.key}'] = [entry] + [EMPTY_ENTRY] * entry.data[ + 'chunk_count' + ] elif entry.metadata['type'] == 'blob_data': blob_chunks.append(entry) @@ -224,22 +169,21 @@ def check_page_entries(nvs_page: NVS_Page, nvs_log: NVS_Logger) -> Dict[str, Lis return seen_written_entires -def filter_namespaces_fake_duplicates(duplicate_entries_dict: Dict[str, List[NVS_Entry]]) -> Dict[str, List[NVS_Entry]]: +def filter_namespaces_fake_duplicates(duplicate_entries_dict: dict[str, list[NVS_Entry]]) -> dict[str, list[NVS_Entry]]: """Takes a dictionary of entries (as written) and returns a new dictionary with "fake" duplicates, where entries which have the same key but under different namespaces are filtered out Use `filter_entry_duplicates()` to properly filter out all duplicates """ - new_duplicate_entries_dict: Dict[str, List[NVS_Entry]] = {} + new_duplicate_entries_dict: dict[str, list[NVS_Entry]] = {} for key, duplicate_entries in duplicate_entries_dict.items(): - seen_entries: List[NVS_Entry] = [] - entry_same_namespace_collisions_list: Set[NVS_Entry] = set() + seen_entries: list[NVS_Entry] = [] + entry_same_namespace_collisions_list: set[NVS_Entry] = set() # Search through the "duplicates" and see if there are real duplicates # E.g. the key can be the same if the namespace is different for entry in duplicate_entries: if entry.metadata['type'] in nvs_const.item_type.values(): - entry_same_namespace_collisions = set() for other_entry in seen_entries: if entry.metadata['namespace'] == other_entry.metadata['namespace']: @@ -251,7 +195,7 @@ def filter_namespaces_fake_duplicates(duplicate_entries_dict: Dict[str, List[NVS seen_entries.append(entry) # Catch real duplicates - new_duplicate_entries: List[NVS_Entry] = [] + new_duplicate_entries: list[NVS_Entry] = [] if len(seen_entries) > 1: for entry in seen_entries: if entry in entry_same_namespace_collisions_list: @@ -263,26 +207,25 @@ def filter_namespaces_fake_duplicates(duplicate_entries_dict: Dict[str, List[NVS return new_duplicate_entries_dict -def filter_blob_related_duplicates(duplicate_entries_dict: Dict[str, List[NVS_Entry]]) -> Dict[str, List[NVS_Entry]]: +def filter_blob_related_duplicates(duplicate_entries_dict: dict[str, list[NVS_Entry]]) -> dict[str, list[NVS_Entry]]: """Takes a dictionary of entries (as written) and returns a new dictionary with "fake" duplicates, where entries related to blob index and blob data under the same namespace are filtered out Use `filter_entry_duplicates()` to properly filter out all duplicates """ - new_duplicate_entries_dict: Dict[str, List[NVS_Entry]] = {} + new_duplicate_entries_dict: dict[str, list[NVS_Entry]] = {} for key, duplicate_entries in duplicate_entries_dict.items(): - seen_blob_index: List[NVS_Entry] = [] - seen_blob_data: List[NVS_Entry] = [] - seen_another_type_data: List[NVS_Entry] = [] - blob_index_chunk_index_collisions_list: Set[NVS_Entry] = set() - blob_data_chunk_index_collisions_list: Set[NVS_Entry] = set() + seen_blob_index: list[NVS_Entry] = [] + seen_blob_data: list[NVS_Entry] = [] + seen_another_type_data: list[NVS_Entry] = [] + blob_index_chunk_index_collisions_list: set[NVS_Entry] = set() + blob_data_chunk_index_collisions_list: set[NVS_Entry] = set() # Search through the "duplicates" and see if there are real duplicates # E.g. the key can be the same for blob_index and blob_data # (and even for more blob_data entries if they have a different chunk_index) for entry in duplicate_entries: if entry.metadata['type'] == 'blob_index': - blob_index_chunk_index_collisions = set() for other_entry in seen_blob_index: if entry.metadata['namespace'] == other_entry.metadata['namespace']: @@ -294,11 +237,12 @@ def filter_blob_related_duplicates(duplicate_entries_dict: Dict[str, List[NVS_En seen_blob_index.append(entry) elif entry.metadata['type'] == 'blob_data': - blob_data_chunk_index_collisions = set() for other_entry in seen_blob_data: - if (entry.metadata['namespace'] == other_entry.metadata['namespace'] - and entry.metadata['chunk_index'] == other_entry.metadata['chunk_index']): + if ( + entry.metadata['namespace'] == other_entry.metadata['namespace'] + and entry.metadata['chunk_index'] == other_entry.metadata['chunk_index'] + ): blob_data_chunk_index_collisions.add(entry) blob_data_chunk_index_collisions.add(other_entry) @@ -310,7 +254,7 @@ def filter_blob_related_duplicates(duplicate_entries_dict: Dict[str, List[NVS_En seen_another_type_data.append(entry) # Catch real duplicates - new_duplicate_entries: List[NVS_Entry] = [] + new_duplicate_entries: list[NVS_Entry] = [] if len(seen_blob_index) > 1: for entry in seen_blob_index: if entry in blob_index_chunk_index_collisions_list: @@ -319,7 +263,7 @@ def filter_blob_related_duplicates(duplicate_entries_dict: Dict[str, List[NVS_En if len(seen_blob_data) > 1: for entry in seen_blob_data: if entry in blob_data_chunk_index_collisions_list: - new_duplicate_entries.append(entry) + new_duplicate_entries.append(entry) for entry in seen_another_type_data: # If there are any duplicates of other types new_duplicate_entries.append(entry) @@ -330,27 +274,24 @@ def filter_blob_related_duplicates(duplicate_entries_dict: Dict[str, List[NVS_En return new_duplicate_entries_dict -def filter_entry_duplicates(entries: Dict[str, List[NVS_Entry]]) -> Dict[str, List[NVS_Entry]]: - """Takes a dictionary of (seen written) entries and outputs a new dictionary with "fake" duplicates filtered out, keeping only real duplicates in +def filter_entry_duplicates(entries: dict[str, list[NVS_Entry]]) -> dict[str, list[NVS_Entry]]: + """Filter fake duplicates; keep real duplicates only (part 2 of duplicate check). - (i.e. duplicate keys under different namespaces and blob index and blob data having the same key under the same namespace are allowed - and should be filtered out) - - Part 2 of duplicate entry check mechanism + Allowed: same key in different namespaces; blob_index and blob_data sharing a key. """ # Only keep seen written entries which have been observerd multiple times (duplicates) duplicate_entries_list = {key: v for key, v in entries.items() if len(v) > 1} # Filter out "fake" duplicates 1 (duplicate keys under different namespaces are allowed) duplicate_entries_list_1 = filter_namespaces_fake_duplicates(duplicate_entries_list) - # Filter out "fake" duplicates 2 (blob index and blob data are allowed to have the same key even in the same namespace) + # Filter out "fake" duplicates 2 (blob index/data may share a key in one namespace) duplicate_entries_list_2 = filter_blob_related_duplicates(duplicate_entries_list_1) return duplicate_entries_list_2 -def print_entry_duplicates(duplicate_entries_list: Dict[str, List[NVS_Entry]], nvs_log: NVS_Logger) -> None: +def print_entry_duplicates(duplicate_entries_list: dict[str, list[NVS_Entry]], nvs_log: NVS_Logger) -> None: if len(duplicate_entries_list) > 0: - nvs_log.info(nvs_log.red('Found duplicate entries:')) - nvs_log.info(nvs_log.red('Entry\tKey\t\t\tType\t\tNamespace idx\tPage\tPage status')) + nvs_log.err('Found duplicate entries:') + nvs_log.err('Entry\tKey\t\t\tType\t\tNamespace idx\tPage\tPage status') for _, duplicate_entries in duplicate_entries_list.items(): # duplicate_entries: List[NVS_Entry] @@ -375,28 +316,24 @@ def print_entry_duplicates(duplicate_entries_list: Dict[str, List[NVS_Entry]], n namepace_tab = '\t' * (2 - namespace_tab_cnt) namespace_str = f'{entry.metadata["namespace"]}' - nvs_log.info( - nvs_log.red( - f'#{entry.index:03d}\t{entry.key}{entry_key_tab}{entry_type}{namepace_tab}{namespace_str}\t\t{page_num}\t{page_status}' - ) + nvs_log.err( + f'#{entry.index:03d}\t{entry.key}{entry_key_tab}{entry_type}{namepace_tab}' + f'{namespace_str}\t\t{page_num}\t{page_status}' ) def assemble_blobs(nvs_log: NVS_Logger) -> None: - """Assembles blob data from blob chunks - """ + """Assembles blob data from blob chunks""" for chunk in blob_chunks: # chunk: NVS_Entry - parent = blobs.get( - f'{chunk.metadata["namespace"]:03d}{chunk.key}', [EMPTY_ENTRY] - )[0] + parent = blobs.get(f'{chunk.metadata["namespace"]:03d}{chunk.key}', [EMPTY_ENTRY])[0] # Blob chunk without blob index check if parent is EMPTY_ENTRY: - nvs_log.info( - nvs_log.red(f'Blob {chunk.key} chunk has no blob index!'), - f'Namespace index: {chunk.metadata["namespace"]:03d}', - f'[{found_namespaces.get(chunk.metadata["namespace"], "undefined")}],', - f'Chunk Index: {chunk.metadata["chunk_index"]:03d}', + nvs_log.err( + f'Blob {chunk.key} chunk has no blob index! ' + f'Namespace index: {chunk.metadata["namespace"]:03d} ' + f'[{found_namespaces.get(chunk.metadata["namespace"], "undefined")}], ' + f'Chunk Index: {chunk.metadata["chunk_index"]:03d}' ) else: blob_key = f'{chunk.metadata["namespace"]:03d}{chunk.key}' @@ -405,8 +342,7 @@ def assemble_blobs(nvs_log: NVS_Logger) -> None: def check_blob_data(nvs_log: NVS_Logger) -> None: - """Checks blob data for missing chunks or data - """ + """Checks blob data for missing chunks or data""" for blob_key in blobs: blob_index = blobs[blob_key][0] blob_chunks = blobs[blob_key][1:] @@ -416,20 +352,20 @@ def check_blob_data(nvs_log: NVS_Logger) -> None: # chunk: NVS_Entry # Blob missing chunk check if chunk is EMPTY_ENTRY: - nvs_log.info( - nvs_log.red(f'Blob {blob_index.key} is missing a chunk!'), - f'Namespace index: {blob_index.metadata["namespace"]:03d}', - f'[{found_namespaces.get(blob_index.metadata["namespace"], "undefined")}],', - f'Chunk Index: {i:03d}', + nvs_log.err( + f'Blob {blob_index.key} is missing a chunk! ' + f'Namespace index: {blob_index.metadata["namespace"]:03d} ' + f'[{found_namespaces.get(blob_index.metadata["namespace"], "undefined")}], ' + f'Chunk Index: {i:03d}' ) else: blob_size -= len(chunk.children) * nvs_const.entry_size # Blob missing data check if blob_size > 0: - nvs_log.info( - nvs_log.red(f'Blob {blob_index.key} is missing {blob_size} B of data!'), - f'Namespace index: {blob_index.metadata["namespace"]:03d}', + nvs_log.err( + f'Blob {blob_index.key} is missing {blob_size} B of data! ' + f'Namespace index: {blob_index.metadata["namespace"]:03d}' ) @@ -441,25 +377,16 @@ def check_blobs(nvs_log: NVS_Logger) -> None: def check_namespaces(nvs_log: NVS_Logger) -> None: - """Checks namespaces (entries using undefined namespace indexes, unused namespaces) - """ + """Checks namespaces (entries using undefined namespace indexes, unused namespaces)""" # Undefined namespace index check for used_ns in used_namespaces: key = found_namespaces.pop(used_ns, None) if key is None: - nvs_log.info( - nvs_log.red('Undefined namespace index!'), - f'Namespace index: {used_ns:03d}', - f'[undefined]', - ) + nvs_log.err(f'Undefined namespace index! Namespace index: {used_ns:03d} [undefined]') # Unused namespace index check for unused_ns in found_namespaces: - nvs_log.info( - nvs_log.yellow('Found unused namespace.'), - f'Namespace index: {unused_ns:03d}', - f'[{found_namespaces[unused_ns]}]', - ) + nvs_log.warn(f'Found unused namespace. Namespace index: {unused_ns:03d} [{found_namespaces[unused_ns]}]') def reset_global_variables() -> None: @@ -474,15 +401,14 @@ def reset_global_variables() -> None: def integrity_check(nvs_partition: NVS_Partition, nvs_log: NVS_Logger) -> None: - """Function for multi-stage integrity check of a NVS partition - """ + """Function for multi-stage integrity check of a NVS partition""" # Partition size check check_partition_size(nvs_partition, nvs_log) # Free/empty page check check_empty_page_present(nvs_partition, nvs_log) - seen_written_entires_all: Dict[str, List[NVS_Entry]] = {} + seen_written_entires_all: dict[str, list[NVS_Entry]] = {} # Loop through all pages in the partition for page in nvs_partition.pages: diff --git a/components/nvs_flash/nvs_partition_tool/nvs_logger.py b/components/nvs_flash/nvs_partition_tool/nvs_logger.py index d44971e071b..6313fe0770f 100644 --- a/components/nvs_flash/nvs_partition_tool/nvs_logger.py +++ b/components/nvs_flash/nvs_partition_tool/nvs_logger.py @@ -1,81 +1,65 @@ #!/usr/bin/env python3 -# SPDX-FileCopyrightText: 2022-2025 Espressif Systems (Shanghai) CO LTD +# SPDX-FileCopyrightText: 2022-2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 import binascii import json import sys from typing import Any -from typing import Dict -from typing import List -from typing import Union -from nvs_parser import nvs_const +from esp_pylib.logger import log from nvs_parser import NVS_Entry from nvs_parser import NVS_Partition +from nvs_parser import nvs_const class NVS_Logger: - ansi = { - 'red': '\033[31m', - 'green': '\033[32m', - 'yellow': '\033[33m', - 'blue': '\033[34m', - 'cyan': '\033[36m', - 'bold': '\033[1m', - 'clear': '\033[0m', - } + """Rich-based logger for NVS partition dump and integrity-check output.""" def __init__(self, *, color: str = 'auto', out_format: str = 'text'): - self.color = color == 'always' or (color == 'auto' and sys.stdout.isatty()) self.output_format = out_format + self.set_color(color) def set_color(self, color: str) -> None: - self.color = color == 'always' or (color == 'auto' and sys.stdout.isatty()) + no_color = color == 'never' or (color == 'auto' and not sys.stdout.isatty()) + self._color_enabled = not no_color def set_format(self, out_format: str) -> None: self.output_format = out_format - def info(self, *args, **kwargs) -> None: # type: ignore - kwargs['file'] = kwargs.get( - 'file', sys.stdout - ) # Set default output to be stdout, but can be overwritten - print(*args, **kwargs) + def info(self, *args: Any, **kwargs: Any) -> None: + log.print(*args, **kwargs) - def error(self, *args, **kwargs) -> None: # type: ignore - kwargs['file'] = kwargs.get( - 'file', sys.stderr - ) # Set default output to be stderr, but can be overwritten - print(*args, **kwargs) + def error(self, *args: Any, **kwargs: Any) -> None: + log.err(*args) + + def warn(self, message: str) -> None: + log.warn(message) + + def err(self, message: str) -> None: + log.err(message) + + def _style(self, text: str, style: str) -> str: + if self._color_enabled: + return f'[{style}]{text}[/{style}]' + return text def red(self, text: str) -> str: - if self.color: - return NVS_Logger.ansi['red'] + text + NVS_Logger.ansi['clear'] - return text + return self._style(text, 'red') def green(self, text: str) -> str: - if self.color: - return NVS_Logger.ansi['green'] + text + NVS_Logger.ansi['clear'] - return text + return self._style(text, 'green') def yellow(self, text: str) -> str: - if self.color: - return NVS_Logger.ansi['yellow'] + text + NVS_Logger.ansi['clear'] - return text + return self._style(text, 'yellow') def blue(self, text: str) -> str: - if self.color: - return NVS_Logger.ansi['blue'] + text + NVS_Logger.ansi['clear'] - return text + return self._style(text, 'blue') def cyan(self, text: str) -> str: - if self.color: - return NVS_Logger.ansi['cyan'] + text + NVS_Logger.ansi['clear'] - return text + return self._style(text, 'cyan') def bold(self, text: str) -> str: - if self.color: - return NVS_Logger.ansi['bold'] + text + NVS_Logger.ansi['clear'] - return text + return self._style(text, 'bold') nvs_log = NVS_Logger() @@ -135,13 +119,9 @@ def dump_everything(nvs_partition: NVS_Partition, written_only: bool = False) -> for page in nvs_partition.pages: # Print page header if page.is_empty: - nvs_log.info( - nvs_log.bold(f'Page Empty, Page address: 0x{page.start_address:x}') - ) + nvs_log.info(nvs_log.bold(f'Page Empty, Page address: 0x{page.start_address:x}')) else: - if ( - page.header['crc']['original'] == page.header['crc']['computed'] - ): # Color CRC32 + if page.header['crc']['original'] == page.header['crc']['computed']: # Color CRC32 crc = nvs_log.green(f'{page.header["crc"]["original"]: >8x}') else: crc = nvs_log.red(f'{page.header["crc"]["original"]: >8x}') @@ -168,21 +148,15 @@ def dump_everything(nvs_partition: NVS_Partition, written_only: bool = False) -> continue # Compress all empty entries - if ( - entry.state == 'Empty' and entry.is_empty - ): # Gather all subsequent empty entries + if entry.state == 'Empty' and entry.is_empty: # Gather all subsequent empty entries empty_entries.append(entry) continue else: # Print the empty entries if len(empty_entries) >= 3: # There is enough entries to compress - nvs_log.info( - nvs_log.bold(f' {empty_entries[0].index:03d}.'), 'Empty' - ) + nvs_log.info(nvs_log.bold(f' {empty_entries[0].index:03d}.'), 'Empty') nvs_log.info(nvs_log.bold(' ...')) - nvs_log.info( - nvs_log.bold(f' {empty_entries[-1].index:03d}.'), 'Empty' - ) + nvs_log.info(nvs_log.bold(f' {empty_entries[-1].index:03d}.'), 'Empty') else: # No need for compression for e in empty_entries: nvs_log.info(nvs_log.bold(f' {e.index:03d}.'), 'Empty') @@ -196,9 +170,7 @@ def dump_everything(nvs_partition: NVS_Partition, written_only: bool = False) -> status = nvs_log.red(f'{status: <7}') crc = '' - if ( - entry.metadata['crc']['original'] == entry.metadata['crc']['computed'] - ): # Color CRC32 + if entry.metadata['crc']['original'] == entry.metadata['crc']['computed']: # Color CRC32 crc = nvs_log.green(f'{entry.metadata["crc"]["original"]: >8x}') else: crc = nvs_log.red(f'{entry.metadata["crc"]["original"]: >8x}') @@ -234,13 +206,8 @@ def dump_everything(nvs_partition: NVS_Partition, written_only: bool = False) -> + f', ChunkStart={entry.data["chunk_start"]}' ) else: - if ( - entry.metadata['crc']['data_original'] - == entry.metadata['crc']['data_computed'] - ): # Color CRC32 - crc = nvs_log.green( - f'{entry.metadata["crc"]["data_original"]:x}' - ) + if entry.metadata['crc']['data_original'] == entry.metadata['crc']['data_computed']: # Color CRC32 + crc = nvs_log.green(f'{entry.metadata["crc"]["data_original"]:x}') else: crc = nvs_log.red(f'{entry.metadata["crc"]["data_original"]:x}') nvs_log.info(f'Size={entry.data["size"]}, CRC32={crc}') @@ -248,9 +215,7 @@ def dump_everything(nvs_partition: NVS_Partition, written_only: bool = False) -> # Dump all children entries if entry.metadata['span'] != 1: for i, data in enumerate(entry.children): - nvs_log.info( - f'{"": >6}0x{(i * nvs_const.entry_size):03x} {data.dump_raw()}' - ) + nvs_log.info(f'{"": >6}0x{(i * nvs_const.entry_size):03x} {data.dump_raw()}') # Dump trailing empty entries if len(empty_entries) >= 3: @@ -277,7 +242,7 @@ def list_namespaces(nvs_partition: NVS_Partition) -> None: ns[entry.data['value']] = entry.key # Print found namespaces - nvs_log.info(nvs_log.bold(f'Index: Namespace')) + nvs_log.info(nvs_log.bold('Index: Namespace')) for ns_index in sorted(ns): nvs_log.info(f' {ns_index:03d}:', nvs_log.cyan(ns[ns_index])) @@ -296,18 +261,11 @@ def dump_key_value_pairs(nvs_partition: NVS_Partition) -> None: if page.is_empty: nvs_log.info(nvs_log.bold('Page Empty')) else: - nvs_log.info( - nvs_log.bold( - f'Page no. {page.header["page_index"]}' - + f', Status: {page.header["status"]}' - ) - ) + nvs_log.info(nvs_log.bold(f'Page no. {page.header["page_index"]}' + f', Status: {page.header["status"]}')) # Print entries for entry in page.entries: - if ( - entry.state == 'Written' and entry.metadata['namespace'] != 0 - ): # Ignore non-written entries + if entry.state == 'Written' and entry.metadata['namespace'] != 0: # Ignore non-written entries chunk_index = '' data = '' if entry.metadata['type'] not in [ @@ -348,9 +306,9 @@ def dump_key_value_pairs(nvs_partition: NVS_Partition) -> None: def dump_written_blobs(nvs_partition: NVS_Partition) -> None: - blobs: Dict = {} - strings: List[NVS_Entry] = [] - legacy_blobs: List[NVS_Entry] = [] + blobs: dict = {} + strings: list[NVS_Entry] = [] + legacy_blobs: list[NVS_Entry] = [] ns = {} empty_entry = NVS_Entry(-1, bytearray(32), 'Erased') @@ -359,9 +317,9 @@ def dump_written_blobs(nvs_partition: NVS_Partition) -> None: for entry in page.entries: if entry.state == 'Written': if entry.metadata['type'] == 'blob_index': - blobs[f'{entry.metadata["namespace"]:03d}{entry.key}'] = [entry] + [ - empty_entry - ] * entry.data['chunk_count'] + blobs[f'{entry.metadata["namespace"]:03d}{entry.key}'] = [entry] + [empty_entry] * entry.data[ + 'chunk_count' + ] elif entry.metadata['type'] == 'blob': legacy_blobs.append(entry) elif entry.metadata['type'] == 'string': @@ -377,30 +335,21 @@ def dump_written_blobs(nvs_partition: NVS_Partition) -> None: if ( entry.state == 'Written' and entry.metadata['type'] != 'blob_index' - and entry.metadata['namespace'] - == blobs[key][0].metadata['namespace'] + and entry.metadata['namespace'] == blobs[key][0].metadata['namespace'] and entry.key == blobs[key][0].key ): - blobs[key][ - 1 - + entry.metadata['chunk_index'] - - blobs[key][0].data['chunk_start'] - ] = entry + blobs[key][1 + entry.metadata['chunk_index'] - blobs[key][0].data['chunk_start']] = entry blob_index = blobs[key][0] blob_chunks = blobs[key][1:] # Print blob info nvs_log.info( - nvs_log.cyan( - ns.get( - blob_index.metadata['namespace'], blob_index.metadata['namespace'] - ) - ) + nvs_log.cyan(ns.get(blob_index.metadata['namespace'], blob_index.metadata['namespace'])) + ':' + nvs_log.yellow(blob_index.key) + ' - ' - + f'Type: Blob (Version 2), ' + + 'Type: Blob (Version 2), ' + f'Size: {blob_index.data["size"]}' ) @@ -416,17 +365,13 @@ def dump_written_blobs(nvs_partition: NVS_Partition) -> None: if entry is empty_entry: nvs_log.info(nvs_log.yellow(f' {"":->63} Missing data {"":-<64}')) else: - nvs_log.info( - f' 0x{(i * nvs_const.entry_size):05x} {entry.dump_raw()}' - ) + nvs_log.info(f' 0x{(i * nvs_const.entry_size):05x} {entry.dump_raw()}') nvs_log.info() # Dump strings for string in strings: nvs_log.info( - nvs_log.cyan( - ns.get(string.metadata['namespace'], string.metadata['namespace']) - ) + nvs_log.cyan(ns.get(string.metadata['namespace'], string.metadata['namespace'])) + ':' + nvs_log.yellow(string.key) + ' - ' @@ -454,15 +399,14 @@ def dump_written_blobs(nvs_partition: NVS_Partition) -> None: def print_json(nvs: NVS_Partition) -> None: class NVSEncoder(json.JSONEncoder): - def default(self, obj: Any) -> Union[Any, Dict[str, Any], str]: + def default(self, obj: Any) -> Any | dict[str, Any] | str: if hasattr(obj, 'toJSON'): return obj.toJSON() if isinstance(obj, bytearray): - return binascii.b2a_base64(obj, newline=False).decode( - 'ascii' - ) # Binary to Base64 ASCII representation + return binascii.b2a_base64(obj, newline=False).decode('ascii') # Binary to Base64 ASCII representation return json.JSONEncoder.default(self, obj) + # Machine-readable JSON on stdout (not via Rich Console — pytest/capture-friendly). print(json.dumps(nvs.toJSON(), cls=NVSEncoder, indent=2)) @@ -501,12 +445,14 @@ def print_minimal_json(nvs_partition: NVS_Partition) -> None: data = binascii.b2a_base64(tmp, newline=False).decode('ascii') if entry.metadata['namespace'] in ns: - key_value_pairs.append({ - 'namespace': ns[entry.metadata['namespace']], - 'key': entry.key, - 'encoding': entry_type, # Add type of data - 'data': data, # Ensure data ends with a newline - 'state': entry.state, - 'is_empty': entry.is_empty if hasattr(entry, 'is_empty') else None, - }) - nvs_log.info(json.dumps(key_value_pairs, indent=4)) + key_value_pairs.append( + { + 'namespace': ns[entry.metadata['namespace']], + 'key': entry.key, + 'encoding': entry_type, # Add type of data + 'data': data, # Ensure data ends with a newline + 'state': entry.state, + 'is_empty': entry.is_empty if hasattr(entry, 'is_empty') else None, + } + ) + print(json.dumps(key_value_pairs, indent=4)) diff --git a/components/nvs_flash/nvs_partition_tool/nvs_tool.py b/components/nvs_flash/nvs_partition_tool/nvs_tool.py index 5f0e26cff37..702bbc6fa87 100755 --- a/components/nvs_flash/nvs_partition_tool/nvs_tool.py +++ b/components/nvs_flash/nvs_partition_tool/nvs_tool.py @@ -1,100 +1,65 @@ #!/usr/bin/env python3 -# SPDX-FileCopyrightText: 2022-2025 Espressif Systems (Shanghai) CO LTD +# SPDX-FileCopyrightText: 2022-2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 -import argparse import os -import sys -import traceback +from collections.abc import Callable +from typing import Any import nvs_check import nvs_logger import nvs_parser +from esp_pylib.logger import log from nvs_logger import nvs_log from nvs_parser import nvs_const - -def program_args() -> argparse.Namespace: - parser = argparse.ArgumentParser( - description='Parse NVS partition', formatter_class=argparse.RawTextHelpFormatter - ) - parser.add_argument('file', help='Path to dumped NVS partition') - parser.add_argument( - '-i', - '--integrity-check', - action='store_true', - help='check partition for potential errors', - ) - tmp = { - 'all': 'print written, erased and empty entries', - 'written': 'print only currently written entries', - 'minimal': 'print only namespace:key=value pairs', - 'blobs': 'print all blobs and strings', - 'namespaces': 'list all written namespaces', - 'storage_info': 'print storage related information (free/used entries, etc)', - 'none': 'do not print anything (if you only want to do integrity check)', - } - parser.add_argument( - '-d', - '--dump', - choices=tmp, - default='all', - metavar='type', - help=( - f"""type: {str(list(tmp.keys()))[1:-1]} -{os.linesep.join([f'{opt} - {tmp[opt]}' for opt in tmp])}""" - ), - ) - parser.add_argument( - '--color', - choices=['never', 'auto', 'always'], - default='auto', - help='Enable color (ANSI)', - ) - tmp = { - 'text': 'print output as a human-readable text', - 'json': 'print output as JSON and exit', - } - parser.add_argument( - '-f', '--format', choices=tmp, default='text', help='Output format' - ) - return parser.parse_args() +_DUMP_CHOICES = [ + 'all', + 'written', + 'minimal', + 'blobs', + 'namespaces', + 'storage_info', + 'none', +] +_DUMP_HELP = { + 'all': 'print written, erased and empty entries', + 'written': 'print only currently written entries', + 'minimal': 'print only namespace:key=value pairs', + 'blobs': 'print all blobs and strings', + 'namespaces': 'list all written namespaces', + 'storage_info': 'print storage related information (free/used entries, etc)', + 'none': 'do not print anything (if you only want to do integrity check)', +} +_FORMAT_CHOICES = ['text', 'json'] -def main() -> None: - args = program_args() +def _run(file: str, integrity_check: bool, dump: str, color: str, out_format: str) -> None: + nvs_log.set_color(color) + nvs_log.set_format(out_format) if nvs_const.entry_size != 32: - raise ValueError(f'Entry size is not 32B! This is currently non negotiable.') - - nvs_log.set_color(args.color) - nvs_log.set_format(args.format) + log.die('Entry size is not 32B! This is currently non negotiable.') try: - with open(args.file, 'rb') as f: + with open(file, 'rb') as f: partition = f.read() - except IndexError: - nvs_log.error('No file given') - raise except FileNotFoundError: - nvs_log.error('Bad filename') - raise + log.die(f'File not found: {file}') + except OSError as e: + log.die(f'Cannot read file {file}: {e}') - nvs = nvs_parser.NVS_Partition(args.file.split('/')[-1], bytearray(partition)) + try: + nvs = nvs_parser.NVS_Partition(os.path.basename(file), bytearray(partition)) + except nvs_parser.NotAlignedError as e: + log.die(str(e)) + except nvs_parser.NVS_Constants.ConstantError as e: + log.die(str(e)) def noop(_: nvs_parser.NVS_Partition) -> None: pass - def format_not_implemented(_: nvs_parser.NVS_Partition) -> None: - raise RuntimeError(f'{args.format} is not implemented') - - def cmd_not_implemented(_: nvs_parser.NVS_Partition) -> None: - raise RuntimeError(f'{args.dump} is not implemented') - - if args.format not in ['text', 'json']: - format_not_implemented(nvs) - - cmds = {} - if args.format == 'text': + cmds: dict[str, Callable[[nvs_parser.NVS_Partition], None]] = {} + if out_format == 'text': cmds = { 'all': nvs_logger.dump_everything, 'written': nvs_logger.dump_written_entries, @@ -104,27 +69,78 @@ def main() -> None: 'storage_info': nvs_logger.storage_stats, 'none': noop, } - - if args.format == 'json': + elif out_format == 'json': cmds = { 'all': nvs_logger.print_json, 'minimal': nvs_logger.print_minimal_json, 'none': noop, } + else: + log.die(f'{out_format} is not implemented') - cmds.get(args.dump, cmd_not_implemented)(nvs) + handler = cmds.get(dump) + if handler is None: + log.die(f'{dump} is not implemented for format {out_format}') + return + handler(nvs) - if args.integrity_check: + if integrity_check: nvs_log.info() nvs_check.integrity_check(nvs, nvs_log) +def _build_cli() -> Any: + import rich_click as click + + dump_help = os.linesep.join(f'{opt} - {_DUMP_HELP[opt]}' for opt in _DUMP_CHOICES) + + @click.command( + context_settings={'help_option_names': ['-h', '--help']}, + help='Parse NVS partition', + ) + @click.argument('file', type=click.Path(exists=True, dir_okay=False)) + @click.option( + '-i', + '--integrity-check', + is_flag=True, + help='Check partition for potential errors', + ) + @click.option( + '-d', + '--dump', + type=click.Choice(_DUMP_CHOICES), + default='all', + show_default=True, + help=f'Dump type:{os.linesep}{dump_help}', + ) + @click.option( + '--color', + type=click.Choice(['never', 'auto', 'always']), + default='auto', + show_default=True, + help='Enable color output', + ) + @click.option( + '-f', + '--format', + 'out_format', + type=click.Choice(_FORMAT_CHOICES), + default='text', + show_default=True, + help='Output format', + ) + def cli(file: str, integrity_check: bool, dump: str, color: str, out_format: str) -> None: + _run(file, integrity_check, dump, color, out_format) + + return cli + + +def main() -> None: + _build_cli()() + + if __name__ == '__main__': - try: - main() - except ValueError: - traceback.print_exc(file=sys.stderr) - sys.exit(1) - except nvs_parser.NVS_Constants.ConstantError: - traceback.print_exc(file=sys.stderr) - sys.exit(1) + from esp_pylib.excepthook import install_exception_reporting + + install_exception_reporting() + main() diff --git a/components/nvs_flash/nvs_partition_tool/test_nvs_gen_check.py b/components/nvs_flash/nvs_partition_tool/test_nvs_gen_check.py index 2244b3b22c1..a3ac6a0870d 100644 --- a/components/nvs_flash/nvs_partition_tool/test_nvs_gen_check.py +++ b/components/nvs_flash/nvs_partition_tool/test_nvs_gen_check.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -# 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 json @@ -15,7 +15,6 @@ import esp_idf_nvs_partition_gen.nvs_partition_gen as nvs_partition_gen import nvs_check as nvs_check import pytest from esp_idf_nvs_partition_gen.nvs_partition_gen import NVS -from nvs_logger import NVS_Logger from nvs_logger import nvs_log from nvs_logger import print_minimal_json from nvs_parser import NVS_Entry @@ -34,16 +33,7 @@ def before() -> None: pytest.skip('pass') -class SilentLogger(NVS_Logger): - def __init__(self) -> None: - super().__init__() - self.color = False - - def info(self, *args, **kwargs) -> None: # type: ignore - pass - - -logger = nvs_log # SilentLogger() +logger = nvs_log LOREM_STRING = """Lorem ipsum dolor sit amet, consectetur adipiscing elit. Nullam eget orci fringilla, cursus nisi sit amet, hendrerit tortor.