DEV Community

Timo Alexander Meer
Timo Alexander Meer

Posted on

Recording the voice using livekit egress

I'm trying to record the voice using livekit but I can't because the TwirpError is occurred consistently.

Here's the recording voice function.

async def start_recording(self, meeting_id: str) -> Optional[str]:
        """
        Start recording a LiveKit room (audio-only)
        Returns the egress ID if successful, None otherwise
        """
        try:                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        
            logger.info(f"Starting recording for meeting {meeting_id}")

            room_name = self.get_room_name(meeting_id)
            logger.debug(f"Room name: {room_name}")


            segment_output = api.SegmentedFileOutput(
                filename_prefix=f"recordings/{meeting_id}",
                playlist_name=f"{meeting_id}.m3u8",
                live_playlist_name=f"{meeting_id}_live.m3u8",
                segment_duration=5,  # 5 second segments for audio
                s3=api.S3Upload(
                    bucket=settings.S3_BUCKET or "",  # Use empty string if not configured
                    region=settings.S3_REGION or "",
                    access_key=settings.S3_ACCESS_KEY or "",
                    secret=settings.S3_SECRET_KEY or "",
                    force_path_style=True,  # Use path-style URLs (required for some S3-compatible services)
                )
            )

            req = api.RoomCompositeEgressRequest(
                room_name=room_name,
                layout="speaker",
                preset=api.EncodingOptionsPreset.H264_720P_30,
                audio_only=True,
                segment_outputs=[segment_output],
            )

            req.segments.CopyFrom(segment_output)

            logger.info(f"RoomCompositeEgressRequest created for room: {room_name}")

            # Start the recording
            logger.info("Calling LiveKit API to start recording")

            egress_info = await self.lk_api.egress.start_room_composite_egress(req)

            if egress_info and egress_info.egress_id:
                logger.info(f"Recording started successfully. Egress ID: {egress_info.egress_id}, Room: {room_name}")
                return egress_info.egress_id
            else:
                logger.warning("Failed to start recording: No egress ID returned")
                return None

        except Exception as e:
            logger.error(f"Error starting LiveKit recording for meeting {meeting_id}: {e}", exc_info=True)
            return None

Enter fullscreen mode Exit fullscreen mode

If anyone who know about this error, please help me fix this

Top comments (0)