how can Livekit transport support video?

[ZaymeShaw]

Is it only daily transport support video currently?

[markbackman]

We're happy to add better support for LiveKit, but we also need some helping hands; ideally, someone with LiveKit expertise. Is anyone available to help out?

[xianqiliu]

@ZaymeShaw
i try to publish video to livekit as the following way and it works for me

# src/pipecat/transports/services/livekit.py

    async def process_frame(self, frame: Frame, direction: FrameDirection):
        """ 重写 livekit 的 process_frame 方法用于处理要发布到 livekit 的视频帧 """
        await super().process_frame(frame, direction)

        if isinstance(frame, OutputImageRawFrame):
            if frame.format == "RGB":
                # 直接使用 RGB 数据
                livekit_video = self._convert_image_to_livekit(frame.image, frame.size)
                await self._client.publish_video(livekit_video) # push video to livekit
        else:
            # 其他类型的帧...
            pass

    def _convert_image_to_livekit(self, image_data: bytes, size: Tuple[int, int]) -> rtc.VideoFrame:
        """将 OutputImageRawFrame 中的图像数据转换为 LiveKit VideoFrame"""
        livekit_frame = rtc.VideoFrame(
            width=size[0],
            height=size[1],
            type=rtc.VideoBufferType.RGB24,
            data=image_data
        )

        return livekit_frame

[xianqiliu]

@ZaymeShaw
yesterday i try to subscribe video from livekit as the following way and it works for me as well, but my purpose is only getting the latest image, it would be great if you have any better idea :)

# src/pipecat/transports/services/livekit.py
    async def _async_on_track_subscribed(
        self,
        track: rtc.Track,
        publication: rtc.RemoteTrackPublication,
        participant: rtc.RemoteParticipant,
    ):
        if track.kind == rtc.TrackKind.KIND_AUDIO:
            logger.info(f"Audio track subscribed: {track.sid} from participant {participant.sid}")
            self._audio_tracks[participant.sid] = track
            audio_stream = rtc.AudioStream(track)
            self._task_manager.create_task(
                self._process_audio_stream(audio_stream, participant.sid),
                f"{self}::_process_audio_stream",
            )
        elif track.kind == rtc.TrackKind.KIND_VIDEO:
            # 视频拉流
            logger.info(f"Video track subscribed: {track.sid} from participant {participant.sid}")
            self._video_tracks[participant.sid] = track
            video_stream = rtc.VideoStream(track)
            self._task_manager.create_task(
                self._process_video_stream(video_stream, participant.sid),
                f"{self}::_process_video_stream",
            )

    async def _process_video_stream(self, video_stream: rtc.VideoStream, participant_id: str):
        """视频处理，获取最新一帧"""
        logger.info(f"Started processing video stream for participant {participant_id}")
        async for event in video_stream:
            if isinstance(event, rtc.VideoFrameEvent):
                self._latest_image = event.frame
                await asyncio.sleep(1)
            else:
                logger.warning(f"Received unexpected event type: {type(event)}")