wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

DailyTransport: replace virtual microphone with custom microphone track

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-05-21 23:03:55 -07:00
parent fcf49e79cc
commit 69ac70eed8
3 changed files with 43 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
CHANGELOG.md
View File

@@ -76,6 +76,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Changed ### Changed
- `DailyTransport` now uses custom microphone audio tracks instead of virtual
microphones. Now, multiple Daily transports can be used in the same process.
- `DailyTransport` now captures audio from individual participants instead of - `DailyTransport` now captures audio from individual participants instead of
the whole room. This allows identifying audio frames per participant. the whole room. This allows identifying audio frames per participant.

2
examples/foundational/04a-transports-daily.py
View File

@@ -37,9 +37,9 @@ async def main():
token, token,
"Respond bot", "Respond bot",
DailyParams( DailyParams(
audio_in_enabled=True,
audio_out_enabled=True, audio_out_enabled=True,
transcription_enabled=True, transcription_enabled=True,
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer(), vad_analyzer=SileroVADAnalyzer(),
), ),
) )

68
src/pipecat/transports/services/daily.py
View File

@@ -44,11 +44,11 @@ try:
AudioData, AudioData,
CallClient, CallClient,
CustomAudioSource, CustomAudioSource,
CustomAudioTrack,
Daily, Daily,
EventHandler, EventHandler,
VideoFrame, VideoFrame,
VirtualCameraDevice, VirtualCameraDevice,
VirtualMicrophoneDevice,
) )
except ModuleNotFoundError as e: except ModuleNotFoundError as e:
logger.error(f"Exception: {e}") logger.error(f"Exception: {e}")
@@ -242,6 +242,12 @@ def completion_callback(future):
return _callback return _callback
@dataclass
class DailyAudioTrack:
source: CustomAudioSource
track: CustomAudioTrack
class DailyTransportClient(EventHandler): class DailyTransportClient(EventHandler):
"""Core client for interacting with Daily's API. """Core client for interacting with Daily's API.
@@ -319,15 +325,12 @@ class DailyTransportClient(EventHandler):
self._out_sample_rate = 0 self._out_sample_rate = 0
self._camera: Optional[VirtualCameraDevice] = None self._camera: Optional[VirtualCameraDevice] = None
self._mic: Optional[VirtualMicrophoneDevice] = None self._microphone_track: Optional[DailyAudioTrack] = None
self._audio_sources: Dict[str, CustomAudioSource] = {} self._custom_audio_tracks: Dict[str, DailyAudioTrack] = {}
def _camera_name(self): def _camera_name(self):
return f"camera-{self}" return f"camera-{self}"
def _mic_name(self):
return f"mic-{self}"
@property @property
def room_url(self) -> str: def room_url(self) -> str:
return self._room_url return self._room_url
@@ -359,19 +362,25 @@ class DailyTransportClient(EventHandler):
await future await future
async def register_audio_destination(self, destination: str): async def register_audio_destination(self, destination: str):
self._audio_sources[destination] = await self.add_custom_audio_track(destination) self._custom_audio_tracks[destination] = await self.add_custom_audio_track(destination)
self._client.update_publishing({"customAudio": {destination: True}}) self._client.update_publishing({"customAudio": {destination: True}})
async def write_raw_audio_frames(self, frames: bytes, destination: Optional[str] = None): async def write_raw_audio_frames(self, frames: bytes, destination: Optional[str] = None):
future = self._get_event_loop().create_future() future = self._get_event_loop().create_future()
if not destination and self._mic:
self._mic.write_frames(frames, completion=completion_callback(future)) audio_source: Optional[CustomAudioSource] = None
elif destination and destination in self._audio_sources: if not destination and self._microphone_track:
source = self._audio_sources[destination] audio_source = self._microphone_track.source
source.write_frames(frames, completion=completion_callback(future)) elif destination and destination in self._custom_audio_tracks:
track = self._custom_audio_tracks[destination]
audio_source = track.source
if audio_source:
audio_source.write_frames(frames, completion=completion_callback(future))
else: else:
logger.warning(f"{self} unable to write audio frames to destination [{destination}]") logger.warning(f"{self} unable to write audio frames to destination [{destination}]")
future.set_result(None) future.set_result(None)
await future await future
async def write_raw_video_frame( async def write_raw_video_frame(
@@ -410,13 +419,10 @@ class DailyTransportClient(EventHandler):
color_format=self._params.video_out_color_format, color_format=self._params.video_out_color_format,
) )
if self._params.audio_out_enabled and not self._mic: if self._params.audio_out_enabled and not self._microphone_track:
self._mic = Daily.create_microphone_device( audio_source = CustomAudioSource(self._out_sample_rate, self._params.audio_out_channels)
self._mic_name(), audio_track = CustomAudioTrack(audio_source)
sample_rate=self._out_sample_rate, self._microphone_track = DailyAudioTrack(source=audio_source, track=audio_track)
channels=self._params.audio_out_channels,
non_blocking=True,
)
async def join(self): async def join(self):
# Transport already joined or joining, ignore. # Transport already joined or joining, ignore.
@@ -501,12 +507,11 @@ class DailyTransportClient(EventHandler):
"microphone": { "microphone": {
"isEnabled": microphone_enabled, "isEnabled": microphone_enabled,
"settings": { "settings": {
"deviceId": self._mic_name(), "customTrack": {
"customConstraints": { "id": self._microphone_track.track.id
"autoGainControl": {"exact": False}, if self._microphone_track
"echoCancellation": {"exact": False}, else "no-microphone-track"
"noiseSuppression": {"exact": False}, }
},
}, },
}, },
}, },
@@ -553,7 +558,7 @@ class DailyTransportClient(EventHandler):
await self._stop_transcription() await self._stop_transcription()
# Remove any custom tracks, if any. # Remove any custom tracks, if any.
for track_name, _ in self._audio_sources.items(): for track_name, _ in self._custom_audio_tracks.items():
await self.remove_custom_audio_track(track_name) await self.remove_custom_audio_track(track_name)
try: try:
@@ -703,19 +708,24 @@ class DailyTransportClient(EventHandler):
color_format=color_format, color_format=color_format,
) )
async def add_custom_audio_track(self, track_name: str) -> CustomAudioSource: async def add_custom_audio_track(self, track_name: str) -> DailyAudioTrack:
future = self._get_event_loop().create_future() future = self._get_event_loop().create_future()
audio_source = CustomAudioSource(self._out_sample_rate, 1) audio_source = CustomAudioSource(self._out_sample_rate, 1)
audio_track = CustomAudioTrack(audio_source)
self._client.add_custom_audio_track( self._client.add_custom_audio_track(
track_name=track_name, track_name=track_name,
audio_source=audio_source, audio_track=audio_track,
completion=completion_callback(future), completion=completion_callback(future),
) )
await future await future
return audio_source track = DailyAudioTrack(source=audio_source, track=audio_track)
return track
async def remove_custom_audio_track(self, track_name: str): async def remove_custom_audio_track(self, track_name: str):
future = self._get_event_loop().create_future() future = self._get_event_loop().create_future()