ruff format

Author: chenosaurus

examples/local-audio/full_duplex.py

@@ -11,14 +11,14 @@ async def main() -> None:
 
     # Load environment variables from a .env file if present
     load_dotenv(find_dotenv())
-    
+
     url = os.getenv("LIVEKIT_URL")
     token = os.getenv("LIVEKIT_TOKEN")
     if not url or not token:
         raise RuntimeError("LIVEKIT_URL and LIVEKIT_TOKEN must be set in env")
 
     room = rtc.Room()
-    
+
     devices = rtc.MediaDevices()
 
     # Open microphone with AEC and prepare a player for remote audio feeding AEC reverse stream
@@ -32,7 +32,9 @@ async def main() -> None:
     streams_by_pub: dict[str, rtc.AudioStream] = {}
     streams_by_participant: dict[str, set[rtc.AudioStream]] = {}
 
-    async def _remove_stream(stream: rtc.AudioStream, participant_sid: str | None = None, pub_sid: str | None = None) -> None:
+    async def _remove_stream(
+        stream: rtc.AudioStream, participant_sid: str | None = None, pub_sid: str | None = None
+    ) -> None:
         try:
             mixer.remove_stream(stream)
         except Exception:
@@ -48,25 +50,50 @@ async def _remove_stream(stream: rtc.AudioStream, participant_sid: str | None =
         if pub_sid is not None:
             streams_by_pub.pop(pub_sid, None)
 
-    def on_track_subscribed(track: rtc.Track, publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant):
+    class _FrameOnlyStream:
+        def __init__(self, inner: rtc.AudioStream) -> None:
+            self._inner = inner
+
+        def __aiter__(self):
+            return self
+
+        async def __anext__(self) -> rtc.AudioFrame:
+            event = await self._inner.__anext__()
+            return event.frame
+
+        async def aclose(self) -> None:
+            await self._inner.aclose()
+
+    def on_track_subscribed(
+        track: rtc.Track,
+        publication: rtc.RemoteTrackPublication,
+        participant: rtc.RemoteParticipant,
+    ):
         if track.kind == rtc.TrackKind.KIND_AUDIO:
-            stream = rtc.AudioStream(track, sample_rate=48000, num_channels=1)
-            streams_by_pub[publication.sid] = stream
-            streams_by_participant.setdefault(participant.sid, set()).add(stream)
-            mixer.add_stream(stream)
+            event_stream = rtc.AudioStream(track, sample_rate=48000, num_channels=1)
+            frame_stream = _FrameOnlyStream(event_stream)
+            streams_by_pub[publication.sid] = frame_stream
+            streams_by_participant.setdefault(participant.sid, set()).add(frame_stream)
+            mixer.add_stream(frame_stream)
             logging.info("subscribed to audio from %s", participant.identity)
 
     room.on("track_subscribed", on_track_subscribed)
 
-    def on_track_unsubscribed(track: rtc.Track, publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant):
+    def on_track_unsubscribed(
+        track: rtc.Track,
+        publication: rtc.RemoteTrackPublication,
+        participant: rtc.RemoteParticipant,
+    ):
         stream = streams_by_pub.get(publication.sid)
         if stream is not None:
             asyncio.create_task(_remove_stream(stream, participant.sid, publication.sid))
             logging.info("unsubscribed from audio of %s", participant.identity)
 
     room.on("track_unsubscribed", on_track_unsubscribed)
 
-    def on_track_unpublished(publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant):
+    def on_track_unpublished(
+        publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant
+    ):
         stream = streams_by_pub.get(publication.sid)
         if stream is not None:
             asyncio.create_task(_remove_stream(stream, participant.sid, publication.sid))
@@ -119,5 +146,3 @@ def on_participant_disconnected(participant: rtc.RemoteParticipant):
 
 if __name__ == "__main__":
     asyncio.run(main())
-
-

examples/local-audio/publish_mic.py

@@ -5,6 +5,7 @@
 
 from livekit import rtc
 
+
 async def main() -> None:
     logging.basicConfig(level=logging.INFO)
 
@@ -47,5 +48,3 @@ async def main() -> None:
 
 if __name__ == "__main__":
     asyncio.run(main())
-
-

livekit-rtc/livekit/rtc/__init__.py

@@ -90,8 +90,10 @@
 from .audio_resampler import AudioResampler, AudioResamplerQuality
 from .audio_mixer import AudioMixer
 from .apm import AudioProcessingModule
+
 try:
     from .media_devices import MediaDevices as MediaDevices
+
     _HAS_MEDIA_DEVICES = True
 except Exception:  # pragma: no cover - optional dependency (sounddevice)
     _HAS_MEDIA_DEVICES = False

livekit-rtc/livekit/rtc/audio_mixer.py

@@ -184,9 +184,6 @@ async def _get_contribution(
             except StopAsyncIteration:
                 exhausted = True
                 break
-            # AudioStream may yield either AudioFrame or AudioFrameEvent; unwrap if needed
-            if hasattr(frame, "frame"):
-                frame = frame.frame  # type: ignore[assignment]
 
             new_data = np.frombuffer(frame.data.tobytes(), dtype=np.int16).reshape(
                 -1, self._num_channels

livekit-rtc/livekit/rtc/media_devices.py

@@ -96,6 +96,7 @@ class MicrophoneCapture:
             instance to `open_output_player` so reverse frames are provided.
         delay_estimator: Internal helper used to combine capture and render delays.
     """
+
     source: AudioSource
     input_stream: sd.InputStream
     task: asyncio.Task
@@ -124,6 +125,7 @@ class OutputPlayer:
     renders (in 10 ms frames) into the APM reverse path so that echo
     cancellation can correlate mic input with speaker output.
     """
+
     def __init__(
         self,
         *,
@@ -343,21 +345,27 @@ def open_microphone(
                 high_pass_filter=high_pass_filter,
                 auto_gain_control=auto_gain_control,
             )
-        delay_estimator: Optional[_APMDelayEstimator] = _APMDelayEstimator() if apm is not None else None
+        delay_estimator: Optional[_APMDelayEstimator] = (
+            _APMDelayEstimator() if apm is not None else None
+        )
         # Store the shared estimator on the device helper so the output player can reuse it
         self._delay_estimator = delay_estimator
 
         # Queue from callback to async task
         q: asyncio.Queue[AudioFrame] = asyncio.Queue(maxsize=queue_capacity)
 
-        def _input_callback(indata: np.ndarray, frame_count: int, time_info: Any, status: Any) -> None:
+        def _input_callback(
+            indata: np.ndarray, frame_count: int, time_info: Any, status: Any
+        ) -> None:
             # Slice into 10 ms frames, optionally APM, enqueue for async capture
             # Compute input (capture) delay using PortAudio timing; combine with last
             # measured output delay to provide APM stream delay in milliseconds.
             if apm is not None:
                 try:
                     input_delay_sec = float(time_info.currentTime - time_info.inputBufferAdcTime)
-                    output_delay_sec = float(delay_estimator.get_output_delay()) if delay_estimator else 0.0
+                    output_delay_sec = (
+                        float(delay_estimator.get_output_delay()) if delay_estimator else 0.0
+                    )
                     total_delay_ms = int(max((input_delay_sec + output_delay_sec) * 1000.0, 0.0))
                     try:
                         apm.set_stream_delay_ms(total_delay_ms)
@@ -444,7 +452,13 @@ async def _pump() -> None:
                     pass
 
         task = asyncio.create_task(_pump())
-        return MicrophoneCapture(source=source, input_stream=input_stream, task=task, apm=apm, delay_estimator=delay_estimator)
+        return MicrophoneCapture(
+            source=source,
+            input_stream=input_stream,
+            task=task,
+            apm=apm,
+            delay_estimator=delay_estimator,
+        )
 
     def open_output_player(
         self,
@@ -466,5 +480,3 @@ def open_output_player(
             output_device=output_device,
             delay_estimator=self._delay_estimator,
         )
-
-