NVS parser WIP

nvs.be

class NC # NVS Constants
    static page_size = 4096
    static entry_size = 32

    static 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',
        }

    static page_status = {
            0xFFFFFFFF: 'Empty',
            0xFFFFFFFE: 'Active',
            0xFFFFFFFC: 'Full',
            0xFFFFFFF8: 'Erasing',
            0x00000000: 'Corrupted',
        }

    static entry_status = {
            3: 'Empty',   # 0b11
            2: 'Written', # 0b10
            0: 'Erased',  # 0b00
        }
end

class NVS_Entry
    var raw, state, is_empty, index, metadata, children, key, data

    def init(index, entry_data, entry_state)
        if size(entry_data) != NC.entry_size
            assert (
                f'Given entry is not aligned to entry size ({size(entry_data)} % {NC.entry_size} = {size(entry_data)%NC.entry_size})'
            )
        end
        # print("entry_data",entry_data)

        def item_convert(i_type, data)
            var byte_size_mask = 0x0F
            var number_sign_mask = 0xF0
            var fixed_entry_length_threshold = (
                0x20  # Fixed length entry type number is always smaller than this
            )
            if NC.item_type.find(i_type) != nil
                # Deal with non variable length entries
                if i_type < fixed_entry_length_threshold
                    var sz = i_type & byte_size_mask
                    var num = data.get(0,sz&7)
                    if i_type & number_sign_mask == false
                        num *= -1
                    end
                    return {'value': num}

                # Deal with variable length entries
                elif  ['string', 'blob_data', 'blob'].find(NC.item_type[i_type]) != nil
                    var sz = data.get(0,2)
                    var crc = data.get(4,4)
                    return {'value': [sz, crc], 'size': sz, 'crc': crc}
                elif NC.item_type[i_type] == 'blob_index'
                    var sz = data.get(0,4)
                    var chunk_count = data[4]
                    var chunk_start = data[5]
                    return {
                        'value': [sz, chunk_count, chunk_start],
                        'size': sz,
                        'chunk_count': chunk_count,
                        'chunk_start': chunk_start,
                    }
                end
            # else
            #    print("unknown i_type",i_type, data)
            end
            return {'value': None}
        end

        def key_decode(data)
            var decoded = ''
            var start = 0
            while data[start] == 0x00 # .rstrip(b'\x00')
                if size(data) == start + 1
                    return nil
                end
                #data = data[1..]
                start += 1
            end
            for char : start..size(data) - 1
                var n = data[char]
                if n < 128
                    decoded += data[char..char].asstring()
                else
                    break
                    #return nil
                end
            end
            return decoded
        end

        self.raw = entry_data
        self.state = entry_state

        var is_empty = true
        for i:0..NC.entry_size-1
            if entry_data[i] != 0xff
                is_empty = false
                break
            end
        end
        self.is_empty = is_empty

        self.index = index

        var namespace = self.raw[0]
        var entry_type = self.raw[1]
        var span = self.raw[2]
        var chunk_index = self.raw[3]
        var crc_val = self.raw[4..8-1]
        var key = self.raw[8..24-1]
        var data = self.raw[24..32-1]
        var raw_without_crc = self.raw[0..4-1] + self.raw[8..32-1]
        import crc
        self.metadata = {
            'namespace': namespace,
            'type': NC.item_type.find(entry_type, f'0x{entry_type:02x}'),
            'span': span,
            'chunk_index': chunk_index,
            'crc': {
                'original': crc_val.get(4,4),
                'computed': crc.crc32(0xFFFFFFFF, raw_without_crc),
                'data_original': data[-4..].get(0,4),
                'data_computed': 0,
            },
        }
        # print(self.metadata)
        self.children = [] # : List['NVS_Entry'] = []
        self.key = key_decode(key)
        if self.key == nil
            self.data = nil
        else
            self.data = item_convert(entry_type, data)
            print(self.key,self.data)
        end
    end

    def child_assign(entry)
        if isinstance(entry, NVS_Entry) == false
            assert ('You can assign only NVS_Entry')
        end
        self.children.push(entry)
    end

    def compute_crc()
        import crc
        if self.metadata['span'] == 1
            return
        end
        # Merge entries into one buffer
        var children_data = bytes()
        for entry : self.children
            children_data..entry.raw
        end
        if self.data
            if self.data['value'] != nil
                if self.data['size']
                    children_data = children_data[0.. self.data['size']]  # Discard padding
                end
            end
        end
        self.metadata['crc']['data_computed'] = crc.crc32(0xFFFFFFFF, children_data)
    end
end


class NVS_Page
    var is_empty, start_address, raw_header, raw_entry_state_bitmap, entries, header

    def init(page_data, address)
        if size(page_data) != NC.page_size
            assert (f'Size of given page does not match page size ({size(page_data)} != {NC.page_size})')
        end

        # Initialize class
        var is_empty = true
        for i:0..NC.entry_size-1
            if page_data[i] != 0xff
                is_empty = false
                break
            end
        end
        self.is_empty = is_empty

        self.start_address = address
        self.raw_header = page_data[0 .. NC.entry_size-1]
        self.raw_entry_state_bitmap = page_data[
            NC.entry_size .. (2 * NC.entry_size)-1
        ]
        self.entries = []


        # Load header
        import crc
        self.header = {
            'status': NC.page_status.find(
                page_data.get(0,4), 'Invalid'
            ),
            'page_index': page_data.get(4,4),
            'version': 256 - page_data[8],
            'crc': {
                'original': page_data.get(28,4),
                'computed': crc.crc32(0xFFFFFFFF, page_data[4..28-1]),
            },
        }
        print("header loaded:",self.header)
        if self.is_empty == true
            return
        end

        # Load entry state bitmap
        var entry_states = []
        for i : 0..size(self.raw_entry_state_bitmap)-1
            var c = self.raw_entry_state_bitmap[i]
            for index : range(0, 8, 2)
                entry_states.push(
                    NC.entry_status.find((c >> index) & 3, 'Invalid')
                )
            end
        end
        entry_states = entry_states[0..-3]

        # Load entries
        var i = 2
        while i < int(NC.page_size / NC.entry_size)  # Loop through every entry
            var span = page_data[(i * NC.entry_size) + 2]
            if [0xFF, 0].find(span) != nil  # 'Default' span length to prevent span overflow
                #print("change span to 1",span)
                span = 1
            end

            # Load an entry
            var entry = NVS_Entry(
                (i - 2),
                page_data[(i * NC.entry_size) .. ((i + 1) * NC.entry_size)-1],
                entry_states[i - 2]
            )
            self.entries.push(entry)

            # Load all children entries
            if span > 1
                for span_idx : range(1, span - 1)
                    var page_addr = i + span_idx
                    var entry_idx = page_addr - 2
                    if (page_addr * NC.entry_size) >= NC.page_size
                        break
                    end
                    var child_entry = NVS_Entry(
                        entry_idx,
                        page_data[
                            page_addr
                            * NC.entry_size ..
                            ((page_addr + 1)* NC.entry_size)-1
                        ],
                        entry_states[entry_idx]
                    )
                    entry.child_assign(child_entry)
                end
                entry.compute_crc()
            end
            i += span
        end
    end

    def toJSON()
        return {
            "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
        }
    end
end

class NP # NVS Partition
    var name, pages
    def init(name, raw_data)
        self.name = name
        self.pages = []
        if size(raw_data) % NC.page_size != 0
            assert (
                f'Given partition data is not aligned to page size ({size(raw_data)} % {NC.page_size} = {size(raw_data)%NC.page_size})'
            )
        end
        # Divide partition into pages
        for i : range(0, size(raw_data)-1, NC.page_size)
            self.pages.push(NVS_Page(raw_data[i.. i + NC.page_size - 1], i))
        end
    end

    def toJSON()
        var json = []
        for i : 0..size(self.pages)-1
            json.push(self.pages[i].toJSON())
        end
        return json
    end
end


import partition_core
var p = partition_core.Partition()
var nvs
for slot : p.slots
    if slot.label == "nvs"
        import flash
        var nvs_partition = flash.read(slot.start,slot.sz)
        nvs = NP("N", nvs_partition)
        
        break
    end
end