Module livekit.rtc.track_publication

Classes

class LocalTrackPublication (owned_info: track_pb2.OwnedTrackPublication)

Expand source code

class LocalTrackPublication(TrackPublication):
    def __init__(self, owned_info: proto_track.OwnedTrackPublication):
        super().__init__(owned_info)
        self._first_subscription: asyncio.Future[None] = asyncio.Future()

    @property
    def track(self) -> Optional[LocalTrack]:
        return cast(Optional[LocalTrack], self._track)

    async def wait_for_subscription(self) -> None:
        await asyncio.shield(self._first_subscription)

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

Ancestors

• TrackPublication

Instance variables

prop track : LocalVideoTrack | LocalAudioTrack | None

Expand source code

@property
def track(self) -> Optional[LocalTrack]:
    return cast(Optional[LocalTrack], self._track)

Methods

async def wait_for_subscription(self) ‑> None

Expand source code

async def wait_for_subscription(self) -> None:
    await asyncio.shield(self._first_subscription)

class RemoteTrackPublication (owned_info: track_pb2.OwnedTrackPublication)

Expand source code

class RemoteTrackPublication(TrackPublication):
    def __init__(self, owned_info: proto_track.OwnedTrackPublication):
        super().__init__(owned_info)
        self._subscribed = False

    @property
    def track(self) -> Optional[RemoteTrack]:
        return cast(Optional[RemoteTrack], self._track)

    @property
    def subscribed(self) -> bool:
        return self._subscribed

    def set_subscribed(self, subscribed: bool) -> None:
        req = proto_ffi.FfiRequest()
        req.set_subscribed.subscribe = subscribed
        req.set_subscribed.publication_handle = self._ffi_handle.handle
        FfiClient.instance.request(req)

    def set_video_quality(self, quality: "VideoQuality.ValueType") -> None:
        """For simulcasted video tracks, request a specific simulcast layer
        from the server. Use one of `rtc.VideoQuality.VIDEO_QUALITY_LOW` (q),
        `VIDEO_QUALITY_MEDIUM` (h), or `VIDEO_QUALITY_HIGH` (f).

        Raises:
            ValueError: if the publication is not a video track, or is not
                simulcasted (selecting a layer requires multiple layers).
        """
        if self.kind != proto_track.TrackKind.KIND_VIDEO:
            raise ValueError(
                f"set_video_quality is only supported on video publications (sid={self.sid})"
            )
        if not self.simulcasted:
            raise ValueError(
                f"set_video_quality requires a simulcasted publication (sid={self.sid})"
            )
        req = proto_ffi.FfiRequest()
        req.set_remote_track_publication_quality.track_publication_handle = self._ffi_handle.handle
        req.set_remote_track_publication_quality.quality = quality
        FfiClient.instance.request(req)

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

Ancestors

• TrackPublication

Instance variables

prop subscribed : bool

Expand source code

@property
def subscribed(self) -> bool:
    return self._subscribed

prop track : RemoteVideoTrack | RemoteAudioTrack | None

Expand source code

@property
def track(self) -> Optional[RemoteTrack]:
    return cast(Optional[RemoteTrack], self._track)

Methods

def set_subscribed(self, subscribed: bool) ‑> None

Expand source code

def set_subscribed(self, subscribed: bool) -> None:
    req = proto_ffi.FfiRequest()
    req.set_subscribed.subscribe = subscribed
    req.set_subscribed.publication_handle = self._ffi_handle.handle
    FfiClient.instance.request(req)

def set_video_quality(self, quality: VideoQuality.ValueType) ‑> None

Expand source code

def set_video_quality(self, quality: "VideoQuality.ValueType") -> None:
    """For simulcasted video tracks, request a specific simulcast layer
    from the server. Use one of `rtc.VideoQuality.VIDEO_QUALITY_LOW` (q),
    `VIDEO_QUALITY_MEDIUM` (h), or `VIDEO_QUALITY_HIGH` (f).

    Raises:
        ValueError: if the publication is not a video track, or is not
            simulcasted (selecting a layer requires multiple layers).
    """
    if self.kind != proto_track.TrackKind.KIND_VIDEO:
        raise ValueError(
            f"set_video_quality is only supported on video publications (sid={self.sid})"
        )
    if not self.simulcasted:
        raise ValueError(
            f"set_video_quality requires a simulcasted publication (sid={self.sid})"
        )
    req = proto_ffi.FfiRequest()
    req.set_remote_track_publication_quality.track_publication_handle = self._ffi_handle.handle
    req.set_remote_track_publication_quality.quality = quality
    FfiClient.instance.request(req)

For simulcasted video tracks, request a specific simulcast layer from the server. Use one of rtc.VideoQuality.VIDEO_QUALITY_LOW (q), VIDEO_QUALITY_MEDIUM (h), or VIDEO_QUALITY_HIGH (f).

Raises

ValueError

if the publication is not a video track, or is not simulcasted (selecting a layer requires multiple layers).

class TrackPublication (owned_info: track_pb2.OwnedTrackPublication)

Expand source code

class TrackPublication:
    def __init__(self, owned_info: proto_track.OwnedTrackPublication):
        self._info = owned_info.info
        self._track: Optional[Track] = None
        self._ffi_handle = FfiHandle(owned_info.handle.id)

    @property
    def track(self) -> Optional[Track]:
        return self._track

    @property
    def sid(self) -> str:
        return self._info.sid

    @property
    def name(self) -> str:
        return self._info.name

    @property
    def kind(self) -> proto_track.TrackKind.ValueType:
        return self._info.kind

    @property
    def source(self) -> proto_track.TrackSource.ValueType:
        return self._info.source

    @property
    def simulcasted(self) -> bool:
        return self._info.simulcasted

    @property
    def width(self) -> int:
        return self._info.width

    @property
    def height(self) -> int:
        return self._info.height

    @property
    def mime_type(self) -> str:
        return self._info.mime_type

    @property
    def muted(self) -> bool:
        return self._info.muted

    @property
    def encryption_type(self) -> proto_e2ee.EncryptionType.ValueType:
        return self._info.encryption_type

    @property
    def audio_features(self) -> List[proto_track.AudioTrackFeature.ValueType]:
        return list(self._info.audio_features)

    @property
    def frame_metadata_features(self) -> List[proto_track.FrameMetadataFeature.ValueType]:
        return list(self._info.frame_metadata_features)

Subclasses

• LocalTrackPublication
• RemoteTrackPublication

Instance variables

prop audio_features : List[int]

Expand source code

@property
def audio_features(self) -> List[proto_track.AudioTrackFeature.ValueType]:
    return list(self._info.audio_features)

prop encryption_type : int

Expand source code

@property
def encryption_type(self) -> proto_e2ee.EncryptionType.ValueType:
    return self._info.encryption_type

prop frame_metadata_features : List[int]

Expand source code

@property
def frame_metadata_features(self) -> List[proto_track.FrameMetadataFeature.ValueType]:
    return list(self._info.frame_metadata_features)

prop height : int

Expand source code

@property
def height(self) -> int:
    return self._info.height

prop kind : int

Expand source code

@property
def kind(self) -> proto_track.TrackKind.ValueType:
    return self._info.kind

prop mime_type : str

Expand source code

@property
def mime_type(self) -> str:
    return self._info.mime_type

prop muted : bool

Expand source code

@property
def muted(self) -> bool:
    return self._info.muted

prop name : str

Expand source code

@property
def name(self) -> str:
    return self._info.name

prop sid : str

Expand source code

@property
def sid(self) -> str:
    return self._info.sid

prop simulcasted : bool

Expand source code

@property
def simulcasted(self) -> bool:
    return self._info.simulcasted

prop source : int

Expand source code

@property
def source(self) -> proto_track.TrackSource.ValueType:
    return self._info.source

prop track : Track | None

Expand source code

@property
def track(self) -> Optional[Track]:
    return self._track

prop width : int

Expand source code

@property
def width(self) -> int:
    return self._info.width