vuer.rtc.types

Core types for the vuer.rtc CRDT-based data store.

This module provides Python equivalents to the TypeScript @vuer-ai/vuer-rtc types, enabling a consistent API across Python and JavaScript clients.

Key structures:

VectorClock: Causal ordering via vector clocks
Operation: Individual CRDT operations (node.insert, vector3.set, etc.)
CRDTMessage: Message envelope containing batched operations
SceneNode: Individual node in the scene graph
SceneGraph: The computed state (map of nodes)
JournalEntry: Committed message with ack/deleted status
EditBuffer: Uncommitted operations awaiting commit
Snapshot: Checkpoint for fast replay
ClientState: Full client state container

create_vector_clock

python

def create_vector_clock() -> VectorClock

Source

Create an empty vector clock.

increment_clock

python

def increment_clock(clock: VectorClock, session_id: str) -> VectorClock

Source

Return a new clock with the session's counter incremented.

merge_clocks

python

def merge_clocks(clock1: VectorClock, clock2: VectorClock) -> VectorClock

Source

Merge two vector clocks, taking the max of each session's counter.

compare_clocks

python

def compare_clocks(clock1: VectorClock, clock2: VectorClock) -> Literal['before', 'after', 'concurrent', 'equal']

Source

Compare two vector clocks for causal ordering. Returns:

"before" if clock1 happened before clock2
"after" if clock1 happened after clock2
"concurrent" if neither happened before the other
"equal" if they are identical

OType

python

class OType(str, Enum)

Source

All supported operation types (dtype.operation format).

python

NUMBER_SET = 'number.set'

python

NUMBER_ADD = 'number.add'

python

NUMBER_MULTIPLY = 'number.multiply'

python

NUMBER_MIN = 'number.min'

python

NUMBER_MAX = 'number.max'

python

STRING_SET = 'string.set'

python

STRING_CONCAT = 'string.concat'

python

TEXT_INIT = 'text.init'

python

TEXT_INSERT = 'text.insert'

python

TEXT_DELETE = 'text.delete'

python

BOOLEAN_SET = 'boolean.set'

python

BOOLEAN_OR = 'boolean.or'

python

BOOLEAN_AND = 'boolean.and'

python

VECTOR3_SET = 'vector3.set'

python

VECTOR3_ADD = 'vector3.add'

python

VECTOR3_MULTIPLY = 'vector3.multiply'

python

VECTOR3_APPLY_EULER = 'vector3.applyEuler'

python

VECTOR3_APPLY_QUATERNION = 'vector3.applyQuaternion'

python

EULER_SET = 'euler.set'

python

EULER_ADD = 'euler.add'

python

QUATERNION_SET = 'quaternion.set'

python

QUATERNION_MULTIPLY = 'quaternion.multiply'

python

COLOR_SET = 'color.set'

python

COLOR_BLEND = 'color.blend'

python

ARRAY_SET = 'array.set'

python

ARRAY_PUSH = 'array.push'

python

ARRAY_UNION = 'array.union'

python

ARRAY_REMOVE = 'array.remove'

python

OBJECT_SET = 'object.set'

python

OBJECT_MERGE = 'object.merge'

python

NODE_INSERT = 'node.insert'

python

NODE_REMOVE = 'node.remove'

python

META_UNDO = 'meta.undo'

python

META_REDO = 'meta.redo'

Operation

python

class Operation

Source

Base operation structure for all CRDT operations.

Attributes: key: Node key (e.g., 'cube-1', 'scene') otype: Operation type in dtype.operation format (e.g., 'vector3.add') path: Property path using dot notation (e.g., 'transform.position') value: Operation-specific value (type depends on otype)

python

key: str

python

otype: str

python

path: str

python

value: Any = None

python

tag: Optional[str] = None

python

parent_key: Optional[str] = None

python

target_msg_id: Optional[str] = None

Operation.to_dict

python

def to_dict(self) -> Dict[str, Any]

Source

Serialize operation to dictionary.

Operation.from_dict

python

def from_dict(cls, data: Dict[str, Any]) -> 'Operation'

Source

Deserialize operation from dictionary.

CRDTMessage

python

class CRDTMessage

Source

Message envelope for batched CRDT operations.

Attributes: id: Unique message identifier (format: sessionId:sequence) session_id: Session that created this message clock: Vector clock for causal ordering lamport_time: Lamport timestamp for total ordering timestamp: Wall-clock time in milliseconds since epoch ops: Array of operations in this batch

python

id: str

python

session_id: str

python

clock: VectorClock

python

lamport_time: int

python

timestamp: float

python

ops: List[Operation]

CRDTMessage.to_dict

python

def to_dict(self) -> Dict[str, Any]

Source

Serialize message to dictionary.

CRDTMessage.from_dict

python

def from_dict(cls, data: Dict[str, Any]) -> 'CRDTMessage'

Source

Deserialize message from dictionary.

generate_message_id

python

def generate_message_id(session_id: str, sequence: int) -> str

Source

Generate a message ID in the format sessionId:sequence.

generate_uuid

python

def generate_uuid() -> str

Source

Generate a UUID string.

SceneNode

python

class SceneNode

Source

Individual node in the scene graph.

Attributes: key: Unique key (human-friendly identifier) tag: Node type (Scene, Mesh, Group, etc.) name: Display name children: List of child node keys clock: Vector clock when created lamport_time: Lamport timestamp created_at: Creation timestamp (ms since epoch) updated_at: Last update timestamp (ms since epoch) deleted_at: Soft delete marker (tombstone), None if not deleted properties: Dynamic properties stored by path

python

key: str

python

tag: str

python

name: str = ''

python

children: List[str] = field(default_factory=list)

python

clock: VectorClock = field(default_factory=dict)

python

lamport_time: int = 0

python

created_at: float = field(default_factory=lambda: time.time() * 1000)

python

updated_at: float = field(default_factory=lambda: time.time() * 1000)

python

deleted_at: Optional[float] = None

python

properties: Dict[str, Any] = field(default_factory=dict)

SceneNode.get_property

python

def get_property(self, path: str, default: Any=None) -> Any

Source

Get a property value by dot-notation path.

SceneNode.set_property

python

def set_property(self, path: str, value: Any) -> None

Source

Set a property value by dot-notation path.

SceneNode.is_deleted

python

def is_deleted(self) -> bool

Source

Check if this node has been soft-deleted.

SceneNode.to_dict

python

def to_dict(self) -> Dict[str, Any]

Source

Serialize node to dictionary.

SceneNode.from_dict

python

def from_dict(cls, data: Dict[str, Any]) -> 'SceneNode'

Source

Deserialize node from dictionary.

SceneNode.copy

python

def copy(self) -> 'SceneNode'

Source

Create a deep copy of this node.

SceneGraph

python

class SceneGraph

Source

The computed state of the scene as a flattened map of nodes.

Attributes: nodes: Dictionary mapping node keys to SceneNode instances root_key: Key of the root node (typically "scene")

python

nodes: Dict[str, SceneNode] = field(default_factory=dict)

python

root_key: str = 'scene'

SceneGraph.get_node

python

def get_node(self, key: str) -> Optional[SceneNode]

Source

Get a node by key.

SceneGraph.set_node

python

def set_node(self, node: SceneNode) -> None

Source

Add or update a node in the graph.

SceneGraph.remove_node

python

def remove_node(self, key: str) -> Optional[SceneNode]

Source

Remove a node from the graph. Returns the removed node or None.

SceneGraph.has_node

python

def has_node(self, key: str) -> bool

Source

Check if a node exists.

SceneGraph.get_children

python

def get_children(self, key: str) -> List[SceneNode]

Source

Get all child nodes of a given node.

SceneGraph.to_dict

python

def to_dict(self) -> Dict[str, Any]

Source

Serialize graph to dictionary.

SceneGraph.from_dict

python

def from_dict(cls, data: Dict[str, Any]) -> 'SceneGraph'

Source

Deserialize graph from dictionary.

SceneGraph.copy

python

def copy(self) -> 'SceneGraph'

Source

Create a deep copy of this graph.

create_empty_graph

python

def create_empty_graph(root_key: str='scene') -> SceneGraph

Source

Create an empty scene graph with a root node.

Args: root_key: Key for the root node (default: "scene")

Returns: A new SceneGraph with an empty root Scene node

JournalEntry

python

class JournalEntry

Source

A committed message with acknowledgment and deletion status.

Attributes: msg: The CRDT message ack: Whether the server has acknowledged this message deleted_at: Timestamp when this entry was undone (None if not undone)

python

msg: CRDTMessage

python

ack: bool = False

python

deleted_at: Optional[float] = None

JournalEntry.is_deleted

python

def is_deleted(self) -> bool

Source

Check if this entry has been undone.

JournalEntry.to_dict

python

def to_dict(self) -> Dict[str, Any]

Source

Serialize entry to dictionary.

JournalEntry.from_dict

python

def from_dict(cls, data: Dict[str, Any]) -> 'JournalEntry'

Source

Deserialize entry from dictionary.

EditBuffer

python

class EditBuffer

Source

Buffer for uncommitted operations awaiting commit.

Attributes: ops: List of pending operations start_graph: Graph state when edits started (for cancel/revert)

python

ops: List[Operation] = field(default_factory=list)

python

start_graph: Optional[SceneGraph] = None

EditBuffer.is_empty

python

def is_empty(self) -> bool

Source

Check if there are no pending edits.

EditBuffer.add

python

def add(self, op: Operation) -> None

Source

Add an operation to the buffer.

EditBuffer.clear

python

def clear(self) -> List[Operation]

Source

Clear the buffer and return the operations.

EditBuffer.to_dict

python

def to_dict(self) -> Dict[str, Any]

Source

Serialize buffer to dictionary.

EditBuffer.from_dict

python

def from_dict(cls, data: Dict[str, Any]) -> 'EditBuffer'

Source

Deserialize buffer from dictionary.

Snapshot

python

class Snapshot

Source

Checkpoint for fast replay.

Attributes: graph: Scene graph state at this checkpoint vector_clock: Vector clock value at checkpoint lamport_time: Max lamport time baked into snapshot journal_index: Number of journal entries baked into snapshot

python

graph: SceneGraph

python

vector_clock: VectorClock

python

lamport_time: int

python

journal_index: int

Snapshot.to_dict

python

def to_dict(self) -> Dict[str, Any]

Source

Serialize snapshot to dictionary.

Snapshot.from_dict

python

def from_dict(cls, data: Dict[str, Any]) -> 'Snapshot'

Source

Deserialize snapshot from dictionary.

create_initial_snapshot

python

def create_initial_snapshot(graph: Optional[SceneGraph]=None) -> Snapshot

Source

Create an initial snapshot with an empty or provided graph.

ClientState

python

class ClientState

Source

Full client state container for the CRDT data store.

Attributes: session_id: Unique identifier for this client session graph: Current computed state journal: List of committed messages with ack status edits: Uncommitted operations awaiting commit snapshot: Checkpoint for fast replay lamport_time: Current Lamport timestamp vector_clock: Current vector clock

python

session_id: str

python

graph: SceneGraph

python

journal: List[JournalEntry]

python

edits: EditBuffer

python

snapshot: Snapshot

python

lamport_time: int

python

vector_clock: VectorClock

ClientState.to_dict

python

def to_dict(self) -> Dict[str, Any]

Source

Serialize state to dictionary.

ClientState.from_dict

python

def from_dict(cls, data: Dict[str, Any]) -> 'ClientState'

Source

Deserialize state from dictionary.

create_initial_state

python

def create_initial_state(session_id: str, snapshot: Optional[Snapshot]=None) -> ClientState

Source

Create an initial client state.

Args: session_id: Unique identifier for this session snapshot: Optional snapshot to restore from

Returns: A new ClientState initialized with the given session ID