diff --git a/docs/sphinx/source/library/fields/common.rst b/docs/sphinx/source/library/fields/common.rst index 02a826b..f039da1 100644 --- a/docs/sphinx/source/library/fields/common.rst +++ b/docs/sphinx/source/library/fields/common.rst @@ -294,6 +294,20 @@ Special Structs .. versionadded:: 2.9.0 +.. autoclass:: caterpillar.fields.Unnamed + :members: + + .. versionadded:: 2.10.0 + +.. autoclass:: caterpillar.fields.sized + :members: + + Inline alternative to ``.sized(length)``, given as the last argument of + ``Unnamed``'s subscript instead: ``Unnamed[TypeA, TypeB, sized[length]]``. + + At most one may be given, and it must be the last argument. + + .. versionadded:: 2.10.0 .. autodata:: caterpillar.fields.Pass diff --git a/docs/sphinx/source/reference/datamodel/standard.rst b/docs/sphinx/source/reference/datamodel/standard.rst index 9a88311..cba8d8d 100644 --- a/docs/sphinx/source/reference/datamodel/standard.rst +++ b/docs/sphinx/source/reference/datamodel/standard.rst @@ -12,6 +12,7 @@ compatibility with older versions of the library, making them particularly impor sequence.rst struct union + unnamed bitfield templates diff --git a/docs/sphinx/source/reference/datamodel/unnamed.rst b/docs/sphinx/source/reference/datamodel/unnamed.rst new file mode 100644 index 0000000..80b3855 --- /dev/null +++ b/docs/sphinx/source/reference/datamodel/unnamed.rst @@ -0,0 +1,71 @@ +.. _unnamed-reference: + +Unnamed +======= + +**Unnamed** provides a different take on the same C-union idea as :ref:`union-reference`: +instead of a whole struct where every field shares memory, :class:`~caterpillar.fields.Unnamed` +is a single *field* whose value may be one of several possible types, all overlaid on the +same fixed-size window of bytes. It is declared with a subscription syntax rather than a +decorator: + +.. code-block:: python + + >>> @struct + ... class Format: + ... value: Unnamed[uint32, CString(4)] + ... + +Unlike :func:`~caterpillar.model.union`, an :class:`~caterpillar.fields.Unnamed` value does +not eagerly decode every alternative. The fixed-size byte window is read once, immediately, +like any other field - but each alternative is only decoded (and then cached) the first time +it is actually requested, via :code:`value[i]` or the shorthand attribute form +:code:`value._i`: + +.. code-block:: python + + >>> obj = unpack(Format, b"\x01\x00\x00\x00ABC\x00") + >>> obj.value[0] + 1 + >>> obj.value._1 + 'ABC' + +Sizing +------ + +By default, the field's size is the *largest* of its member types' sizes - every member +must therefore have a statically-known, fixed :code:`__size__`. An explicit size (an +:code:`int` or a context lambda) can be supplied via :code:`.sized(...)` to override this, +which is required if any member has a greedy or otherwise unsized :code:`__size__`: + +.. code-block:: python + + >>> @struct + ... class Format: + ... n: uint16 + ... value: Unnamed[uint32, uint16].sized(this.n) + ... + +Access +------ + +* :code:`value[i]` / :code:`value._i` - decode (or return the cached) member ``i``. + Negative indices are supported. +* :code:`value[i:j]` - returns a plain :code:`tuple` of the decoded members in that range. +* :code:`len(value)` - the number of members. +* Iterating a value decodes and yields every member in order. + +An :class:`~caterpillar.fields.Unnamed` value can also be constructed directly, without a +prior unpack - this is how you produce a value for packing from scratch: + +>>> Unnamed() # nothing assigned yet +Unnamed() +>>> Unnamed(1, 2, 3) # positional: fills index 0, 1, 2 +Unnamed(1, 2, 3) +>>> Unnamed(_3=0xBEEF) # keyword: only index 3 is set +Unnamed(?, ?, ?, 48879) + +Positional arguments fill indices left to right starting at ``0``; keyword arguments named +``_`` (e.g. ``_0``, ``_3``) set an arbitrary index directly. + +For more information and examples, see :ref:`tutorial-unnamed`. diff --git a/docs/sphinx/source/tutorial/advanced/index.rst b/docs/sphinx/source/tutorial/advanced/index.rst index e53f49b..c43c094 100644 --- a/docs/sphinx/source/tutorial/advanced/index.rst +++ b/docs/sphinx/source/tutorial/advanced/index.rst @@ -22,6 +22,7 @@ create and manipulate your own custom struct classes in Python. actions bitfield union + unnamed templates diff --git a/docs/sphinx/source/tutorial/advanced/unnamed.rst b/docs/sphinx/source/tutorial/advanced/unnamed.rst new file mode 100644 index 0000000..45f3414 --- /dev/null +++ b/docs/sphinx/source/tutorial/advanced/unnamed.rst @@ -0,0 +1,86 @@ +.. _tutorial-unnamed: + +Unnamed +======= + +The previous section introduced :ref:`tutorial-union`, which mimics a C union at the level +of a whole struct: every field of the struct shares the same memory, and assigning one field +updates all the others. :class:`~caterpillar.fields.Unnamed` brings the same idea down to a +single *field*: one position in a struct that may hold any one of several possible types, all +read from - and written to - the very same window of bytes. + +Where a *union* struct eagerly keeps every field in sync, an :class:`~caterpillar.fields.Unnamed` +field is lazy: the fixed-size bytes are read once, immediately, like any other field, but no +particular interpretation of those bytes is decoded until you actually ask for it. + +A basic example +---------------- + +Suppose a record stores a 4-byte value that should sometimes be read as a 32-bit integer, and +sometimes as a short ASCII tag. Instead of declaring two separate fields and manually keeping +track of which one is meaningful, declare a single :code:`Unnamed[...]` field listing both +possible types: + +.. code-block:: python + :caption: A field that can be read as either an integer or a tag + + @struct + class Record: + kind: uint16 + value: Unnamed[uint32, CString(4)] + +Unpacking reads the 2-byte :code:`kind` field plus exactly 4 bytes for :code:`value` - the +size of the largest member - up front. Neither :code:`uint32` nor :code:`CString` is decoded +yet at this point: + +>>> obj = unpack(Record, b"\x01\x00ABCD") +>>> obj.value[0] # decode as uint32, then cache it +1145258561 +>>> obj.value._1 # decode as CString, then cache it +'ABCD' +>>> obj.value +Unnamed(1145258561, 'ABCD') + +Both :code:`value[i]` (works with any integer, including negative indices) and the shorthand +attribute form :code:`value._i` are equivalent. + +Sizing +------ + +By default, the field is sized as the *largest* member (so every member must have a fixed, +statically-known size). Use :code:`.sized(...)` to declare an explicit size instead - a plain +:code:`int`, or a context lambda evaluated against the enclosing struct, exactly like sizing a +sequence: + +.. code-block:: python + :caption: Explicit, context-dependent sizing + + @struct + class Record: + length: uint16 + value: Unnamed[uint32, uint16].sized(this.length) + # using sized[...] expression is also possible (must be last item) + .. value: Unnamed[uint32, uint16, sized[this.length]] + +Building a value for packing +------------------------------ + +An :class:`~caterpillar.fields.Unnamed` value does not have to come from :code:`unpack(...)` - +you can construct one directly to pack it. All of the following are valid: + +>>> Unnamed() # nothing assigned yet - cannot be packed like this +Unnamed() +>>> Unnamed(1, 2, 3) # positional arguments fill index 0, 1, 2, ... +Unnamed(1, 2, 3) +>>> Unnamed(_1="XY") # keyword arguments (``_``) set one index directly +Unnamed(?, 'XY') + +Since a positional-then-keyword call is processed left to right, whichever index was assigned +*last* is the one that ends up encoded when the value is packed. + +>>> pack(Record(kind=1, value=Unnamed(_1="XY"))) +b'\x01\x00XY\x00\x00' +>>> pack(Record(kind=2, value=Unnamed(0xCAFEBABE))) +b'\x02\x00\xbe\xba\xfe\xca' + + diff --git a/src/caterpillar/_common.py b/src/caterpillar/_common.py index 9ae22fd..396d39f 100644 --- a/src/caterpillar/_common.py +++ b/src/caterpillar/_common.py @@ -61,7 +61,10 @@ def __enter__(self) -> None: self.context[self.name] = self.value def __exit__( - self, exc_type: type, exc_value: Exception, traceback: TracebackType + self, + exc_type: type | None, + exc_value: Exception | None, + traceback: TracebackType | None, ) -> None: self.context[self.name] = self.old_value # We have to apply the right field as instance of the Field class diff --git a/src/caterpillar/fields/_base.py b/src/caterpillar/fields/_base.py index ea85052..3d01f27 100644 --- a/src/caterpillar/fields/_base.py +++ b/src/caterpillar/fields/_base.py @@ -65,6 +65,21 @@ def singleton(cls: type[_T]) -> _T: INVALID_DEFAULT: object = object() DEFAULT_OPTION: object = object() +# Constant representing a default value that caterpillar must not use as a +# concrete value. +IGNORED_DEFAULT: object = object() + + +def has_default(value: object) -> bool: + """ + Check whether *value* is a concrete, usable default value. + + :param value: The value to check, typically a :attr:`Field.default`. + :return: True if *value* can be safely used as a fallback/packing value. + """ + return value is not INVALID_DEFAULT and value is not IGNORED_DEFAULT + + class Field(Generic[_IT, _OT], PackMixin[_IT], UnpackMixin[_OT]): """Represents a field in a data structure. @@ -526,7 +541,7 @@ def __unpack__(self, context: _ContextLike) -> _OT: if not isinstance(exc, StructException): exc = StructException(str(exc), context) value = self.default - if value is INVALID_DEFAULT or isinstance(exc, ValidationError): + if not has_default(value) or isinstance(exc, ValidationError): raise exc return value @@ -561,7 +576,7 @@ def __unpack__(self, context: _ContextLike) -> _OT: exc = StructException(str(exc), context) # Any exception leads to a default value if configured value = self.default - if value is INVALID_DEFAULT or isinstance(exc, ValidationError): + if not has_default(value) or isinstance(exc, ValidationError): raise exc else: # Context functions should be executed with top priority diff --git a/src/caterpillar/fields/common.py b/src/caterpillar/fields/common.py index c328d2a..033198e 100644 --- a/src/caterpillar/fields/common.py +++ b/src/caterpillar/fields/common.py @@ -53,7 +53,7 @@ from caterpillar.options import GLOBAL_FIELD_FLAGS, Flag from caterpillar.shared import ATTR_BYTEORDER, getstruct, typeof -from ._base import INVALID_DEFAULT, Field, singleton +from ._base import Field, INVALID_DEFAULT, singleton, has_default from ._mixin import ByteOrderMixin, FieldStruct # Explicitly report deprecation warnings @@ -924,7 +924,7 @@ def decode(self, parsed: _IT, context: ContextLike) -> _EnumT: default = self.default field = context.get(CTX_FIELD) if default is INVALID_DEFAULT and field: - default = field.default + default = field.default if has_default(default) else INVALID_DEFAULT if default is INVALID_DEFAULT: if (field and field.has_flag(ENUM_STRICT)) or self.strict: diff --git a/src/caterpillar/fields/unnamed.py b/src/caterpillar/fields/unnamed.py new file mode 100644 index 0000000..9108a14 --- /dev/null +++ b/src/caterpillar/fields/unnamed.py @@ -0,0 +1,444 @@ +# Copyright (C) MatrixEditor 2023-2026 +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +import re +from io import BytesIO +from typing import TYPE_CHECKING, Annotated, Any, Generic, get_args, get_origin + +from typing_extensions import TypeVarTuple, Unpack, override + +from caterpillar._common import WithoutContextVar, read_exact +from caterpillar.abc import ( + _AnnotationT, + _ArchLike, + _ContextLambda, + _ContextLike, + _EndianLike, + _StructLike, +) +from caterpillar.byteorder import ByteOrder, DynByteOrder +from caterpillar.context import ( + CTX_ARCH, + CTX_FIELD, + CTX_ORDER, + CTX_PATH, + CTX_SEQ, + CTX_STREAM, + O_CONTEXT_FACTORY, + Context, +) +from caterpillar.exception import ValidationError +from caterpillar.fields import Field +from caterpillar.registry import TypeConverter, annotation_registry, to_struct +from caterpillar.shared import getstruct + +from ._mixin import FieldStruct + +__all__ = [ + "Unnamed", + "sized", +] + +_Tys = TypeVarTuple("_Tys") + +# Matches the "._" convenience attribute access, e.g. "_0", "_12". +_ATTR_RE = re.compile(r"^_([0-9]+)$") + + +def _resolve_member(annotation: _AnnotationT) -> _StructLike[Any, Any]: + """ + Resolve a single ``Unnamed`` member annotation into a ``_StructLike``. + """ + if get_origin(annotation) is Annotated: + _, struct_obj, *_ = get_args(annotation) + annotation = struct_obj + return to_struct(getstruct(annotation, annotation)) + + +class UnnamedAlias(Generic[Unpack[_Tys]]): + """ + Lightweight marker object produced by ``Unnamed[...]``. + + This is *not* a ``_StructLike`` itself - :class:`_UnnamedTypeConverter` + is what turns it into a real, usable struct (:class:`UnnamedStruct`) + once it is encountered as a field annotation. + """ + + __slots__: tuple[str, ...] = ("length", "member_annotations") + + def __init__( + self, + member_annotations: tuple[_AnnotationT, ...], + length: "_ContextLambda[int] | int | None" = None, + ) -> None: + self.member_annotations: tuple[_AnnotationT, ...] = member_annotations + self.length: _ContextLambda[int] | int | None = length + + def sized( + self, length: "_ContextLambda[int] | int" + ) -> "UnnamedAlias[Unpack[_Tys]]": + """Override the default (max-of-members) size for this field.""" + return UnnamedAlias(self.member_annotations, length) + + +class UnnamedStruct( + FieldStruct["Unnamed[Unpack[_Tys]]", "Unnamed[Unpack[_Tys]]"], Generic[Unpack[_Tys]] +): + """The real ``_StructLike`` behind an ``Unnamed[...]`` field.""" + + __slots__: tuple[str, ...] = ("length", "members") + + def __init__( + self, + members: tuple[_StructLike[Any, Any], ...], + length: "_ContextLambda[int] | int | None", + ) -> None: + self.members: tuple[_StructLike[Any, Any], ...] = members + self.length: _ContextLambda[int] | int | None = length + + def __type__(self) -> type: + return Unnamed + + def __size__(self, context: _ContextLike) -> int: + if self.length is not None: + return self.length(context) if callable(self.length) else self.length + + sizes = [m.__size__(context) for m in self.members] + if any(s is Ellipsis for s in sizes): + raise ValidationError( + "Unnamed members must have a fixed size unless an " + + "explicit size is given via .sized(...)", + context, + ) + return max(sizes) + + @override + def unpack_single(self, context: _ContextLike) -> "Unnamed[Unpack[_Tys]]": + root: _ContextLike = context._root + field: Field | None = context.get(CTX_FIELD) + order: _EndianLike | None = ( + field.order + if field is not None and field.has_order() + else context.get(CTX_ORDER, root.get(CTX_ORDER)) + ) + if isinstance(order, DynByteOrder): + order = ByteOrder(order.name, order.getch(context)) + + arch: _ArchLike | None = context.get(CTX_ARCH, root.get(CTX_ARCH)) + path: str = context[CTX_PATH] + size = self.__size__(context) + raw = read_exact(context, size, "Unnamed field") + return Unnamed.from_raw(self.members, raw, order, arch, path) + + @override + def pack_single(self, obj: "Unnamed[Unpack[_Tys]]", context: _ContextLike) -> None: + size = self.__size__(context) + stream = context[CTX_STREAM] + + if obj._raw is not None: + data = obj._raw # pyright: ignore[reportPrivateUsage] + if len(data) != size: + raise ValidationError( + f"Unnamed field expected {size} bytes, got {len(data)}", + context, + ) + stream.write(data) + return + + if obj._active_index is None: # pyright: ignore[reportPrivateUsage] + raise ValidationError( + "Packing a fresh Unnamed requires at least one member " + + "value to be set, e.g. Unnamed(value) or " + + "Unnamed(_2=value)", + context, + ) + + index = obj._active_index # pyright: ignore[reportPrivateUsage] + if not (0 <= index < len(self.members)): + raise ValidationError( + f"member index {index} is out of range for this Unnamed " + + f"field (0-{len(self.members) - 1})", + context, + ) + value = obj._cache[index] # pyright: ignore[reportPrivateUsage] + member = self.members[index] + buf = BytesIO() + with ( + WithoutContextVar(context, CTX_STREAM, buf), + WithoutContextVar(context, CTX_SEQ, False), + ): + member.__pack__(value, context) + data = buf.getvalue() + if len(data) > size: + raise ValidationError( + f"encoded member {index} does not fit in {size} bytes (got {len(data)})", + context, + ) + stream.write(data.ljust(size, b"\x00")) + + +class _SizedMarker: + """ + Internal, dedicated marker produced by :class:`sized`. + """ + + __slots__: tuple[str, ...] = ("length",) + + def __init__(self, length: "_ContextLambda[int] | int") -> None: + self.length: _ContextLambda[int] | int = length + + @override + def __repr__(self) -> str: + return f"sized[{self.length!r}]" + + +class sized: + """ + Inline marker for a fixed (or context-lambda) size, usable as the last + argument of :class:`Unnamed`'s subscript:: + + payload: Unnamed[uint8_t, uint16_t, sized[10]] + + which is a runtime-equivalent shorthand for:: + + payload: Unnamed[uint8_t, uint16_t].sized(10) + + .. warning:: + ``sized[...]`` occupies one of ``Unnamed``'s generic positions, so + static type checkers count it as an extra member type + + At most one ``sized[...]`` marker may appear in ``Unnamed[...]``, and it + must be the last argument. + """ + + def __class_getitem__(cls, length: "_ContextLambda[int] | int") -> _SizedMarker: + return _SizedMarker(length) + + +class Unnamed(tuple[Unpack[_Tys]], Generic[Unpack[_Tys]]): + """ + A fixed-size, lazily-decoded, positional view over a shared byte window. + + Instances are normally produced by unpacking an ``Unnamed[...]``-typed + field. Individual members are decoded (and cached) on first access via + ``value[i]`` or ``value._i``. + + Fresh instances may also be constructed directly for packing purposes, + without a prior unpack: + + >>> Unnamed() # nothing assigned yet + Unnamed() + >>> Unnamed(1, 2, 3) # positional: index 0, 1, 2 + Unnamed(1, 2, 3) + >>> Unnamed(_3=0xBEEF) # keyword: only index 3 is set + Unnamed(?, ?, ?, 48879) + + Since all members share the exact same bytes, only one member can ever + be *packed*: whichever index was assigned last (in call order) is the + one actually encoded. Assigning the same index twice in one call is + an error. + """ + + if TYPE_CHECKING: + + @classmethod + def sized( + cls, length: _ContextLambda[int] | int + ) -> "Unnamed[Unpack[_Tys]]": ... + + _raw: bytes | None + _members: tuple[_StructLike[Any, Any], ...] | None + _order: _EndianLike | None + _arch: _ArchLike | None + _path: str + _cache: dict[int, Any] + _active_index: int | None + + def __new__(cls, *args: Any, **kwargs: Any) -> "Unnamed[Unpack[_Tys]]": + self = tuple.__new__(cls, ()) + self._raw = None + self._members = None + self._order = None + self._arch = None + self._path = "" + self._cache = {} + # Index of the member that was assigned last (in call order) - + # this is the one that gets packed when there is no raw data yet. + self._active_index = None + + for index, value in enumerate(args): + self._cache[index] = value + self._active_index = index + + for key, value in kwargs.items(): + m = _ATTR_RE.match(key) + if not m: + raise TypeError( + f"{cls.__name__}() got an unexpected keyword argument " + + f"{key!r} (expected e.g. _0, _1, ...)" + ) + index = int(m.group(1)) + if index in self._cache: + raise TypeError( + f"{cls.__name__}() got multiple values for index {index}" + ) + self._cache[index] = value + self._active_index = index + + return self + + @classmethod + def from_raw( + cls, + members: tuple[_StructLike[Any, Any], ...], + raw: bytes, + order: "_EndianLike | None", + arch: "_ArchLike | None", + path: str, + ) -> "Unnamed[Unpack[_Tys]]": + self = tuple.__new__(cls, ()) + self._raw = raw + self._members = members + self._order = order + self._arch = arch + self._path = path + self._cache = {} + self._active_index = None + return self + + @override + def __len__(self) -> int: + if self._members is not None: + return len(self._members) + if self._cache: + return max(self._cache) + 1 + return 0 + + def get(self, index: int) -> Any: + assert self._members is not None and self._raw is not None + member = self._members[index] + factory = O_CONTEXT_FACTORY.value or Context + fresh_context = factory( + _io=BytesIO(self._raw), + _is_seq=False, + _order=self._order, + _arch=self._arch, + _path=f"{self._path}[{index}]", + ) + return member.__unpack__(fresh_context) + + @override + def __getitem__(self, index: Any) -> Any: + if isinstance(index, slice): + return tuple(self[i] for i in range(*index.indices(len(self)))) + + n = len(self) + i = index + n if index < 0 else index + if not (0 <= i < n): + raise IndexError(index) + + if i in self._cache: + return self._cache[i] + + if self._raw is None: + raise AttributeError( + f"member {i} was not assigned on this Unnamed and " + + "there is no raw data available to decode it from" + ) + + value = self.get(i) + self._cache[i] = value + return value + + @override + def __iter__(self): + for i in range(len(self)): + yield self[i] + + def __getattr__(self, name: str) -> Any: + m = _ATTR_RE.match(name) + if not m: + raise AttributeError(name) + return self[int(m.group(1))] + + @override + def __repr__(self) -> str: + n = len(self) + parts = (repr(self._cache[i]) if i in self._cache else "?" for i in range(n)) + return f"Unnamed({', '.join(parts)})" + + @override + def __eq__(self, other: object) -> bool: + if not isinstance(other, Unnamed): + return NotImplemented + if self._raw is not None and other._raw is not None: + return self._raw == other._raw + return len(self) == len(other) and all( + self[i] == other[i] for i in range(len(self)) + ) + + @override + def __ne__(self, other: object) -> bool: + result = self.__eq__(other) + if result is NotImplemented: + return result + return not result + + @override + def __hash__(self) -> int: + if self._raw is not None: + return hash(self._raw) + return hash(tuple(self)) + + def __class_getitem__(cls, args: object) -> UnnamedAlias: # pyright: ignore[reportIncompatibleMethodOverride] + if not isinstance(args, tuple): + args = (args,) + if not args: + raise TypeError("Unnamed[...] requires at least one member type") + + length: _ContextLambda[int] | int | None = None + marker_count = sum(1 for a in args if isinstance(a, _SizedMarker)) + if marker_count: + last = args[-1] + if marker_count > 1 or not isinstance(last, _SizedMarker): + raise TypeError( + "Unnamed[...] accepts at most one sized[...] marker, " + + "and it must be the last argument" + ) + length = last.length + args = args[:-1] + if not args: + raise TypeError( + "Unnamed[...] requires at least one member type " + + "(besides the trailing sized[...] marker)" + ) + + return UnnamedAlias(args, length) + + +class _UnnamedTypeConverter(TypeConverter): + @override + def matches(self, annotation: object) -> bool: + return isinstance(annotation, UnnamedAlias) + + @override + def convert( + self, annotation: object, kwargs: dict[str, Any] + ) -> _StructLike[Any, Any]: + assert isinstance(annotation, UnnamedAlias) + members = tuple(_resolve_member(a) for a in annotation.member_annotations) + return UnnamedStruct(members, annotation.length) + + +annotation_registry.insert(0, _UnnamedTypeConverter()) diff --git a/src/caterpillar/model/_base.py b/src/caterpillar/model/_base.py index 985cb3f..c8f3022 100644 --- a/src/caterpillar/model/_base.py +++ b/src/caterpillar/model/_base.py @@ -15,6 +15,8 @@ # pyright: reportPrivateUsage=false, reportAny=false, reportExplicitAny=false from caterpillar.byteorder import Inherit import re +import dataclasses as dc + from collections.abc import Iterable from typing import Annotated, Any, Generic, get_args, get_origin @@ -55,6 +57,7 @@ ) from caterpillar.shared import ATTR_ACTION_PACK, ATTR_ACTION_UNPACK, Action +from caterpillar.fields._base import has_default, IGNORED_DEFAULT class _Member: def __init__( @@ -269,15 +272,10 @@ def _included(self, name: str, default: object, annotation: _AnnotationT) -> boo :param default: The default value of the field. :return: True if the field should be included, else False. """ - if self.has_option(S_DISCARD_UNNAMED): - if re.match(r"^_[0-9]*$", name): - return False - - if self.has_option(S_DISCARD_CONST): - if default != INVALID_DEFAULT: - return False + if self.has_option(S_DISCARD_UNNAMED) and re.match(r"^_[0-9]*$", name): + return False - return True + return not (self.has_option(S_DISCARD_CONST) and has_default(default)) def _set_default(self, name: str, value: object) -> None: pass @@ -286,6 +284,7 @@ def _process_default( self, name: str, annotation: _AnnotationT, had_default: bool = False ) -> Any: default: Any = getattr(self.model, name, INVALID_DEFAULT) + raw_default = default # constant values that are not in the form of fields, structs or types should # be wrapped into constant values. For more information, see _process_field if isinstance(annotation, Field): @@ -305,6 +304,15 @@ def _process_default( # Unions will get none as default value for all fields default = None + if default is raw_default and isinstance(raw_default, dc.Field): + # The class attribute is still an untouched dataclasses.field(...) + return ( + IGNORED_DEFAULT + if raw_default.default is not dc.MISSING + or raw_default.default_factory is not dc.MISSING + else INVALID_DEFAULT + ) + if default != INVALID_DEFAULT: self._set_default(name, default) return default @@ -663,7 +671,7 @@ def pack_one(self, obj: _SeqIT, context: _ContextLike) -> None: ) name = union_field.get_name() - context[ctx_path] = ".".join([base_path, ""]) + context[ctx_path] = f"{base_path}." # REVISIT: are constant values allowed here? + name validation? value = self.get_value(obj, name, union_field) union_field.__pack__(value, context) diff --git a/test/_Py/fields/test_py_unnamed.py b/test/_Py/fields/test_py_unnamed.py new file mode 100644 index 0000000..18a0c47 --- /dev/null +++ b/test/_Py/fields/test_py_unnamed.py @@ -0,0 +1,217 @@ +import pytest + +from caterpillar.fields.unnamed import Unnamed, sized +from caterpillar.py import ( + StructDefMixin, + ValidationError, + f, + pack, + struct, + this, + unpack, +) +from caterpillar.types import uint8_t, uint16_t, uint32_t + + +@struct +class Packet: + tag: uint8_t + payload: Unnamed[uint8_t, uint16_t, uint32_t] + + +@struct +class LengthPrefixed: + n: uint8_t + payload: Unnamed[uint8_t, uint16_t].sized(this.n) + + +@struct +class SizedMarkerPacket: + tag: uint8_t + payload: Unnamed[uint8_t, uint16_t, sized[4]] + + +@struct +class SizedMarkerContextLambda: + n: uint8_t + payload: Unnamed[uint8_t, uint16_t, sized[this.n]] + + +@struct +class Header: + a: uint8_t + b: uint8_t + + +@struct +class Nested: + payload: Unnamed[Header, uint16_t] + + +@struct +class Elem(StructDefMixin): + tag: uint8_t + payload: Unnamed[uint8_t, uint16_t] + + +@struct +class Container: + items: f[list[Elem], Elem[4]] + + +def test_lazy_decode_via_getitem_and_attr(): + p = unpack(Packet, bytes([0xAB, 0x11, 0x22, 0x33, 0x44])) + + assert repr(p.payload) == "Unnamed(?, ?, ?)" + assert p.payload[0] == 0x11 + assert p.payload._1 == 0x2211 + assert p.payload[2] == 0x44332211 + assert len(p.payload) == 3 + + +def test_decoded_members_are_cached(): + p = unpack(Packet, bytes([0xAB, 0x11, 0x22, 0x33, 0x44])) + first = p.payload[0] + second = p.payload[0] + assert first == second == 0x11 + + +def test_negative_index_and_iteration(): + p = unpack(Packet, bytes([0xAB, 0x11, 0x22, 0x33, 0x44])) + assert p.payload[-1] == p.payload[2] + assert list(p.payload) == [p.payload[0], p.payload[1], p.payload[2]] + + +def test_slice_returns_plain_tuple(): + p = unpack(Packet, bytes([0xAB, 0x11, 0x22, 0x33, 0x44])) + assert p.payload[0:2] == (p.payload[0], p.payload[1]) + + +def test_out_of_range_index_raises(): + p = unpack(Packet, bytes([0xAB, 0x11, 0x22, 0x33, 0x44])) + with pytest.raises(IndexError): + p.payload[3] + + +def test_context_lambda_sizing(): + data = bytes([2, 0x01, 0x02]) + p = unpack(LengthPrefixed, data) + assert p.payload[0] == 1 + assert p.payload[1] == 0x0201 + assert pack(p) == data + + +def test_sized_marker_overrides_default_size(): + # sized[4] overrides the default (max-of-members == 2) size to 4 bytes, + # and the marker itself is excluded from the real member count. + data = bytes([0xAB, 0x11, 0x22, 0x33, 0x44]) + p = unpack(SizedMarkerPacket, data) + assert len(p.payload) == 2 + assert p.payload[0] == 0x11 + assert p.payload[1] == 0x2211 + assert pack(p) == data + + +def test_sized_marker_with_context_lambda(): + data = bytes([3, 0x01, 0x02, 0x03]) + p = unpack(SizedMarkerContextLambda, data) + assert p.payload[0] == 1 + assert p.payload[1] == 0x0201 + assert pack(p) == data + + +def test_sized_marker_alone_raises(): + with pytest.raises(TypeError): + Unnamed[sized[10]] + + +def test_sized_marker_duplicate_raises(): + with pytest.raises(TypeError): + Unnamed[uint8_t, sized[4], sized[8]] + + +def test_sized_marker_not_last_raises(): + with pytest.raises(TypeError): + Unnamed[sized[4], uint8_t, uint16_t] + + +def test_nested_struct_member(): + data = bytes([0x10, 0x20]) + p = unpack(Nested, data) + assert p.payload[0] == Header(a=0x10, b=0x20) + assert isinstance(p.payload[0], Header) + assert p.payload[1] == 0x2010 + assert pack(p) == data + + +def test_array_of_unions_independence(): + data = bytes( + [ + 0x01, + 0xAA, + 0xBB, + 0x02, + 0xCC, + 0xDD, + 0x03, + 0xEE, + 0xFF, + 0x04, + 0x11, + 0x22, + ] + ) + c = unpack(Container, data) + + assert c.items[3].payload[0] == 0x11 + assert c.items[0].payload[0] == 0xAA + assert c.items[2].payload[1] == 0xFFEE + assert c.items[1].payload[0] == 0xCC + assert pack(c) == data + + +def test_empty_constr(): + u = Unnamed() + assert len(u) == 0 + assert list(u) == [] + assert repr(u) == "Unnamed()" + + +def test_positional_constr(): + u = Unnamed(1, 2, 3) + assert len(u) == 3 + assert u[0] == 1 + assert u[1] == 2 + assert u._2 == 3 + assert tuple(u) == (1, 2, 3) + + +def test_keyword_constr(): + u = Unnamed(_3=0xBEEF) + assert len(u) == 4 + assert u[3] == 0xBEEF + with pytest.raises(AttributeError): + u[0] + + +def test_duplicate_index_assignment(): + with pytest.raises(TypeError): + Unnamed(1, 2, _0=99) + + +def test_unknown_keyword_raises(): + with pytest.raises(TypeError): + Unnamed(x=1) + + +def test_last_assigned_value_is_packed(): + u = Unnamed(1, 2, 3) + out = pack(Packet(tag=9, payload=u)) + reparsed = unpack(Packet, out) + assert reparsed.payload[2] == 3 + assert len(out) == 5 + + +def test_pack_with_nothing_set_raises(): + with pytest.raises(ValidationError): + pack(Packet(tag=1, payload=Unnamed()))