|
| 1 | +"""Egress manager for recording dual-channel audio to S3.""" |
| 2 | + |
| 3 | +import os |
| 4 | + |
| 5 | +from livekit import api |
| 6 | +from livekit.protocol import egress as egress_proto |
| 7 | +from loguru import logger |
| 8 | + |
| 9 | + |
| 10 | +class EgressConfig: |
| 11 | + """Configuration for egress recordings.""" |
| 12 | + |
| 13 | + def __init__( |
| 14 | + self, |
| 15 | + s3_bucket: str, |
| 16 | + s3_prefix: str = "", |
| 17 | + aws_access_key: str | None = None, |
| 18 | + aws_secret_key: str | None = None, |
| 19 | + aws_region: str | None = None, |
| 20 | + livekit_url: str | None = None, |
| 21 | + livekit_api_key: str | None = None, |
| 22 | + livekit_api_secret: str | None = None, |
| 23 | + ): |
| 24 | + """Initialize egress configuration. |
| 25 | +
|
| 26 | + Args: |
| 27 | + s3_bucket: S3 bucket name for recordings |
| 28 | + s3_prefix: Prefix/path within the bucket |
| 29 | + aws_access_key: AWS access key (defaults to env var) |
| 30 | + aws_secret_key: AWS secret key (defaults to env var) |
| 31 | + aws_region: AWS region (defaults to env var or us-east-1) |
| 32 | + livekit_url: LiveKit server URL (defaults to env var) |
| 33 | + livekit_api_key: LiveKit API key (defaults to env var) |
| 34 | + livekit_api_secret: LiveKit API secret (defaults to env var) |
| 35 | + """ |
| 36 | + self.s3_bucket = s3_bucket |
| 37 | + self.s3_prefix = s3_prefix.rstrip("/") |
| 38 | + |
| 39 | + # AWS credentials |
| 40 | + self.aws_access_key = aws_access_key or os.environ.get("AWS_ACCESS_KEY_ID", "") |
| 41 | + self.aws_secret_key = aws_secret_key or os.environ.get( |
| 42 | + "AWS_SECRET_ACCESS_KEY", "" |
| 43 | + ) |
| 44 | + self.aws_region = aws_region or os.environ.get("AWS_REGION", "us-east-1") |
| 45 | + |
| 46 | + # LiveKit credentials |
| 47 | + self.livekit_url = livekit_url or os.environ.get("LIVEKIT_URL", "") |
| 48 | + self.livekit_api_key = livekit_api_key or os.environ.get("LIVEKIT_API_KEY", "") |
| 49 | + self.livekit_api_secret = livekit_api_secret or os.environ.get( |
| 50 | + "LIVEKIT_API_SECRET", "" |
| 51 | + ) |
| 52 | + |
| 53 | + |
| 54 | +class EgressManager: |
| 55 | + """Manages LiveKit egress for dual-channel audio recording to S3.""" |
| 56 | + |
| 57 | + def __init__(self, config: EgressConfig): |
| 58 | + """Initialize the egress manager. |
| 59 | +
|
| 60 | + Args: |
| 61 | + config: Egress configuration |
| 62 | + """ |
| 63 | + self.config = config |
| 64 | + self._api: api.LiveKitAPI | None = None |
| 65 | + self._egress_id: str | None = None |
| 66 | + |
| 67 | + @property |
| 68 | + def livekit_api(self) -> api.LiveKitAPI: |
| 69 | + """Lazily initialize LiveKit API client.""" |
| 70 | + if self._api is None: |
| 71 | + self._api = api.LiveKitAPI( |
| 72 | + url=self.config.livekit_url, |
| 73 | + api_key=self.config.livekit_api_key, |
| 74 | + api_secret=self.config.livekit_api_secret, |
| 75 | + ) |
| 76 | + return self._api |
| 77 | + |
| 78 | + @property |
| 79 | + def egress_id(self) -> str | None: |
| 80 | + """Get the current egress ID if recording is active.""" |
| 81 | + return self._egress_id |
| 82 | + |
| 83 | + def _create_s3_upload(self) -> egress_proto.S3Upload: |
| 84 | + """Create S3 upload configuration.""" |
| 85 | + return egress_proto.S3Upload( |
| 86 | + access_key=self.config.aws_access_key, |
| 87 | + secret=self.config.aws_secret_key, |
| 88 | + bucket=self.config.s3_bucket, |
| 89 | + region=self.config.aws_region, |
| 90 | + ) |
| 91 | + |
| 92 | + async def start_dual_channel_recording(self, room_name: str) -> str | None: |
| 93 | + """Start dual-channel audio recording for a room. |
| 94 | +
|
| 95 | + The agent's audio will be on one channel, and all other participants |
| 96 | + (users) will be on the other channel. |
| 97 | +
|
| 98 | + Args: |
| 99 | + room_name: Name of the LiveKit room to record |
| 100 | +
|
| 101 | + Returns: |
| 102 | + Egress ID if started successfully, None on failure |
| 103 | + """ |
| 104 | + if self._egress_id: |
| 105 | + logger.warning( |
| 106 | + f"Egress already active with ID {self._egress_id}, skipping start" |
| 107 | + ) |
| 108 | + return self._egress_id |
| 109 | + |
| 110 | + try: |
| 111 | + s3_upload = self._create_s3_upload() |
| 112 | + |
| 113 | + # Build the filepath with prefix |
| 114 | + filepath_prefix = ( |
| 115 | + f"{self.config.s3_prefix}/audio" if self.config.s3_prefix else "audio" |
| 116 | + ) |
| 117 | + filepath = f"{filepath_prefix}/{{room_name}}-{{time}}.ogg" |
| 118 | + |
| 119 | + file_output = egress_proto.EncodedFileOutput( |
| 120 | + filepath=filepath, |
| 121 | + s3=s3_upload, |
| 122 | + ) |
| 123 | + |
| 124 | + # Start room composite egress with dual-channel audio |
| 125 | + # DUAL_CHANNEL_AGENT puts agent audio on one channel, all other participants on the other |
| 126 | + info = await self.livekit_api.egress.start_room_composite_egress( |
| 127 | + egress_proto.RoomCompositeEgressRequest( |
| 128 | + room_name=room_name, |
| 129 | + audio_only=True, |
| 130 | + audio_mixing=egress_proto.AudioMixing.DUAL_CHANNEL_AGENT, |
| 131 | + file_outputs=[file_output], |
| 132 | + ) |
| 133 | + ) |
| 134 | + |
| 135 | + self._egress_id = info.egress_id |
| 136 | + logger.info( |
| 137 | + f"Started dual-channel egress recording for room {room_name}, " |
| 138 | + f"egress_id={self._egress_id}" |
| 139 | + ) |
| 140 | + return self._egress_id |
| 141 | + |
| 142 | + except Exception as e: |
| 143 | + logger.error(f"Failed to start egress recording: {e}") |
| 144 | + return None |
| 145 | + |
| 146 | + async def stop_recording(self) -> bool: |
| 147 | + """Stop the active egress recording. |
| 148 | +
|
| 149 | + Returns: |
| 150 | + True if stopped successfully or no active recording, False on error |
| 151 | + """ |
| 152 | + if not self._egress_id: |
| 153 | + logger.debug("No active egress to stop") |
| 154 | + return True |
| 155 | + |
| 156 | + try: |
| 157 | + await self.livekit_api.egress.stop_egress( |
| 158 | + egress_proto.StopEgressRequest(egress_id=self._egress_id) |
| 159 | + ) |
| 160 | + logger.info(f"Stopped egress recording, egress_id={self._egress_id}") |
| 161 | + self._egress_id = None |
| 162 | + return True |
| 163 | + except Exception as e: |
| 164 | + logger.error(f"Failed to stop egress recording: {e}") |
| 165 | + return False |
| 166 | + |
| 167 | + async def close(self) -> None: |
| 168 | + """Clean up resources.""" |
| 169 | + if self._api: |
| 170 | + await self._api.aclose() |
| 171 | + self._api = None |
| 172 | + |
| 173 | + |
| 174 | +def create_default_egress_manager() -> EgressManager: |
| 175 | + """Create an egress manager with default configuration for the target S3 bucket. |
| 176 | +
|
| 177 | + Returns: |
| 178 | + Configured EgressManager instance |
| 179 | + """ |
| 180 | + config = EgressConfig( |
| 181 | + s3_bucket="audivi-audio-recordings", |
| 182 | + s3_prefix="livekit-demos", |
| 183 | + ) |
| 184 | + return EgressManager(config) |
0 commit comments