Module livekit.rtc.data_track

Classes

class DataTrackFrame (payload: bytes, user_timestamp: Optional[int] = None)

Expand source code

@dataclass
class DataTrackFrame:
    """A frame published on a data track, consisting of a payload and optional metadata."""

    payload: bytes
    """The frame's payload."""

    user_timestamp: Optional[int] = None
    """The frame's user timestamp, if one is associated."""

A frame published on a data track, consisting of a payload and optional metadata.

Instance variables

var payload : bytes

The frame's payload.

var user_timestamp : int | None

The frame's user timestamp, if one is associated.

class DataTrackInfo (sid: str, name: str, uses_e2ee: bool)

Expand source code

@dataclass
class DataTrackInfo:
    """Information about a published data track."""

    sid: str
    """Unique track identifier assigned by the SFU.

    This identifier may change if a reconnect occurs. Use ``name`` if a
    stable identifier is needed.
    """

    name: str
    """Name of the track assigned by the publisher."""

    uses_e2ee: bool
    """Whether or not frames sent on the track use end-to-end encryption."""

Information about a published data track.

Instance variables

var name : str

Name of the track assigned by the publisher.

var sid : str

Unique track identifier assigned by the SFU.

This identifier may change if a reconnect occurs. Use name if a stable identifier is needed.

var uses_e2ee : bool

Whether or not frames sent on the track use end-to-end encryption.

class DataTrackStream (owned_info: proto_data_track.OwnedDataTrackStream)

Expand source code

class DataTrackStream:
    """An active subscription to a remote data track.

    Use as an async iterator to receive frames::

        stream = remote_track.subscribe()
        async for frame in stream:
            process(frame.payload)

    Dropping or closing the stream unsubscribes from the track.

    If subscribing to the track fails, :class:`SubscribeDataTrackError`
    is raised when iteration ends instead of a normal ``StopAsyncIteration``.
    """

    def __init__(self, owned_info: proto_data_track.OwnedDataTrackStream) -> None:
        self._ffi_handle = FfiHandle(owned_info.handle.id)
        handle_id = owned_info.handle.id

        self._queue = FfiClient.instance.queue.subscribe(
            filter_fn=lambda e: (
                e.WhichOneof("message") == "data_track_stream_event"
                and e.data_track_stream_event.stream_handle == handle_id
            ),
        )
        self._closed = False

    async def read(self) -> Optional[DataTrackFrame]:
        """Read a single frame, or ``None`` if the stream has ended."""
        try:
            return await self.__anext__()
        except StopAsyncIteration:
            return None

    def __aiter__(self) -> AsyncIterator[DataTrackFrame]:
        return self

    async def __anext__(self) -> DataTrackFrame:
        if self._closed:
            raise StopAsyncIteration

        self._send_read_request()
        event: proto_ffi.FfiEvent = await self._queue.get()
        stream_event = event.data_track_stream_event
        detail = stream_event.WhichOneof("detail")

        if detail == "frame_received":
            proto_frame = stream_event.frame_received.frame
            user_ts: Optional[int] = None
            if proto_frame.HasField("user_timestamp"):
                user_ts = proto_frame.user_timestamp
            return DataTrackFrame(
                payload=proto_frame.payload,
                user_timestamp=user_ts,
            )
        elif detail == "eos":
            self._close()
            if stream_event.eos.HasField("error"):
                raise SubscribeDataTrackError(stream_event.eos.error.message)
            raise StopAsyncIteration
        else:
            self._close()
            raise StopAsyncIteration

    def _send_read_request(self) -> None:
        req = proto_ffi.FfiRequest()
        req.data_track_stream_read.stream_handle = self._ffi_handle.handle
        FfiClient.instance.request(req)

    def _close(self) -> None:
        if not self._closed:
            self._closed = True
            FfiClient.instance.queue.unsubscribe(self._queue)

    def close(self) -> None:
        """Explicitly close the subscription and unsubscribe."""
        self._close()
        self._ffi_handle.dispose()

    async def aclose(self) -> None:
        self.close()

An active subscription to a remote data track.

Use as an async iterator to receive frames::

stream = remote_track.subscribe()
async for frame in stream:
    process(frame.payload)

Dropping or closing the stream unsubscribes from the track.

If subscribing to the track fails, :class:SubscribeDataTrackError is raised when iteration ends instead of a normal StopAsyncIteration.

Methods

async def aclose(self) ‑> None

Expand source code

async def aclose(self) -> None:
    self.close()

def close(self) ‑> None

Expand source code

def close(self) -> None:
    """Explicitly close the subscription and unsubscribe."""
    self._close()
    self._ffi_handle.dispose()

Explicitly close the subscription and unsubscribe.

async def read(self) ‑> DataTrackFrame | None

Expand source code

async def read(self) -> Optional[DataTrackFrame]:
    """Read a single frame, or ``None`` if the stream has ended."""
    try:
        return await self.__anext__()
    except StopAsyncIteration:
        return None

Read a single frame, or None if the stream has ended.

class LocalDataTrack (owned_info: proto_data_track.OwnedLocalDataTrack)

Expand source code

class LocalDataTrack:
    """Data track published by the local participant."""

    def __init__(self, owned_info: proto_data_track.OwnedLocalDataTrack) -> None:
        self._info = DataTrackInfo(
            sid=owned_info.info.sid,
            name=owned_info.info.name,
            uses_e2ee=owned_info.info.uses_e2ee,
        )
        self._ffi_handle = FfiHandle(owned_info.handle.id)

    @property
    def info(self) -> DataTrackInfo:
        """Information about the data track."""
        return self._info

    def try_push(self, frame: DataTrackFrame) -> None:
        """Try pushing a frame to subscribers of the track.

        See :class:`DataTrackFrame` for how to construct a frame and attach metadata.

        Args:
            frame: The data track frame to send.

        Raises:
            PushFrameError: If the push fails.
        """
        proto_frame = proto_data_track.DataTrackFrame(payload=bytes(frame.payload))
        if frame.user_timestamp is not None:
            proto_frame.user_timestamp = frame.user_timestamp

        req = proto_ffi.FfiRequest()
        req.local_data_track_try_push.track_handle = self._ffi_handle.handle
        req.local_data_track_try_push.frame.CopyFrom(proto_frame)

        resp = FfiClient.instance.request(req)
        if resp.local_data_track_try_push.HasField("error"):
            raise PushFrameError(resp.local_data_track_try_push.error.message)

    def is_published(self) -> bool:
        """Whether or not the track is still published."""
        req = proto_ffi.FfiRequest()
        req.local_data_track_is_published.track_handle = self._ffi_handle.handle

        resp = FfiClient.instance.request(req)
        return bool(resp.local_data_track_is_published.is_published)

    async def unpublish(self) -> None:
        """Unpublishes the track."""
        req = proto_ffi.FfiRequest()
        req.local_data_track_unpublish.track_handle = self._ffi_handle.handle
        FfiClient.instance.request(req)

    def __repr__(self) -> str:
        return f"rtc.LocalDataTrack(sid={self._info.sid}, name={self._info.name})"

Data track published by the local participant.

Instance variables

prop info : DataTrackInfo

Expand source code

@property
def info(self) -> DataTrackInfo:
    """Information about the data track."""
    return self._info

Information about the data track.

Methods

def is_published(self) ‑> bool

Expand source code

def is_published(self) -> bool:
    """Whether or not the track is still published."""
    req = proto_ffi.FfiRequest()
    req.local_data_track_is_published.track_handle = self._ffi_handle.handle

    resp = FfiClient.instance.request(req)
    return bool(resp.local_data_track_is_published.is_published)

Whether or not the track is still published.

def try_push(self,
frame: DataTrackFrame) ‑> None

Expand source code

def try_push(self, frame: DataTrackFrame) -> None:
    """Try pushing a frame to subscribers of the track.

    See :class:`DataTrackFrame` for how to construct a frame and attach metadata.

    Args:
        frame: The data track frame to send.

    Raises:
        PushFrameError: If the push fails.
    """
    proto_frame = proto_data_track.DataTrackFrame(payload=bytes(frame.payload))
    if frame.user_timestamp is not None:
        proto_frame.user_timestamp = frame.user_timestamp

    req = proto_ffi.FfiRequest()
    req.local_data_track_try_push.track_handle = self._ffi_handle.handle
    req.local_data_track_try_push.frame.CopyFrom(proto_frame)

    resp = FfiClient.instance.request(req)
    if resp.local_data_track_try_push.HasField("error"):
        raise PushFrameError(resp.local_data_track_try_push.error.message)

Try pushing a frame to subscribers of the track.

See :class:DataTrackFrame for how to construct a frame and attach metadata.

Args

frame

The data track frame to send.

Raises

PushFrameError

If the push fails.

async def unpublish(self) ‑> None

Expand source code

async def unpublish(self) -> None:
    """Unpublishes the track."""
    req = proto_ffi.FfiRequest()
    req.local_data_track_unpublish.track_handle = self._ffi_handle.handle
    FfiClient.instance.request(req)

Unpublishes the track.

class PushFrameError (message: str)

Expand source code

class PushFrameError(Exception):
    """Frame could not be pushed to a data track.

    Pushing a frame can fail for several reasons:

    - The track has been unpublished by the local participant or SFU
    - The room is no longer connected
    - Frames are being pushed too fast
    """

    def __init__(self, message: str) -> None:
        self.message = message

Frame could not be pushed to a data track.

Pushing a frame can fail for several reasons:

• The track has been unpublished by the local participant or SFU
• The room is no longer connected
• Frames are being pushed too fast

Ancestors

• builtins.Exception
• builtins.BaseException

class RemoteDataTrack (owned_info: proto_data_track.OwnedRemoteDataTrack)

Expand source code

class RemoteDataTrack:
    """Data track published by a remote participant."""

    def __init__(self, owned_info: proto_data_track.OwnedRemoteDataTrack) -> None:
        self._info = DataTrackInfo(
            sid=owned_info.info.sid,
            name=owned_info.info.name,
            uses_e2ee=owned_info.info.uses_e2ee,
        )
        self._ffi_handle = FfiHandle(owned_info.handle.id)
        self._publisher_identity = owned_info.publisher_identity

    @property
    def info(self) -> DataTrackInfo:
        """Information about the data track."""
        return self._info

    @property
    def publisher_identity(self) -> str:
        """Identity of the participant who published the track."""
        return self._publisher_identity

    def subscribe(self, *, buffer_size: Optional[int] = None) -> DataTrackStream:
        """Subscribes to the data track to receive frames.

        Args:
            buffer_size: Maximum number of received frames to buffer internally.
                When ``None``, the default buffer size is used.
                Zero is not a valid buffer size; if a value of zero is provided, it will be clamped to one.

        Returns a :class:`DataTrackStream` that yields
        :class:`DataTrackFrame` instances as they arrive. If the
        subscription encounters an error, it is raised as
        :class:`SubscribeDataTrackError` when iteration ends.
        """
        opts = proto_data_track.DataTrackSubscribeOptions()
        if buffer_size is not None:
            opts.buffer_size = buffer_size

        req = proto_ffi.FfiRequest()
        req.subscribe_data_track.track_handle = self._ffi_handle.handle
        req.subscribe_data_track.options.CopyFrom(opts)

        resp = FfiClient.instance.request(req)
        return DataTrackStream(resp.subscribe_data_track.stream)

    def is_published(self) -> bool:
        """Whether or not the track is still published."""
        req = proto_ffi.FfiRequest()
        req.remote_data_track_is_published.track_handle = self._ffi_handle.handle

        resp = FfiClient.instance.request(req)
        return bool(resp.remote_data_track_is_published.is_published)

    def set_pipeline_options(self, *, max_partial_frames: Optional[int] = None) -> None:
        """Configures options for the pipeline handling incoming packets for this track.

        These options apply to all current and future subscriptions of this track, and
        may be set at any time. New options take effect with the next received packet.

        Args:
            max_partial_frames: Maximum number of partial frames the depacketizer will
                track concurrently for this track. Defaults to 1. Higher values give more
                out-of-order tolerance for high-frequency senders at the cost of additional
                buffering. Zero is not a valid value; if a value of zero is provided, it
                will be clamped to one. When ``None``, the current value is kept.
        """
        opts = proto_data_track.RemoteDataTrackPipelineOptions()
        if max_partial_frames is not None:
            opts.max_partial_frames = max_partial_frames

        req = proto_ffi.FfiRequest()
        req.remote_data_track_set_pipeline_options.track_handle = self._ffi_handle.handle
        req.remote_data_track_set_pipeline_options.options.CopyFrom(opts)

        FfiClient.instance.request(req)

    def __repr__(self) -> str:
        return (
            f"rtc.RemoteDataTrack(sid={self._info.sid}, name={self._info.name}, "
            f"publisher_identity={self._publisher_identity})"
        )

Data track published by a remote participant.

Instance variables

prop info : DataTrackInfo

Expand source code

@property
def info(self) -> DataTrackInfo:
    """Information about the data track."""
    return self._info

Information about the data track.

prop publisher_identity : str

Expand source code

@property
def publisher_identity(self) -> str:
    """Identity of the participant who published the track."""
    return self._publisher_identity

Identity of the participant who published the track.

Methods

def is_published(self) ‑> bool

Expand source code

def is_published(self) -> bool:
    """Whether or not the track is still published."""
    req = proto_ffi.FfiRequest()
    req.remote_data_track_is_published.track_handle = self._ffi_handle.handle

    resp = FfiClient.instance.request(req)
    return bool(resp.remote_data_track_is_published.is_published)

Whether or not the track is still published.

def set_pipeline_options(self, *, max_partial_frames: Optional[int] = None) ‑> None

Expand source code

def set_pipeline_options(self, *, max_partial_frames: Optional[int] = None) -> None:
    """Configures options for the pipeline handling incoming packets for this track.

    These options apply to all current and future subscriptions of this track, and
    may be set at any time. New options take effect with the next received packet.

    Args:
        max_partial_frames: Maximum number of partial frames the depacketizer will
            track concurrently for this track. Defaults to 1. Higher values give more
            out-of-order tolerance for high-frequency senders at the cost of additional
            buffering. Zero is not a valid value; if a value of zero is provided, it
            will be clamped to one. When ``None``, the current value is kept.
    """
    opts = proto_data_track.RemoteDataTrackPipelineOptions()
    if max_partial_frames is not None:
        opts.max_partial_frames = max_partial_frames

    req = proto_ffi.FfiRequest()
    req.remote_data_track_set_pipeline_options.track_handle = self._ffi_handle.handle
    req.remote_data_track_set_pipeline_options.options.CopyFrom(opts)

    FfiClient.instance.request(req)

Configures options for the pipeline handling incoming packets for this track.

These options apply to all current and future subscriptions of this track, and may be set at any time. New options take effect with the next received packet.

Args

max_partial_frames

Maximum number of partial frames the depacketizer will track concurrently for this track. Defaults to 1. Higher values give more out-of-order tolerance for high-frequency senders at the cost of additional buffering. Zero is not a valid value; if a value of zero is provided, it will be clamped to one. When None, the current value is kept.

def subscribe(self, *, buffer_size: Optional[int] = None) ‑> DataTrackStream

Expand source code

def subscribe(self, *, buffer_size: Optional[int] = None) -> DataTrackStream:
    """Subscribes to the data track to receive frames.

    Args:
        buffer_size: Maximum number of received frames to buffer internally.
            When ``None``, the default buffer size is used.
            Zero is not a valid buffer size; if a value of zero is provided, it will be clamped to one.

    Returns a :class:`DataTrackStream` that yields
    :class:`DataTrackFrame` instances as they arrive. If the
    subscription encounters an error, it is raised as
    :class:`SubscribeDataTrackError` when iteration ends.
    """
    opts = proto_data_track.DataTrackSubscribeOptions()
    if buffer_size is not None:
        opts.buffer_size = buffer_size

    req = proto_ffi.FfiRequest()
    req.subscribe_data_track.track_handle = self._ffi_handle.handle
    req.subscribe_data_track.options.CopyFrom(opts)

    resp = FfiClient.instance.request(req)
    return DataTrackStream(resp.subscribe_data_track.stream)

Subscribes to the data track to receive frames.

Args

buffer_size

Maximum number of received frames to buffer internally. When None, the default buffer size is used. Zero is not a valid buffer size; if a value of zero is provided, it will be clamped to one.

Returns a :class:DataTrackStream that yields :class:DataTrackFrame instances as they arrive. If the subscription encounters an error, it is raised as :class:SubscribeDataTrackError when iteration ends.

class SubscribeDataTrackError (message: str)

Expand source code

class SubscribeDataTrackError(Exception):
    """An error that can occur when subscribing to a data track."""

    def __init__(self, message: str) -> None:
        self.message = message

An error that can occur when subscribing to a data track.

Ancestors

• builtins.Exception
• builtins.BaseException