# This is a generated file! Please edit source .ksy file and use kaitai-struct-compiler to rebuild # type: ignore import kaitaistruct from kaitaistruct import KaitaiStruct, KaitaiStream, BytesIO from enum import IntEnum if getattr(kaitaistruct, 'API_VERSION', (0, 9)) < (0, 11): raise Exception("Incompatible Kaitai Struct Python API: 0.11 or later is required, but you have %s" % (kaitaistruct.__version__)) class Zchunk(KaitaiStruct): """ .. seealso:: Source - https://github.com/zchunk/zchunk/blob/99e51afa38c723e7c25834c2c3b305d20ef55d04/zchunk_format.txt """ class ChecksumTypes(IntEnum): sha1 = 0 sha256 = 1 sha512 = 2 sha512_128 = 3 class CompressionTypes(IntEnum): none = 0 zstd = 2 def __init__(self, _io, _parent=None, _root=None): super(Zchunk, self).__init__(_io) self._parent = _parent self._root = _root or self self._read() def _read(self): self.lead = Zchunk.HeaderLead(self._io, self, self._root) self._raw_header_rest = self._io.read_bytes(self.lead.len_header_rest.value) _io__raw_header_rest = KaitaiStream(BytesIO(self._raw_header_rest)) self.header_rest = Zchunk.HeaderWithoutLead(_io__raw_header_rest, self, self._root) self.dict = self._io.read_bytes(self.header_rest.index.len_dict.value) if (not (self.lead.is_detached_header)): pass self.chunks = [] for i in range(len(self.header_rest.index.chunks_metadata)): self.chunks.append(self._io.read_bytes(self.header_rest.index.chunks_metadata[i].len_chunk.value)) def _fetch_instances(self): pass self.lead._fetch_instances() self.header_rest._fetch_instances() if (not (self.lead.is_detached_header)): pass for i in range(len(self.chunks)): pass class ChecksumType(KaitaiStruct): def __init__(self, _io, _parent=None, _root=None): super(Zchunk.ChecksumType, self).__init__(_io) self._parent = _parent self._root = _root self._read() def _read(self): self.raw = Zchunk.CompressedInteger(self._io, self, self._root) _ = self.raw if not self.len_checksum != 0: raise kaitaistruct.ValidationExprError(self.raw, self._io, u"/types/checksum_type/seq/0") def _fetch_instances(self): pass self.raw._fetch_instances() @property def len_checksum(self): if hasattr(self, '_m_len_checksum'): return self._m_len_checksum self._m_len_checksum = (20 if self.value == Zchunk.ChecksumTypes.sha1 else (32 if self.value == Zchunk.ChecksumTypes.sha256 else (64 if self.value == Zchunk.ChecksumTypes.sha512 else (16 if self.value == Zchunk.ChecksumTypes.sha512_128 else 0)))) return getattr(self, '_m_len_checksum', None) @property def value(self): if hasattr(self, '_m_value'): return self._m_value self._m_value = KaitaiStream.resolve_enum(Zchunk.ChecksumTypes, self.raw.value) return getattr(self, '_m_value', None) class Chunk(KaitaiStruct): def __init__(self, len_checksum, has_data_streams, has_uncompressed_source, _io, _parent=None, _root=None): super(Zchunk.Chunk, self).__init__(_io) self._parent = _parent self._root = _root self.len_checksum = len_checksum self.has_data_streams = has_data_streams self.has_uncompressed_source = has_uncompressed_source self._read() def _read(self): if self.has_data_streams: pass self.chunk_stream = Zchunk.CompressedInteger(self._io, self, self._root) self.chunk_checksum = self._io.read_bytes(self.len_checksum) if self.has_uncompressed_source: pass self.uncompressed_chunk_checksum = self._io.read_bytes(self.len_checksum) self.len_chunk = Zchunk.CompressedInteger(self._io, self, self._root) self.len_uncompressed_chunk = Zchunk.CompressedInteger(self._io, self, self._root) def _fetch_instances(self): pass if self.has_data_streams: pass self.chunk_stream._fetch_instances() if self.has_uncompressed_source: pass self.len_chunk._fetch_instances() self.len_uncompressed_chunk._fetch_instances() class CompressedInteger(KaitaiStruct): """Like `/common/vlq_base128_le` (LEB128), but the logic of the "continuation" flag in the most significant bit is inverted, so instead of `has_next`, it is called `is_last` (if the highest bit is set to zero, it means "continue", whereas in standard LEB128, the highest bit set to **one** means "continue"). Therefore, we cannot simply import `/common/vlq_base128_le` and use it, because it is incompatible. """ def __init__(self, _io, _parent=None, _root=None): super(Zchunk.CompressedInteger, self).__init__(_io) self._parent = _parent self._root = _root self._read() def _read(self): self.groups = [] i = 0 while True: _ = Zchunk.CompressedInteger.Group(i, self._io, self, self._root) self.groups.append(_) if _.is_last: break i += 1 def _fetch_instances(self): pass for i in range(len(self.groups)): pass self.groups[i]._fetch_instances() class Group(KaitaiStruct): """One byte group, clearly divided into 7-bit "value" chunk and 1-bit "continuation" flag. """ def __init__(self, idx, _io, _parent=None, _root=None): super(Zchunk.CompressedInteger.Group, self).__init__(_io) self._parent = _parent self._root = _root self.idx = idx self._read() def _read(self): self.is_last = self._io.read_bits_int_be(1) != 0 if not self.is_last == (True if self.idx == 9 else self.is_last): raise kaitaistruct.ValidationNotEqualError((True if self.idx == 9 else self.is_last), self.is_last, self._io, u"/types/compressed_integer/types/group/seq/0") self.value = self._io.read_bits_int_be(7) if not self.value <= (1 if self.idx == 9 else 127): raise kaitaistruct.ValidationGreaterThanError((1 if self.idx == 9 else 127), self.value, self._io, u"/types/compressed_integer/types/group/seq/1") def _fetch_instances(self): pass @property def len(self): if hasattr(self, '_m_len'): return self._m_len self._m_len = len(self.groups) return getattr(self, '_m_len', None) @property def value(self): """Resulting unsigned value as normal integer.""" if hasattr(self, '_m_value'): return self._m_value self._m_value = (((((((((self.groups[0].value | (self.groups[1].value << 7 if self.len >= 2 else 0)) | (self.groups[2].value << 14 if self.len >= 3 else 0)) | (self.groups[3].value << 21 if self.len >= 4 else 0)) | (self.groups[4].value << 28 if self.len >= 5 else 0)) | (self.groups[5].value << 35 if self.len >= 6 else 0)) | (self.groups[6].value << 42 if self.len >= 7 else 0)) | (self.groups[7].value << 49 if self.len >= 8 else 0)) | (self.groups[8].value << 56 if self.len >= 9 else 0)) | (self.groups[9].value << 63 if self.len >= 10 else 0)) return getattr(self, '_m_value', None) class HeaderLead(KaitaiStruct): def __init__(self, _io, _parent=None, _root=None): super(Zchunk.HeaderLead, self).__init__(_io) self._parent = _parent self._root = _root self._read() def _read(self): self.magic = self._io.read_bytes(5) if not ((self.magic == b"\x00\x5A\x43\x4B\x31") or (self.magic == b"\x00\x5A\x48\x52\x31")) : raise kaitaistruct.ValidationNotAnyOfError(self.magic, self._io, u"/types/header_lead/seq/0") self.overall_checksum_type = Zchunk.ChecksumType(self._io, self, self._root) self.len_header_rest = Zchunk.CompressedInteger(self._io, self, self._root) self.header_checksum = self._io.read_bytes(self.overall_checksum_type.len_checksum) def _fetch_instances(self): pass self.overall_checksum_type._fetch_instances() self.len_header_rest._fetch_instances() @property def is_detached_header(self): """Determines whether this file is a zchunk detached header (`.zhr`). If not, it is a complete zchunk file (`.zck`). """ if hasattr(self, '_m_is_detached_header'): return self._m_is_detached_header self._m_is_detached_header = KaitaiStream.byte_array_index(self.magic, 2) == 72 return getattr(self, '_m_is_detached_header', None) class HeaderWithoutLead(KaitaiStruct): def __init__(self, _io, _parent=None, _root=None): super(Zchunk.HeaderWithoutLead, self).__init__(_io) self._parent = _parent self._root = _root self._read() def _read(self): self.preface = Zchunk.Preface(self._io, self, self._root) self.len_index = Zchunk.CompressedInteger(self._io, self, self._root) self._raw_index = self._io.read_bytes(self.len_index.value) _io__raw_index = KaitaiStream(BytesIO(self._raw_index)) self.index = Zchunk.Index(_io__raw_index, self, self._root) self.num_signatures = Zchunk.CompressedInteger(self._io, self, self._root) _ = self.num_signatures if not _.value == 0: raise kaitaistruct.ValidationExprError(self.num_signatures, self._io, u"/types/header_without_lead/seq/3") def _fetch_instances(self): pass self.preface._fetch_instances() self.len_index._fetch_instances() self.index._fetch_instances() self.num_signatures._fetch_instances() class Index(KaitaiStruct): def __init__(self, _io, _parent=None, _root=None): super(Zchunk.Index, self).__init__(_io) self._parent = _parent self._root = _root self._read() def _read(self): self.chunk_checksum_type = Zchunk.ChecksumType(self._io, self, self._root) self.num_chunks = Zchunk.CompressedInteger(self._io, self, self._root) _ = self.num_chunks if not _.value >= 1: raise kaitaistruct.ValidationExprError(self.num_chunks, self._io, u"/types/index/seq/1") if self._parent.preface.has_data_streams: pass self.dict_stream = Zchunk.CompressedInteger(self._io, self, self._root) _ = self.dict_stream if not _.value == 0: raise kaitaistruct.ValidationExprError(self.dict_stream, self._io, u"/types/index/seq/2") self.dict_checksum = self._io.read_bytes(self.chunk_checksum_type.len_checksum) if self._parent.preface.has_uncompressed_source: pass self.uncompressed_dict_checksum = self._io.read_bytes(self.chunk_checksum_type.len_checksum) self.len_dict = Zchunk.CompressedInteger(self._io, self, self._root) self.len_uncompressed_dict = Zchunk.CompressedInteger(self._io, self, self._root) self.chunks_metadata = [] for i in range(self.num_data_chunks): self.chunks_metadata.append(Zchunk.Chunk(self.chunk_checksum_type.len_checksum, self._parent.preface.has_data_streams, self._parent.preface.has_uncompressed_source, self._io, self, self._root)) def _fetch_instances(self): pass self.chunk_checksum_type._fetch_instances() self.num_chunks._fetch_instances() if self._parent.preface.has_data_streams: pass self.dict_stream._fetch_instances() if self._parent.preface.has_uncompressed_source: pass self.len_dict._fetch_instances() self.len_uncompressed_dict._fetch_instances() for i in range(len(self.chunks_metadata)): pass self.chunks_metadata[i]._fetch_instances() @property def num_data_chunks(self): """Number of data chunks. `num_chunks` counts the dictionary as chunk 0, so it is one greater than this number. """ if hasattr(self, '_m_num_data_chunks'): return self._m_num_data_chunks self._m_num_data_chunks = self.num_chunks.value - 1 return getattr(self, '_m_num_data_chunks', None) class OptionalElement(KaitaiStruct): def __init__(self, _io, _parent=None, _root=None): super(Zchunk.OptionalElement, self).__init__(_io) self._parent = _parent self._root = _root self._read() def _read(self): self.element_id = Zchunk.CompressedInteger(self._io, self, self._root) self.len_data = Zchunk.CompressedInteger(self._io, self, self._root) self.data = self._io.read_bytes(self.len_data.value) def _fetch_instances(self): pass self.element_id._fetch_instances() self.len_data._fetch_instances() class Preface(KaitaiStruct): def __init__(self, _io, _parent=None, _root=None): super(Zchunk.Preface, self).__init__(_io) self._parent = _parent self._root = _root self._read() def _read(self): self.data_checksum = self._io.read_bytes(self._root.lead.overall_checksum_type.len_checksum) self.flags = Zchunk.CompressedInteger(self._io, self, self._root) _ = self.flags if not _.value <= 7: raise kaitaistruct.ValidationExprError(self.flags, self._io, u"/types/preface/seq/1") self.compression_type_int = Zchunk.CompressedInteger(self._io, self, self._root) _ = self.compression_type_int if not ((_.value == int(Zchunk.CompressionTypes.none)) or (_.value == int(Zchunk.CompressionTypes.zstd))) : raise kaitaistruct.ValidationExprError(self.compression_type_int, self._io, u"/types/preface/seq/2") if self.has_optional_elements: pass self.num_optional_elements = Zchunk.CompressedInteger(self._io, self, self._root) _ = self.num_optional_elements if not _.value >= 1: raise kaitaistruct.ValidationExprError(self.num_optional_elements, self._io, u"/types/preface/seq/3") if self.has_optional_elements: pass self.optional_elements = [] for i in range(self.num_optional_elements.value): self.optional_elements.append(Zchunk.OptionalElement(self._io, self, self._root)) def _fetch_instances(self): pass self.flags._fetch_instances() self.compression_type_int._fetch_instances() if self.has_optional_elements: pass self.num_optional_elements._fetch_instances() if self.has_optional_elements: pass for i in range(len(self.optional_elements)): pass self.optional_elements[i]._fetch_instances() @property def compression_type(self): if hasattr(self, '_m_compression_type'): return self._m_compression_type self._m_compression_type = KaitaiStream.resolve_enum(Zchunk.CompressionTypes, self.compression_type_int.value) return getattr(self, '_m_compression_type', None) @property def has_data_streams(self): if hasattr(self, '_m_has_data_streams'): return self._m_has_data_streams self._m_has_data_streams = self.flags.value & 1 != 0 return getattr(self, '_m_has_data_streams', None) @property def has_optional_elements(self): if hasattr(self, '_m_has_optional_elements'): return self._m_has_optional_elements self._m_has_optional_elements = self.flags.value & 2 != 0 return getattr(self, '_m_has_optional_elements', None) @property def has_uncompressed_source(self): """The file may be applied against an uncompressed source. This adds an uncompressed checksum to every index entry, including the dictionary. """ if hasattr(self, '_m_has_uncompressed_source'): return self._m_has_uncompressed_source self._m_has_uncompressed_source = self.flags.value & 4 != 0 return getattr(self, '_m_has_uncompressed_source', None)