jam.utils.codec.composite package

Submodules

Module Contents

Composite type codecs for JAM protocol.

This module provides codecs for composite types including: - Arrays (fixed length) - Options (nullable values) - Vectors (dynamic length) - Dictionaries (key-value pairs) - Bit sequences (sequences of bits) - Dataclasses (structured data)

class jam.utils.codec.composite.ArrayCodec(length: int)[source]

Bases: Codec[Sequence[Codable]]

Codec for fixed-length arrays/sequences.

Arrays are encoded by concatenating their encoded elements in order. The length is fixed and known at encoding/decoding time.

MAX_SIZE = 1000

__init__(length: int)[source]

length: int

encode_size(value: Sequence[Codable]) → int[source]: Calculate the number of bytes needed to encode the value.

encode_into(value: Sequence[Codable], buffer: bytearray, offset: int = 0) → int[source]: Encode the value into the provided buffer at the specified offset.

static decode_from(length: int, codable_class: type[Codable], buffer: bytes | bytearray | memoryview, offset: int = 0) → Tuple[List, int][source]: Decode a value from the provided buffer starting at the specified offset.

class jam.utils.codec.composite.ChoiceCodec(choices: ~typing.Dict[str, ~typing.Type[~jam.utils.codec.codable.Codable[~jam.utils.codec.composite.choices.T]]], _ChoiceCodec__tag_codec: ~jam.utils.codec.codec.Codec[int] = <jam.utils.codec.primitives.integers.GeneralCodec object>)[source]

Bases: Codec[T], Generic[T]

Codec for choice/union values.

Choice values are encoded with a tag byte indicating the selected type, followed by the encoded value of that type.

The tag is encoded as a general integer, followed by the encoded value of the selected type. The tag value corresponds to the index of the type in the choices list.

Parameters:

choices – A list of types that are allowed for this choice. Their index will be used as the tag.
tag_codec – A codec for the tag. Defaults to GeneralCodec() [Best for most cases]. Alternatively we can use FixedInt(U8) for a fixed size tag if choices > 128.

Raises:

ValueError – If choices list is empty

__init__(choices: ~typing.Dict[str, ~typing.Type[~jam.utils.codec.codable.Codable[~jam.utils.codec.composite.choices.T]]], _ChoiceCodec__tag_codec: ~jam.utils.codec.codec.Codec[int] = <jam.utils.codec.primitives.integers.GeneralCodec object>)[source]

Initialize ChoiceCodec.

Parameters:: choices – A list of types that are allowed for this choice. Their index will be used as the tag.
Raises:: ValueError – If choices list is empty

encode_size(_value: Dict[str, Codable[T]]) → int[source]

Calculate encoded size for value.

Parameters:: value – Value to encode
Returns:: Number of bytes needed for encoding
Raises:: EncodeError – If value type is not in choices list

encode_into(_value: Dict[str, Codable[T]], buffer: bytearray, offset: int = 0) → int[source]

Encode value into buffer.

Parameters:

value – Value to encode
buffer – Target buffer
offset – Starting offset

Returns:

Number of bytes written

Raises:

EncodeError – If value type is not in choices list or buffer is too small

static decode_from(choices: Dict[str, Type[Codable[T]]], buffer: bytes | bytearray | memoryview, offset: int = 0) → Tuple[Dict[str, Codable[T]], int][source]

Decode choice value from buffer.

Parameters:

choices – List of possible types
buffer – Source buffer
offset – Starting offset

Returns:

Tuple of (decoded value, bytes read)

Raises:

DecodeError – If buffer is invalid/too short or tag is invalid
ValueError – If choices list is empty

class jam.utils.codec.composite.VectorCodec[source]

Bases: Codec[Sequence[Codable[T]]], Generic[T]

Codec for dynamic-length sequences (vectors).

Vectors are encoded with a variable-length prefix indicating size, followed by the concatenated encoded elements.

encode_size(value: Sequence[Codable[T]]) → int[source]

Calculate number of bytes needed to encode vector.

Parameters:: value – Sequence to encode
Returns:: Number of bytes needed
Raises:: EncodeError – If sequence is invalid type or too long

encode_into(value: Sequence[Codable[T]], buffer: bytearray, offset: int = 0) → int[source]

Encode vector into buffer.

Parameters:

value – Sequence to encode
buffer – Target buffer
offset – Starting position in buffer

Returns:

Number of bytes written

Raises:

EncodeError – If sequence invalid or buffer too small

static decode_from(codable_class: Type[Codable[T]], buffer: bytes | bytearray | memoryview, offset: int = 0, max_length: int = 9223372036854775807) → tuple[List[T], int][source]

Decode vector from buffer.

Parameters:

buffer – Source buffer
offset – Starting position in buffer

Returns:

Tuple of (decoded list, bytes read)

Raises:

DecodeError – If buffer too small or invalid encoding

class jam.utils.codec.composite.DictionaryCodec[source]

Bases: Codec[Mapping[K, V]], Generic[K, V]

Codec for key-value mappings.

Dictionaries are encoded as length-prefixed sequences of key-value pairs, sorted by encoded key bytes for deterministic encoding.

_encode_pair(key: Codable[K], value: Codable[V], buffer: bytearray, offset: int) → Tuple[bytes, int][source]

Encode a single key-value pair into buffer.

Parameters:

key – Key to encode
value – Value to encode
buffer – Target buffer
offset – Starting position in buffer

Returns:

Tuple of (key_bytes for sorting, bytes written)

Raises:

EncodeError – If key/value invalid or buffer too small

encode_size(value: Mapping[Codable[K], Codable[V]]) → int[source]

Calculate number of bytes needed to encode dictionary.

Parameters:: value – Dictionary to encode
Returns:: Number of bytes needed
Raises:: EncodeError – If dictionary contains invalid types

encode_into(value: Mapping[Codable[K], Codable[V]], buffer: bytearray, offset: int = 0) → int[source]

Encode dictionary into buffer.

Parameters:

value – Dictionary to encode
buffer – Target buffer
offset – Starting position in buffer

Returns:

Number of bytes written

Raises:

EncodeError – If dictionary invalid or buffer too small

static decode_from(key_codable_class: Type[Codable[K]], value_codable_class: Type[Codable[V]], buffer: bytes | bytearray | memoryview, offset: int = 0) → Tuple[Dict[K, V], int][source]

Decode dictionary from buffer.

Parameters:

buffer – Source buffer
offset – Starting position in buffer

Returns:

Tuple of (decoded dict, bytes read)

Raises:

DecodeError – If buffer too small or invalid encoding

class jam.utils.codec.composite.BitSequenceCodec(bit_length: int | None = None, bit_order: Literal['msb', 'lsb'] = 'msb')[source]

Bases: Codec[Sequence[bool]]

Codec for encoding and decoding sequences of bits.

Bits are packed into octets (bytes) from least significant to most significant. IMP: It adds bit length encoded as a single byte at the beginning of the sequence ONLY if the bit length is not provided.

If dynamic length sequence:

Initialise with None
No need to pass bit length to decode_from

If fixed length sequence:

Initialise with bit length
Pass bit length to decode_from

__init__(bit_length: int | None = None, bit_order: Literal['msb', 'lsb'] = 'msb')[source]

bit_length: int | None = None

bit_order: Literal['msb', 'lsb'] = 'msb'

encode_size(value: Sequence[bool]) → int[source]: Calculate the number of bytes needed to encode the value.

encode_into(value: Sequence[bool], buffer: bytearray, offset: int = 0) → int[source]: Encode the value into the provided buffer at the specified offset.

static decode_from(buffer: bytes | bytearray | memoryview, offset: int = 0, bit_length: int | None = None, bit_order: Literal['msb', 'lsb'] = 'msb') → Tuple[Sequence[bool], int][source]

Decode bit sequence from buffer.

Parameters:

buffer – Source buffer
offset – Starting offset
bit_length – Expected number of bits (required)

Returns:

Tuple of (decoded bit list, bytes read)

Raises:

DecodeError – If buffer too small or bit_length not specified