#!/usr/bin/env python3 # SPDX-FileCopyrightText: 2022-2023 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 import binascii import json import sys from typing import Any, Dict, List, Union from nvs_parser import NVS_Entry, NVS_Partition, 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', } 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 def set_color(self, color: str) -> None: self.color = color == 'always' or (color == 'auto' and sys.stdout.isatty()) 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 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 red(self, text: str) -> str: if self.color: return NVS_Logger.ansi['red'] + text + NVS_Logger.ansi['clear'] return text def green(self, text: str) -> str: if self.color: return NVS_Logger.ansi['green'] + text + NVS_Logger.ansi['clear'] return text def yellow(self, text: str) -> str: if self.color: return NVS_Logger.ansi['yellow'] + text + NVS_Logger.ansi['clear'] return text def blue(self, text: str) -> str: if self.color: return NVS_Logger.ansi['blue'] + text + NVS_Logger.ansi['clear'] return text def cyan(self, text: str) -> str: if self.color: return NVS_Logger.ansi['cyan'] + text + NVS_Logger.ansi['clear'] return text def bold(self, text: str) -> str: if self.color: return NVS_Logger.ansi['bold'] + text + NVS_Logger.ansi['clear'] return text nvs_log = NVS_Logger() def storage_stats(nvs_partition: NVS_Partition) -> None: global_stats = { 'written_entries': 0, 'free_entries': 0, 'erased_entries': 0, 'invalid_entries': 0, } for page in nvs_partition.pages: written_e = 0 free_e = 0 erased_e = 0 invalid_e = 0 for entry in page.entries: if entry.state == 'Written': written_e += 1 elif entry.state == 'Empty': free_e += 1 elif entry.state == 'Erased': erased_e += 1 else: invalid_e += 1 nvs_log.info(nvs_log.bold(f'Page {page.header["status"]}')) nvs_log.info(' Found entries:') nvs_log.info(f' Written: {written_e: 5d}') nvs_log.info(f' Erased: {erased_e: 5d}') nvs_log.info(f' Empty: {free_e: 5d}') nvs_log.info(f' Invalid: {invalid_e: 5d}') nvs_log.info(f' Total: {written_e + free_e + erased_e + invalid_e: 5d}') nvs_log.info() global_stats['written_entries'] += written_e global_stats['erased_entries'] += erased_e global_stats['free_entries'] += free_e global_stats['invalid_entries'] += invalid_e nvs_log.info(nvs_log.bold('Global')) nvs_log.info(' Config:') nvs_log.info(f' Page size: {nvs_const.page_size: 5d}') nvs_log.info(f' Entry size: {nvs_const.entry_size: 5d}') nvs_log.info(f' Total pages: {len(nvs_partition.pages): 5d}') nvs_log.info(' Entries:') nvs_log.info(f' Written: {global_stats["written_entries"]: 5d}') nvs_log.info(f' Erased: {global_stats["erased_entries"]: 5d}') nvs_log.info(f' Empty: {global_stats["free_entries"]: 5d}') nvs_log.info(f' Invalid: {global_stats["invalid_entries"]: 5d}') nvs_log.info(f' Total: {sum([global_stats[key] for key in global_stats]): 5d}') nvs_log.info() def dump_everything(nvs_partition: NVS_Partition, written_only: bool = False) -> None: 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}') ) else: 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}') nvs_log.info( nvs_log.bold( f'Page no. {page.header["page_index"]}' + f', Status: {page.header["status"]}' + f', Version: {page.header["version"]}' + f', CRC32: {crc}' ) + nvs_log.bold(f', Page address: 0x{page.start_address:x}') ) nvs_log.info(nvs_log.bold(' Entry state bitmap: '), end='') for x in page.raw_entry_state_bitmap: nvs_log.info(f'{x:02x} ', end='') nvs_log.info() # Dump entries empty_entries = [] for entry in page.entries: # Skip non-written entries if needed if written_only and not entry.state == 'Written': continue # Compress all 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(' ...')) 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') empty_entries.clear() # Dump a single entry status = entry.state if status == 'Written': status = nvs_log.green(f'{status: <7}') elif status == 'Erased': status = nvs_log.red(f'{status: <7}') crc = '' 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}') nvs_log.info( nvs_log.bold(f' {entry.index:03d}.') + ' ' + status + f', Namespace Index: {entry.metadata["namespace"]:03d}' + f', Type: {entry.metadata["type"]:<10}' + f', Span: {entry.metadata["span"]:03d}' + f', Chunk Index: {entry.metadata["chunk_index"]:03d}' + f', CRC32: {crc}' + f' | {entry.key} : ', end='', ) if entry.metadata['type'] not in [ 'string', 'blob_data', 'blob_index', 'blob', ]: # Entry is non-variable length if entry.data is not None: nvs_log.info(entry.data['value']) else: nvs_log.info(entry.data) # None else: if entry.metadata['type'] == 'blob_index': nvs_log.info( f'Size={entry.data["size"]}' + f', ChunkCount={entry.data["chunk_count"]}' + 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}' ) else: crc = nvs_log.red(f'{entry.metadata["crc"]["data_original"]:x}') nvs_log.info(f'Size={entry.data["size"]}, CRC32={crc}') # 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()}' ) # Dump trailing empty entries if len(empty_entries) >= 3: 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') else: for e in empty_entries: nvs_log.info(nvs_log.bold(f' {e.index:03d}.'), 'Empty') empty_entries.clear() nvs_log.info() def dump_written_entries(nvs_partition: NVS_Partition) -> None: dump_everything(nvs_partition, True) def list_namespaces(nvs_partition: NVS_Partition) -> None: # Gather namespaces ns = {} for page in nvs_partition.pages: for entry in page.entries: if entry.state == 'Written' and entry.metadata['namespace'] == 0: ns[entry.data['value']] = entry.key # Print found namespaces nvs_log.info(nvs_log.bold(f'Index : Namespace')) for ns_index in sorted(ns): nvs_log.info(f' {ns_index:03d} :', nvs_log.cyan(ns[ns_index])) def dump_key_value_pairs(nvs_partition: NVS_Partition) -> None: # Get namespace list ns = {} for page in nvs_partition.pages: for entry in page.entries: if entry.state == 'Written' and entry.metadata['namespace'] == 0: ns[entry.data['value']] = entry.key # Print key-value pairs for page in nvs_partition.pages: # Print page header 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"]}' ) ) # Print entries for entry in page.entries: if ( entry.state == 'Written' and entry.metadata['namespace'] != 0 ): # Ignore non-written entries chunk_index = '' data = '' if entry.metadata['type'] not in [ 'string', 'blob_data', 'blob_index', 'blob', ]: # Non-variable length entry data = entry.data['value'] elif entry.metadata['type'] == 'blob_index': continue else: # Variable length entries tmp = b'' for e in entry.children: # Merge all children entries tmp += bytes(e.raw) tmp = tmp[: entry.data['size']] # Discard padding if entry.metadata['type'] == 'blob_data': if entry.metadata['chunk_index'] >= 128: # Get real chunk index chunk_index = f'[{entry.metadata["chunk_index"] - 128}]' else: chunk_index = f'[{entry.metadata["chunk_index"]}]' data = str(tmp) if entry.metadata['namespace'] not in ns: continue else: nvs_log.info( ' ' + nvs_log.cyan(ns[entry.metadata['namespace']]) + ':' + nvs_log.yellow(entry.key) + f'{chunk_index} = {data}' ) nvs_log.info() def dump_written_blobs(nvs_partition: NVS_Partition) -> None: blobs: Dict = {} strings: List[NVS_Entry] = [] legacy_blobs: List[NVS_Entry] = [] ns = {} empty_entry = NVS_Entry(-1, bytearray(32), 'Erased') # Gather namespaces, blob indexes and legacy blobs for page in nvs_partition.pages: 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'] elif entry.metadata['type'] == 'blob': legacy_blobs.append(entry) elif entry.metadata['type'] == 'string': strings.append(entry) elif entry.metadata['namespace'] == 0: ns[entry.data['value']] = entry.key # Dump blobs for key in blobs: for page in nvs_partition.pages: for entry in page.entries: # Gather all blob chunks if ( entry.state == 'Written' and entry.metadata['type'] != 'blob_index' 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 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.yellow(blob_index.key) + ' - ' + f'Type: Blob (Version 2), ' + f'Size: {blob_index.data["size"]}' ) # Print blob data raw_entries = [] for kid in blob_chunks: # Gather all chunk entries if kid is empty_entry: raw_entries += [empty_entry] else: raw_entries += kid.children for i, entry in enumerate(raw_entries): 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() # Dump strings for string in strings: nvs_log.info( nvs_log.cyan( ns.get(string.metadata['namespace'], string.metadata['namespace']) ) + ':' + nvs_log.yellow(string.key) + ' - ' + 'Type: String, ' + f'Size: {string.data["size"]}' ) for i, entry in enumerate(string.children): nvs_log.info(f' 0x{(i * nvs_const.entry_size):05x} {entry.dump_raw()}') nvs_log.info() # Dump legacy blobs for blob in legacy_blobs: nvs_log.info( nvs_log.cyan(ns.get(blob.metadata['namespace'], blob.metadata['namespace'])) + ':' + nvs_log.yellow(blob.key) + ' - ' + 'Type: Blob (Version 1), ' + f'Size: {blob.data["size"]}' ) for i, entry in enumerate(blob.children): nvs_log.info(f' 0x{(i * nvs_const.entry_size):05x} {entry.dump_raw()}') nvs_log.info() def print_json(nvs: NVS_Partition) -> None: class NVSEncoder(json.JSONEncoder): def default(self, obj: Any) -> Union[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 json.JSONEncoder.default(self, obj) print(json.dumps(nvs.toJSON(), cls=NVSEncoder, indent=2))