Mirrors/esp-idf
1
0
Fork 0
mirror of https://github.com/espressif/esp-idf.git synced 2026-10-02 11:10:54 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(nvs_flash): normalize nvs partition tool CLI output

Browse Source
This commit is contained in:
sonika.rathi
2026-06-22 10:42:51 +02:00
parent be650228e8
commit e9315fa19f
4 changed files with 268 additions and 390 deletions
Download Patch File Download Diff File Expand all files Collapse all files
components/nvs_flash/nvs_partition_tool/nvs_check.py
+96 -170
View File
@@ -1,62 +1,41 @@
#!/usr/bin/env python3 #!/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 # 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_logger import NVS_Logger
from nvs_parser import nvs_const
from nvs_parser import NVS_Entry from nvs_parser import NVS_Entry
from nvs_parser import NVS_Page from nvs_parser import NVS_Page
from nvs_parser import NVS_Partition from nvs_parser import NVS_Partition
from nvs_parser import nvs_const
EMPTY_ENTRY = NVS_Entry(-1, bytearray(32), 'Erased') EMPTY_ENTRY = NVS_Entry(-1, bytearray(32), 'Erased')
used_namespaces: Dict[int, Optional[str]] = {} used_namespaces: dict[int, str | None] = {}
found_namespaces: Dict[int, str] = {} found_namespaces: dict[int, str] = {}
blobs: Dict = {} blobs: dict = {}
blob_chunks: List[NVS_Entry] = [] blob_chunks: list[NVS_Entry] = []
def check_partition_size(nvs_partition: NVS_Partition, nvs_log: NVS_Logger, read_only: bool = False) -> bool: 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 """Checks if the partition is large enough and has enough pages"""
"""
if len(nvs_partition.raw_data) // 0x1000 < 3 and not read_only: if len(nvs_partition.raw_data) // 0x1000 < 3 and not read_only:
nvs_log.info( nvs_log.warn('NVS Partition size must be at least 0x3000 (4kiB * 3 pages == 12kiB)!')
nvs_log.yellow(
'NVS Partition size must be at least 0x3000 (4kiB * 3 pages == 12kiB)!'
)
)
return False return False
if len(nvs_partition.raw_data) % 0x1000 != 0: if len(nvs_partition.raw_data) % 0x1000 != 0:
nvs_log.info( nvs_log.warn('NVS Partition size must be a multiple of 0x1000 (4kiB)!')
nvs_log.yellow(
'NVS Partition size must be a multiple of 0x1000 (4kiB)!'
)
)
return False return False
if len(nvs_partition.pages) < 3 and not read_only: if len(nvs_partition.pages) < 3 and not read_only:
nvs_log.info( nvs_log.warn('NVS Partition must contain 3 pages (sectors) at least to function properly!')
nvs_log.yellow(
'NVS Partition must contain 3 pages (sectors) at least to function properly!'
)
)
return False return False
return True return True
def check_empty_page_present(nvs_partition: NVS_Partition, nvs_log: NVS_Logger) -> bool: 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): if not any(page.header['status'] == 'Empty' for page in nvs_partition.pages):
nvs_log.info( nvs_log.err(
nvs_log.red( 'No free (empty) page found in the NVS partition, at least one free page is required for proper function!'
'''No free (empty) page found in the NVS partition,
at least one free page is required for proper function!'''
) )
) nvs_log.err('NVS partition possibly truncated?')
nvs_log.info(nvs_log.red('NVS partition possibly truncated?\n'))
return False return False
return True 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: if nvs_page.raw_entry_state_bitmap != bytearray({0xFF}) * nvs_const.entry_size:
result = False result = False
nvs_log.info( nvs_log.err('The page is reported as Empty but its entry state bitmap is not empty!')
nvs_log.red(
'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]): if any([not e.is_empty for e in nvs_page.entries]):
result = False result = False
nvs_log.info( nvs_log.err('The page is reported as Empty but there are data written!')
nvs_log.red('The page is reported as Empty but there are data written!')
)
return result return result
def check_page_crc(nvs_page: NVS_Page, nvs_log: NVS_Logger) -> bool: def check_page_crc(nvs_page: NVS_Page, nvs_log: NVS_Logger) -> bool:
if nvs_page.header['crc']['original'] == nvs_page.header['crc']['computed']: if nvs_page.header['crc']['original'] == nvs_page.header['crc']['computed']:
nvs_log.info( nvs_log.info(nvs_log.cyan(f'Page no. {nvs_page.header["page_index"]}'), '\tCRC32: OK')
nvs_log.cyan(f'Page no. {nvs_page.header["page_index"]}'), '\tCRC32: OK'
)
return True return True
else: else:
nvs_log.info( nvs_log.info(
nvs_log.cyan(f'Page no. {nvs_page.header["page_index"]}'), 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}'), 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}'), nvs_log.green(f'{nvs_page.header["crc"]["computed"]:x}'),
) )
return False 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 """Identifies and logs written entries
Part 1 of duplicate entry check mechanism 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 return entry_dict
def check_page_entries(nvs_page: NVS_Page, nvs_log: NVS_Logger) -> 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 (entry state, children CRC32, entry type, span and gathers blobs and namespaces) """Checks entries in the given page (state, CRC32, type, span; gathers blobs and namespaces)."""
""" seen_written_entires: dict[str, list[NVS_Entry]] = {}
seen_written_entires: Dict[str, List[NVS_Entry]] = {}
for entry in nvs_page.entries: for entry in nvs_page.entries:
# entry: NVS_Entry # 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 # Duplicate entry check (1) - same key, different index - find duplicates
seen_written_entires = identify_entry_duplicates(entry, seen_written_entires) 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.is_empty:
if entry.state == 'Written': if entry.state == 'Written':
nvs_log.info( nvs_log.err(f' Entry #{entry.index:03d} is reported as Written but it is empty!')
nvs_log.red(
f' Entry #{entry.index:03d} is reported as Written but it is empty!'
)
)
continue continue
elif entry.state == 'Erased': elif entry.state == 'Erased':
nvs_log.info( nvs_log.warn(
nvs_log.yellow( f' Entry #{entry.index:03d} is reported as Erased but it is empty! '
f' Entry #{entry.index:03d} is reported as Erased but it is empty! (Only entries reported as Empty should be empty)' '(Only entries reported as Empty should be empty)'
)
) )
if entry.state == 'Written': if entry.state == 'Written':
# Entry CRC32 check # Entry CRC32 check
if ( if entry.metadata['crc']['original'] != entry.metadata['crc']['computed']:
entry.metadata['crc']['original']
!= entry.metadata['crc']['computed']
):
nvs_log.info( nvs_log.info(
nvs_log.red( nvs_log.red(f' Entry #{entry.index:03d} {entry.key} has wrong CRC32!'),
f' Entry #{entry.index:03d} {entry.key} has wrong CRC32!{"": <5}' 'Written:',
),
f'Written:',
nvs_log.red(f'{entry.metadata["crc"]["original"]:x}'), nvs_log.red(f'{entry.metadata["crc"]["original"]:x}'),
f'Generated:', 'Generated:',
nvs_log.green(f'{entry.metadata["crc"]["computed"]:x}'), nvs_log.green(f'{entry.metadata["crc"]["computed"]:x}'),
) )
# Entry children CRC32 check # Entry children CRC32 check
if ( if entry.metadata['span'] > 1 and (
entry.metadata['span'] > 1 entry.metadata['crc']['data_original'] != entry.metadata['crc']['data_computed']
and (entry.metadata['crc']['data_original'] != entry.metadata['crc']['data_computed'])
): ):
nvs_log.info( nvs_log.info(
nvs_log.red( nvs_log.red(f' Entry #{entry.index:03d} {entry.key} data (string, blob) has wrong CRC32!'),
f' Entry #{entry.index:03d} {entry.key} data (string, blob) has wrong CRC32!' 'Written:',
),
f'Written:',
nvs_log.red(f'{entry.metadata["crc"]["data_original"]:x}'), nvs_log.red(f'{entry.metadata["crc"]["data_original"]:x}'),
f'Generated:', 'Generated:',
nvs_log.green(f'{entry.metadata["crc"]["data_computed"]:x}'), nvs_log.green(f'{entry.metadata["crc"]["data_computed"]:x}'),
) )
# Entry type check # Entry type check
if entry.metadata['type'] not in [ if entry.metadata['type'] not in [nvs_const.item_type[key] for key in nvs_const.item_type]:
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"]}'
nvs_log.info(
nvs_log.yellow(
f' Type of entry #{entry.index:03d} {entry.key} is unrecognized!'
),
f'Type: {entry.metadata["type"]}',
) )
# Span check # Span check
if ( if entry.index + entry.metadata['span'] - 1 >= int(nvs_const.page_size / nvs_const.entry_size) - 2:
entry.index + entry.metadata['span'] - 1 nvs_log.err(f' Variable length entry #{entry.index:03d} {entry.key} is out of bounds!')
>= 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!'
)
)
# Spanned entry state checks # Spanned entry state checks
elif entry.metadata['span'] > 1: elif entry.metadata['span'] > 1:
parent_state = entry.state parent_state = entry.state
for kid in entry.children: for kid in entry.children:
if parent_state != kid.state: if parent_state != kid.state:
nvs_log.info( nvs_log.warn(
nvs_log.yellow(' Inconsistent data state!'), f' Inconsistent data state! Entry #{entry.index:03d} {entry.key} '
f'Entry #{entry.index:03d} {entry.key} state: {parent_state},', f'state: {parent_state}, Data entry #{kid.index:03d} {entry.key} '
f'Data entry #{kid.index:03d} {entry.key} state: {kid.state}', f'state: {kid.state}'
) )
# Gather blobs & namespaces # Gather blobs & namespaces
if entry.metadata['type'] == 'blob_index': if entry.metadata['type'] == 'blob_index':
blobs[f'{entry.metadata["namespace"]:03d}{entry.key}'] = [entry] + [ blobs[f'{entry.metadata["namespace"]:03d}{entry.key}'] = [entry] + [EMPTY_ENTRY] * entry.data[
EMPTY_ENTRY 'chunk_count'
] * entry.data['chunk_count'] ]
elif entry.metadata['type'] == 'blob_data': elif entry.metadata['type'] == 'blob_data':
blob_chunks.append(entry) 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 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, """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 where entries which have the same key but under different namespaces are filtered out
Use `filter_entry_duplicates()` to properly filter out all duplicates 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(): for key, duplicate_entries in duplicate_entries_dict.items():
seen_entries: List[NVS_Entry] = [] seen_entries: list[NVS_Entry] = []
entry_same_namespace_collisions_list: Set[NVS_Entry] = set() entry_same_namespace_collisions_list: set[NVS_Entry] = set()
# Search through the "duplicates" and see if there are real duplicates # Search through the "duplicates" and see if there are real duplicates
# E.g. the key can be the same if the namespace is different # E.g. the key can be the same if the namespace is different
for entry in duplicate_entries: for entry in duplicate_entries:
if entry.metadata['type'] in nvs_const.item_type.values(): if entry.metadata['type'] in nvs_const.item_type.values():
entry_same_namespace_collisions = set() entry_same_namespace_collisions = set()
for other_entry in seen_entries: for other_entry in seen_entries:
if entry.metadata['namespace'] == other_entry.metadata['namespace']: 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) seen_entries.append(entry)
# Catch real duplicates # Catch real duplicates
new_duplicate_entries: List[NVS_Entry] = [] new_duplicate_entries: list[NVS_Entry] = []
if len(seen_entries) > 1: if len(seen_entries) > 1:
for entry in seen_entries: for entry in seen_entries:
if entry in entry_same_namespace_collisions_list: 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 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, """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 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 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(): for key, duplicate_entries in duplicate_entries_dict.items():
seen_blob_index: List[NVS_Entry] = [] seen_blob_index: list[NVS_Entry] = []
seen_blob_data: List[NVS_Entry] = [] seen_blob_data: list[NVS_Entry] = []
seen_another_type_data: List[NVS_Entry] = [] seen_another_type_data: list[NVS_Entry] = []
blob_index_chunk_index_collisions_list: Set[NVS_Entry] = set() blob_index_chunk_index_collisions_list: set[NVS_Entry] = set()
blob_data_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 # 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 # 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) # (and even for more blob_data entries if they have a different chunk_index)
for entry in duplicate_entries: for entry in duplicate_entries:
if entry.metadata['type'] == 'blob_index': if entry.metadata['type'] == 'blob_index':
blob_index_chunk_index_collisions = set() blob_index_chunk_index_collisions = set()
for other_entry in seen_blob_index: for other_entry in seen_blob_index:
if entry.metadata['namespace'] == other_entry.metadata['namespace']: 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) seen_blob_index.append(entry)
elif entry.metadata['type'] == 'blob_data': elif entry.metadata['type'] == 'blob_data':
blob_data_chunk_index_collisions = set() blob_data_chunk_index_collisions = set()
for other_entry in seen_blob_data: for other_entry in seen_blob_data:
if (entry.metadata['namespace'] == other_entry.metadata['namespace'] if (
and entry.metadata['chunk_index'] == other_entry.metadata['chunk_index']): 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(entry)
blob_data_chunk_index_collisions.add(other_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) seen_another_type_data.append(entry)
# Catch real duplicates # Catch real duplicates
new_duplicate_entries: List[NVS_Entry] = [] new_duplicate_entries: list[NVS_Entry] = []
if len(seen_blob_index) > 1: if len(seen_blob_index) > 1:
for entry in seen_blob_index: for entry in seen_blob_index:
if entry in blob_index_chunk_index_collisions_list: if entry in blob_index_chunk_index_collisions_list:
@@ -330,27 +274,24 @@ def filter_blob_related_duplicates(duplicate_entries_dict: Dict[str, List[NVS_En
return new_duplicate_entries_dict return new_duplicate_entries_dict
def filter_entry_duplicates(entries: Dict[str, List[NVS_Entry]]) -> Dict[str, List[NVS_Entry]]: 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 """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 Allowed: same key in different namespaces; blob_index and blob_data sharing a key.
and should be filtered out)
Part 2 of duplicate entry check mechanism
""" """
# Only keep seen written entries which have been observerd multiple times (duplicates) # 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} 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) # Filter out "fake" duplicates 1 (duplicate keys under different namespaces are allowed)
duplicate_entries_list_1 = filter_namespaces_fake_duplicates(duplicate_entries_list) 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) duplicate_entries_list_2 = filter_blob_related_duplicates(duplicate_entries_list_1)
return duplicate_entries_list_2 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: if len(duplicate_entries_list) > 0:
nvs_log.info(nvs_log.red('Found duplicate entries:')) nvs_log.err('Found duplicate entries:')
nvs_log.info(nvs_log.red('Entry\tKey\t\t\tType\t\tNamespace idx\tPage\tPage status')) nvs_log.err('Entry\tKey\t\t\tType\t\tNamespace idx\tPage\tPage status')
for _, duplicate_entries in duplicate_entries_list.items(): for _, duplicate_entries in duplicate_entries_list.items():
# duplicate_entries: List[NVS_Entry] # 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) namepace_tab = '\t' * (2 - namespace_tab_cnt)
namespace_str = f'{entry.metadata["namespace"]}' namespace_str = f'{entry.metadata["namespace"]}'
nvs_log.info( nvs_log.err(
nvs_log.red( f'#{entry.index:03d}\t{entry.key}{entry_key_tab}{entry_type}{namepace_tab}'
f'#{entry.index:03d}\t{entry.key}{entry_key_tab}{entry_type}{namepace_tab}{namespace_str}\t\t{page_num}\t{page_status}' f'{namespace_str}\t\t{page_num}\t{page_status}'
)
) )
def assemble_blobs(nvs_log: NVS_Logger) -> None: 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: for chunk in blob_chunks:
# chunk: NVS_Entry # chunk: NVS_Entry
parent = blobs.get( parent = blobs.get(f'{chunk.metadata["namespace"]:03d}{chunk.key}', [EMPTY_ENTRY])[0]
f'{chunk.metadata["namespace"]:03d}{chunk.key}', [EMPTY_ENTRY]
)[0]
# Blob chunk without blob index check # Blob chunk without blob index check
if parent is EMPTY_ENTRY: if parent is EMPTY_ENTRY:
nvs_log.info( nvs_log.err(
nvs_log.red(f'Blob {chunk.key} chunk has no blob index!'), f'Blob {chunk.key} chunk has no blob index! '
f'Namespace index: {chunk.metadata["namespace"]:03d}', f'Namespace index: {chunk.metadata["namespace"]:03d} '
f'[{found_namespaces.get(chunk.metadata["namespace"], "undefined")}],', f'[{found_namespaces.get(chunk.metadata["namespace"], "undefined")}], '
f'Chunk Index: {chunk.metadata["chunk_index"]:03d}', f'Chunk Index: {chunk.metadata["chunk_index"]:03d}'
) )
else: else:
blob_key = f'{chunk.metadata["namespace"]:03d}{chunk.key}' 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: 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: for blob_key in blobs:
blob_index = blobs[blob_key][0] blob_index = blobs[blob_key][0]
blob_chunks = blobs[blob_key][1:] blob_chunks = blobs[blob_key][1:]
@@ -416,20 +352,20 @@ def check_blob_data(nvs_log: NVS_Logger) -> None:
# chunk: NVS_Entry # chunk: NVS_Entry
# Blob missing chunk check # Blob missing chunk check
if chunk is EMPTY_ENTRY: if chunk is EMPTY_ENTRY:
nvs_log.info( nvs_log.err(
nvs_log.red(f'Blob {blob_index.key} is missing a chunk!'), f'Blob {blob_index.key} is missing a chunk! '
f'Namespace index: {blob_index.metadata["namespace"]:03d}', f'Namespace index: {blob_index.metadata["namespace"]:03d} '
f'[{found_namespaces.get(blob_index.metadata["namespace"], "undefined")}],', f'[{found_namespaces.get(blob_index.metadata["namespace"], "undefined")}], '
f'Chunk Index: {i:03d}', f'Chunk Index: {i:03d}'
) )
else: else:
blob_size -= len(chunk.children) * nvs_const.entry_size blob_size -= len(chunk.children) * nvs_const.entry_size
# Blob missing data check # Blob missing data check
if blob_size > 0: if blob_size > 0:
nvs_log.info( nvs_log.err(
nvs_log.red(f'Blob {blob_index.key} is missing {blob_size} B of data!'), f'Blob {blob_index.key} is missing {blob_size} B of data! '
f'Namespace index: {blob_index.metadata["namespace"]:03d}', 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: 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 # Undefined namespace index check
for used_ns in used_namespaces: for used_ns in used_namespaces:
key = found_namespaces.pop(used_ns, None) key = found_namespaces.pop(used_ns, None)
if key is None: if key is None:
nvs_log.info( nvs_log.err(f'Undefined namespace index! Namespace index: {used_ns:03d} [undefined]')
nvs_log.red('Undefined namespace index!'),
f'Namespace index: {used_ns:03d}',
f'[undefined]',
)
# Unused namespace index check # Unused namespace index check
for unused_ns in found_namespaces: for unused_ns in found_namespaces:
nvs_log.info( nvs_log.warn(f'Found unused namespace. Namespace index: {unused_ns:03d} [{found_namespaces[unused_ns]}]')
nvs_log.yellow('Found unused namespace.'),
f'Namespace index: {unused_ns:03d}',
f'[{found_namespaces[unused_ns]}]',
)
def reset_global_variables() -> None: 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: 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 # Partition size check
check_partition_size(nvs_partition, nvs_log) check_partition_size(nvs_partition, nvs_log)
# Free/empty page check # Free/empty page check
check_empty_page_present(nvs_partition, nvs_log) 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 # Loop through all pages in the partition
for page in nvs_partition.pages: for page in nvs_partition.pages:
components/nvs_flash/nvs_partition_tool/nvs_logger.py
+60 -114
View File
@@ -1,81 +1,65 @@
#!/usr/bin/env python3 #!/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 # SPDX-License-Identifier: Apache-2.0
import binascii import binascii
import json import json
import sys import sys
from typing import Any 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_Entry
from nvs_parser import NVS_Partition from nvs_parser import NVS_Partition
from nvs_parser import nvs_const
class NVS_Logger: class NVS_Logger:
ansi = { """Rich-based logger for NVS partition dump and integrity-check output."""
'red': '\033[31m',
'green': '\033[32m',
'yellow': '\033[33m',
'blue': '\033[34m',
'cyan': '\033[36m',
'bold': '\033[1m',
'clear': '\033[0m',
}
def __init__(self, *, color: str = 'auto', out_format: str = 'text'): 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.output_format = out_format
self.set_color(color)
def set_color(self, color: str) -> None: 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: def set_format(self, out_format: str) -> None:
self.output_format = out_format self.output_format = out_format
def info(self, *args, **kwargs) -> None: # type: ignore def info(self, *args: Any, **kwargs: Any) -> None:
kwargs['file'] = kwargs.get( log.print(*args, **kwargs)
'file', sys.stdout
) # Set default output to be stdout, but can be overwritten
print(*args, **kwargs)
def error(self, *args, **kwargs) -> None: # type: ignore def error(self, *args: Any, **kwargs: Any) -> None:
kwargs['file'] = kwargs.get( log.err(*args)
'file', sys.stderr
) # Set default output to be stderr, but can be overwritten def warn(self, message: str) -> None:
print(*args, **kwargs) 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: def red(self, text: str) -> str:
if self.color: return self._style(text, 'red')
return NVS_Logger.ansi['red'] + text + NVS_Logger.ansi['clear']
return text
def green(self, text: str) -> str: def green(self, text: str) -> str:
if self.color: return self._style(text, 'green')
return NVS_Logger.ansi['green'] + text + NVS_Logger.ansi['clear']
return text
def yellow(self, text: str) -> str: def yellow(self, text: str) -> str:
if self.color: return self._style(text, 'yellow')
return NVS_Logger.ansi['yellow'] + text + NVS_Logger.ansi['clear']
return text
def blue(self, text: str) -> str: def blue(self, text: str) -> str:
if self.color: return self._style(text, 'blue')
return NVS_Logger.ansi['blue'] + text + NVS_Logger.ansi['clear']
return text
def cyan(self, text: str) -> str: def cyan(self, text: str) -> str:
if self.color: return self._style(text, 'cyan')
return NVS_Logger.ansi['cyan'] + text + NVS_Logger.ansi['clear']
return text
def bold(self, text: str) -> str: def bold(self, text: str) -> str:
if self.color: return self._style(text, 'bold')
return NVS_Logger.ansi['bold'] + text + NVS_Logger.ansi['clear']
return text
nvs_log = NVS_Logger() 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: for page in nvs_partition.pages:
# Print page header # Print page header
if page.is_empty: if page.is_empty:
nvs_log.info( nvs_log.info(nvs_log.bold(f'Page Empty, Page address: 0x{page.start_address:x}'))
nvs_log.bold(f'Page Empty, Page address: 0x{page.start_address:x}')
)
else: else:
if ( if page.header['crc']['original'] == page.header['crc']['computed']: # Color CRC32
page.header['crc']['original'] == page.header['crc']['computed']
): # Color CRC32
crc = nvs_log.green(f'{page.header["crc"]["original"]: >8x}') crc = nvs_log.green(f'{page.header["crc"]["original"]: >8x}')
else: else:
crc = nvs_log.red(f'{page.header["crc"]["original"]: >8x}') 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 continue
# Compress all empty entries # Compress all empty entries
if ( if entry.state == 'Empty' and entry.is_empty: # Gather all subsequent empty entries
entry.state == 'Empty' and entry.is_empty
): # Gather all subsequent empty entries
empty_entries.append(entry) empty_entries.append(entry)
continue continue
else: else:
# Print the empty entries # Print the empty entries
if len(empty_entries) >= 3: # There is enough entries to compress if len(empty_entries) >= 3: # There is enough entries to compress
nvs_log.info( nvs_log.info(nvs_log.bold(f' {empty_entries[0].index:03d}.'), 'Empty')
nvs_log.bold(f' {empty_entries[0].index:03d}.'), 'Empty'
)
nvs_log.info(nvs_log.bold(' ...')) nvs_log.info(nvs_log.bold(' ...'))
nvs_log.info( nvs_log.info(nvs_log.bold(f' {empty_entries[-1].index:03d}.'), 'Empty')
nvs_log.bold(f' {empty_entries[-1].index:03d}.'), 'Empty'
)
else: # No need for compression else: # No need for compression
for e in empty_entries: for e in empty_entries:
nvs_log.info(nvs_log.bold(f' {e.index:03d}.'), 'Empty') 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}') status = nvs_log.red(f'{status: <7}')
crc = '' crc = ''
if ( if entry.metadata['crc']['original'] == entry.metadata['crc']['computed']: # Color CRC32
entry.metadata['crc']['original'] == entry.metadata['crc']['computed']
): # Color CRC32
crc = nvs_log.green(f'{entry.metadata["crc"]["original"]: >8x}') crc = nvs_log.green(f'{entry.metadata["crc"]["original"]: >8x}')
else: else:
crc = nvs_log.red(f'{entry.metadata["crc"]["original"]: >8x}') 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"]}' + f', ChunkStart={entry.data["chunk_start"]}'
) )
else: else:
if ( if entry.metadata['crc']['data_original'] == entry.metadata['crc']['data_computed']: # Color CRC32
entry.metadata['crc']['data_original'] crc = nvs_log.green(f'{entry.metadata["crc"]["data_original"]:x}')
== entry.metadata['crc']['data_computed']
): # Color CRC32
crc = nvs_log.green(
f'{entry.metadata["crc"]["data_original"]:x}'
)
else: else:
crc = nvs_log.red(f'{entry.metadata["crc"]["data_original"]:x}') crc = nvs_log.red(f'{entry.metadata["crc"]["data_original"]:x}')
nvs_log.info(f'Size={entry.data["size"]}, CRC32={crc}') 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 # Dump all children entries
if entry.metadata['span'] != 1: if entry.metadata['span'] != 1:
for i, data in enumerate(entry.children): for i, data in enumerate(entry.children):
nvs_log.info( nvs_log.info(f'{"": >6}0x{(i * nvs_const.entry_size):03x} {data.dump_raw()}')
f'{"": >6}0x{(i * nvs_const.entry_size):03x} {data.dump_raw()}'
)
# Dump trailing empty entries # Dump trailing empty entries
if len(empty_entries) >= 3: if len(empty_entries) >= 3:
@@ -277,7 +242,7 @@ def list_namespaces(nvs_partition: NVS_Partition) -> None:
ns[entry.data['value']] = entry.key ns[entry.data['value']] = entry.key
# Print found namespaces # 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): for ns_index in sorted(ns):
nvs_log.info(f' {ns_index:03d}:', nvs_log.cyan(ns[ns_index])) 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: if page.is_empty:
nvs_log.info(nvs_log.bold('Page Empty')) nvs_log.info(nvs_log.bold('Page Empty'))
else: else:
nvs_log.info( nvs_log.info(nvs_log.bold(f'Page no. {page.header["page_index"]}' + f', Status: {page.header["status"]}'))
nvs_log.bold(
f'Page no. {page.header["page_index"]}'
+ f', Status: {page.header["status"]}'
)
)
# Print entries # Print entries
for entry in page.entries: for entry in page.entries:
if ( if entry.state == 'Written' and entry.metadata['namespace'] != 0: # Ignore non-written entries
entry.state == 'Written' and entry.metadata['namespace'] != 0
): # Ignore non-written entries
chunk_index = '' chunk_index = ''
data = '' data = ''
if entry.metadata['type'] not in [ 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: def dump_written_blobs(nvs_partition: NVS_Partition) -> None:
blobs: Dict = {} blobs: dict = {}
strings: List[NVS_Entry] = [] strings: list[NVS_Entry] = []
legacy_blobs: List[NVS_Entry] = [] legacy_blobs: list[NVS_Entry] = []
ns = {} ns = {}
empty_entry = NVS_Entry(-1, bytearray(32), 'Erased') 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: for entry in page.entries:
if entry.state == 'Written': if entry.state == 'Written':
if entry.metadata['type'] == 'blob_index': if entry.metadata['type'] == 'blob_index':
blobs[f'{entry.metadata["namespace"]:03d}{entry.key}'] = [entry] + [ blobs[f'{entry.metadata["namespace"]:03d}{entry.key}'] = [entry] + [empty_entry] * entry.data[
empty_entry 'chunk_count'
] * entry.data['chunk_count'] ]
elif entry.metadata['type'] == 'blob': elif entry.metadata['type'] == 'blob':
legacy_blobs.append(entry) legacy_blobs.append(entry)
elif entry.metadata['type'] == 'string': elif entry.metadata['type'] == 'string':
@@ -377,30 +335,21 @@ def dump_written_blobs(nvs_partition: NVS_Partition) -> None:
if ( if (
entry.state == 'Written' entry.state == 'Written'
and entry.metadata['type'] != 'blob_index' and entry.metadata['type'] != 'blob_index'
and entry.metadata['namespace'] and entry.metadata['namespace'] == blobs[key][0].metadata['namespace']
== blobs[key][0].metadata['namespace']
and entry.key == blobs[key][0].key and entry.key == blobs[key][0].key
): ):
blobs[key][ blobs[key][1 + entry.metadata['chunk_index'] - blobs[key][0].data['chunk_start']] = entry
1
+ entry.metadata['chunk_index']
- blobs[key][0].data['chunk_start']
] = entry
blob_index = blobs[key][0] blob_index = blobs[key][0]
blob_chunks = blobs[key][1:] blob_chunks = blobs[key][1:]
# Print blob info # Print blob info
nvs_log.info( nvs_log.info(
nvs_log.cyan( nvs_log.cyan(ns.get(blob_index.metadata['namespace'], blob_index.metadata['namespace']))
ns.get(
blob_index.metadata['namespace'], blob_index.metadata['namespace']
)
)
+ ':' + ':'
+ nvs_log.yellow(blob_index.key) + nvs_log.yellow(blob_index.key)
+ ' - ' + ' - '
+ f'Type: Blob (Version 2), ' + 'Type: Blob (Version 2), '
+ f'Size: {blob_index.data["size"]}' + f'Size: {blob_index.data["size"]}'
) )
@@ -416,17 +365,13 @@ def dump_written_blobs(nvs_partition: NVS_Partition) -> None:
if entry is empty_entry: if entry is empty_entry:
nvs_log.info(nvs_log.yellow(f' {"":->63} Missing data {"":-<64}')) nvs_log.info(nvs_log.yellow(f' {"":->63} Missing data {"":-<64}'))
else: else:
nvs_log.info( nvs_log.info(f' 0x{(i * nvs_const.entry_size):05x} {entry.dump_raw()}')
f' 0x{(i * nvs_const.entry_size):05x} {entry.dump_raw()}'
)
nvs_log.info() nvs_log.info()
# Dump strings # Dump strings
for string in strings: for string in strings:
nvs_log.info( nvs_log.info(
nvs_log.cyan( nvs_log.cyan(ns.get(string.metadata['namespace'], string.metadata['namespace']))
ns.get(string.metadata['namespace'], string.metadata['namespace'])
)
+ ':' + ':'
+ nvs_log.yellow(string.key) + 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: def print_json(nvs: NVS_Partition) -> None:
class NVSEncoder(json.JSONEncoder): 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'): if hasattr(obj, 'toJSON'):
return obj.toJSON() return obj.toJSON()
if isinstance(obj, bytearray): if isinstance(obj, bytearray):
return binascii.b2a_base64(obj, newline=False).decode( return binascii.b2a_base64(obj, newline=False).decode('ascii') # Binary to Base64 ASCII representation
'ascii'
) # Binary to Base64 ASCII representation
return json.JSONEncoder.default(self, obj) 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)) 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') data = binascii.b2a_base64(tmp, newline=False).decode('ascii')
if entry.metadata['namespace'] in ns: if entry.metadata['namespace'] in ns:
key_value_pairs.append({ key_value_pairs.append(
{
'namespace': ns[entry.metadata['namespace']], 'namespace': ns[entry.metadata['namespace']],
'key': entry.key, 'key': entry.key,
'encoding': entry_type, # Add type of data 'encoding': entry_type, # Add type of data
'data': data, # Ensure data ends with a newline 'data': data, # Ensure data ends with a newline
'state': entry.state, 'state': entry.state,
'is_empty': entry.is_empty if hasattr(entry, 'is_empty') else None, 'is_empty': entry.is_empty if hasattr(entry, 'is_empty') else None,
}) }
nvs_log.info(json.dumps(key_value_pairs, indent=4)) )
print(json.dumps(key_value_pairs, indent=4))
components/nvs_flash/nvs_partition_tool/nvs_tool.py
+93 -77
View File
@@ -1,30 +1,27 @@
#!/usr/bin/env python3 #!/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 # SPDX-License-Identifier: Apache-2.0
import argparse
import os import os
import sys from collections.abc import Callable
import traceback from typing import Any
import nvs_check import nvs_check
import nvs_logger import nvs_logger
import nvs_parser import nvs_parser
from esp_pylib.logger import log
from nvs_logger import nvs_log from nvs_logger import nvs_log
from nvs_parser import nvs_const from nvs_parser import nvs_const
_DUMP_CHOICES = [
def program_args() -> argparse.Namespace: 'all',
parser = argparse.ArgumentParser( 'written',
description='Parse NVS partition', formatter_class=argparse.RawTextHelpFormatter 'minimal',
) 'blobs',
parser.add_argument('file', help='Path to dumped NVS partition') 'namespaces',
parser.add_argument( 'storage_info',
'-i', 'none',
'--integrity-check', ]
action='store_true', _DUMP_HELP = {
help='check partition for potential errors',
)
tmp = {
'all': 'print written, erased and empty entries', 'all': 'print written, erased and empty entries',
'written': 'print only currently written entries', 'written': 'print only currently written entries',
'minimal': 'print only namespace:key=value pairs', 'minimal': 'print only namespace:key=value pairs',
@@ -33,68 +30,36 @@ def program_args() -> argparse.Namespace:
'storage_info': 'print storage related information (free/used entries, etc)', 'storage_info': 'print storage related information (free/used entries, etc)',
'none': 'do not print anything (if you only want to do integrity check)', 'none': 'do not print anything (if you only want to do integrity check)',
} }
parser.add_argument( _FORMAT_CHOICES = ['text', 'json']
'-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()
def main() -> None: def _run(file: str, integrity_check: bool, dump: str, color: str, out_format: str) -> None:
args = program_args() nvs_log.set_color(color)
nvs_log.set_format(out_format)
if nvs_const.entry_size != 32: if nvs_const.entry_size != 32:
raise ValueError(f'Entry size is not 32B! This is currently non negotiable.') log.die('Entry size is not 32B! This is currently non negotiable.')
nvs_log.set_color(args.color)
nvs_log.set_format(args.format)
try: try:
with open(args.file, 'rb') as f: with open(file, 'rb') as f:
partition = f.read() partition = f.read()
except IndexError:
nvs_log.error('No file given')
raise
except FileNotFoundError: except FileNotFoundError:
nvs_log.error('Bad filename') log.die(f'File not found: {file}')
raise 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: def noop(_: nvs_parser.NVS_Partition) -> None:
pass pass
def format_not_implemented(_: nvs_parser.NVS_Partition) -> None: cmds: dict[str, Callable[[nvs_parser.NVS_Partition], None]] = {}
raise RuntimeError(f'{args.format} is not implemented') if out_format == 'text':
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 = { cmds = {
'all': nvs_logger.dump_everything, 'all': nvs_logger.dump_everything,
'written': nvs_logger.dump_written_entries, 'written': nvs_logger.dump_written_entries,
@@ -104,27 +69,78 @@ def main() -> None:
'storage_info': nvs_logger.storage_stats, 'storage_info': nvs_logger.storage_stats,
'none': noop, 'none': noop,
} }
elif out_format == 'json':
if args.format == 'json':
cmds = { cmds = {
'all': nvs_logger.print_json, 'all': nvs_logger.print_json,
'minimal': nvs_logger.print_minimal_json, 'minimal': nvs_logger.print_minimal_json,
'none': noop, '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_log.info()
nvs_check.integrity_check(nvs, nvs_log) 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__': if __name__ == '__main__':
try: from esp_pylib.excepthook import install_exception_reporting
install_exception_reporting()
main() 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)
components/nvs_flash/nvs_partition_tool/test_nvs_gen_check.py
+2 -12
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3 #!/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 # SPDX-License-Identifier: Apache-2.0
import base64 import base64
import json 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 nvs_check as nvs_check
import pytest import pytest
from esp_idf_nvs_partition_gen.nvs_partition_gen import NVS 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 nvs_log
from nvs_logger import print_minimal_json from nvs_logger import print_minimal_json
from nvs_parser import NVS_Entry from nvs_parser import NVS_Entry
@@ -34,16 +33,7 @@ def before() -> None:
pytest.skip('pass') pytest.skip('pass')
class SilentLogger(NVS_Logger): logger = nvs_log
def __init__(self) -> None:
super().__init__()
self.color = False
def info(self, *args, **kwargs) -> None: # type: ignore
pass
logger = nvs_log # SilentLogger()
LOREM_STRING = """Lorem ipsum dolor sit amet, consectetur adipiscing elit. LOREM_STRING = """Lorem ipsum dolor sit amet, consectetur adipiscing elit.
Nullam eget orci fringilla, cursus nisi sit amet, hendrerit tortor. Nullam eget orci fringilla, cursus nisi sit amet, hendrerit tortor.