Codec

Type-safe serialization framework implementing the JAM protocol specification.

Core Components

Codec[T]

Generic codec interface for type-safe encoding/decoding:

• encode(value: T) -> bytes

• decode(buffer: bytes) -> T

• encode_into(value: T, buffer: bytearray, offset: int) -> int

• decode_from(buffer: bytes, offset: int) -> Tuple[T, int]

Codable

Base class for encodable types:

• Direct codec usage

• Sequence-based encoding

• Custom encoding logic

• JSON serialization support

Encoding Format

Length Prefixes

• 0x00-0xFC: Direct value (1 byte)

• 0xFF: u16 value (3 bytes)

• 0xFE: u24 value (4 bytes)

• 0xFD: u32 value (5 bytes)

Type Encoding

Primitive Types:

• bool: Single byte (0=false, 1=true)

• trivial int: Fixed width unsigned integers

• general int: Variable length unsigned integers

• str: Length-prefixed UTF-8

• bytes: Length-prefixed raw bytes

• null: Zero bytes

Container Types:

• Array[T,N]: Fixed-length [T; N]

• Vector[T]: Length-prefixed [len][T…]

• Dict[K,V]: Length-prefixed sorted pairs

• Option[T]: Tagged union (0=None, 1=Some(T))

• BitSequence: Packed bits

Error Handling

Exception Types

• CodecError: Base exception class

• BufferError: Buffer operation errors

• EncodeError: Encoding failures

• DecodeError: Decoding failures

Common Error Cases

1. Buffer underflow/overflow

2. Invalid length prefixes

3. Type conversion errors

4. Sequence validation errors

Examples

Basic Usage

from jam.utils.codec.primitives import IntegerCodec, StringCodec
from jam.utils.codec.composite import VectorCodec

# Primitive types
u32_codec = IntegerCodec(4)
encoded_int = u32_codec.encode(42)  # -> [2A 00 00 00]
decoded_int = u32_codec.decode(encoded_int)  # -> 42

string_codec = StringCodec()
encoded_str = string_codec.encode("hello")  # -> [05 68 65 6C 6C 6F]
decoded_str = string_codec.decode(encoded_str)  # -> "hello"

# Container types
vec_codec = VectorCodec(int, u32_codec)
encoded_vec = vec_codec.encode([1, 2])  # -> [02 01 00 00 00 02 00 00 00]
decoded_vec = vec_codec.decode(encoded_vec)  # -> [1, 2]

Custom Types

from dataclasses import dataclass
from jam.utils.codec.decorators import decodable_dataclass

@decodable_dataclass
@dataclass
class Point:
    x: int
    y: int

point = Point(1, 2)
encoded = point.encode()  # -> [01 00 00 00 02 00 00 00]
decoded = Point.decode(encoded)  # -> Point(x=1, y=2)

Error Handling

try:
    u32_codec.decode(bytes([0x01]))  # Too short
except BufferError as e:
    print(f"Expected {e.expected} bytes, got {e.actual}")

try:
    string_codec.decode(bytes([0xFF, 0x01]))  # Invalid UTF-8
except DecodeError as e:
    print(f"Failed to decode string: {e}")

Submodules

• jam.utils.codec.codec
  • Interface
  • Methods
  • Implementation
  • Examples
  • API Reference
• jam.utils.codec.codable
  • Interface
  • Usage Patterns
  • Implementation Details
  • Examples
  • API Reference
• jam.utils.codec.errors
  • Exception Types
  • Common Error Cases
  • Examples
  • API Reference
• jam.utils.codec.primitives package
  • Submodules
  • Module Contents
• jam.utils.codec.composite package
  • Submodules
  • Module Contents
• jam.utils.codec.decorators package
  • Submodules
  • Module Contents

Module Contents

Type-safe serialization framework 📦

Core: * Codec[T]: Generic codec interface * Codable: Base class for encodable types * primitives: Basic type codecs * composite: Container type codecs * decorators: Automatic codec generation * errors: Error handling

class jam.utils.codec.Codec

Bases: ABC, Generic[T]

Abstract base class defining the interface for encoding and decoding data.

abstract encode_size(value: T) → int

Calculate the number of bytes needed to encode the value.

abstract encode_into(value: T, buffer: bytearray, offset: int = 0) → int

Encode the value into the provided buffer at the specified offset.

encode(value: T) → bytes

Encode the value into a new bytes object.

abstract decode_from(buffer: bytes | bytearray | memoryview, offset: int = 0) → Tuple[T, int]

Decode a value from the provided buffer starting at the specified offset.

class jam.utils.codec.Codable(codec: Codec[Any] | None = None)

Bases: Generic[T]

Base class for all codable types.

Can be used in three ways: 1. With a codec: Initialize with codec=some_codec 2. With sequence: Initialize with enc_sequence=lambda: [field1, field2, …] 3. With JSON: Use to_json() and from_json() for JSON serialization

value: Any = None

__init__(codec: Codec[Any] | None = None)

Initialize the Codable.

Parameters:

• codec – Optional codec to use for encoding/decoding

• enc_sequence – Optional function that returns sequence of fields to encode

codec: Codec[Any] | None = None

encode_size() → int

Calculate number of bytes needed to encode.

encode() → bytes

Encode into bytes.

encode_into(buffer: bytearray, offset: int = 0) → int

Encode into provided buffer.

static decode_from(buffer: bytes | bytearray | memoryview, offset: int = 0) → Tuple[Any, int]

Decode from buffer. Must be implemented by subclasses or added via decorator.

Parameters:

• buffer – Buffer to decode from

• offset – Starting position in buffer

Returns:

• The decoded value

• Number of bytes read

Return type:

Tuple containing

exception jam.utils.codec.CodecError

Bases: Exception

Base codec exception.

exception jam.utils.codec.BufferError(expected: int, actual: int, message: str = 'Buffer error')

Bases: CodecError

Buffer operation error with expected vs actual size.

expected: int

actual: int

message: str = 'Buffer error'

__init__(expected: int, actual: int, message: str = 'Buffer error') → None

exception jam.utils.codec.EncodeError(expected: int, actual: int, message: str = 'Buffer error')

Bases: BufferError

Exception raised when encoding fails.

exception jam.utils.codec.DecodeError(expected: int, actual: int, message: str = 'Buffer error')

Bases: BufferError

Exception raised when decoding fails.