#!/usr/bin/env python3 # SPDX-FileCopyrightText: 2022-2023 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 from typing import Any, Dict, List, Optional from zlib import crc32 # Constants class NVS_Constants: class ConstantError(AttributeError): pass def __init__(self) -> None: self.page_size = 4096 self.entry_size = 32 self.item_type = { 0x01: 'uint8_t', 0x11: 'int8_t', 0x02: 'uint16_t', 0x12: 'int16_t', 0x04: 'uint32_t', 0x14: 'int32_t', 0x08: 'uint64_t', 0x18: 'int64_t', 0x21: 'string', 0x41: 'blob', 0x42: 'blob_data', 0x48: 'blob_index', } self.page_status = { 0xFFFFFFFF: 'Empty', 0xFFFFFFFE: 'Active', 0xFFFFFFFC: 'Full', 0xFFFFFFF8: 'Erasing', 0x00000000: 'Corrupted', } self.entry_status = { 0b11: 'Empty', 0b10: 'Written', 0b00: 'Erased', } def __setattr__(self, key: str, val: Any) -> None: if self.__dict__.get(key, None) is None: self.__dict__[key] = val else: raise NVS_Constants.ConstantError('Cannot change a constant!') nvs_const = NVS_Constants() class NotAlignedError(ValueError): pass class NVS_Partition: def __init__(self, name: str, raw_data: bytearray): if len(raw_data) % nvs_const.page_size != 0: raise NotAlignedError( f'Given partition data is not aligned to page size ({len(raw_data)} % {nvs_const.page_size} = {len(raw_data)%nvs_const.page_size})' ) # Divide partition into pages self.name = name self.pages = [] for i in range(0, len(raw_data), nvs_const.page_size): self.pages.append(NVS_Page(raw_data[i: i + nvs_const.page_size], i)) def toJSON(self) -> Dict[str, Any]: return dict(name=self.name, pages=self.pages) class NVS_Page: def __init__(self, page_data: bytearray, address: int): if len(page_data) != nvs_const.page_size: raise NotAlignedError( f'Size of given page does not match page size ({len(page_data)} != {nvs_const.page_size})' ) # Initialize class self.is_empty = ( page_data[0: nvs_const.entry_size] == bytearray({0xFF}) * nvs_const.entry_size ) self.start_address = address self.raw_header = page_data[0: nvs_const.entry_size] self.raw_entry_state_bitmap = page_data[ nvs_const.entry_size: 2 * nvs_const.entry_size ] self.entries = [] # Load header self.header: Dict[str, Any] = { 'status': nvs_const.page_status.get( int.from_bytes(page_data[0:4], byteorder='little'), 'Invalid' ), 'page_index': int.from_bytes(page_data[4:8], byteorder='little'), 'version': 256 - page_data[8], 'crc': { 'original': int.from_bytes(page_data[28:32], byteorder='little'), 'computed': crc32(page_data[4:28], 0xFFFFFFFF), }, } # Load entry state bitmap entry_states = [] for c in self.raw_entry_state_bitmap: for index in range(0, 8, 2): entry_states.append( nvs_const.entry_status.get((c >> index) & 3, 'Invalid') ) entry_states = entry_states[:-2] # Load entries i = 2 while i < int( nvs_const.page_size / nvs_const.entry_size ): # Loop through every entry span = page_data[(i * nvs_const.entry_size) + 2] if span in [0xFF, 0]: # 'Default' span length to prevent span overflow span = 1 # Load an entry entry = NVS_Entry( i - 2, page_data[i * nvs_const.entry_size: (i + 1) * nvs_const.entry_size], entry_states[i - 2], ) self.entries.append(entry) # Load all children entries if span != 1: for span_idx in range(1, span): page_addr = i + span_idx entry_idx = page_addr - 2 if page_addr * nvs_const.entry_size >= nvs_const.page_size: break child_entry = NVS_Entry( entry_idx, page_data[ page_addr * nvs_const.entry_size: (page_addr + 1) * nvs_const.entry_size ], entry_states[entry_idx], ) entry.child_assign(child_entry) entry.compute_crc() i += span def toJSON(self) -> Dict[str, Any]: return dict( is_empty=self.is_empty, start_address=self.start_address, raw_header=self.raw_header, raw_entry_state_bitmap=self.raw_entry_state_bitmap, header=self.header, entries=self.entries, ) class NVS_Entry: def __init__(self, index: int, entry_data: bytearray, entry_state: str): if len(entry_data) != nvs_const.entry_size: raise NotAlignedError( f'Given entry is not aligned to entry size ({len(entry_data)} % {nvs_const.entry_size} = {len(entry_data)%nvs_const.entry_size})' ) def item_convert(i_type: int, data: bytearray) -> Dict: byte_size_mask = 0x0F number_sign_mask = 0xF0 fixed_entry_length_threshold = ( 0x20 # Fixed length entry type number is always smaller than this ) if i_type in nvs_const.item_type: # Deal with non variable length entries if i_type < fixed_entry_length_threshold: size = i_type & byte_size_mask num = int.from_bytes( data[:size], byteorder='little', signed=bool(i_type & number_sign_mask), ) return {'value': num} # Deal with variable length entries if nvs_const.item_type[i_type] in ['string', 'blob_data', 'blob']: size = int.from_bytes(data[:2], byteorder='little') crc = int.from_bytes(data[4:8], byteorder='little') return {'value': [size, crc], 'size': size, 'crc': crc} if nvs_const.item_type[i_type] == 'blob_index': size = int.from_bytes(data[:4], byteorder='little') chunk_count = data[4] chunk_start = data[5] return { 'value': [size, chunk_count, chunk_start], 'size': size, 'chunk_count': chunk_count, 'chunk_start': chunk_start, } return {'value': None} def key_decode(data: bytearray) -> Optional[str]: decoded = '' for n in data.rstrip(b'\x00'): char = chr(n) if char.isascii(): decoded += char else: return None return decoded self.raw = entry_data self.state = entry_state self.is_empty = self.raw == bytearray({0xFF}) * nvs_const.entry_size self.index = index namespace = self.raw[0] entry_type = self.raw[1] span = self.raw[2] chunk_index = self.raw[3] crc = self.raw[4:8] key = self.raw[8:24] data = self.raw[24:32] raw_without_crc = self.raw[:4] + self.raw[8:32] self.metadata: Dict[str, Any] = { 'namespace': namespace, 'type': nvs_const.item_type.get(entry_type, f'0x{entry_type:02x}'), 'span': span, 'chunk_index': chunk_index, 'crc': { 'original': int.from_bytes(crc, byteorder='little'), 'computed': crc32(raw_without_crc, 0xFFFFFFFF), 'data_original': int.from_bytes(data[-4:], byteorder='little'), 'data_computed': 0, }, } self.children: List['NVS_Entry'] = [] self.key = key_decode(key) if self.key is None: self.data = None else: self.data = item_convert(entry_type, data) def dump_raw(self) -> str: hex_bytes = '' decoded = '' for i, c in enumerate(self.raw): middle_index = int(len(self.raw) / 2) if i == middle_index: # Add a space in the middle hex_bytes += ' ' decoded += ' ' hex_bytes += f'{c:02x} ' decoded += chr(c) if chr(c).isprintable() else '.' return hex_bytes + ' ' + decoded def child_assign(self, entry: 'NVS_Entry') -> None: if not isinstance(entry, type(self)): raise ValueError('You can assign only NVS_Entry') self.children.append(entry) def compute_crc(self) -> None: if self.metadata['span'] == 1: return # Merge entries into one buffer children_data = bytearray() for entry in self.children: children_data += entry.raw if self.data: children_data = children_data[: self.data['size']] # Discard padding self.metadata['crc']['data_computed'] = crc32(children_data, 0xFFFFFFFF) def toJSON(self) -> Dict[str, Any]: return dict( raw=self.raw, state=self.state, is_empty=self.is_empty, index=self.index, metadata=self.metadata, children=self.children, key=self.key, data=self.data, )