fox-it__dissect.cstruct-66

{
  "status": "ok",
  "stdout": "48:class {name}(Structure):\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "dissect/cstruct/compiler.py",
  "content": "from __future__ import annotations\n\nimport struct\nfrom collections import OrderedDict\nfrom textwrap import dedent\nfrom typing import TYPE_CHECKING, List\n\nfrom dissect.cstruct.bitbuffer import BitBuffer\nfrom dissect.cstruct.expression import Expression\nfrom dissect.cstruct.types import (\n    Array,\n    BytesInteger,\n    CharType,\n    Enum,\n    EnumInstance,\n    Field,\n    Flag,\n    FlagInstance,\n    Instance,\n    PackedType,\n    Pointer,\n    PointerInstance,\n    Structure,\n    Union,\n    WcharType,\n)\n\nif TYPE_CHECKING:\n    from dissect.cstruct import cstruct\n\n\nclass Compiler:\n    \"\"\"Compiler for cstruct structures. Creates somewhat optimized parsing code.\"\"\"\n\n    TYPES = (\n        Structure,\n        Pointer,\n        Enum,\n        Flag,\n        Array,\n        PackedType,\n        CharType,\n        WcharType,\n        BytesInteger,\n    )\n\n    COMPILE_TEMPLATE = \"\"\"\nclass {name}(Structure):\n    def __init__(self, cstruct, structure, source=None):\n        self.structure = structure\n        self.source = source\n        super().__init__(cstruct, structure.name, structure.fields, anonymous=structure.anonymous)\n\n    def _read(self, stream, context=None):\n        r = OrderedDict()\n        sizes = {{}}\n        bitreader = BitBuffer(stream, self.cstruct.endian)\n\n{read_code}\n\n        return Instance(self, r, sizes)\n\n    def add_field(self, name, type_, offset=None):\n        raise NotImplementedError(\"Can't add fields to a compiled structure\")\n\n    def __repr__(self):\n        return '<Structure {name} +compiled>'\n\"\"\"\n\n    def __init__(self, cstruct: cstruct):\n        self.cstruct = cstruct\n\n    def compile(self, structure: Structure) -> Structure:\n        if isinstance(structure, Union) or structure.align:\n            return structure\n\n        structure_name = structure.name\n\n        try:\n            # Generate struct class based on provided structure type\n            source = self.gen_struct_class(structure_name, structure)\n        except TypeError:\n            return structure\n\n        # Create code object that can be executed later on\n        code_object = compile(\n            source,\n            f\"<compiled {structure_name}>\",\n            \"exec\",\n        )\n\n        env = {\n            \"OrderedDict\": OrderedDict,\n            \"Structure\": Structure,\n            \"Instance\": Instance,\n            \"Expression\": Expression,\n            \"EnumInstance\": EnumInstance,\n            \"FlagInstance\": FlagInstance,\n            \"PointerInstance\": PointerInstance,\n            \"BytesInteger\": BytesInteger,\n            \"BitBuffer\": BitBuffer,\n            \"struct\": struct,\n            \"range\": range,\n        }\n\n        exec(code_object, env)\n        return env[structure_name](self.cstruct, structure, source)\n\n    def gen_struct_class(self, name: str, structure: Structure) -> str:\n        blocks = []\n        classes = []\n        cur_block = []\n        read_size = 0\n        prev_was_bits = False\n\n        for field in structure.fields:\n            field_type = self.cstruct.resolve(field.type)\n\n            if not isinstance(field_type, self.TYPES):\n                raise TypeError(f\"Unsupported type for compiler: {field_type}\")\n\n            if isinstance(field_type, Structure) or (\n                isinstance(field_type, Array) and isinstance(field_type.type, (Structure, Array))\n            ):\n                blocks.append(self.gen_read_block(read_size, cur_block))\n\n                struct_read = \"s = stream.tell()\\n\"\n                if isinstance(field_type, Array):\n                    num = field_type.count\n\n                    if isinstance(num, Expression):\n                        num = f\"max(0, Expression(self.cstruct, '{num.expression}').evaluate(r))\"\n\n                    struct_read += dedent(\n                        f\"\"\"\n                        r['{field.name}'] = []\n                        for _ in range({num}):\n                            r['{field.name}'].append(self.lookup['{field.name}'].type.type._read(stream, context=r))\n                        sizes['{field.name}'] = stream.tell() - s\n                        \"\"\"\n                    )\n                elif isinstance(field_type, Structure) and field_type.anonymous:\n                    struct_read += dedent(\n                        f\"\"\"\n                        v = self.lookup[\"{field.name}\"].type._read(stream, context=r)\n                        r.update(v._values)\n                        sizes.update(v._sizes)\n                        \"\"\"\n                    )\n                else:",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 408,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "dissect/cstruct/compiler.py",
  "content": "                    struct_read += dedent(\n                        f\"\"\"\n                        r['{field.name}'] = self.lookup['{field.name}'].type._read(stream, context=r)\n                        sizes['{field.name}'] = stream.tell() - s\n                        \"\"\"\n                    )\n\n                blocks.append(struct_read)\n                read_size = 0\n                cur_block = []\n                continue\n\n            if field.bits:\n                blocks.append(self.gen_read_block(read_size, cur_block))\n                if isinstance(field_type, Enum):\n                    bitfield_read = dedent(\n                        f\"\"\"\n                        r['{field.name}'] = self.cstruct.{field.type.name}(\n                            bitreader.read(self.cstruct.{field.type.type.name}, {field.bits})\n                        )\n                        \"\"\"\n                    )\n                else:\n                    bitfield_read = f\"r['{field.name}'] = bitreader.read(self.cstruct.{field.type.name}, {field.bits})\"\n                blocks.append(bitfield_read)\n\n                read_size = 0\n                cur_block = []\n                prev_was_bits = True\n                continue\n\n            if prev_was_bits:\n                blocks.append(\"bitreader.reset()\")\n                prev_was_bits = False\n\n            try:\n                count = len(field_type)\n                read_size += count\n                cur_block.append(field)\n            except TypeError:\n                if cur_block:\n                    blocks.append(self.gen_read_block(read_size, cur_block))\n\n                blocks.append(self.gen_dynamic_block(field))\n                read_size = 0\n                cur_block = []\n\n        if len(cur_block):\n            blocks.append(self.gen_read_block(read_size, cur_block))\n\n        read_code = \"\\n\\n\".join(blocks)\n        read_code = \"\\n\".join([\"    \" * 2 + line for line in read_code.split(\"\\n\")])\n\n        classes.append(self.COMPILE_TEMPLATE.format(name=name, read_code=read_code))\n        return \"\\n\\n\".join(classes)\n\n    def gen_read_block(self, size: int, block: List[str]) -> str:\n        template = dedent(\n            f\"\"\"\n            buf = stream.read({size})\n            if len(buf) != {size}: raise EOFError()\n            data = struct.unpack(self.cstruct.endian + '{{}}', buf)\n            {{}}\n            \"\"\"\n        )\n\n        read_code = []\n        fmt = []\n\n        cur_type = None\n        cur_count = 0\n\n        buf_offset = 0\n        data_offset = 0\n\n        for field in block:\n            field_type = self.cstruct.resolve(field.type)\n            read_type = field_type\n\n            count = 1\n            data_count = 1\n\n            if isinstance(read_type, (Enum, Flag)):\n                read_type = read_type.type\n            elif isinstance(read_type, Pointer):\n                read_type = self.cstruct.pointer\n\n            if isinstance(field_type, Array):\n                count = read_type.count\n                data_count = count\n                read_type = read_type.type\n\n                if isinstance(read_type, (Enum, Flag)):\n                    read_type = read_type.type\n                elif isinstance(read_type, Pointer):\n                    read_type = self.cstruct.pointer\n\n                if isinstance(read_type, (CharType, WcharType, BytesInteger)):\n                    read_slice = f\"{buf_offset}:{buf_offset + + (count * read_type.size)}\"\n                else:\n                    read_slice = f\"{data_offset}:{data_offset + count}\"\n            elif isinstance(read_type, CharType):\n                read_slice = f\"{buf_offset}:{buf_offset + 1}\"\n            elif isinstance(read_type, (WcharType, BytesInteger)):\n                read_slice = f\"{buf_offset}:{buf_offset + read_type.size}\"\n            else:\n                read_slice = str(data_offset)\n\n            if not cur_type:\n                if isinstance(read_type, PackedType):\n                    cur_type = read_type.packchar\n                else:\n                    cur_type = \"x\"\n\n            if isinstance(read_type, (PackedType, CharType, WcharType, BytesInteger, Enum, Flag)):\n                char_count = count\n\n                if isinstance(read_type, (CharType, WcharType, BytesInteger)):\n                    data_count = 0\n                    pack_char = \"x\"\n                    char_count *= read_type.size\n                else:\n                    pack_char = read_type.packchar\n\n                if cur_type != pack_char:\n                    fmt.append(f\"{cur_count}{cur_type}\")\n                    cur_count = 0\n\n                cur_count += char_count\n                cur_type = pack_char\n\n            if isinstance(read_type, BytesInteger):\n                getter = \"BytesInteger.parse(buf[{slice}], {size}, {count}, {signed}, self.cstruct.endian){data_slice}\"\n\n                getter = getter.format(\n                    slice=read_slice,\n                    size=read_type.size,\n                    count=count,\n                    signed=read_type.signed,\n                    data_slice=\"[0]\" if count == 1 else \"\",\n                )\n            elif isinstance(read_type, (CharType, WcharType)):\n                getter = f\"buf[{read_slice}]\"\n\n                if isinstance(read_type, WcharType):\n                    getter += \".decode('utf-16-le' if self.cstruct.endian == '<' else 'utf-16-be')\"\n            else:\n                getter = f\"data[{read_slice}]\"\n\n            if isinstance(field_type, (Enum, Flag)):",
  "start_line": 151,
  "end_line": 300,
  "total_lines": 408,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "dissect/cstruct/compiler.py",
  "content": "                enum_type = field_type.__class__.__name__\n                getter = f\"{enum_type}Instance(self.cstruct.{field_type.name}, {getter})\"\n            elif isinstance(field_type, Array) and isinstance(field_type.type, (Enum, Flag)):\n                enum_type = field_type.type.__class__.__name__\n                getter = f\"[{enum_type}Instance(self.cstruct.{field_type.type.name}, d) for d in {getter}]\"\n            elif isinstance(field_type, Pointer):\n                getter = f\"PointerInstance(self.cstruct.{field_type.type.name}, stream, {getter}, r)\"\n            elif isinstance(field_type, Array) and isinstance(field_type.type, Pointer):\n                getter = f\"[PointerInstance(self.cstruct.{field_type.type.type.name}, stream, d, r) for d in {getter}]\"\n            elif isinstance(field_type, Array) and isinstance(read_type, PackedType):\n                getter = f\"list({getter})\"\n\n            read_code.append(f\"r['{field.name}'] = {getter}\")\n            read_code.append(f\"sizes['{field.name}'] = {count * read_type.size}\")\n\n            data_offset += data_count\n            buf_offset += count * read_type.size\n\n        if cur_count:\n            fmt.append(f\"{cur_count}{cur_type}\")\n\n        return template.format(\"\".join(fmt), \"\\n\".join(read_code))\n\n    def gen_dynamic_block(self, field: Field) -> str:\n        if not isinstance(field.type, Array):\n            raise TypeError(f\"Only Array can be dynamic, got {field.type!r}\")\n\n        field_type = self.cstruct.resolve(field.type.type)\n        reader = None\n\n        if isinstance(field_type, (Enum, Flag)):\n            field_type = field_type.type\n\n        if not field.type.count:  # Null terminated\n            if isinstance(field_type, PackedType):\n                reader = dedent(\n                    f\"\"\"\n                    t = []\n                    while True:\n                        d = stream.read({field_type.size})\n                        if len(d) != {field_type.size}: raise EOFError()\n                        v = struct.unpack(self.cstruct.endian + '{field_type.packchar}', d)[0]\n                        if v == 0: break\n                        t.append(v)\n                    \"\"\"\n                )\n\n            elif isinstance(field_type, (CharType, WcharType)):\n                null = \"\\\\x00\" * field_type.size\n                reader = dedent(\n                    f\"\"\"\n                    t = []\n                    while True:\n                        c = stream.read({field_type.size})\n                        if len(c) != {field_type.size}: raise EOFError()\n                        if c == b'{null}': break\n                        t.append(c)\n                    t = b''.join(t)\"\"\"  # It's important there's no newline here because of the optional decode\n                )\n\n                if isinstance(field_type, WcharType):\n                    reader += \".decode('utf-16-le' if self.cstruct.endian == '<' else 'utf-16-be')\"\n            elif isinstance(field_type, BytesInteger):\n                reader = dedent(\n                    f\"\"\"\n                    t = []\n                    while True:\n                        d = stream.read({field_type.size})\n                        if len(d) != {field_type.size}: raise EOFError()\n                        v = BytesInteger.parse(d, {field_type.size}, 1, {field_type.signed}, self.cstruct.endian)\n                        if v == 0: break\n                        t.append(v)\n                    \"\"\"\n                )\n\n            if isinstance(field_type, (Enum, Flag)):\n                enum_type = field_type.__class__.__name__\n                reader += f\"\\nt = [{enum_type}Instance(self.cstruct.{field_type.name}, d) for d in t]\"\n\n            if not reader:\n                raise TypeError(f\"Couldn't compile a reader for array {field!r}, {field_type!r}.\")\n\n            return f\"s = stream.tell()\\n{reader}\\nr['{field.name}'] = t\\nsizes['{field.name}'] = stream.tell() - s\"\n\n        expr_read = dedent(\n            f\"\"\"\n            dynsize = max(0, Expression(self.cstruct, \"{field.type.count.expression}\").evaluate(r))\n            buf = stream.read(dynsize * {field_type.size})\n            if len(buf) != dynsize * {field_type.size}: raise EOFError()\n            r['{field.name}'] = {{reader}}\n            sizes['{field.name}'] = dynsize * {field_type.size}\n            \"\"\"\n        )\n\n        if isinstance(field_type, PackedType):\n            reader = f\"list(struct.unpack(self.cstruct.endian + f'{{dynsize}}{field_type.packchar}', buf))\"\n        elif isinstance(field_type, (CharType, WcharType)):\n            reader = \"buf\"\n            if isinstance(field_type, WcharType):\n                reader += \".decode('utf-16-le' if self.cstruct.endian == '<' else 'utf-16-be')\"\n        elif isinstance(field_type, BytesInteger):\n            reader = f\"BytesInteger.parse(buf, {field_type.size}, dynsize, {field_type.signed}, self.cstruct.endian)\"\n\n        if isinstance(field_type, (Enum, Flag)):\n            enum_type = field_type.__class__.__name__\n            reader += f\"[{enum_type}Instance(self.cstruct.{field_type.name}, d) for d in {reader}]\"\n\n        return expr_read.format(reader=reader, size=None)",
  "start_line": 301,
  "end_line": 408,
  "total_lines": 408,
  "is_truncated": false
}

{
  "status": "ok",
  "stdout": "",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "stdout": "__init__.py\nbase.py\nbytesinteger.py\nchartype.py\nenum.py\nflag.py\ninstance.py\npackedtype.py\npointer.py\nstructure.py\nvoidtype.py\nwchartype.py\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "dissect/cstruct/types/structure.py",
  "content": "from __future__ import annotations\n\nimport io\nfrom collections import OrderedDict\nfrom typing import TYPE_CHECKING, Any, BinaryIO, List\n\nfrom dissect.cstruct.bitbuffer import BitBuffer\nfrom dissect.cstruct.types import BaseType, Enum, Instance\n\nif TYPE_CHECKING:\n    from dissect.cstruct import cstruct\n\n\nclass Field:\n    \"\"\"Holds a structure field.\"\"\"\n\n    def __init__(self, name: str, type_: BaseType, bits: int = None, offset: int = None):\n        self.name = name\n        self.type = type_\n        self.bits = bits\n        self.offset = offset\n        self.alignment = type_.alignment\n\n    def __repr__(self):\n        bits_str = f\" : {self.bits}\" if self.bits else \"\"\n        return f\"<Field {self.name} {self.type}{bits_str}>\"\n\n\nclass Structure(BaseType):\n    \"\"\"Type class for structures.\"\"\"\n\n    def __init__(\n        self, cstruct: cstruct, name: str, fields: List[Field] = None, align: bool = False, anonymous: bool = False\n    ):\n        super().__init__(cstruct)\n        self.name = name\n        self.size = None\n        self.alignment = None\n\n        self.lookup = OrderedDict()\n        self.fields = fields\n\n        self.align = align\n        self.anonymous = anonymous\n        self.dynamic = False\n\n        for field in self.fields:\n            self.lookup[field.name] = field\n            if isinstance(field.type, Structure) and field.type.anonymous:\n                self.lookup.update(field.type.lookup)\n\n        self._calc_size_and_offsets()\n\n    def __len__(self) -> int:\n        if self.dynamic:\n            raise TypeError(\"Dynamic size\")\n\n        if self.size is None:\n            self._calc_size_and_offsets()\n\n        return self.size\n\n    def __repr__(self) -> str:\n        return f\"<Structure {self.name}>\"\n\n    def _calc_size_and_offsets(self) -> None:\n        \"\"\"Iterate all fields in this structure to calculate the field offsets and total structure size.\n\n        If a structure has a dynamic field, further field offsets will be set to None and self.dynamic\n        will be set to True.\n        \"\"\"\n        # The current offset, set to None if we become dynamic\n        offset = 0\n        # The current alignment for this structure\n        alignment = 0\n\n        # The current bit field type\n        bits_type = None\n        # The offset of the current bit field, set to None if we become dynamic\n        bits_field_offset = 0\n        # How many bits we have left in the current bit field\n        bits_remaining = 0\n\n        for field in self.fields:\n            if field.offset is not None:\n                # If a field already has an offset, it's leading\n                offset = field.offset\n\n            if self.align and offset is not None:\n                # Round to next alignment\n                offset += -offset & (field.alignment - 1)\n\n            # The alignment of this struct is equal to its largest members' alignment\n            alignment = max(alignment, field.type.alignment)\n\n            if field.bits:\n                field_type = field.type\n\n                if isinstance(field_type, Enum):\n                    field_type = field_type.type\n\n                # Bit fields have special logic\n                if (\n                    # Exhausted a bit field\n                    bits_remaining == 0\n                    # Moved to a bit field of another type, e.g. uint16 f1 : 8, uint32 f2 : 8;\n                    or field_type != bits_type\n                    # Still processing a bit field, but it's at a different offset due to alignment or a manual offset\n                    or (bits_type is not None and offset > bits_field_offset + bits_type.size)\n                ):\n                    # ... if any of this is true, we have to move to the next field\n                    bits_type = field_type\n                    bits_count = bits_type.size * 8\n                    bits_remaining = bits_count\n                    bits_field_offset = offset\n\n                    if offset is not None:\n                        # We're not dynamic, update the structure size and current offset\n                        offset += bits_type.size\n\n                    field.offset = bits_field_offset\n\n                bits_remaining -= field.bits\n\n                if bits_remaining < 0:\n                    raise ValueError(\"Straddled bit fields are unsupported\")\n            else:\n                # Reset bits stuff\n                bits_type = None\n                bits_field_offset = bits_remaining = 0\n\n                field.offset = offset\n\n                if offset is not None:\n                    # We're not dynamic, update the structure size and current offset\n                    try:\n                        field_len = len(field.type)\n                    except TypeError:\n                        # This field is dynamic\n                        offset = None\n                        self.dynamic = True\n                        continue\n\n                    offset += field_len\n\n        if self.align and offset is not None:\n            # Add \"tail padding\" if we need to align\n            # This bit magic rounds up to the next alignment boundary\n            # E.g. offset = 3; alignment = 8; -offset & (alignment - 1) = 5\n            offset += -offset & (alignment - 1)",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 376,
  "is_truncated": true
}

{
  "status": "ok",
  "stdout": "Successfully compiled!\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "stdout": "Caught expected SyntaxError: invalid syntax (<compiled in>, line 2)\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "stdout": "57:        bitreader = BitBuffer(stream, self.cstruct.endian)\n71:        self.cstruct = cstruct\n107:        return env[structure_name](self.cstruct, structure, source)\n117:            field_type = self.cstruct.resolve(field.type)\n132:                        num = f\"max(0, Expression(self.cstruct, '{num.expression}').evaluate(r))\"\n168:                        r['{field.name}'] = self.cstruct.{field.type.name}(\n169:                            bitreader.read(self.cstruct.{field.type.type.name}, {field.bits})\n174:                    bitfield_read = f\"r['{field.name}'] = bitreader.read(self.cstruct.{field.type.name}, {field.bits})\"\n212:            data = struct.unpack(self.cstruct.endian + '{{}}', buf)\n227:            field_type = self.cstruct.resolve(field.type)\n236:                read_type = self.cstruct.pointer\n246:                    read_type = self.cstruct.pointer\n283:                getter = \"BytesInteger.parse(buf[{slice}], {size}, {count}, {signed}, self.cstruct.endian){data_slice}\"\n296:                    getter += \".decode('utf-16-le' if self.cstruct.endian == '<' else 'utf-16-be')\"\n302:                getter = f\"{enum_type}Instance(self.cstruct.{field_type.name}, {getter})\"\n305:                getter = f\"[{enum_type}Instance(self.cstruct.{field_type.type.name}, d) for d in {getter}]\"\n307:                getter = f\"PointerInstance(self.cstruct.{field_type.type.name}, stream, {getter}, r)\"\n309:                getter = f\"[PointerInstance(self.cstruct.{field_type.type.type.name}, stream, d, r) for d in {getter}]\"\n328:        field_type = self.cstruct.resolve(field.type.type)\n342:                        v = struct.unpack(self.cstruct.endian + '{field_type.packchar}', d)[0]\n362:                    reader += \".decode('utf-16-le' if self.cstruct.endian == '<' else 'utf-16-be')\"\n370:                        v = BytesInteger.parse(d, {field_type.size}, 1, {field_type.signed}, self.cstruct.endian)\n378:                reader += f\"\\nt = [{enum_type}Instance(self.cstruct.{field_type.name}, d) for d in t]\"\n387:            dynsize = max(0, Expression(self.cstruct, \"{field.type.count.expression}\").evaluate(r))\n396:            reader = f\"list(struct.unpack(self.cstruct.endian + f'{{dynsize}}{field_type.packchar}', buf))\"\n400:                reader += \".decode('utf-16-le' if self.cstruct.endian == '<' else 'utf-16-be')\"\n402:            reader = f\"BytesInteger.parse(buf, {field_type.size}, dynsize, {field_type.signed}, self.cstruct.endian)\"\n406:            reader += f\"[{enum_type}Instance(self.cstruct.{field_type.name}, d) for d in {reader}]\"\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "dissect/cstruct/types/base.py",
  "content": "from __future__ import annotations\n\nfrom io import BytesIO\nfrom typing import TYPE_CHECKING, Any, BinaryIO, List\n\nfrom dissect.cstruct.exceptions import ArraySizeError\nfrom dissect.cstruct.expression import Expression\n\nif TYPE_CHECKING:\n    from dissect.cstruct import cstruct\n\n\nclass BaseType:\n    \"\"\"Base class for cstruct type classes.\"\"\"\n\n    def __init__(self, cstruct: cstruct):\n        self.cstruct = cstruct\n\n    def __getitem__(self, count: int) -> Array:\n        return Array(self.cstruct, self, count)\n\n    def __call__(self, *args, **kwargs) -> Any:\n        if len(args) > 0:\n            return self.read(*args, **kwargs)\n\n        result = self.default()\n        if kwargs:\n            for k, v in kwargs.items():\n                setattr(result, k, v)\n\n        return result\n\n    def reads(self, data: bytes) -> Any:\n        \"\"\"Parse the given data according to the type that implements this class.\n\n        Args:\n            data: Byte string to parse.\n\n        Returns:\n            The parsed value of this type.\n        \"\"\"\n\n        return self._read(BytesIO(data))\n\n    def dumps(self, data: Any) -> bytes:\n        \"\"\"Dump the given data according to the type that implements this class.\n\n        Args:\n            data: Data to dump.\n\n        Returns:\n            The resulting bytes.\n\n        Raises:\n            ArraySizeError: Raised when ``len(data)`` does not match the size of a statically sized array field.\n        \"\"\"\n        out = BytesIO()\n        self._write(out, data)\n        return out.getvalue()\n\n    def read(self, obj: BinaryIO, *args, **kwargs) -> Any:\n        \"\"\"Parse the given data according to the type that implements this class.\n\n        Args:\n            obj: Data to parse. Can be a (byte) string or a file-like object.\n\n        Returns:\n            The parsed value of this type.\n        \"\"\"\n        if isinstance(obj, (bytes, memoryview, bytearray)):\n            return self.reads(obj)\n\n        return self._read(obj)\n\n    def write(self, stream: BinaryIO, data: Any) -> int:\n        \"\"\"Write the given data to a writable file-like object according to the\n        type that implements this class.\n\n        Args:\n            stream: Writable file-like object to write to.\n            data: Data to write.\n\n        Returns:\n            The amount of bytes written.\n\n        Raises:\n            ArraySizeError: Raised when ``len(data)`` does not match the size of a statically sized array field.\n        \"\"\"\n        return self._write(stream, data)\n\n    def _read(self, stream: BinaryIO, context: dict[str, Any] = None) -> Any:\n        raise NotImplementedError()\n\n    def _read_array(self, stream: BinaryIO, count: int, context: dict[str, Any] = None) -> List[Any]:\n        return [self._read(stream, context) for _ in range(count)]\n\n    def _read_0(self, stream: BinaryIO, context: dict[str, Any] = None) -> List[Any]:\n        raise NotImplementedError()\n\n    def _write(self, stream: BinaryIO, data: Any) -> int:\n        raise NotImplementedError()\n\n    def _write_array(self, stream: BinaryIO, data: Any) -> int:\n        num = 0\n        for i in data:\n            num += self._write(stream, i)\n\n        return num\n\n    def _write_0(self, stream: BinaryIO, data: Any) -> int:\n        raise NotImplementedError()\n\n    def default(self) -> Any:\n        \"\"\"Return a default value of this type.\"\"\"\n        raise NotImplementedError()\n\n    def default_array(self, count: int) -> List[Any]:\n        \"\"\"Return a default array of this type.\"\"\"\n        return [self.default() for _ in range(count)]\n\n\nclass Array(BaseType):\n    \"\"\"Implements a fixed or dynamically sized array type.\n\n    Example:\n        When using the default C-style parser, the following syntax is supported:\n\n            x[3] -> 3 -> static length.\n            x[] -> None -> null-terminated.\n            x[expr] -> expr -> dynamic length.\n    \"\"\"\n\n    def __init__(self, cstruct: cstruct, type_: BaseType, count: int):\n        self.type = type_\n        self.count = count\n        self.null_terminated = self.count is None\n        self.dynamic = isinstance(self.count, Expression)\n        self.alignment = type_.alignment\n        super().__init__(cstruct)\n\n    def __repr__(self) -> str:\n        if self.null_terminated:\n            return f\"{self.type}[]\"\n\n        return f\"{self.type}[{self.count}]\"\n\n    def __len__(self) -> int:\n        if self.dynamic or self.null_terminated:\n            raise TypeError(\"Dynamic size\")\n",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 209,
  "is_truncated": true
}