Merge pull request #1 from ModernMAK/dunder-magic

Move codebase to dunder-methods, metaclasses, and protocols

Author: ModernMAK

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = struct-lib
-version = 2022.0-b1
+version = 2022.1-b0
 author = Marcus Kertesz
 description = An object-oriented alternative to the built-in struct module. Still uses struct under the hood, but allows defining user structures, and packing into/from binaryio/bytesio types.
 long_description = file: README.md

setup.py

@@ -1,4 +1,5 @@
 from setuptools import setup
 
 if __name__ == "__main__":
-    setup()
+    setup()
+

src/structlib/abc_/__init__.py

@@ -0,0 +1,188 @@
+from abc import ABC, abstractmethod
+from typing import Tuple, BinaryIO, Any
+from structlib.io import bufferio, streamio
+from structlib.protocols.packing import Packable, IterPackable, StructPackable, PrimitivePackable, TPrim, DataclassPackable, DClassType, DClass, ConstPackable
+from structlib.protocols.typedef import align_of, size_of, TypeDefAlignable, TypeDefSizable, T
+from structlib.typing_ import WritableBuffer, ReadableBuffer, ReadableStream, WritableStream
+
+
+class PackableABC(Packable, TypeDefSizable, TypeDefAlignable, ABC):
+    """
+    A Packable which uses pack/unpack and a fixed size typedef to perform buffer/stream operations.
+    """
+
+    def pack_buffer(self, buffer: WritableBuffer, *args: Any, offset: int, origin: int) -> int:
+        packed = self.pack(*args)
+        alignment = align_of(self)
+        return bufferio.write(buffer, packed, alignment, offset, origin)
+
+    def unpack_buffer(self, buffer: ReadableBuffer, *, offset: int, origin: int) -> Tuple[int, Any]:
+        size = size_of(self)
+        alignment = align_of(self)
+        read, packed = bufferio.read(buffer, size, alignment, offset, origin)
+        unpacked = self.unpack(packed)
+        return read, unpacked
+
+    def pack_stream(self, stream: BinaryIO, *args: Any, origin: int) -> int:
+        packed = self.pack(*args)
+        alignment = align_of(self)
+        return streamio.write(stream, packed, alignment, origin)
+
+    def unpack_stream(self, stream: BinaryIO, *, origin: int) -> Tuple[int, Any]:
+        size = size_of(self)
+        alignment = align_of(self)
+        read, packed = streamio.read(stream, size, alignment, origin)
+        unpacked = self.unpack(packed)
+        return read, unpacked
+
+
+class ConstPackableABC(ConstPackable, TypeDefSizable, TypeDefAlignable, ABC):
+    """
+    A ConstPackable which uses pack/unpack and a fixed size typedef to perform buffer/stream operations.
+    """
+
+    def const_pack_buffer(self, buffer: WritableBuffer, offset: int, origin: int) -> int:
+        packed = self.const_pack()
+        alignment = align_of(self)
+        return bufferio.write(buffer, packed, alignment, offset, origin)
+
+    def const_unpack_buffer(self, buffer: ReadableBuffer, *, offset: int, origin: int) -> Tuple[int, Any]:
+        size = size_of(self)
+        alignment = align_of(self)
+        read, packed = bufferio.read(buffer, size, alignment, offset, origin)
+        unpacked = self.const_unpack(packed)
+        return read, unpacked
+
+    def const_pack_stream(self, stream: BinaryIO, *args: Any, origin: int) -> int:
+        packed = self.const_pack()
+        alignment = align_of(self)
+        return streamio.write(stream, packed, alignment, origin)
+
+    def const_unpack_stream(self, stream: BinaryIO, *, origin: int) -> Tuple[int, Any]:
+        size = size_of(self)
+        alignment = align_of(self)
+        read, packed = streamio.read(stream, size, alignment, origin)
+        unpacked = self.const_unpack(packed)
+        return read, unpacked
+
+
+class IterPackableABC(IterPackable, TypeDefSizable, TypeDefAlignable, ABC):
+    """
+    An IterPackable which uses iter_pack/iter_unpack to perform buffer/stream operations.
+    """
+
+    def iter_pack_buffer(self, buffer: WritableBuffer, *args: Any, offset: int, origin: int) -> int:
+        packed = self.iter_pack(*args)
+        alignment = align_of(self)
+        return bufferio.write(buffer, packed, alignment, offset, origin)
+
+    def iter_unpack_buffer(self, buffer: ReadableBuffer, iter_count: int, *, offset: int, origin: int) -> Tuple[int, Any]:
+        size = size_of(self) * iter_count
+        alignment = align_of(self)
+        read, packed = bufferio.read(buffer, size, alignment, offset, origin)
+        unpacked = self.iter_unpack(packed, iter_count)
+        return read, unpacked
+
+    def iter_pack_stream(self, stream: BinaryIO, *args: Any, origin: int) -> int:
+        packed = self.iter_pack(*args)
+        alignment = align_of(self)
+        return streamio.write(stream, packed, alignment, origin)
+
+    def iter_unpack_stream(self, stream: BinaryIO, iter_count: int, *, origin: int) -> Tuple[int, Any]:
+        size = size_of(self) * iter_count
+        alignment = align_of(self)
+        read, packed = streamio.read(stream, size, alignment, origin)
+        unpacked = self.iter_unpack(packed, iter_count)
+        return read, unpacked
+
+
+class StructPackableABC(StructPackable, TypeDefSizable, TypeDefAlignable, ABC):
+    """
+    A Packable which uses pack/unpack to perform buffer/stream operations.
+    """
+
+    def struct_pack_buffer(self, buffer: WritableBuffer, *args: Any, offset: int, origin: int) -> int:
+        packed = self.struct_pack(*args)
+        alignment = align_of(self)
+        return bufferio.write(buffer, packed, alignment, offset, origin)
+
+    def struct_unpack_buffer(self, buffer: ReadableBuffer, *, offset: int, origin: int) -> Tuple[int, Any]:
+        size = size_of(self)
+        alignment = align_of(self)
+        read, packed = bufferio.read(buffer, size, alignment, offset, origin)
+        unpacked = self.struct_unpack(packed)
+        return read, unpacked
+
+    def struct_pack_stream(self, stream: BinaryIO, *args: Any, origin: int) -> int:
+        packed = self.struct_pack(*args)
+        alignment = align_of(self)
+        return streamio.write(stream, packed, alignment, origin)
+
+    def struct_unpack_stream(self, stream: BinaryIO, *, origin: int) -> Tuple[int, Any]:
+        size = size_of(self)
+        alignment = align_of(self)
+        read, packed = streamio.read(stream, size, alignment, origin)
+        unpacked = self.struct_unpack(packed)
+        return read, unpacked
+
+
+class PrimitivePackableABC(PrimitivePackable, TypeDefSizable, TypeDefAlignable, ABC):
+    """
+    A Packable which uses pack/unpack to perform buffer/stream operations.
+    """
+
+    def prim_pack_buffer(self, buffer: WritableBuffer, arg: TPrim, *, offset: int = 0, origin: int = 0) -> int:
+        packed = self.prim_pack(arg)
+        alignment = align_of(self)
+        return bufferio.write(buffer, packed, alignment, offset, origin)
+
+    def unpack_prim_buffer(self, buffer: ReadableBuffer, *, offset: int = 0, origin: int = 0) -> Tuple[int, TPrim]:
+        size = size_of(self)
+        alignment = align_of(self)
+        read, packed = bufferio.read(buffer, size, alignment, offset, origin)
+        unpacked = self.unpack_prim(packed)
+        return read, unpacked
+
+    def prim_pack_stream(self, stream: WritableStream, arg: TPrim, *, origin: int = 0) -> int:
+        packed = self.prim_pack(arg)
+        alignment = align_of(self)
+        return streamio.write(stream, packed, alignment, origin)
+
+    def unpack_prim_stream(self, stream: ReadableStream, *, origin: int = 0) -> Tuple[int, TPrim]:
+        size = size_of(self)
+        alignment = align_of(self)
+        read, packed = streamio.read(stream, size, alignment, origin)
+        unpacked = self.unpack_prim(packed)
+        return read, unpacked
+
+
+class DataclassPackableABC(DataclassPackable, TypeDefSizable, TypeDefAlignable, ABC):
+    """
+    A Packable which uses pack/unpack to perform buffer/stream operations.
+    """
+
+    def dclass_pack_buffer(self, buffer: WritableBuffer, *, offset: int = 0, origin: int = 0) -> int:
+        packed = self.dclass_pack()
+        alignment = align_of(self)
+        return bufferio.write(buffer, packed, alignment, offset, origin)
+
+    @classmethod
+    def dclass_unpack_buffer(cls: DClassType, buffer: ReadableBuffer, *, offset: int = 0, origin: int = 0) -> Tuple[int, DClass]:
+        size = size_of(cls)
+        alignment = align_of(cls)
+        read, packed = bufferio.read(buffer, size, alignment, offset, origin)
+        unpacked = cls.dclass_unpack(packed)
+        return read, unpacked
+
+    def dclass_pack_stream(self, stream: WritableStream, *, origin: int = 0) -> int:
+        packed = self.dclass_pack()
+        alignment = align_of(self)
+        return streamio.write(stream, packed, alignment, origin)
+
+    @classmethod
+    def dclass_unpack_stream(cls: DClassType, stream: ReadableStream, *, origin: int = 0) -> Tuple[int, DClass]:
+        size = size_of(cls)
+        alignment = align_of(cls)
+        read, packed = streamio.read(stream, size, alignment, origin)
+        unpacked = cls.dclass_unpack(packed)
+        return read, unpacked

src/structlib/abc_/packing.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from copy import copy
+from typing import TypeVar
+
+from structlib.byteorder import ByteOrder
+from structlib.protocols.typedef import TypeDefAlignable, TypeDefByteOrder, TypeDefSizable
+
+T = TypeVar("T")
+
+
+class TypeDefAlignableABC(TypeDefAlignable):
+    def __init__(self, alignment: int):
+        self.__typedef_alignment__ = alignment
+
+    def __typedef_align_as__(self: T, alignment: int) -> T:
+        if self.__typedef_alignment__ == alignment:
+            return self
+        else:
+            inst = copy(self)
+            inst.__typedef_alignment__ = alignment
+            return inst
+
+
+class TypeDefSizableABC(TypeDefSizable):
+    def __init__(self, native_size: int):
+        self.__typedef_native_size__ = native_size
+
+
+class TypeDefByteOrderABC(TypeDefByteOrder):
+    def __init__(self, byteorder: ByteOrder):
+        self.__typedef_byteorder__ = byteorder
+
+    def __typedef_byteorder_as__(self: T, byteorder: ByteOrder) -> T:
+        if self.__typedef_byteorder__ == byteorder:
+            return self
+        else:
+            inst = copy(self)
+            inst.__typedef_byteorder__ = byteorder
+            return inst
+

src/structlib/abc_/typedef.py

@@ -0,0 +1,23 @@
+import sys
+from typing import Literal, Optional
+
+ByteOrder = Literal["little", "big"]
+
+LittleEndian: Literal["little"] = "little"
+BigEndian: Literal["big"] = "big"
+NetworkEndian: ByteOrder = BigEndian
+NativeEndian: ByteOrder = sys.byteorder
+
+
+def resolve_byteorder(*byteorder: Optional[ByteOrder], default: ByteOrder = NativeEndian):
+    """
+    Will resolve to the first non-None byteorder, if all byteorder objects are None, default will be used.
+    :param byteorder:
+    :param default:
+    :return:
+    """
+    for bom in byteorder:
+        if bom is None:
+            continue
+        return bom
+    return default