otio_sync_core

otio_sync_core is the core library for coordinating OTIO timeline synchronisation across a network session. It provides the SyncManager that drives synchronisation, UDP and RabbitMQ network backends, a transparent proxy for intercepting attribute writes, typed protocol messages, and the colour-pipeline and annotation codecs.

OTIO Sync core library.

Provides SyncManager for coordinating OTIO timeline synchronisation across a network session, along with UDP and RabbitMQ network backends and a transparent proxy for intercepting attribute writes.

manager

The SyncManager is the central coordinator: it owns the OTIO timeline, applies incoming sync events, and emits outgoing ones over the configured network backend.

Core synchronisation manager for the OTIO Sync protocol.

SyncManager maintains a GUID-indexed map of every OTIO object in the shared session and coordinates mutations across a pluggable network layer. It implements the master-election handshake, delta buffering during join, and all broadcast helpers defined in the OTIO Sync Protocol v1 proposal.

otio_sync_core.manager.STATE_DISCOVERING = 'DISCOVERING'

Broadcasting WHO_IS_MASTER; waiting for a response.

otio_sync_core.manager.STATE_JOINING = 'JOINING'

Master found; waiting for a full state snapshot.

otio_sync_core.manager.STATE_NONE = 'NONE'

Session has not yet started.

otio_sync_core.manager.STATE_SYNCED = 'SYNCED'

Snapshot received and applied; fully participating in the session.

class otio_sync_core.manager.SyncManager(session_id: str = 'default_session', self_guid: str | None = None, network: SyncNetworkProtocol | None = None)

Bases: object

Coordinates OTIO object synchronisation across a network session.

The manager maintains two complementary data structures:

• _object_map — a flat {guid: otio_object} index for O(1) lookup by GUID.

• _timelines — a {guid: Timeline} map of every registered top-level timeline.

All mutations (inserts, removals, property changes) are applied locally and broadcast to peers via the injected network backend. Incoming messages are applied through apply_patch(), which also fires registered observer callbacks so that the host application (e.g. the RV plugin) can react to remote changes.

Session lifecycle

1. Call start_session() — status transitions to STATE_DISCOVERING.

2. The caller polls receive_and_apply_all() until a master_found action is returned, then calls request_state().

3. Status transitions to STATE_JOINING; incoming non-session messages are buffered in _delta_buffer.

4. When a state_snapshot_received action is returned, the caller invokes apply_snapshot() which applies the full state and replays buffered deltas before transitioning to STATE_SYNCED.

If no master responds within the discovery timeout (implemented in the caller), the caller elects itself master and calls broadcast_master_response().

Parameters:

• session_id – Logical session identifier; scopes all network messages.

• self_guid – Stable GUID for this peer; auto-generated when not provided.

• network – Network backend satisfying SyncNetworkProtocol. May be set or replaced after construction.

property active_clip_guid: str | None

Sequence clip GUID if the active timeline is a single-clip timeline, else None.

Return type:

str or None

annotation_clip_guid_at(clip_guid: str, frame: int) → str | None

Return the sync GUID of the annotation clip at (clip_guid, frame).

Convenience wrapper around annotation_track_guid_for_clip() and _find_annotation_clip_at() that returns the clip’s own GUID rather than the object itself.

Parameters:

• clip_guid – GUID of the media clip being annotated.

• frame – 0-indexed clip-local frame number.

Returns:

Annotation clip GUID, or None if not found.

Return type:

str or None

annotation_track_guid_for_clip(clip_guid: str, preferred_timeline_guid: str | None = None) → str | None

Return the GUID of the Annotations track in the same timeline as clip_guid.

Searches every non-annotation track for clip_guid, then returns the first track whose name contains "annotation" (case-insensitive) from that same timeline.

When preferred_timeline_guid is provided (e.g. the current active_timeline_guid), that timeline is searched first. This ensures that annotations are written to the clip timeline’s annotation track while in clip mode, rather than the sequence timeline’s track.

Parameters:

• clip_guid – Sync GUID of the media clip.

• preferred_timeline_guid – GUID of the timeline to search first; falls back to all timelines if not found there.

Returns:

Annotation track GUID, or None if not found.

Return type:

str or None

apply_patch(payload: dict[str, Any]) → tuple[str, Any] | None

Apply a single incoming message from the network.

Dispatches on command_schema and event fields. Returns an (action, data) tuple when the caller needs to act (e.g. to update RV state), or None when the message was fully handled internally.

Messages from self_guid are silently discarded. Messages arriving during STATE_JOINING are buffered (except session messages) and replayed by apply_snapshot().

Parameters:

payload – Parsed message envelope received from the network.

Returns:

(action_name, action_data) or None.

apply_snapshot(snapshot_data: dict[str, Any]) → list[tuple[str, Any]]

Replace local state with a full snapshot and replay buffered deltas.

Clears _object_map and _timelines, deserialises the timelines from snapshot_data, then replays any buffered messages whose sync_timestamp is newer than the snapshot. Transitions status to STATE_SYNCED.

Parameters:

snapshot_data – payload dict from a STATE_SNAPSHOT message.

Returns:

List of (action, data) tuples produced by replaying buffered deltas; to be handled by the caller in the same way as the return value of receive_and_apply_all().

broadcast_add_annotation(annotation_track_guid: str, clip_guid: str, clip_local_time: RationalTime, events: list[dict[str, Any]]) → str | None

Build an annotation clip and insert it via the standard patch path.

Annotations are expressed as insert_child patches so that all peers apply them through the same code path as any other timeline mutation.

The annotation track mirrors the structure produced by ORIAnnotations.ReviewItem._export_otio_media(): each annotated frame is a 1-frame Clip and the gaps between annotated frames are Gap objects whose duration is frame − track_end frames. A second stroke on an already-annotated frame merges its commands into the existing clip rather than inserting a duplicate.

Parameters:

• annotation_track_guid – GUID of the target Annotations track.

• clip_guid – GUID of the media clip being annotated.

• clip_local_time – 0-indexed time within the clip’s source range.

• events – Serialised OTIO SyncEvent dicts (PaintStart.1, PaintPoints.1) as produced by otio.adapters.write_to_string.

Returns:

The sync GUID of the annotation clip that was created or merged into, or None if the operation could not be completed.

Return type:

str or None

broadcast_add_timeline(tl_guid: str) → None

Broadcast a timeline to all peers so they can register it.

Works for both sequence timelines (new playlist / new sequence) and single-clip annotation timelines. Call once immediately after register_timeline() to propagate a locally-created timeline to all connected peers. Peers that already hold the same GUID silently ignore the message.

Parameters:

tl_guid – GUID of the timeline to broadcast.

broadcast_clip_timeline(tl_guid: str) → None

Broadcast a clip timeline to all peers so they can register its annotation track.

Should be called once per clip timeline, immediately after get_or_create_clip_timeline() returns a new GUID. Peers that already have the timeline (same deterministic GUID) will skip the ADD_TIMELINE message.

Delegates to broadcast_add_timeline().

Parameters:

tl_guid – GUID of the clip timeline to broadcast.

broadcast_display_state(state_dict: dict[str, Any]) → None

Broadcast the current display state to all peers and persist it.

Expected keys in state_dict:

• pan — [x, y] normalised pan offset.

• zoom — zoom multiplier (1.0 = no zoom).

• exposure — exposure adjustment in stops (0.0 = no change).

• channel — active channel string: "RGBA", "R", "G", "B", or "A".

The state is also written into the active timeline’s metadata["display_settings"] so it survives a full session teardown if the OTIO file is saved to disk.

Parameters:

state_dict – Display state fields as listed above.

broadcast_master_discovery() → None

Broadcast a WHO_IS_MASTER session message.

broadcast_master_response() → None

Broadcast an I_AM_MASTER session message.

Called after self-election (discovery timeout) or when an existing master receives a WHO_IS_MASTER it should answer.

broadcast_move_child(parent_uuid: str, child_uuid: str, to_index: int) → None

Move child_uuid to to_index within its parent and broadcast the change.

Applies the reorder locally before broadcasting so the local OTIO model stays consistent regardless of network round-trip time.

Parameters:

• parent_uuid – GUID of the parent container.

• child_uuid – GUID of the child to move.

• to_index – Target position in the parent’s child list.

broadcast_partial_annotation(clip_guid: str, frame: float, fps: float, events: list) → None

Broadcast a mid-stroke partial annotation to peers (visual only, no timeline persistence).

Called periodically while the user is drawing a stroke, before pen-up. Peers render the stroke visually but do not write it to the OTIO timeline — that happens on pen-up via broadcast_add_annotation().

Parameters:

• clip_guid – Sync GUID of the media clip being annotated.

• frame – 0-indexed clip-local frame number.

• fps – Frame rate used to interpret frame.

• events – Serialised SyncEvent dicts (PaintStart.1, PaintPoints.1).

broadcast_playback_state(state_dict: dict[str, Any], timeline_guid: str | None = None) → None

Broadcast the current playback state to all peers.

Parameters:

• state_dict – Playback state fields (playing, current_time, looping, etc.) as defined by the protocol.

• timeline_guid – GUID of the timeline being played; falls back to active_timeline_guid.

broadcast_remove_child(parent_uuid: str, child_uuid: str) → None

Remove child_uuid from its parent and broadcast the change.

The child is removed from both the parent container and _object_map.

Parameters:

• parent_uuid – GUID of the parent container.

• child_uuid – GUID of the child to remove.

broadcast_replace_annotation_commands(annotation_clip_guid: str, events: list) → None

Replace all annotation_commands on an existing clip and broadcast to peers.

Used when the user edits text in an annotation in place — the command count stays the same but the text content changes. Sends a REPLACE_ANNOTATION_COMMANDS message so peers replace the full command list rather than appending a delta.

Parameters:

• annotation_clip_guid – Sync GUID of the annotation clip to update.

• events – Full replacement list of SyncEvent objects (strokes + captions) representing the current annotation state.

broadcast_selection(clip_guid: str, view_mode: str = 'source') → None

Broadcast the selected clip GUID to all peers.

Parameters:

• clip_guid – OTIO sync GUID of the selected clip. Receivers map this back to their local representation (RV source group, xStudio playlist position, etc.) before applying.

• view_mode (str) – View mode string (“source” or “sequence”).

broadcast_timeline_rename(tl_guid: str, new_name: str) → None

Rename a timeline locally and broadcast the change to all peers.

Updates the timeline’s name attribute in _timelines immediately, then sends a RENAME_TIMELINE message so all connected peers apply the same rename.

Parameters:

• tl_guid – GUID of the timeline to rename.

• new_name – New display name for the timeline.

close() → None

Stop the network backend and release all resources.

count_annotation_commands(clip_guid: str, frame: int) → tuple[int, int]

Return (n_strokes, n_captions) already committed for (clip_guid, frame).

Counts PaintStart events (strokes) and TextAnnotation events (captions) in the annotation track. Accumulates across all matching clips at the same frame so that old snapshots containing per-stroke clips are handled correctly.

Parameters:

• clip_guid – GUID of the media clip being annotated.

• frame – 0-indexed clip-local frame number.

Returns:

(n_strokes, n_captions) already in the annotation track.

Return type:

tuple

display_state: dict[str, Any]

Last received display state dict; empty until the first display message. Keys: pan ([x, y] normalised), zoom (float), exposure (stops), channel ("RGBA", "R", "G", "B", or "A").

get_or_create_clip_timeline(clip_guid: str) → str | None

Return the GUID of the single-clip timeline for clip_guid, creating it lazily.

All peers independently derive the same GUIDs via _derive_guid(), so no coordination message is required before clips can be used across peers. Callers should broadcast the timeline via broadcast_clip_timeline() the first time it is created so that peers without local creation can register the annotation track in their _object_map (required for receiving annotation INSERT_CHILD patches).

The clip copy inside the clip timeline shares the same sync GUID as the sequence clip. _traverse_and_map_preserve() ensures the sequence clip remains canonical in _object_map so that range_in_parent() returns the sequence-level position.

Parameters:

clip_guid – Sync GUID of the target sequence clip.

Returns:

GUID of the clip timeline, or None if clip_guid is not a known Clip.

Return type:

str or None

insert_child(parent_uuid: str, child_obj: SerializableObject, index: int = -1) → None

Insert child_obj into the parent container and broadcast the change.

A GUID is assigned to child_obj if it does not already have one. Use index=-1 to append.

Parameters:

• parent_uuid – GUID of the parent container (Track or Stack).

• child_obj – OTIO object to insert.

• index – Position at which to insert; -1 appends.

property is_syncing: bool

True while a snapshot or incoming delta is being applied locally.

Callers can read this to suppress outgoing broadcasts that would echo changes back to their source.

Return type:

bool

property object_map: dict[str, SerializableObject]

Read-only view of the flat GUID → OTIO object index.

on_display_changed(callback: Callable[[dict[str, Any]], None]) → Callable[[dict[str, Any]], None]

Register a callback fired whenever a display-state message arrives.

The callback receives the raw display state dict (same structure as display_state). Also usable as a decorator.

Parameters:

callback – Callable receiving the display state dict.

Returns:

The callback unchanged (decorator-compatible).

on_hierarchy_changed(callback: Callable[[str, str, str], None]) → Callable[[str, str, str], None]

Register a callback for hierarchy change events.

Fires for both locally-initiated and remotely-applied structural changes. May be used as a decorator.

Parameters:

callback – Callable receiving (parent_uuid, action, child_uuid) where action is one of "insert_child" or "remove_child".

Returns:

The callback unchanged (decorator-compatible).

on_playback_changed(callback: Callable[[dict[str, Any]], None]) → Callable[[dict[str, Any]], None]

Register a callback fired whenever a playback-state message arrives.

The callback receives the raw playback state dict (same structure as playback_state). Also usable as a decorator.

Parameters:

callback – Callable receiving the playback state dict.

Returns:

The callback unchanged (decorator-compatible).

on_property_changed(callback: Callable[[str, str, Any], None]) → Callable[[str, str, Any], None]

Register a callback for property change events.

Fires for both locally-initiated and remotely-applied property changes. May be used as a decorator.

Parameters:

callback – Callable receiving (target_uuid, path, new_value).

Returns:

The callback unchanged (decorator-compatible).

on_status_changed(callback: Callable[[str], None]) → Callable[[str], None]

Register a callback fired whenever status transitions.

Parameters:

callback – Callable receiving the new status string.

Returns:

The callback unchanged (decorator-compatible).

on_synced(callback: Callable[[], None]) → Callable[[], None]

Register a callback fired once when the session reaches STATE_SYNCED.

Fires both when this peer self-elects as master and when it finishes joining an existing master. Also usable as a decorator.

Parameters:

callback – Zero-argument callable.

Returns:

The callback unchanged (decorator-compatible).

playback_state: dict[str, Any]

Last received playback state dict; empty until the first playback message.

receive_and_apply_all() → list[tuple[str, Any]]

Drain the network and apply every pending message.

Returns:

List of (action, data) tuples for messages that require a response from the caller (e.g. to update RV state). Empty when all messages were handled internally or no messages were waiting.

register_timeline(timeline: Timeline) → OTIOSyncProxy

Register a timeline, assign GUIDs to all its objects, and index them.

Sets active_timeline_guid to the new timeline’s GUID if no active timeline exists yet.

Parameters:

timeline – The Timeline to register.

Returns:

An OTIOSyncProxy wrapping timeline so that attribute writes are automatically broadcast.

request_state() → None

Send a STATE_REQUEST to the master and enter STATE_JOINING.

Non-session messages received while joining are buffered in _delta_buffer and replayed by apply_snapshot().

reset_timelines() → None

Clear all registered timelines, the object map, and the active GUID.

Used during master re-initialisation when the timeline data must be rebuilt from scratch (e.g. after the RV node graph settles).

property root_timeline: Timeline | None

The active timeline, or the first registered timeline when none is active.

Returns:

Active Timeline, or None if no timelines have been registered.

selected_clip_guid: str | None

GUID of the clip most recently selected by a remote peer via a SELECTION broadcast. None when the selection is cleared.

send_state_snapshot(target_guid: str, playback_state: dict[str, Any] | None = None) → None

Serialise all registered timelines and send a full snapshot to a joiner.

Only the master should call this method. The snapshot is broadcast to the whole session (not unicast), but only the peer whose GUID matches target_guid will act on it.

Parameters:

• target_guid – GUID of the requesting peer.

• playback_state – Optional current playback state dict to include so the joiner can immediately seek to the right position.

property sequence_timeline_guid: str | None

GUID of the first registered timeline that is not a clip timeline.

Return type:

str or None

set_property(target_uuid: str, path: str, value: Any) → None

Set property path to value on object target_uuid and broadcast.

Property paths are either plain attributes (e.g. "name") or metadata sub-paths starting with "metadata/" (e.g. "metadata/annotations").

Parameters:

• target_uuid – GUID of the target object.

• path – Target property or metadata sub-key path.

• value – New value; must be a primitive type.

start_session() → None

Begin the join process by broadcasting a master-discovery message.

Transitions status to STATE_DISCOVERING. The caller is responsible for timing out and calling the appropriate method if no master responds (see class docstring for the full lifecycle).

tick() → list[tuple[str, Any]]

Poll the network and auto-advance the session handshake.

This is the recommended entry point for new client integrations. It wraps receive_and_apply_all() and handles the session state machine automatically:

• master_found → calls request_state() internally.

• state_snapshot_received → calls apply_snapshot() internally.

• state_request_received → returned to caller; the master must respond by calling send_state_snapshot().

Application-level events (playback_settings, selection_changed, annotation_*, insert_child, …) are returned so the caller can react to them. Playback updates are also delivered through the on_playback_changed() callback if one is registered.

Compare with receive_and_apply_all(), which returns every raw action tuple and leaves the handshake entirely to the caller.

Returns:

List of (action, data) tuples requiring application action (subset of what receive_and_apply_all() would return).

property timelines: dict[str, Timeline]

Read-only view of all registered timelines, keyed by sync GUID.

otio_sync_core.manager.sync_event_schema(cmd: Any) → str

Return the OTIO schema name for a SyncEvent object or a serialised dict.

Centralises the hasattr(cmd, "schema_name") / isinstance(cmd, dict) pattern that appears throughout annotation-handling code.

Parameters:

cmd – A deserialised SyncEvent object or a raw dict whose "OTIO_SCHEMA" key carries the schema name.

Returns:

Schema name string (e.g. "PaintStart.1"), or "" if cmd is neither.

Return type:

str

protocol_messages

Typed dataclasses describing the wire-format messages exchanged between sync participants.

Typed protocol message definitions for the OTIO Sync transport layer.

Each message class defined here is the single source of truth for one transport-layer message: its command_schema, its event name, and the shape of its payload. This mirrors how SyncEvent is the source of truth for the OTIO message layer, and lets a documentation generator describe the protocol directly from these classes (see docs/ generator).

Design constraints (see the typed-protocol-messages change design doc):

• Messages are pure data — handler logic lives in the manager/patcher, not on the message classes — so the classes stay importable in isolation for documentation.

• Registration is explicit via the register() decorator, keyed on (SCHEMA, EVENT), so the receive-side dispatch registry cannot drift from the definitions.

• Serialization is explicit: to_payload() builds a plain dict without reflective whole-object walking (no dataclasses.asdict()) and without per-message isinstance validation, so hot-path messages (PartialAnnotation, PlaybackSettingsSet) stay cheap.

• The settings messages declare their known fields for documentation but tolerate unknown fields (carried in extras) for forward-compatibility with independent producers.

class otio_sync_core.protocol_messages.AddTimeline(timeline_guid: str, timeline: Any, sync_timestamp: float | None = None)

Bases: ProtocolMessage

Registers a new timeline (sequence or single-clip) with all peers.

EVENT: ClassVar[str] = 'ADD_TIMELINE'

SCHEMA: ClassVar[str] = 'TIMELINE_1.0'

as_otio() → Any

Return the OTIO timeline, deserializing if still in wire form.

classmethod from_payload(data: dict[str, Any]) → AddTimeline

Reconstruct a message instance from a received command.payload.

sync_timestamp: float | None = None

timeline: Any

timeline_guid: str

to_payload() → dict[str, Any]

Return the command.payload dict for this message.

class otio_sync_core.protocol_messages.DisplaySettingsSet(pan: list | None = None, zoom: float | None = None, exposure: float | None = None, channel: str | None = None, sync_timestamp: float | None = None, extras: dict = <factory>)

Bases: ProtocolMessage

Display state broadcast (pan/zoom/exposure/channel).

Known fields are declared for documentation; additional producer fields are preserved in extras.

EVENT: ClassVar[str] = 'SET'

SCHEMA: ClassVar[str] = 'DISPLAY_SETTINGS_1.0'

channel: str | None = None

exposure: float | None = None

extras: dict

classmethod from_payload(data: dict[str, Any]) → DisplaySettingsSet

Reconstruct a message instance from a received command.payload.

pan: list | None = None

sync_timestamp: float | None = None

to_payload() → dict[str, Any]

Return the command.payload dict for this message.

zoom: float | None = None

class otio_sync_core.protocol_messages.IAmMaster(master_guid: str)

Bases: ProtocolMessage

Master’s response to discovery, announcing itself as session master.

ENVELOPE_SCHEMA: ClassVar[str | None] = 'SYNC_REVIEW_1.0'

Legacy top-level envelope schema preserved for older peers.

EVENT: ClassVar[str] = 'I_AM_MASTER'

SCHEMA: ClassVar[str] = 'LiveSession.1'

classmethod from_payload(data: dict[str, Any]) → IAmMaster

Reconstruct a message instance from a received command.payload.

master_guid: str

to_payload() → dict[str, Any]

Return the command.payload dict for this message.

class otio_sync_core.protocol_messages.InsertChild(parent_uuid: str, child_data: Any, index: int = -1, sync_timestamp: float | None = None)

Bases: ProtocolMessage

Inserts a child object into a parent container.

EVENT: ClassVar[str] = 'INSERT_CHILD'

SCHEMA: ClassVar[str] = 'OTIO_SESSION_1.0'

as_otio() → Any

Return the OTIO child object, deserializing if still in wire form.

child_data: Any

classmethod from_payload(data: dict[str, Any]) → InsertChild

Reconstruct a message instance from a received command.payload.

index: int = -1

parent_uuid: str

sync_timestamp: float | None = None

to_payload() → dict[str, Any]

Return the command.payload dict for this message.

class otio_sync_core.protocol_messages.MoveChild(parent_uuid: str, child_uuid: str, to_index: int = 0, sync_timestamp: float | None = None)

Bases: ProtocolMessage

Moves a child to a new index within its parent container.

EVENT: ClassVar[str] = 'MOVE_CHILD'

SCHEMA: ClassVar[str] = 'OTIO_SESSION_1.0'

child_uuid: str

classmethod from_payload(data: dict[str, Any]) → MoveChild

Reconstruct a message instance from a received command.payload.

parent_uuid: str

sync_timestamp: float | None = None

to_index: int = 0

to_payload() → dict[str, Any]

Return the command.payload dict for this message.

class otio_sync_core.protocol_messages.NewParticipant

Bases: ProtocolMessage

Announces that a new participant has joined the sync review.

EVENT: ClassVar[str] = 'NEW_PARTICIPANT'

SCHEMA: ClassVar[str] = 'LiveSession.1'

classmethod from_payload(data: dict[str, Any]) → NewParticipant

Reconstruct a message instance from a received command.payload.

to_payload() → dict[str, Any]

Return the command.payload dict for this message.

class otio_sync_core.protocol_messages.NewPresenter(presenter_hash: str)

Bases: ProtocolMessage

Announces that a peer has become the session presenter.

EVENT: ClassVar[str] = 'NEW_PRESENTER'

SCHEMA: ClassVar[str] = 'LiveSession.1'

classmethod from_payload(data: dict[str, Any]) → NewPresenter

Reconstruct a message instance from a received command.payload.

presenter_hash: str

to_payload() → dict[str, Any]

Return the command.payload dict for this message.

class otio_sync_core.protocol_messages.PartialAnnotation(clip_guid: str, frame: float, fps: float, events: list = <factory>)

Bases: ProtocolMessage

Mid-stroke partial annotation (visual preview, not persisted).

Hot path: fires repeatedly while a stroke is being drawn. No validation or reflective serialization is performed.

EVENT: ClassVar[str] = 'PARTIAL'

SCHEMA: ClassVar[str] = 'Annotation.1'

clip_guid: str

events: list

fps: float

frame: float

classmethod from_payload(data: dict[str, Any]) → PartialAnnotation

Reconstruct a message instance from a received command.payload.

to_payload() → dict[str, Any]

Return the command.payload dict for this message.

class otio_sync_core.protocol_messages.PlaybackSettingsSet(playing: bool | None = None, current_time: dict | None = None, looping: bool | None = None, timeline_guid: str | None = None, sync_timestamp: float | None = None, extras: dict = <factory>)

Bases: ProtocolMessage

Playback state broadcast.

Hot path: fires on frame change during playback/scrubbing. Known fields are declared for documentation; any additional producer fields are preserved in extras and round-tripped unchanged.

EVENT: ClassVar[str] = 'SET'

SCHEMA: ClassVar[str] = 'PLAYBACK_SETTINGS_1.0'

current_time: dict | None = None

extras: dict

classmethod from_payload(data: dict[str, Any]) → PlaybackSettingsSet

Reconstruct a message instance from a received command.payload.

looping: bool | None = None

playing: bool | None = None

sync_timestamp: float | None = None

timeline_guid: str | None = None

to_payload() → dict[str, Any]

Return the command.payload dict for this message.

class otio_sync_core.protocol_messages.ProtocolMessage

Bases: object

Base class for all transport-layer protocol messages.

Subclasses are @dataclass-decorated and @register-ed. They set the class-level SCHEMA and EVENT constants and implement to_payload() / from_payload() explicitly.

Variables:

• SCHEMA – The envelope command_schema for this message.

• EVENT – The envelope command.event for this message.

• ENVELOPE_SCHEMA – Optional top-level schema key written on the envelope (only IAmMaster uses this for legacy compatibility).

ENVELOPE_SCHEMA: ClassVar[str | None] = None

EVENT: ClassVar[str] = ''

SCHEMA: ClassVar[str] = ''

classmethod doc_fields() → list[tuple[str, str, str]]

Return (name, type, description) triples for documentation.

Default implementation reads the dataclass fields, skipping the extras catch-all used by tolerant messages.

Returns:

List of (field_name, type_name, doc) tuples.

classmethod from_payload(data: dict[str, Any]) → ProtocolMessage

Reconstruct a message instance from a received command.payload.

to_payload() → dict[str, Any]

Return the command.payload dict for this message.

class otio_sync_core.protocol_messages.RemoveChild(parent_uuid: str, child_uuid: str, sync_timestamp: float | None = None)

Bases: ProtocolMessage

Removes a child from its parent container.

EVENT: ClassVar[str] = 'REMOVE_CHILD'

SCHEMA: ClassVar[str] = 'OTIO_SESSION_1.0'

child_uuid: str

classmethod from_payload(data: dict[str, Any]) → RemoveChild

Reconstruct a message instance from a received command.payload.

parent_uuid: str

sync_timestamp: float | None = None

to_payload() → dict[str, Any]

Return the command.payload dict for this message.

class otio_sync_core.protocol_messages.RenameTimeline(timeline_guid: str, name: str, sync_timestamp: float | None = None)

Bases: ProtocolMessage

Renames an existing timeline on all peers.

EVENT: ClassVar[str] = 'RENAME_TIMELINE'

SCHEMA: ClassVar[str] = 'TIMELINE_1.0'

classmethod from_payload(data: dict[str, Any]) → RenameTimeline

Reconstruct a message instance from a received command.payload.

name: str

sync_timestamp: float | None = None

timeline_guid: str

to_payload() → dict[str, Any]

Return the command.payload dict for this message.

class otio_sync_core.protocol_messages.ReplaceAnnotationCommands(annotation_clip_guid: str, commands: list = <factory>, sync_timestamp: float | None = None)

Bases: ProtocolMessage

Replaces the full annotation-command list on an annotation clip.

EVENT: ClassVar[str] = 'REPLACE_ANNOTATION_COMMANDS'

SCHEMA: ClassVar[str] = 'OTIO_SESSION_1.0'

annotation_clip_guid: str

as_otio() → list

Return the OTIO SyncEvent list, deserializing any wire-form entries.

commands: list

classmethod from_payload(data: dict[str, Any]) → ReplaceAnnotationCommands

Reconstruct a message instance from a received command.payload.

sync_timestamp: float | None = None

to_payload() → dict[str, Any]

Return the command.payload dict for this message.

class otio_sync_core.protocol_messages.SelectionSet(clip_guid: str, view_mode: str = 'source', sync_timestamp: float | None = None)

Bases: ProtocolMessage

Broadcasts the clip the master has selected and the active view mode.

EVENT: ClassVar[str] = 'SET'

SCHEMA: ClassVar[str] = 'SELECTION_1.0'

clip_guid: str

classmethod from_payload(data: dict[str, Any]) → SelectionSet

Reconstruct a message instance from a received command.payload.

sync_timestamp: float | None = None

to_payload() → dict[str, Any]

Return the command.payload dict for this message.

view_mode: str = 'source'

class otio_sync_core.protocol_messages.SetProperty(target_uuid: str, path: str, value: Any, sync_timestamp: float | None = None)

Bases: ProtocolMessage

Sets a property or metadata path on an object.

EVENT: ClassVar[str] = 'SET_PROPERTY'

SCHEMA: ClassVar[str] = 'OTIO_SESSION_1.0'

classmethod from_payload(data: dict[str, Any]) → SetProperty

Reconstruct a message instance from a received command.payload.

path: str

sync_timestamp: float | None = None

target_uuid: str

to_payload() → dict[str, Any]

Return the command.payload dict for this message.

value: Any

class otio_sync_core.protocol_messages.SharedKeyRequest(key: str)

Bases: ProtocolMessage

Requests the session’s shared key from a peer.

EVENT: ClassVar[str] = 'SHARED_KEY_REQUEST'

SCHEMA: ClassVar[str] = 'LiveSession.1'

classmethod from_payload(data: dict[str, Any]) → SharedKeyRequest

Reconstruct a message instance from a received command.payload.

key: str

to_payload() → dict[str, Any]

Return the command.payload dict for this message.

class otio_sync_core.protocol_messages.SharedKeyResponse(key: str)

Bases: ProtocolMessage

Responds to a shared-key request with the session’s shared key.

EVENT: ClassVar[str] = 'SHARED_KEY_RESPONSE'

SCHEMA: ClassVar[str] = 'LiveSession.1'

classmethod from_payload(data: dict[str, Any]) → SharedKeyResponse

Reconstruct a message instance from a received command.payload.

key: str

to_payload() → dict[str, Any]

Return the command.payload dict for this message.

class otio_sync_core.protocol_messages.StateRequest(target_guid: str, requester_guid: str)

Bases: ProtocolMessage

Joiner’s request to the master for a full state snapshot.

EVENT: ClassVar[str] = 'STATE_REQUEST'

SCHEMA: ClassVar[str] = 'LiveSession.1'

classmethod from_payload(data: dict[str, Any]) → StateRequest

Reconstruct a message instance from a received command.payload.

requester_guid: str

target_guid: str

to_payload() → dict[str, Any]

Return the command.payload dict for this message.

class otio_sync_core.protocol_messages.StateSnapshot(target_guid: str, timelines: dict = <factory>, active_timeline_guid: str | None = None, snapshot_timestamp: float | None = None, playback_state: dict | None = None, display_state: dict | None = None)

Bases: ProtocolMessage

Master’s full session snapshot sent in response to a state request.

EVENT: ClassVar[str] = 'STATE_SNAPSHOT'

SCHEMA: ClassVar[str] = 'LiveSession.1'

active_timeline_guid: str | None = None

as_otio() → dict[str, Any]

Return {guid: OTIO timeline}, deserializing any wire-form entries.

display_state: dict | None = None

classmethod from_payload(data: dict[str, Any]) → StateSnapshot

Reconstruct a message instance from a received command.payload.

playback_state: dict | None = None

snapshot_timestamp: float | None = None

target_guid: str

timelines: dict

to_payload() → dict[str, Any]

Return the command.payload dict for this message.

class otio_sync_core.protocol_messages.WhoIsMaster(requester_guid: str)

Bases: ProtocolMessage

Master-discovery broadcast asking any existing master to identify itself.

EVENT: ClassVar[str] = 'WHO_IS_MASTER'

SCHEMA: ClassVar[str] = 'LiveSession.1'

classmethod from_payload(data: dict[str, Any]) → WhoIsMaster

Reconstruct a message instance from a received command.payload.

requester_guid: str

to_payload() → dict[str, Any]

Return the command.payload dict for this message.

otio_sync_core.protocol_messages.doc_field(*, default: ~typing.Any = <dataclasses._MISSING_TYPE object>, default_factory: ~typing.Any = <dataclasses._MISSING_TYPE object>, doc: str = '')

Declare a dataclass field carrying a documentation string in metadata.

The documentation generator reads field.metadata["doc"] for each field.

Parameters:

• default – Default value (mutually exclusive with default_factory).

• default_factory – Zero-arg callable producing the default.

• doc – Human-readable description of the field.

otio_sync_core.protocol_messages.message_for(command_schema: str, event: str) → type[ProtocolMessage] | None

Return the message class for (command_schema, event), or None.

Parameters:

• command_schema – Envelope command_schema value.

• event – Envelope command.event value.

Returns:

The registered ProtocolMessage subclass, or None when the pair is unknown (caller should ignore the message safely).

otio_sync_core.protocol_messages.register(cls: type[ProtocolMessage]) → type[ProtocolMessage]

Register cls in the protocol registry keyed on (SCHEMA, EVENT).

Used as a class decorator. Raises if two classes claim the same (SCHEMA, EVENT) pair, so collisions surface at import time.

Parameters:

cls – A ProtocolMessage subclass with SCHEMA/EVENT set.

Returns:

cls unchanged (decorator-compatible).

otio_sync_core.protocol_messages.registered_messages() → dict[tuple[str, str], type[ProtocolMessage]]

Return a copy of the full (schema, event) -> class registry.

Used by the documentation generator to enumerate every protocol message.

network

The UDP network backend and the network-protocol interface implemented by all backends.

UDP broadcast network backend and shared network Protocol for OTIO Sync.

class otio_sync_core.network.SyncNetworkProtocol(*args, **kwargs)

Bases: Protocol

Structural interface that all network backends must satisfy.

Both UDPNetwork and RabbitMQNetwork conform to this protocol, allowing SyncManager to accept either without a concrete base class.

receive_payloads() → list[dict[str, Any]]

Return all payloads received since the last call, without blocking.

send_payload(payload: dict[str, Any]) → None

Broadcast payload to all peers in the session.

stop() → None

Shut down the network connection and release resources.

class otio_sync_core.network.UDPNetwork(port: int = 9999, broadcast_ip: str | None = None, self_guid: str | None = None)

Bases: object

LAN broadcast network backend using UDP.

Opens a non-blocking receive socket bound to port and a separate send socket with SO_BROADCAST set. All peers on the same LAN segment that bind to the same port will receive every message.

Self-filtering is done via self_guid: any received payload whose source_guid matches is silently discarded.

Parameters:

• port – UDP port to bind and broadcast on.

• broadcast_ip – Explicit broadcast address; auto-detected when None.

• self_guid – GUID of the local peer used to filter own messages.

close() → None

Close both sockets immediately.

receive_payloads() → list[dict[str, Any]]

Drain all available UDP datagrams and return them as parsed dicts.

Non-blocking; returns an empty list when no data is waiting. Own messages (matched by source_guid) are silently dropped.

Returns:

List of received payload dicts.

send_payload(payload: dict[str, Any]) → None

Broadcast payload as JSON to the LAN.

Injects source_guid into the payload if not already present.

Parameters:

payload – Message envelope to broadcast.

stop() → None

Alias for close(); satisfies SyncNetworkProtocol.

otio_sync_core.network.get_local_broadcast() → str

Derive the LAN broadcast address from the default route interface.

Falls back to 255.255.255.255 if the address cannot be determined.

Returns:

Broadcast IP address string, e.g. "192.168.1.255".

rabbitmq_network

A RabbitMQ-backed network transport, an alternative to the UDP backend.

RabbitMQ fanout-exchange network backend for OTIO Sync.

class otio_sync_core.rabbitmq_network.RabbitMQNetwork(host: str = '127.0.0.1', port: int = 5672, session_id: str = 'otio-sync-default', self_guid: str | None = None)

Bases: object

RabbitMQ network backend for OTIO Sync.

Uses a fanout exchange so that every peer bound to the same exchange receives every published message. The exchange name is derived from session_id, which implicitly scopes peers to a session without any server-side configuration.

Two dedicated background threads handle I/O so callers are never blocked by pika:

• _consumer_thread — owns a BlockingConnection for receiving; pushes decoded payloads onto _incoming_queue.

• _publisher_thread — owns a separate BlockingConnection for sending; drains _send_queue in a tight loop with automatic reconnection on failure.

send_payload therefore never touches a socket directly; it is always non-blocking for the caller.

Self-filtering is applied in the consumer callback: any message whose source_guid matches self_guid is silently discarded before being enqueued.

Parameters:

• host – RabbitMQ broker hostname or IP.

• port – RabbitMQ broker AMQP port.

• session_id – Logical session name; used to derive the exchange name.

• self_guid – GUID of the local peer used to filter own messages. Auto-generated if not provided.

receive_payloads() → list[dict[str, Any]]

Drain the internal queue and return all pending payloads.

Non-blocking; returns an empty list when nothing is waiting. Messages are populated by the background consumer thread.

Returns:

List of received payload dicts.

send_payload(payload: dict[str, Any]) → None

Enqueue payload for publishing to the fanout exchange.

Non-blocking: the actual socket write happens on the publisher thread. Injects source_guid into the payload if not already present.

Parameters:

payload – Message envelope to broadcast.

stop() → None

Signal background threads to exit and wait for them to finish.

Blocks for up to 2 seconds per thread.

wait_until_ready(timeout: float = 5.0) → bool

Block until the consumer queue is bound and ready to receive messages.

Parameters:

timeout – Maximum seconds to wait before returning False.

Returns:

True if the consumer became ready within timeout, else False.

Return type:

bool

proxy

A transparent proxy that intercepts attribute writes on wrapped OTIO objects so that local edits can be turned into outgoing sync events.

Transparent proxy that intercepts attribute writes and forwards them to OTIOPatcher.

class otio_sync_core.proxy.OTIOSyncProxy(obj: otio.core.SerializableObject, patcher: OTIOPatcher, parent_path: str = '')

Bases: object

Transparent proxy around an OTIO SerializableObject.

Attribute reads are passed through to the wrapped object unchanged. Attribute writes are applied to the wrapped object and forwarded to the OTIOPatcher as a set_property change, so that remote peers stay in sync without the caller needing to be aware of the sync layer.

Child OTIO objects returned by __getattr__ are themselves wrapped in a new OTIOSyncProxy so that nested attribute writes are also captured.

Parameters:

• obj – The OTIO object to wrap.

• patcher – The OTIOPatcher (or compatible) that owns this object.

• parent_path – Dot-separated path prefix used when constructing the property path for nested objects.

patcher

Helpers for converting between OTIO objects and plain dictionaries and for applying patches to an OTIO timeline.

Transport-agnostic patching engine for OpenTimelineIO (OTIO) graphs.

class otio_sync_core.patcher.OTIOPatcher

Bases: object

Manages the lifecycle of OTIO graph patches.

Tracks object GUIDs, observes mutations, and applies patch events (such as property changes or hierarchy insertions/moves/removals) to the local graph.

apply_patch(msg: ProtocolMessage) → tuple[str, Any] | None

Apply an OTIO-session mutation message to the local graph.

Dispatches on the concrete message type, so the same class that built the payload (in set_property(), insert_child(), etc.) is the one used to consume it.

Parameters:

msg – A reconstructed OTIO-session ProtocolMessage: SetProperty, MoveChild, RemoveChild, ReplaceAnnotationCommands, or InsertChild.

Returns:

An (action_name, action_data) tuple when the caller needs to act, or None.

Return type:

tuple or None

ensure_guid_and_map(obj: Any) → None

Assign a sync GUID to obj if absent, then add it to object_map.

Non-SerializableObject values are ignored.

Parameters:

obj – Candidate OTIO object.

insert_child(parent_uuid: str, child_obj: SerializableObject, index: int = -1) → ProtocolMessage | None

Insert child_obj into the parent container locally.

Parameters:

• parent_uuid – GUID of the parent container.

• child_obj – The OTIO object to insert.

• index – Position at which to insert; -1 appends.

Returns:

The generated patch payload, or None if parent_uuid is not found.

Return type:

dict or None

move_child(parent_uuid: str, child_uuid: str, to_index: int) → ProtocolMessage | None

Move child_uuid within its parent container locally.

Parameters:

• parent_uuid – GUID of the parent container.

• child_uuid – GUID of the child to move.

• to_index – Target position in the parent’s child list.

Returns:

The generated patch payload, or None if parent/child is not found or index is unchanged.

Return type:

dict or None

on_hierarchy_changed(callback: Callable[[str, str, str], None]) → Callable[[str, str, str], None]

Register a callback for hierarchy change events.

Parameters:

callback – Callable receiving (parent_uuid, action, child_uuid) where action is one of "insert_child", "remove_child", or "move_child".

Returns:

The callback unchanged (decorator-compatible).

Return type:

Callable

on_property_changed(callback: Callable[[str, str, Any], None]) → Callable[[str, str, Any], None]

Register a callback for property change events.

Parameters:

callback – Callable receiving (target_uuid, path, new_value).

Returns:

The callback unchanged (decorator-compatible).

Return type:

Callable

remove_child(parent_uuid: str, child_uuid: str) → ProtocolMessage | None

Remove child_uuid from its parent container locally.

Parameters:

• parent_uuid – GUID of the parent container.

• child_uuid – GUID of the child to remove.

Returns:

The generated patch payload, or None if parent or child is not found.

Return type:

dict or None

set_property(target_uuid: str, path: str, value: Any) → ProtocolMessage | None

Set property path to value on object target_uuid locally.

Parameters:

• target_uuid – GUID of the target object.

• path – Target property or metadata sub-key path (e.g. "name" or "metadata/custom").

• value – New value; must be a primitive type.

Returns:

The generated patch payload, or None if target_uuid is not found.

Return type:

dict or None

traverse_and_map(item: SerializableObject) → None

Recursively assign GUIDs to all OTIO objects under item and index them.

Parameters:

item – Root OTIO object to traverse.

traverse_and_map_preserve(item: SerializableObject) → None

Recursively assign GUIDs to all OTIO objects under item without overwriting existing entries.

Parameters:

item – Root OTIO object to traverse.

color

Colour-pipeline metadata helpers, including recognition of the special OCIO colorspace.

Color pipeline metadata schema, vocabulary parsing, and resolution.

This module bridges the OTIO Color Pipeline Model RFC through OTIO metadata until the native fields land in OTIO core. The metadata keys mirror the RFC field names verbatim so a future migration to native Timeline.color / Composable.color_space fields is a key move, not a reshape:

• Timeline.metadata["color"] — a config group with config, working_space and output_space string entries.

• Composable.metadata["color_space"] — a clip’s input colorspace as a single vocabulary-prefixed string (e.g. "ocio:ACEScg").

The module carries no rendering logic and resolves nothing to a transform — it only parses names, reads the metadata defensively, and applies the hierarchical resolution rule. Host adapters (OpenRV, xStudio) turn the resolved name into an actual color transform against their own OCIO config.

It deliberately does not import opentimelineio so it stays importable in isolation; it only touches the .metadata mapping of the objects passed in.

otio_sync_core.color.COLOR_GROUP = 'color'

timeline.metadata[COLOR_GROUP].

Type:

Key of the timeline-level color config group

otio_sync_core.color.COLOR_SPACE = 'color_space'

clip.metadata[COLOR_SPACE].

Type:

Key of a composable’s input colorspace

otio_sync_core.color.CONFIG = 'config'

Sub-keys of the timeline color group.

otio_sync_core.color.DEFAULT_VOCABULARY = 'ocio'

Vocabulary assumed for a bare (unprefixed) name.

otio_sync_core.color.RESOLVED_VOCABULARIES = ('ocio', 'interop')

Vocabularies actively resolved by host adapters in v1. Other tags (cicp, resolve, aces, custom, …) are preserved verbatim.

otio_sync_core.color.is_resolved_vocabulary(value: str) → bool

Return whether value’s vocabulary is one a host adapter resolves in v1.

Unknown vocabularies are valid and MUST be preserved verbatim, but host adapters are only expected to resolve RESOLVED_VOCABULARIES.

Parameters:

value – A colorspace string.

Returns:

True if the vocabulary is in RESOLVED_VOCABULARIES.

otio_sync_core.color.parse_colorspace(value: str) → tuple[str, str]

Split a colorspace string into (vocabulary, name).

The text before the first : is the vocabulary tag iff it is a valid tag (ASCII [a-z0-9_]); the remainder is the name. A string with no colon — or whose leading segment is not a valid tag — is treated as a bare name in DEFAULT_VOCABULARY. This matches the RFC rule that names legitimately containing a colon (e.g. "ocio:Utility - Curve - sRGB") must be prefixed, while unprefixed names never are.

Parameters:

value – The colorspace string (e.g. "ocio:ACEScg", "ACEScg").

Returns:

(vocabulary, name). vocabulary is the lowercased tag, or DEFAULT_VOCABULARY for a bare name.

otio_sync_core.color.read_color_space(obj: Any) → str | None

Return a composable’s color_space string, or None if unset.

Parameters:

obj – An OTIO Composable (or any object with .metadata).

Returns:

The non-empty colorspace string, or None.

otio_sync_core.color.read_timeline_color(timeline: Any) → dict

Return the timeline color group as a plain dict (empty if unset).

Parameters:

timeline – An OTIO Timeline (or any object with .metadata).

Returns:

A dict possibly containing CONFIG, WORKING_SPACE and OUTPUT_SPACE. Empty when the timeline is unmanaged.

otio_sync_core.color.resolve_input_colorspace(clip: Any, timeline: Any | None = None, host_default: str | None = None) → str | None

Resolve a clip’s effective input colorspace.

Resolution order, per the color-pipeline-sync capability:

1. the clip’s own color_space if set;

2. otherwise the timeline’s working_space;

3. otherwise host_default.

No media-reference field or provenance data is consulted.

Parameters:

• clip – The OTIO Clip whose input space is being resolved.

• timeline – The owning Timeline, consulted for working_space. Optional; pass None to skip straight to host_default.

• host_default – The host’s fallback colorspace name.

Returns:

The effective colorspace string, or None when nothing is set.

xs_annotation_codec

Encoding and decoding of xStudio annotations to and from the OTIO sync format.

Bidirectional codec: xStudio pen-stroke dicts ↔ OTIO SyncEvent objects.

Both conversion directions are pure functions with no xStudio SDK dependency. Callers are responsible for registering the SyncEvent schemadef in OTIO_PLUGIN_MANIFEST_PATH before importing this module.

Coordinate systems

System	x range	y	origin
xStudio native	[−1, +1] (W-normalised)	down	centre
OTIO SyncEvent / RV paint	[−aspect/2, +aspect/2] (H-normalised)	up	centre

Scale factor: aspect_half = W / (2 × H). For 16:9 media: 1920 / (2 × 1080) ≈ 0.8889.

xStudio → OTIO: x_otio =  x_xs * aspect_half, y_otio = −y_xs * aspect_half OTIO → xStudio: x_xs   =  x_otio / aspect_half, y_xs   = −y_otio / aspect_half

otio_sync_core.xs_annotation_codec.sync_events_to_xs_captions(commands: list, aspect_half: float) → list

Convert TextAnnotation SyncEvent objects to xStudio caption dicts.

Parameters:

• commands – Sequence of SyncEvent objects; only TextAnnotation entries are processed.

• aspect_half – W / (2H) coordinate scale factor.

Returns:

List of xStudio caption dicts suitable for Bookmark.set_annotation(captions=...).

Return type:

list

otio_sync_core.xs_annotation_codec.sync_events_to_xs_strokes(commands: list, aspect_half: float) → list

Convert a PaintStart / PaintPoints command sequence to xStudio stroke dicts.

Inverts the H-normalised / Y-up (OTIO/RV) coordinate system back to the W-normalised / Y-down system that xStudio expects:

x_xs = x_otio / aspect_half
y_xs = −y_otio / aspect_half

Parameters:

• commands – Sequence of SyncEvent objects from an annotation clip (PaintStart, PaintPoints, and TextAnnotation entries are all accepted; only the paint entries are processed here).

• aspect_half – W / (2H) derived from the target media resolution.

Returns:

List of xStudio pen-stroke dicts suitable for Bookmark.set_annotation(strokes=...).

Return type:

list

otio_sync_core.xs_annotation_codec.xs_captions_to_sync_events(captions: list, aspect_half: float, existing_uuids: List[str] | None = None) → list

Convert xStudio captions dicts to a sequence of OTIO SyncEvent objects.

Parameters:

• captions – List of xStudio caption dicts from Bookmark.annotation_data["Data"]["captions"].

• aspect_half – W / (2H) coordinate scale factor.

• existing_uuids – When provided, reuse these UUID strings (by index) instead of generating fresh ones. Pass the existing UUIDs from the OTIO clip when building a replacement command list so that RV can update text nodes in place.

Returns:

List of TextAnnotation SyncEvent objects.

Return type:

list

otio_sync_core.xs_annotation_codec.xs_strokes_to_sync_events(pen_strokes: list, aspect_half: float, uuid_list: List[str] | None = None) → list

Convert xStudio pen_strokes dicts to a sequence of OTIO SyncEvent objects.

Each xStudio stroke dict becomes a PaintStart + PaintPoints pair. Point coordinates are converted from W-normalised / Y-down to H-normalised / Y-up by multiplying x/y by aspect_half / −aspect_half.

xStudio V4 stroke dicts use "colour": [r, g, b] and "type": "Brush"/"Pen"/"Erase" (the legacy V3 keys "r", "g", "b" and "is_erase_stroke" are automatically upgraded by xStudio’s annotation deserialiser before they reach Python).

Parameters:

• pen_strokes – List of xStudio pen-stroke dicts as returned by Bookmark.annotation_data["Data"]["pen_strokes"].

• aspect_half – W / (2H) coordinate scale factor.

• uuid_list – Optional list of stable UUID strings, one per stroke. When given, uuid_list[i] is used for stroke i instead of a freshly generated UUID. Pass this from _stroke_uuid_cache when repeated partial broadcasts of the same frame must share stable UUIDs.

Returns:

List of SyncEvent objects (interleaved PaintStart / PaintPoints entries).

Return type:

list