libmuscle.snapshot module

class libmuscle.snapshot.MsgPackSnapshot(triggers: List[str], wallclock_time: float, port_message_counts: Dict[str, List[int]], is_final_snapshot: bool, message: Message | None, settings_overlay: Settings)

Bases: Snapshot

Snapshot stored in messagepack format

SNAPSHOT_VERSION_BYTE = b'1'

static bytes_to_message(data: bytes) → Message | None

Use MPPMessage deserializer for serializing the message object

classmethod from_bytes(data: bytes) → Snapshot

Create a snapshot object from binary data.

Parameters:

data – binary data representing the snapshot. Note that this must exclude the versioning byte.

static message_to_bytes(message: Message | None) → bytes

Use MPPMessage serializer for serializing the message object

to_bytes() → bytes

Convert the snapshot object to binary data.

Returns:

Binary data representing the snapshot. Note that this must

exclude the versioning byte.

class libmuscle.snapshot.Snapshot(triggers: List[str], wallclock_time: float, port_message_counts: Dict[str, List[int]], is_final_snapshot: bool, message: Message | None, settings_overlay: Settings)

Bases: ABC

Snapshot data structure.

This is an abstract base class, implementations are provided by subclasses.

SNAPSHOT_VERSION_BYTE = b'\x00'

abstract classmethod from_bytes(data: bytes) → Snapshot

Create a snapshot object from binary data.

Parameters:

data – binary data representing the snapshot. Note that this must exclude the versioning byte.

abstract to_bytes() → bytes

Convert the snapshot object to binary data.

Returns:

Binary data representing the snapshot. Note that this must

exclude the versioning byte.

class libmuscle.snapshot.SnapshotMetadata(triggers: List[str], wallclock_time: float, timestamp: float, next_timestamp: float | None, port_message_counts: Dict[str, List[int]], is_final_snapshot: bool, snapshot_filename: str)

Bases: object

Metadata of a snapshot for sending to the muscle_manager.

static from_snapshot(snapshot: Snapshot, snapshot_filename: str) → SnapshotMetadata

Create snapshot metadata from the given snapshot and filename

is_final_snapshot: bool

next_timestamp: float | None

port_message_counts: Dict[str, List[int]]

snapshot_filename: str

timestamp: float

triggers: List[str]

wallclock_time: float