"""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 :mod:`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 :func:`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 :func:`dataclasses.asdict`) and without per-message ``isinstance`` validation, so hot-path messages (:class:`PartialAnnotation`, :class:`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. """ from __future__ import annotations import json from dataclasses import MISSING, dataclass, field, fields from typing import Any, ClassVar # --------------------------------------------------------------------------- # OTIO wire conversion # # These helpers are the single place that converts between OTIO objects and # their wire form. ``opentimelineio`` is imported lazily *inside* them so this # module stays importable without OTIO installed (the documentation generator # only reads class/field metadata, never calls these). The format must stay # byte-identical to the prior call-site serialization (``otio_json``, # ``indent=-1``) so peers on older code keep interoperating. # --------------------------------------------------------------------------- def _to_wire(obj: Any) -> Any: """Serialize an OTIO object to its wire ``dict``; pass through a ``dict``. :param obj: An OTIO ``SerializableObject`` or an already-serialized dict. :returns: The wire-form dict. """ if isinstance(obj, dict): return obj import opentimelineio as otio return json.loads(otio.adapters.write_to_string(obj, "otio_json", indent=-1)) def _from_wire(data: Any) -> Any: """Deserialize a wire ``dict`` to an OTIO object; pass through a non-dict. :param data: A wire-form dict, or an already-deserialized OTIO object. :returns: The OTIO ``SerializableObject``. """ if not isinstance(data, dict): return data import opentimelineio as otio return otio.adapters.read_from_string(json.dumps(data), "otio_json") # --------------------------------------------------------------------------- # Registry # --------------------------------------------------------------------------- #: Maps ``(command_schema, event)`` to the message class that defines it. _REGISTRY: dict[tuple[str, str], type["ProtocolMessage"]] = {} def 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. :param cls: A :class:`ProtocolMessage` subclass with ``SCHEMA``/``EVENT`` set. :returns: *cls* unchanged (decorator-compatible). """ key = (cls.SCHEMA, cls.EVENT) if not cls.SCHEMA or not cls.EVENT: raise ValueError(f"{cls.__name__} must define non-empty SCHEMA and EVENT") if key in _REGISTRY: raise ValueError( f"Duplicate protocol message registration for {key}: " f"{_REGISTRY[key].__name__} and {cls.__name__}" ) _REGISTRY[key] = cls return cls def message_for(command_schema: str, event: str) -> "type[ProtocolMessage] | None": """Return the message class for ``(command_schema, event)``, or ``None``. :param command_schema: Envelope ``command_schema`` value. :param event: Envelope ``command.event`` value. :returns: The registered :class:`ProtocolMessage` subclass, or ``None`` when the pair is unknown (caller should ignore the message safely). """ return _REGISTRY.get((command_schema, event)) def 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. """ return dict(_REGISTRY) def doc_field( *, default: Any = MISSING, default_factory: Any = MISSING, doc: str = "", ): """Declare a dataclass field carrying a documentation string in metadata. The documentation generator reads ``field.metadata["doc"]`` for each field. :param default: Default value (mutually exclusive with *default_factory*). :param default_factory: Zero-arg callable producing the default. :param doc: Human-readable description of the field. """ if default_factory is not MISSING: return field(default_factory=default_factory, metadata={"doc": doc}) if default is not MISSING: return field(default=default, metadata={"doc": doc}) return field(metadata={"doc": doc}) # --------------------------------------------------------------------------- # Base class # --------------------------------------------------------------------------- class ProtocolMessage: """Base class for all transport-layer protocol messages. Subclasses are ``@dataclass``-decorated and ``@register``-ed. They set the class-level :attr:`SCHEMA` and :attr:`EVENT` constants and implement :meth:`to_payload` / :meth:`from_payload` explicitly. :cvar SCHEMA: The envelope ``command_schema`` for this message. :cvar EVENT: The envelope ``command.event`` for this message. :cvar ENVELOPE_SCHEMA: Optional top-level ``schema`` key written on the envelope (only :class:`IAmMaster` uses this for legacy compatibility). """ SCHEMA: ClassVar[str] = "" EVENT: ClassVar[str] = "" ENVELOPE_SCHEMA: ClassVar["str | None"] = None def to_payload(self) -> dict[str, Any]: """Return the ``command.payload`` dict for this message.""" raise NotImplementedError @classmethod def from_payload(cls, data: dict[str, Any]) -> "ProtocolMessage": """Reconstruct a message instance from a received ``command.payload``.""" raise NotImplementedError @classmethod def doc_fields(cls) -> 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. """ out: list[tuple[str, str, str]] = [] for f in fields(cls): # type: ignore[arg-type] if f.name == "extras": continue type_name = getattr(f.type, "__name__", str(f.type)) out.append((f.name, type_name, f.metadata.get("doc", ""))) return out # --------------------------------------------------------------------------- # Session family — LiveSession.1 # --------------------------------------------------------------------------- @register @dataclass class WhoIsMaster(ProtocolMessage): """Master-discovery broadcast asking any existing master to identify itself.""" SCHEMA = "LiveSession.1" EVENT = "WHO_IS_MASTER" requester_guid: str = doc_field(doc="GUID of the peer asking who the master is.") def to_payload(self) -> dict[str, Any]: return {"requester_guid": self.requester_guid} @classmethod def from_payload(cls, data: dict[str, Any]) -> "WhoIsMaster": return cls(requester_guid=data.get("requester_guid")) @register @dataclass class IAmMaster(ProtocolMessage): """Master's response to discovery, announcing itself as session master.""" SCHEMA = "LiveSession.1" EVENT = "I_AM_MASTER" #: Legacy top-level envelope schema preserved for older peers. ENVELOPE_SCHEMA = "SYNC_REVIEW_1.0" master_guid: str = doc_field(doc="GUID of the peer that is the session master.") def to_payload(self) -> dict[str, Any]: return {"master_guid": self.master_guid} @classmethod def from_payload(cls, data: dict[str, Any]) -> "IAmMaster": return cls(master_guid=data.get("master_guid")) @register @dataclass class PeerAnnounce(ProtocolMessage): """Peer identity broadcast: who I am and what I can be authoritative for. Feeds the peer table that host election reads. Election is a pure function of that table, so every peer must learn of every other peer — not just of the master. Sent on joining and **periodically thereafter**. The periodic send is what makes silence meaningful: a peer may legitimately go quiet for a whole session, so only the absence of announcements distinguishes one that is idle from one that has died. Peers age out anyone they have not heard from within the liveness timeout. Nobody answers an announcement. Answering used to be how a joiner discovered peers that had long since gone quiet; a joiner now learns them from the roster in :class:`StateSnapshot`, and any it misses from their next periodic announcement. Dropping the answer removes the only step in this protocol whose message count grew with the size of the session. """ SCHEMA = "LiveSession.1" EVENT = "PEER_ANNOUNCE" peer_guid: str = doc_field(doc="GUID of the announcing peer.") app: str = doc_field( default="", doc='Application name, e.g. "xstudio" or "openrv". Ranks the peer for ' "host election; an unranked name is still eligible.", ) capabilities: list = doc_field( default_factory=list, doc='Roles this peer can hold, e.g. ["visibility"].', ) role: "str | None" = doc_field( default=None, doc="Session role of the announcing peer: `driver`, `reviewer`, or " "`viewer` — what this participant is permitted to emit at all, " "which is a different question from who holds the canonical state " "(master), who chooses what the session looks at (host), or who is " "broadcasting a category right now (the write leases). Omitted when " "the peer declares none, and an absent role means **the session's " "default role**, not the most restrictive one: a peer running code " "that predates roles must not read as ineligible, or one old peer " "would make a session with drivers in it look driverless. " "Self-declared on the same terms as `app`, and enforced by the " "*sender*: a receiving peer applies a message without checking the " "sender's role.", ) identity: "dict | None" = doc_field( default=None, doc="Who is on the other end: {user, first_name, last_name, host, " "source}. Every field is optional and the whole section is omitted " "when the peer has no identity — a peer without one is a full " "participant, labelled by app and GUID. Self-declared and " "**unverified**, on the same terms as `app`: it identifies " "cooperating participants, it does not authenticate them. " "`source` records where it came from (`local`, `override`, or a " "future authenticated provider). The displayed name is derived " "from these fields by the receiver and is not transmitted.", ) def to_payload(self) -> dict[str, Any]: payload = { "peer_guid": self.peer_guid, "app": self.app, "capabilities": list(self.capabilities), } if self.role: payload["role"] = self.role if self.identity is not None: payload["identity"] = dict(self.identity) return payload @classmethod def from_payload(cls, data: dict[str, Any]) -> "PeerAnnounce": return cls( peer_guid=data.get("peer_guid"), app=data.get("app") or "", capabilities=list(data.get("capabilities") or []), role=data.get("role"), identity=data.get("identity"), ) @register @dataclass class PeerDepart(ProtocolMessage): """Peer's notice that it is leaving the session. Removes the sender from every other peer's peer table, so a role elected from that table — host, in particular — moves off a peer that has gone. Without it the table is append-only and a departed host keeps visibility authority, freezing the session's view with no peer permitted to change it. **Best-effort.** It is sent once, on a path with no delivery guarantee, and a peer that crashes never sends it at all. Correctness therefore does not rest on it: peers also age out anyone they have not heard announce within the liveness timeout, and this message only makes the common case prompt. Do not add a consumer that assumes arrival. """ SCHEMA = "LiveSession.1" EVENT = "PEER_DEPART" peer_guid: str = doc_field(doc="GUID of the departing peer.") def to_payload(self) -> dict[str, Any]: return {"peer_guid": self.peer_guid} @classmethod def from_payload(cls, data: dict[str, Any]) -> "PeerDepart": return cls(peer_guid=data.get("peer_guid")) @register @dataclass class SetPeerRole(ProtocolMessage): """A driver's grant of a session role to a named participant (``session-role-administration``). Two properties are not inferable from the field list alone: **Broadcast, not addressed to its target alone.** Every peer — issuer, target, and everyone else — merges ``{user: role}`` into its own copy of the session's identity-keyed role memory. This is what makes the grant reach the master's memory (and so every later joiner's :class:`StateSnapshot`) without a routing hop, and what makes it survive the target's reconnection: a unicast to the target alone would leave the master's memory unaware the grant ever happened. **Applied by its target, which then re-announces.** A receiving peer does **not** write ``role`` into its peer-table entry for ``user`` on receipt of this message. Only the peer whose own identity matches ``user`` adopts the role, for itself, and re-announces it — :class:`PeerAnnounce` remains the single write path into every peer's table. Every other peer learns the new role from that subsequent announcement, the same way it learns a role on joining. """ SCHEMA = "LiveSession.1" EVENT = "SET_PEER_ROLE" user: str = doc_field(doc="Identity key of the participant being granted a role — the " "same `identity[\"user\"]` value the session's role memory is " "keyed on, never a peer GUID.") role: str = doc_field(doc="The role being granted: `driver`, `reviewer`, or `viewer`.") issuer_guid: "str | None" = doc_field( default=None, doc="GUID of the peer that issued the grant, carried for logging and " "provenance only. Not consulted by any receiving peer: who may " "issue a grant is checked by the issuer, before sending, against " "the same role table the broadcast guard uses.", ) def to_payload(self) -> dict[str, Any]: payload = {"user": self.user, "role": self.role} if self.issuer_guid: payload["issuer_guid"] = self.issuer_guid return payload @classmethod def from_payload(cls, data: dict[str, Any]) -> "SetPeerRole": return cls( user=data.get("user"), role=data.get("role"), issuer_guid=data.get("issuer_guid"), ) @register @dataclass class StateRequest(ProtocolMessage): """Joiner's request to the master for a full state snapshot.""" SCHEMA = "LiveSession.1" EVENT = "STATE_REQUEST" target_guid: str = doc_field(doc="GUID of the master the request is aimed at.") requester_guid: str = doc_field(doc="GUID of the joining peer.") def to_payload(self) -> dict[str, Any]: return {"target_guid": self.target_guid, "requester_guid": self.requester_guid} @classmethod def from_payload(cls, data: dict[str, Any]) -> "StateRequest": return cls( target_guid=data.get("target_guid"), requester_guid=data.get("requester_guid"), ) @register @dataclass class StateSnapshot(ProtocolMessage): """Master's full session snapshot sent in response to a state request.""" SCHEMA = "LiveSession.1" EVENT = "STATE_SNAPSHOT" target_guid: str = doc_field(doc="GUID of the joining peer this snapshot is for.") timelines: dict = doc_field( default_factory=dict, doc="Map of timeline GUID to OTIO timeline (objects on send, wire dicts on receive).", ) active_timeline_guid: "str | None" = doc_field( default=None, doc="GUID of the active timeline at snapshot time." ) snapshot_timestamp: "float | None" = doc_field( default=None, doc="Epoch seconds when the snapshot was taken." ) playback_state: "dict | None" = doc_field( default=None, doc="Optional current playback state to seed the joiner." ) display_state: "dict | None" = doc_field( default=None, doc="Optional current display state to seed the joiner." ) host_guid: "str | None" = doc_field( default=None, doc="GUID of the session host (visibility authority) at snapshot time, " "so a joiner does not assume it is host and fight the real one.", ) peers: dict = doc_field( default_factory=dict, doc="Peers present at snapshot time, as {guid: {app, capabilities, role, identity}}, so " "a joiner learns the peer set without every peer answering its " "announcement. Not the only discovery path: a joiner that receives " "no snapshot learns peers from their periodic announcements. " "Carries no liveness stamp — that is the receiver's own clock. " "`identity` is the same optional section PEER_ANNOUNCE carries, on " "the same terms — self-declared and **unverified** — and is present " "here so a peer that has gone quiet can still be named by a joiner " "that has never heard it announce. `role` is carried for the same " "reason and on the same terms as it is on PEER_ANNOUNCE — host " "eligibility is evaluated against this table, and a peer that has " "gone quiet is known to a joiner only through this roster, so a " "role omitted here would make that peer look role-less until its " "next heartbeat. An absent role means the session's default role.", ) session_roles: "dict | None" = doc_field( default=None, doc="Session role policy at snapshot time, as {\"default_role\": str, " '"peer_roles": {user: role}}. `default_role` is what a participant ' "the session does not recognise is given; `peer_roles` is the " "session's memory of who has held a role, keyed on the identity's " "`user` rather than on peer GUID — a driver who drops and rejoins " "has a new GUID, which is the only case that memory exists for. " "**Omitted when the session declares no policy**, exactly as a " "free channel is omitted from `broadcast_ownership` and an unset " "host is omitted from `host_guid`: an absent section means \"no " "declared policy\", not an empty one, so a peer predating this " "field cannot clear a session's policy by relaying state. Policy " "lives for the session only and is not persisted anywhere.", ) broadcast_ownership: "dict | None" = doc_field( default=None, doc="Per-channel write-lease state at snapshot time, as " '{"position"|"display"|"structure": {"owner_guid": str, ' '"remaining_ms": float}}. A channel with no live owner is omitted, ' "not sent with a null owner, so a peer predating this field — or a " "snapshot taken while every channel happened to be free — cannot " "be read as clearing a lease another peer already holds.", ) def to_payload(self) -> dict[str, Any]: payload: dict[str, Any] = { "target_guid": self.target_guid, "timelines": {g: _to_wire(tl) for g, tl in self.timelines.items()}, "active_timeline_guid": self.active_timeline_guid, "snapshot_timestamp": self.snapshot_timestamp, } if self.playback_state is not None: payload["playback_state"] = self.playback_state if self.display_state is not None: payload["display_state"] = self.display_state if self.host_guid is not None: payload["host_guid"] = self.host_guid if self.peers: payload["peers"] = {g: dict(p) for g, p in self.peers.items()} if self.session_roles: payload["session_roles"] = dict(self.session_roles) if self.broadcast_ownership: payload["broadcast_ownership"] = { ch: dict(info) for ch, info in self.broadcast_ownership.items() } return payload @classmethod def from_payload(cls, data: dict[str, Any]) -> "StateSnapshot": return cls( target_guid=data.get("target_guid"), timelines=data.get("timelines", {}), active_timeline_guid=data.get("active_timeline_guid"), snapshot_timestamp=data.get("snapshot_timestamp"), playback_state=data.get("playback_state"), display_state=data.get("display_state"), host_guid=data.get("host_guid"), peers=dict(data.get("peers") or {}), session_roles=data.get("session_roles"), broadcast_ownership=data.get("broadcast_ownership"), ) def as_otio(self) -> dict[str, Any]: """Return ``{guid: OTIO timeline}``, deserializing any wire-form entries.""" return {g: _from_wire(tl) for g, tl in self.timelines.items()} @register @dataclass class NewPresenter(ProtocolMessage): """Announces that a peer has become the session presenter.""" SCHEMA = "LiveSession.1" EVENT = "NEW_PRESENTER" presenter_hash: str = doc_field(doc="Hash identifying the new presenter.") def to_payload(self) -> dict[str, Any]: return {"presenter_hash": self.presenter_hash} @classmethod def from_payload(cls, data: dict[str, Any]) -> "NewPresenter": return cls(presenter_hash=data.get("presenter_hash")) @register @dataclass class NewParticipant(ProtocolMessage): """Announces that a new participant has joined the sync review.""" SCHEMA = "LiveSession.1" EVENT = "NEW_PARTICIPANT" def to_payload(self) -> dict[str, Any]: return {} @classmethod def from_payload(cls, data: dict[str, Any]) -> "NewParticipant": return cls() @register @dataclass class SharedKeyRequest(ProtocolMessage): """Requests the session's shared key from a peer.""" SCHEMA = "LiveSession.1" EVENT = "SHARED_KEY_REQUEST" key: str = doc_field(doc="The shared key being requested.") def to_payload(self) -> dict[str, Any]: return {"key": self.key} @classmethod def from_payload(cls, data: dict[str, Any]) -> "SharedKeyRequest": return cls(key=data.get("key")) @register @dataclass class SharedKeyResponse(ProtocolMessage): """Responds to a shared-key request with the session's shared key.""" SCHEMA = "LiveSession.1" EVENT = "SHARED_KEY_RESPONSE" key: str = doc_field(doc="The shared key being returned.") def to_payload(self) -> dict[str, Any]: return {"key": self.key} @classmethod def from_payload(cls, data: dict[str, Any]) -> "SharedKeyResponse": return cls(key=data.get("key")) # --------------------------------------------------------------------------- # Timeline origin # # A timeline's sync model is selected by its origin, carried inside the # timeline's own ``metadata.sync.origin`` (so it travels for free inside any # serialized timeline — StateSnapshot, AddTimeline, ReplaceTimeline). Peers # read it to route: OTIO-imported timelines use whole-OTIO snapshot pushes for # topology changes; native timelines use per-child patches. A timeline lacking # the marker is treated as native for backward-compatibility with older peers. # --------------------------------------------------------------------------- #: Origin marker value for timelines expanded from an imported ``.otio`` file. ORIGIN_OTIO_IMPORT = "otio_import" #: Origin marker value (and default) for ad-hoc / native clip-list timelines. ORIGIN_NATIVE = "native" def timeline_origin(timeline: Any) -> str: """Return a timeline's ``metadata.sync.origin``, defaulting to native. Accepts either an OTIO ``Timeline`` object or its wire-form dict, so the rule lives in exactly one place for both send and receive sides. Any timeline without the marker (older peers, native timelines) reads as :data:`ORIGIN_NATIVE`. :param timeline: An OTIO ``Timeline`` (has ``.metadata``) or a wire dict. :returns: The origin marker string. """ metadata = getattr(timeline, "metadata", None) if metadata is None and isinstance(timeline, dict): metadata = timeline.get("metadata") try: origin = metadata["sync"]["origin"] except (KeyError, TypeError): return ORIGIN_NATIVE return origin or ORIGIN_NATIVE # --------------------------------------------------------------------------- # Timeline family — TIMELINE_1.0 # --------------------------------------------------------------------------- @register @dataclass class AddTimeline(ProtocolMessage): """Registers a new timeline (sequence or single-clip) with all peers.""" SCHEMA = "TIMELINE_1.0" EVENT = "ADD_TIMELINE" timeline_guid: str = doc_field(doc="GUID of the timeline being added.") timeline: Any = doc_field( doc="OTIO timeline (object on send, wire dict on receive)." ) sync_timestamp: "float | None" = doc_field( default=None, doc="Epoch seconds when the message was sent." ) def to_payload(self) -> dict[str, Any]: return { "timeline_guid": self.timeline_guid, "timeline": _to_wire(self.timeline), "sync_timestamp": self.sync_timestamp, } @classmethod def from_payload(cls, data: dict[str, Any]) -> "AddTimeline": return cls( timeline_guid=data.get("timeline_guid"), timeline=data.get("timeline"), sync_timestamp=data.get("sync_timestamp"), ) def as_otio(self) -> Any: """Return the OTIO timeline, deserializing if still in wire form.""" return _from_wire(self.timeline) @register @dataclass class RenameTimeline(ProtocolMessage): """Renames an existing timeline on all peers.""" SCHEMA = "TIMELINE_1.0" EVENT = "RENAME_TIMELINE" timeline_guid: str = doc_field(doc="GUID of the timeline to rename.") name: str = doc_field(doc="New display name for the timeline.") sync_timestamp: "float | None" = doc_field( default=None, doc="Epoch seconds when the message was sent." ) def to_payload(self) -> dict[str, Any]: return { "timeline_guid": self.timeline_guid, "name": self.name, "sync_timestamp": self.sync_timestamp, } @classmethod def from_payload(cls, data: dict[str, Any]) -> "RenameTimeline": return cls( timeline_guid=data.get("timeline_guid"), name=data.get("name", ""), sync_timestamp=data.get("sync_timestamp"), ) @register @dataclass class RemoveTimeline(ProtocolMessage): """Removes an existing timeline from all peers. Carries only the GUID — peers already hold the timeline, so no OTIO payload is needed. Receivers that do not hold the GUID treat it as a no-op. """ SCHEMA = "TIMELINE_1.0" EVENT = "REMOVE_TIMELINE" timeline_guid: str = doc_field(doc="GUID of the timeline to remove.") sync_timestamp: "float | None" = doc_field( default=None, doc="Epoch seconds when the message was sent." ) def to_payload(self) -> dict[str, Any]: return { "timeline_guid": self.timeline_guid, "sync_timestamp": self.sync_timestamp, } @classmethod def from_payload(cls, data: dict[str, Any]) -> "RemoveTimeline": return cls( timeline_guid=data.get("timeline_guid"), sync_timestamp=data.get("sync_timestamp"), ) @register @dataclass class ReplaceTimeline(ProtocolMessage): """Wholesale replacement of a timeline's structure ("brute-force push"). Carries a complete OTIO timeline and replaces the target's structure on each peer in one shot, rather than as incremental child mutations. Used for topology changes (clip insert/remove, large re-edit) on OTIO-origin timelines, where reconstructing the structure via RV's native OTIO reader is cheaper and higher-fidelity than a stream of per-child patches. Distinct from :class:`AddTimeline` (which models a *new* timeline and is a no-op when the GUID already exists): ``REPLACE_TIMELINE`` deliberately overwrites an existing timeline. Each object's ``metadata.sync.guid`` in the pushed timeline is preserved so attribute patches and annotations stay resolvable across the replace. Applying to an unknown GUID creates it. """ SCHEMA = "TIMELINE_1.0" EVENT = "REPLACE_TIMELINE" timeline_guid: str = doc_field(doc="GUID of the timeline to replace (or create).") timeline: Any = doc_field( doc="Full OTIO timeline (object on send, wire dict on receive)." ) sync_timestamp: "float | None" = doc_field( default=None, doc="Epoch seconds when the message was sent." ) def to_payload(self) -> dict[str, Any]: return { "timeline_guid": self.timeline_guid, "timeline": _to_wire(self.timeline), "sync_timestamp": self.sync_timestamp, } @classmethod def from_payload(cls, data: dict[str, Any]) -> "ReplaceTimeline": return cls( timeline_guid=data.get("timeline_guid"), timeline=data.get("timeline"), sync_timestamp=data.get("sync_timestamp"), ) def as_otio(self) -> Any: """Return the OTIO timeline, deserializing if still in wire form.""" return _from_wire(self.timeline) # --------------------------------------------------------------------------- # Settings family — declare known fields, tolerate extras (hot paths) # --------------------------------------------------------------------------- @register @dataclass class PlaybackSettingsSet(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. """ SCHEMA = "PLAYBACK_SETTINGS_1.0" EVENT = "SET" playing: "bool | None" = doc_field(default=None, doc="Whether playback is running.") current_time: "dict | None" = doc_field( default=None, doc="Current position as a serialized RationalTime." ) playback_mode: "str | None" = doc_field( default=None, doc='Playback mode: "play-once", "loop", or "ping-pong".', ) timeline_guid: "str | None" = doc_field( default=None, doc="GUID of the timeline being viewed/played." ) view_mode: "str | None" = doc_field( default=None, doc='View mode: "sequence" (position authoritative, clip derived from ' 'the frame) or "source" (clip_guid authoritative, current_time is the ' 'in-clip offset).', ) clip_guid: "str | None" = doc_field( default=None, doc="Active clip sync GUID. Authoritative in source mode; confirmation/" "highlight only in sequence mode (never seeked to).", ) sync_timestamp: "float | None" = doc_field( default=None, doc="Epoch seconds when the message was sent." ) extras: dict = field(default_factory=dict) #: Field names modelled explicitly (everything else falls into ``extras``). _KNOWN: ClassVar[tuple[str, ...]] = ( "playing", "current_time", "playback_mode", "timeline_guid", "view_mode", "clip_guid", "sync_timestamp", ) def to_payload(self) -> dict[str, Any]: payload: dict[str, Any] = {} if self.playing is not None: payload["playing"] = self.playing if self.current_time is not None: payload["current_time"] = self.current_time if self.playback_mode is not None: payload["playback_mode"] = self.playback_mode if self.timeline_guid is not None: payload["timeline_guid"] = self.timeline_guid if self.view_mode is not None: payload["view_mode"] = self.view_mode if self.clip_guid is not None: payload["clip_guid"] = self.clip_guid if self.sync_timestamp is not None: payload["sync_timestamp"] = self.sync_timestamp payload.update(self.extras) return payload @classmethod def from_payload(cls, data: dict[str, Any]) -> "PlaybackSettingsSet": extras = {k: v for k, v in data.items() if k not in cls._KNOWN} return cls( playing=data.get("playing"), current_time=data.get("current_time"), playback_mode=data.get("playback_mode"), timeline_guid=data.get("timeline_guid"), view_mode=data.get("view_mode"), clip_guid=data.get("clip_guid"), sync_timestamp=data.get("sync_timestamp"), extras=extras, ) @register @dataclass class DisplaySettingsSet(ProtocolMessage): """Display state broadcast (pan/zoom/exposure/channel). Known fields are declared for documentation; additional producer fields are preserved in ``extras``. """ SCHEMA = "DISPLAY_SETTINGS_1.0" EVENT = "SET" pan: "list | None" = doc_field(default=None, doc="Normalised [x, y] pan offset.") zoom: "float | None" = doc_field(default=None, doc="Zoom multiplier (1.0 = none).") exposure: "float | None" = doc_field( default=None, doc="Exposure adjustment in stops (0.0 = none)." ) channel: "str | None" = doc_field( default=None, doc='Active channel: "RGBA", "R", "G", "B", or "A".' ) sync_timestamp: "float | None" = doc_field( default=None, doc="Epoch seconds when the message was sent." ) extras: dict = field(default_factory=dict) _KNOWN: ClassVar[tuple[str, ...]] = ( "pan", "zoom", "exposure", "channel", "sync_timestamp", ) def to_payload(self) -> dict[str, Any]: payload: dict[str, Any] = {} if self.pan is not None: payload["pan"] = self.pan if self.zoom is not None: payload["zoom"] = self.zoom if self.exposure is not None: payload["exposure"] = self.exposure if self.channel is not None: payload["channel"] = self.channel if self.sync_timestamp is not None: payload["sync_timestamp"] = self.sync_timestamp payload.update(self.extras) return payload @classmethod def from_payload(cls, data: dict[str, Any]) -> "DisplaySettingsSet": extras = {k: v for k, v in data.items() if k not in cls._KNOWN} return cls( pan=data.get("pan"), zoom=data.get("zoom"), exposure=data.get("exposure"), channel=data.get("channel"), sync_timestamp=data.get("sync_timestamp"), extras=extras, ) # --------------------------------------------------------------------------- # Selection family — RETIRED # # SELECTION_1.0 / SelectionSet was removed in the unify-view-state-sync change. # Selection state (active clip + sequence/source view mode) is now carried by the # view-state fields ``view_mode`` and ``clip_guid`` on PLAYBACK_SETTINGS_1.0, so # there is a single authoritative view-state message instead of two channels. # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Annotation family — Annotation.1 (hot path) # --------------------------------------------------------------------------- @register @dataclass class PartialAnnotation(ProtocolMessage): """Mid-stroke partial annotation (visual preview, not persisted). Hot path: fires repeatedly while a stroke is being drawn (before pen-up). Peers render the transient stroke visually, but do not write it to the local OTIO timeline. On pen-up/completion, a full stroke is committed via :class:`InsertChild` instead. No validation or reflective serialization is performed. """ SCHEMA = "Annotation.1" EVENT = "PARTIAL" clip_guid: str = doc_field(doc="Sync GUID of the clip being annotated.") frame: float = doc_field(doc="0-indexed clip-local frame number.") fps: float = doc_field(doc="Frame rate used to interpret 'frame'.") # Deliberately kept as serialized SyncEvent dicts (NOT typed OTIO objects): # this is the hottest path and the host codec already produces dicts, so a # typed field would force wasteful deserialize-then-reserialize churn. events: list = doc_field( default_factory=list, doc="Serialized SyncEvent dicts for the in-progress stroke." ) def to_payload(self) -> dict[str, Any]: return { "clip_guid": self.clip_guid, "frame": self.frame, "fps": self.fps, "events": self.events, } @classmethod def from_payload(cls, data: dict[str, Any]) -> "PartialAnnotation": return cls( clip_guid=data.get("clip_guid"), frame=data.get("frame"), fps=data.get("fps"), events=data.get("events", []), ) # --------------------------------------------------------------------------- # OTIO session family — OTIO_SESSION_1.0 # Single definition: built and consumed by patcher.py. Payloads carry the # wire form (already-serialized child_data / commands), so to_payload is a cheap # field copy. # --------------------------------------------------------------------------- @register @dataclass class SetProperty(ProtocolMessage): """Sets a property or metadata path on an object.""" SCHEMA = "OTIO_SESSION_1.0" EVENT = "SET_PROPERTY" target_uuid: str = doc_field(doc="GUID of the target object.") path: str = doc_field(doc="Property name or 'metadata/...' sub-path.") value: Any = doc_field(doc="New primitive value.") sync_timestamp: "float | None" = doc_field( default=None, doc="Epoch seconds when the mutation occurred." ) def to_payload(self) -> dict[str, Any]: return { "target_uuid": self.target_uuid, "path": self.path, "value": self.value, "sync_timestamp": self.sync_timestamp, } @classmethod def from_payload(cls, data: dict[str, Any]) -> "SetProperty": return cls( target_uuid=data.get("target_uuid"), path=data.get("path"), value=data.get("value"), sync_timestamp=data.get("sync_timestamp"), ) @register @dataclass class InsertChild(ProtocolMessage): """Inserts a child object into a parent container.""" SCHEMA = "OTIO_SESSION_1.0" EVENT = "INSERT_CHILD" parent_uuid: str = doc_field(doc="GUID of the parent container.") child_data: Any = doc_field( doc="OTIO child object (object on send, wire dict on receive)." ) index: int = doc_field(default=-1, doc="Insert position; -1 appends.") sync_timestamp: "float | None" = doc_field( default=None, doc="Epoch seconds when the mutation occurred." ) def to_payload(self) -> dict[str, Any]: return { "parent_uuid": self.parent_uuid, "index": self.index, "child_data": _to_wire(self.child_data), "sync_timestamp": self.sync_timestamp, } @classmethod def from_payload(cls, data: dict[str, Any]) -> "InsertChild": return cls( parent_uuid=data.get("parent_uuid"), child_data=data.get("child_data"), index=data.get("index", -1), sync_timestamp=data.get("sync_timestamp"), ) def as_otio(self) -> Any: """Return the OTIO child object, deserializing if still in wire form.""" return _from_wire(self.child_data) @register @dataclass class MoveChild(ProtocolMessage): """Moves a child to a new index within its parent container.""" SCHEMA = "OTIO_SESSION_1.0" EVENT = "MOVE_CHILD" parent_uuid: str = doc_field(doc="GUID of the parent container.") child_uuid: str = doc_field(doc="GUID of the child to move.") to_index: int = doc_field(default=0, doc="Target position in the parent.") sync_timestamp: "float | None" = doc_field( default=None, doc="Epoch seconds when the mutation occurred." ) def to_payload(self) -> dict[str, Any]: return { "parent_uuid": self.parent_uuid, "child_uuid": self.child_uuid, "to_index": self.to_index, "sync_timestamp": self.sync_timestamp, } @classmethod def from_payload(cls, data: dict[str, Any]) -> "MoveChild": return cls( parent_uuid=data.get("parent_uuid"), child_uuid=data.get("child_uuid"), to_index=data.get("to_index", 0), sync_timestamp=data.get("sync_timestamp"), ) @register @dataclass class RemoveChild(ProtocolMessage): """Removes a child from its parent container.""" SCHEMA = "OTIO_SESSION_1.0" EVENT = "REMOVE_CHILD" parent_uuid: str = doc_field(doc="GUID of the parent container.") child_uuid: str = doc_field(doc="GUID of the child to remove.") sync_timestamp: "float | None" = doc_field( default=None, doc="Epoch seconds when the mutation occurred." ) def to_payload(self) -> dict[str, Any]: return { "parent_uuid": self.parent_uuid, "child_uuid": self.child_uuid, "sync_timestamp": self.sync_timestamp, } @classmethod def from_payload(cls, data: dict[str, Any]) -> "RemoveChild": return cls( parent_uuid=data.get("parent_uuid"), child_uuid=data.get("child_uuid"), sync_timestamp=data.get("sync_timestamp"), ) @register @dataclass class ReplaceAnnotationCommands(ProtocolMessage): """Replaces the full annotation-command list on an annotation clip. Used when modifying/updating existing committed annotations in-place (e.g., editing text/captions or dragging/moving them), rather than appending a delta or committing a new stroke. """ SCHEMA = "OTIO_SESSION_1.0" EVENT = "REPLACE_ANNOTATION_COMMANDS" annotation_clip_guid: str = doc_field(doc="GUID of the annotation clip to update.") commands: list = doc_field( default_factory=list, doc="Full replacement list of OTIO SyncEvents (objects on send, wire dicts on receive).", ) sync_timestamp: "float | None" = doc_field( default=None, doc="Epoch seconds when the mutation occurred." ) def to_payload(self) -> dict[str, Any]: return { "annotation_clip_guid": self.annotation_clip_guid, "commands": [_to_wire(c) for c in self.commands], "sync_timestamp": self.sync_timestamp, } @classmethod def from_payload(cls, data: dict[str, Any]) -> "ReplaceAnnotationCommands": return cls( annotation_clip_guid=data.get("annotation_clip_guid"), commands=data.get("commands", []), sync_timestamp=data.get("sync_timestamp"), ) def as_otio(self) -> list: """Return the OTIO SyncEvent list, deserializing any wire-form entries.""" return [_from_wire(c) for c in self.commands] # --------------------------------------------------------------------------- # Broadcast ownership family — BROADCAST_OWNERSHIP_1.0 # --------------------------------------------------------------------------- @register @dataclass class ClaimOwnership(ProtocolMessage): """A peer's claim to the write lease for one broadcast-ownership channel. Sent both when a peer claims a free channel and when it re-claims a channel it already holds (refreshing the lease). ``claim_ts`` is a wall clock reading, not a local monotonic one — it is compared against another peer's ``claim_ts`` for the deterministic tiebreak (earlier wins, lower ``peer_guid`` breaks an exact tie), which only works if every peer evaluates the same two values (design.md D2). Every peer, including the claimant itself, resolves this message through the same rule. """ SCHEMA = "BROADCAST_OWNERSHIP_1.0" EVENT = "CLAIM_OWNERSHIP" category: str = doc_field( doc='Lease channel being claimed: "position", "display", or "structure".' ) peer_guid: str = doc_field(doc="GUID of the claiming peer.") claim_ts: "float | None" = doc_field( default=None, doc="Wall-clock epoch seconds when the claim was made; drives the " "deterministic tiebreak between simultaneous claims.", ) def to_payload(self) -> dict[str, Any]: return { "category": self.category, "peer_guid": self.peer_guid, "claim_ts": self.claim_ts, } @classmethod def from_payload(cls, data: dict[str, Any]) -> "ClaimOwnership": return cls( category=data.get("category"), peer_guid=data.get("peer_guid"), claim_ts=data.get("claim_ts"), ) @register @dataclass class ReleaseOwnership(ProtocolMessage): """A peer's explicit release of a broadcast-ownership channel it holds. Frees the channel (or promotes a pending claimant) immediately rather than waiting for the lease to expire. Best-effort: a peer that disconnects without sending this is still handled — the lease simply expires through the ordinary silence-based path. """ SCHEMA = "BROADCAST_OWNERSHIP_1.0" EVENT = "RELEASE_OWNERSHIP" category: str = doc_field( doc='Lease channel being released: "position", "display", or "structure".' ) peer_guid: str = doc_field(doc="GUID of the releasing peer.") def to_payload(self) -> dict[str, Any]: return {"category": self.category, "peer_guid": self.peer_guid} @classmethod def from_payload(cls, data: dict[str, Any]) -> "ReleaseOwnership": return cls(category=data.get("category"), peer_guid=data.get("peer_guid"))