Module livekit.rtc.track

Classes

class LocalAudioTrack (info: track_pb2.OwnedTrack)

Expand source code

class LocalAudioTrack(Track):
    def __init__(self, info: proto_track.OwnedTrack):
        super().__init__(info)

    @staticmethod
    def create_audio_track(name: str, source: "AudioSource") -> "LocalAudioTrack":
        req = proto_ffi.FfiRequest()
        req.create_audio_track.name = name
        req.create_audio_track.source_handle = source._ffi_handle.handle

        resp = FfiClient.instance.request(req)
        return LocalAudioTrack(resp.create_audio_track.track)

    def mute(self):
        req = proto_ffi.FfiRequest()
        req.local_track_mute.track_handle = self._ffi_handle.handle
        req.local_track_mute.mute = True
        FfiClient.instance.request(req)
        self._info.muted = True

    def unmute(self):
        req = proto_ffi.FfiRequest()
        req.local_track_mute.track_handle = self._ffi_handle.handle
        req.local_track_mute.mute = False
        FfiClient.instance.request(req)
        self._info.muted = False

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

Ancestors

• Track

Static methods

def create_audio_track(name: str, source: AudioSource) ‑> LocalAudioTrack

Expand source code

@staticmethod
def create_audio_track(name: str, source: "AudioSource") -> "LocalAudioTrack":
    req = proto_ffi.FfiRequest()
    req.create_audio_track.name = name
    req.create_audio_track.source_handle = source._ffi_handle.handle

    resp = FfiClient.instance.request(req)
    return LocalAudioTrack(resp.create_audio_track.track)

Methods

def mute(self)

Expand source code

def mute(self):
    req = proto_ffi.FfiRequest()
    req.local_track_mute.track_handle = self._ffi_handle.handle
    req.local_track_mute.mute = True
    FfiClient.instance.request(req)
    self._info.muted = True

def unmute(self)

Expand source code

def unmute(self):
    req = proto_ffi.FfiRequest()
    req.local_track_mute.track_handle = self._ffi_handle.handle
    req.local_track_mute.mute = False
    FfiClient.instance.request(req)
    self._info.muted = False

class LocalVideoTrack (info: track_pb2.OwnedTrack)

Expand source code

class LocalVideoTrack(Track):
    def __init__(self, info: proto_track.OwnedTrack):
        super().__init__(info)

    @staticmethod
    def create_video_track(name: str, source: "VideoSource") -> "LocalVideoTrack":
        req = proto_ffi.FfiRequest()
        req.create_video_track.name = name
        req.create_video_track.source_handle = source._ffi_handle.handle

        resp = FfiClient.instance.request(req)
        return LocalVideoTrack(resp.create_video_track.track)

    def mute(self):
        req = proto_ffi.FfiRequest()
        req.local_track_mute.track_handle = self._ffi_handle.handle
        req.local_track_mute.mute = True
        FfiClient.instance.request(req)
        self._info.muted = True

    def unmute(self):
        req = proto_ffi.FfiRequest()
        req.local_track_mute.track_handle = self._ffi_handle.handle
        req.local_track_mute.mute = False
        FfiClient.instance.request(req)
        self._info.muted = False

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

Ancestors

• Track

Static methods

def create_video_track(name: str, source: VideoSource) ‑> LocalVideoTrack

Expand source code

@staticmethod
def create_video_track(name: str, source: "VideoSource") -> "LocalVideoTrack":
    req = proto_ffi.FfiRequest()
    req.create_video_track.name = name
    req.create_video_track.source_handle = source._ffi_handle.handle

    resp = FfiClient.instance.request(req)
    return LocalVideoTrack(resp.create_video_track.track)

Methods

def mute(self)

Expand source code

def mute(self):
    req = proto_ffi.FfiRequest()
    req.local_track_mute.track_handle = self._ffi_handle.handle
    req.local_track_mute.mute = True
    FfiClient.instance.request(req)
    self._info.muted = True

def unmute(self)

Expand source code

def unmute(self):
    req = proto_ffi.FfiRequest()
    req.local_track_mute.track_handle = self._ffi_handle.handle
    req.local_track_mute.mute = False
    FfiClient.instance.request(req)
    self._info.muted = False

class RemoteAudioTrack (info: track_pb2.OwnedTrack)

Expand source code

class RemoteAudioTrack(Track):
    def __init__(self, info: proto_track.OwnedTrack):
        super().__init__(info)

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

Ancestors

• Track

class RemoteVideoTrack (info: track_pb2.OwnedTrack)

Expand source code

class RemoteVideoTrack(Track):
    def __init__(self, info: proto_track.OwnedTrack):
        super().__init__(info)

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

Ancestors

• Track

class Track (owned_info: track_pb2.OwnedTrack)

Expand source code

class Track:
    def __init__(self, owned_info: proto_track.OwnedTrack):
        self._info = owned_info.info
        self._ffi_handle = FfiHandle(owned_info.handle.id)

    @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 stream_state(self) -> proto_track.StreamState.ValueType:
        return self._info.stream_state

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

    async def get_stats(self) -> List[proto_stats.RtcStats]:
        req = proto_ffi.FfiRequest()
        req.get_stats.track_handle = self._ffi_handle.handle

        queue = FfiClient.instance.queue.subscribe()
        try:
            resp = FfiClient.instance.request(req)
            cb: proto_ffi.FfiEvent = await queue.wait_for(
                lambda e: e.get_stats.async_id == resp.get_stats.async_id
            )
        finally:
            FfiClient.instance.queue.unsubscribe(queue)

        if cb.get_stats.error:
            raise Exception(cb.get_stats.error)

        return list(cb.get_stats.stats)

Subclasses

• LocalAudioTrack
• LocalVideoTrack
• RemoteAudioTrack
• RemoteVideoTrack

Instance variables

prop kind : int

Expand source code

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

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 stream_state : int

Expand source code

@property
def stream_state(self) -> proto_track.StreamState.ValueType:
    return self._info.stream_state

Methods

async def get_stats(self) ‑> List[stats_pb2.RtcStats]

Expand source code

async def get_stats(self) -> List[proto_stats.RtcStats]:
    req = proto_ffi.FfiRequest()
    req.get_stats.track_handle = self._ffi_handle.handle

    queue = FfiClient.instance.queue.subscribe()
    try:
        resp = FfiClient.instance.request(req)
        cb: proto_ffi.FfiEvent = await queue.wait_for(
            lambda e: e.get_stats.async_id == resp.get_stats.async_id
        )
    finally:
        FfiClient.instance.queue.unsubscribe(queue)

    if cb.get_stats.error:
        raise Exception(cb.get_stats.error)

    return list(cb.get_stats.stats)