Vuer

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