pebble/tools/pbpack.py

#!/usr/bin/env python
# Copyright 2024 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


import argparse
import stm32_crc
import struct
import time


class ResourcePackTableEntry(object):
    TABLE_ENTRY_FMT = '<IIII'

    def __init__(self, content_index, offset, length, crc):
        # The index into the contents array that holds our data in the ResourcePack object
        self.content_index = content_index
        # The offset into the body of the pbpack in bytes that we can find the content
        self.offset = offset
        # The length of the content
        self.length = length
        # The CRC of the content
        self.crc = crc

    def serialize(self, file_id):
        fmt = self.TABLE_ENTRY_FMT
        return struct.pack(fmt, file_id, self.offset, self.length, self.crc)

    @classmethod
    def deserialize(cls, table_entry_data):
        """
        Return a tuple of file_id, ResourcePackTableEntry object for the given data.
        """

        fmt = cls.TABLE_ENTRY_FMT
        file_id, offset, length, crc = struct.unpack(fmt, table_entry_data)

        entry = ResourcePackTableEntry(-1, offset, length, crc)
        return file_id, entry

    def __repr__(self):
        return str(self.__dict__)



class ResourcePack(object):
    """ Pebble resource pack file format (de)serialization tools.

        An instance of this class is an in-memory representation of a resource
        pack. The class has a number of methods to facilitate (de)serialization
        of .pbpack files.

    """

    TABLE_ENTRY_SIZE_BYTES = 16
    MANIFEST_FMT = '<III'
    MANIFEST_SIZE_BYTES = 12

    def get_content_crc(self):
        all_contents = self.serialize_content()
        return stm32_crc.crc32(all_contents)

    def serialize_manifest(self, crc=None, timestamp=None):
        fmt = self.MANIFEST_FMT

        return struct.pack(fmt, len(self.table_entries), self.crc, self.timestamp)

    def serialize_table(self):
        # Serialize these entries into table_data
        cur_file_id = 1
        table_data = ''
        for cur_file_id, table_entry in enumerate(self.table_entries, start=1):
            table_data += table_entry.serialize(cur_file_id)

        # Pad the rest of the table_data up to table_size
        for i in xrange(cur_file_id, self.table_size):
            table_data += ResourcePackTableEntry(0, 0, 0, 0).serialize(0)

        return table_data

    def serialize_content(self):
        """
        Serialize the content in the order dictated by offsets in the table entries
        """

        serialized_content_indexes = set()
        serialized_content = []
        for entry in sorted(self.table_entries, key=lambda e: e.offset):
            if entry.content_index in serialized_content_indexes:
                continue

            serialized_content_indexes.add(entry.content_index)

            serialized_content.append(self.contents[entry.content_index])

        return b"".join(serialized_content)

    @classmethod
    def deserialize(cls, f_in, is_system=True):
        resource_pack = cls(is_system)

        # Parse manifest:
        manifest_data = f_in.read(cls.MANIFEST_SIZE_BYTES)
        (num_files, crc, timestamp) = struct.unpack(cls.MANIFEST_FMT, manifest_data)
        resource_pack.num_files = num_files
        resource_pack.crc = crc
        resource_pack.timestamp = timestamp

        # Parse table entries:
        resource_pack.table_entries = []
        for n in xrange(num_files):
            table_entry = f_in.read(cls.TABLE_ENTRY_SIZE_BYTES)
            file_id, entry = ResourcePackTableEntry.deserialize(table_entry)

            if file_id == 0:
                # No more entries
                break

            if file_id != n + 1:
                raise Exception("File ID is expected to be %u, but was %u" %
                                (n + 1, file_id))

            resource_pack.table_entries.append(entry)

        if len(resource_pack.table_entries) != num_files:
            raise Exception("Number of files in manifest is %u, but actual"
                            "number is %u" % (num_files, n))

        # Figure out which content_index to assign to each table entry. Find all unique offset and
        # length combinations and then assign content indexes appropriately. We need to include the
        # length because we allow zero length resources and a zero length resource will have the
        # same offset as a non-zero length resource
        unique_offsets = set()
        for e in resource_pack.table_entries:
            unique_offsets.add((e.offset, e.length))
        unique_offsets = list(unique_offsets)

        for e in resource_pack.table_entries:
            e.content_index = unique_offsets.index((e.offset, e.length))

        # Fetch the contents, make sure we only load each unique piece of content once
        loaded_content_indexes = set()
        for entry in sorted(resource_pack.table_entries, key=lambda e: e.content_index):
            if entry.content_index in loaded_content_indexes:
                # Already loaded, just skip
                continue

            loaded_content_indexes.add(entry.content_index)

            f_in.seek(entry.offset + resource_pack.content_start)
            content = f_in.read(entry.length)

            calculated_crc = stm32_crc.crc32(content)

            if calculated_crc != entry.crc:
                raise Exception("Entry %s does not match CRC of content (%u). "
                                "Hint: try with%s the --app flag"
                                % (entry, calculated_crc, "" if is_system else "out"))

            resource_pack.contents.append(content)

        resource_pack.finalized = True

        return resource_pack

    def finalize(self):
        """
        Take all the resources that have been added using add_resource and finalize the pack. Once
        a pack is finalized no more resources may be added.
        """

        if (len(self.table_entries) > self.table_size):
            raise Exception("Exceeded max number of resources. Must have %d or "
                            "fewer" % self.table_size)

        # Assign offsets to each of the entries in reverse order. The reason we do this in reverse
        # is so that the resource at the end is guaranteed to be at the end of the pack.
        #
        # This is required because the firmware looks at the offset + length of the final resource
        # in the table in order to determine the total size of the pack. If the final resource
        # is a duplicate of an earlier resource and if we assigned offsets starting at the
        # beginning of the table, that final resource would end up pointing to an assigned
        # offset somewhere in the middle of the pack, causing the pack to appear to be truncated.
        current_offset = sum((len(c) for c in self.contents))
        for table_entry in reversed(self.table_entries):
            if table_entry.offset == -1:
                # This entry doesn't have an offset in the output file yet, assign one to all
                # entries that share the same content index

                current_offset -= table_entry.length

                for e in self.table_entries:
                    if e.content_index == table_entry.content_index:
                        e.offset = current_offset

        self.crc = self.get_content_crc()

        self.finalized = True

    def serialize(self, f_out):
        if not self.finalized:
            self.finalize()

        f_out.write(self.serialize_manifest(self.crc))
        f_out.write(self.serialize_table())
        for c in self.serialize_content():
            f_out.write(c)

        return self.crc

    def add_resource(self, content):
        if self.finalized:
            raise Exception("Cannot add additional resource, " +
                            "resource pack has already been finalized")

        # If resource already is present, add to table only
        try:
            # Try to find the index for this content if it already exists. If it doesn't we'll
            # throw the ValueError.
            content_index = self.contents.index(content)
        except ValueError:
            # This content is completely new, add it to the contents list.
            self.contents.append(content)
            content_index = len(self.contents) - 1

        crc = stm32_crc.crc32(content)

        # Use -1 as the offset as we don't assign offsets until serialize_table
        self.table_entries.append(ResourcePackTableEntry(content_index, -1, len(content), crc))

    def dump(self):
        """
        Dump a bunch of information about this pbpack to stdout
        """

        print 'Manifest CRC: 0x%x' % self.crc
        print 'Calculated CRC: 0x%x' % self.get_content_crc()
        print 'Num Items: %u' % len(self.table_entries)
        for i, entry in enumerate(self.table_entries, start=1):
            print '  %u: Offset %u Length %u CRC 0x%x' % (i, entry.offset, entry.length, entry.crc)

    def __init__(self, is_system):
        self.table_size = 512 if is_system else 256
        self.content_start = self.MANIFEST_SIZE_BYTES + self.table_size * self.TABLE_ENTRY_SIZE_BYTES

        # Note that we never actually set the timestamp in newly generated pbpacks. The timestamp
        # field in the manifest is unused in practice and we'll probably reuse that field for
        # another purpose in the future.
        self.timestamp = 0

        # CRC of the content. Note that it only covers the content itself and not the table nor
        # the manifest. Calculated during finalize().
        self.crc = 0

        # List of binary content, where each entry is the raw processed data for each unique
        # resource.
        self.contents = []

        # List of resources that are in the pack. Note that this list may be longer than the
        # self.contents list if there are duplicates, duplicated entries (exact same data) will
        # not be repeated in self.contents. Each entry is a ResourcePackTableEntry
        self.table_entries = []

        # Indicates that the ResourcePack has been built and no more resources can be added.
        self.finalized = False


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='dump pbpack metadata')

    parser.add_argument('pbpack_path', help='path to pbpack to dump')
    parser.add_argument('--app', default=False, action='store_true',
                        help='Indicate this pbpack is an app pbpack')

    args = parser.parse_args()

    with open(args.pbpack_path, 'rb') as f:
        pack = ResourcePack.deserialize(f, is_system=not args.app)

    pack.dump()