jam.state.components.rho module

class jam.state.components.rho.WorkReportState(report: WorkReport, time: U32)

Bases: Codable

Work report state

report: WorkReport

time: U32

__init__(report: WorkReport, time: U32) → None

Initialize the Codable.

Parameters:

• codec – Optional codec to use for encoding/decoding

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

static decode_from(buffer: bytes | bytearray | memoryview, offset: int = 0) → Tuple[T, 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

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

Encode into provided buffer.

encode_size() → int

Calculate number of bytes needed to encode.

class jam.state.components.rho.OptionalWorkReportState(initial: Dict[str, Codable[T]] | Codable[T])

Bases: Choice

Work report state

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

Decode option from buffer.

Parameters:

• optional_type – Type of the optional value

• buffer – Source buffer

• offset – Starting offset

Returns:

Tuple of (decoded value, bytes read)

Raises:

• DecodeError – If buffer is invalid or too short

• ValueError – If types list is empty

class jam.state.components.rho.Rho(initial: Sequence[T] = [], codec: Codec[T] | None = None)

Bases: Array[OptionalWorkReportState]

Work report state array

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