E2E Test: select video quality when publish simulcast video track.

livekit-rtc/livekit/rtc/__init__.py

@@ -45,6 +45,7 @@
     ParticipantTrackPermission,
 )
 from ._proto.video_frame_pb2 import FrameMetadata, VideoBufferType, VideoCodec, VideoRotation
+from ._proto.track_publication_pb2 import VideoQuality
 from .audio_frame import AudioFrame
 from .audio_source import AudioSource
 from .audio_stream import AudioFrameEvent, AudioStream, NoiseCancellationOptions
@@ -188,6 +189,7 @@
     "TranscriptionSegment",
     "VideoCodec",
     "VideoEncoding",
+    "VideoQuality",
     "VideoFrame",
     "VideoFrameEvent",
     "VideoSource",

livekit-rtc/livekit/rtc/track_publication.py

@@ -19,8 +19,11 @@
 from ._proto import e2ee_pb2 as proto_e2ee
 from ._proto import ffi_pb2 as proto_ffi
 from ._proto import track_pb2 as proto_track
+from ._proto import track_publication_pb2 as proto_track_pub
 from .track import Track, LocalTrack, RemoteTrack
 
+VideoQuality = proto_track_pub.VideoQuality
+
 
 class TrackPublication:
     def __init__(self, owned_info: proto_track.OwnedTrackPublication):
@@ -116,5 +119,27 @@ def set_subscribed(self, subscribed: bool):
         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})"