diff --git a/CHANGELOG.md b/CHANGELOG.md index 4dc4273be..7bf00326d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,33 @@ All notable changes to **pipecat** will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Added + +- Added new `LmntTTSService` text-to-speech service. + (see https://www.lmnt.com/) + +- Added `TTSModelUpdateFrame`, `TTSLanguageUpdateFrame`, `STTModelUpdateFrame`, + and `STTLanguageUpdateFrame` frames to allow you to switch models, language + and voices in TTS and STT services. + +- Added new `transcriptions.Language` enum. + +### Changed + +- `DailyTransport.on_joined` event now returns the full session data instead of + just the participant. + +- `CartesiaTTSService` is now a subclass of `TTSService`. + +- `DeepgramSTTService` is now a subclass of `STTService`. + +- `WhisperSTTService` is now a subclass of `SegmentedSTTService`. A + `SegmentedSTTService` is a `STTService` where the provided audio is given in a + big chunk (i.e. from when the user starts speaking until the user stops + speaking) instead of a continous stream. + ## [0.0.41] - 2024-08-22 ### Added diff --git a/README.md b/README.md index 40f96636c..681fd3b91 100644 --- a/README.md +++ b/README.md @@ -38,7 +38,7 @@ pip install "pipecat-ai[option,...]" Your project may or may not need these, so they're made available as optional requirements. Here is a list: -- **AI services**: `anthropic`, `azure`, `deepgram`, `gladia`, `google`, `fal`, `moondream`, `openai`, `openpipe`, `playht`, `silero`, `whisper`, `xtts` +- **AI services**: `anthropic`, `azure`, `deepgram`, `gladia`, `google`, `fal`, `lmnt`, `moondream`, `openai`, `openpipe`, `playht`, `silero`, `whisper`, `xtts` - **Transports**: `local`, `websocket`, `daily` ## Code examples diff --git a/examples/foundational/07k-interruptible-lmnt.py b/examples/foundational/07k-interruptible-lmnt.py index 790c4794e..6e68564ea 100644 --- a/examples/foundational/07k-interruptible-lmnt.py +++ b/examples/foundational/07k-interruptible-lmnt.py @@ -50,7 +50,7 @@ async def main(): tts = LmntTTSService( api_key=os.getenv("LMNT_API_KEY"), - voice="morgan" + voice_id="morgan" ) llm = OpenAILLMService( diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index e3b2aab37..13c2f53f1 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -8,6 +8,7 @@ from typing import Any, List, Mapping, Optional, Tuple from dataclasses import dataclass, field +from pipecat.transcriptions.language import Language from pipecat.utils.utils import obj_count, obj_id from pipecat.vad.vad_analyzer import VADParams @@ -131,9 +132,10 @@ class TranscriptionFrame(TextFrame): """ user_id: str timestamp: str + language: Language | None = None def __str__(self): - return f"{self.name}(user: {self.user_id}, text: {self.text}, timestamp: {self.timestamp})" + return f"{self.name}(user: {self.user_id}, text: {self.text}, language: {self.language}, timestamp: {self.timestamp})" @dataclass @@ -142,9 +144,10 @@ class InterimTranscriptionFrame(TextFrame): the transport's receive queue when a participant speaks.""" user_id: str timestamp: str + language: Language | None = None def __str__(self): - return f"{self.name}(user: {self.user_id}, text: {self.text}, timestamp: {self.timestamp})" + return f"{self.name}(user: {self.user_id}, text: {self.text}, language: {self.language}, timestamp: {self.timestamp})" @dataclass @@ -433,6 +436,13 @@ class LLMModelUpdateFrame(ControlFrame): model: str +@dataclass +class TTSModelUpdateFrame(ControlFrame): + """A control frame containing a request to update the TTS model. + """ + model: str + + @dataclass class TTSVoiceUpdateFrame(ControlFrame): """A control frame containing a request to update to a new TTS voice. @@ -440,6 +450,31 @@ class TTSVoiceUpdateFrame(ControlFrame): voice: str +@dataclass +class TTSLanguageUpdateFrame(ControlFrame): + """A control frame containing a request to update to a new TTS language and + optional voice. + + """ + language: Language + + +@dataclass +class STTModelUpdateFrame(ControlFrame): + """A control frame containing a request to update the STT model and optional + language. + + """ + model: str + + +@dataclass +class STTLanguageUpdateFrame(ControlFrame): + """A control frame containing a request to update to STT language. + """ + language: Language + + @dataclass class FunctionCallInProgressFrame(SystemFrame): """A frame signaling that a function call is in progress. diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 7ca1c3620..653a883d0 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -81,11 +81,6 @@ class RTVIAction(BaseModel): return super().model_post_init(__context) -# -# Client -> Pipecat messages. -# - - class RTVIServiceOptionConfig(BaseModel): name: str value: Any @@ -100,6 +95,16 @@ class RTVIConfig(BaseModel): config: List[RTVIServiceConfig] +# +# Client -> Pipecat messages. +# + + +class RTVIUpdateConfig(BaseModel): + config: List[RTVIServiceConfig] + interrupt: bool = False + + class RTVIActionRunArgument(BaseModel): name: str value: Any @@ -489,8 +494,8 @@ class RTVIProcessor(FrameProcessor): case "get-config": await self._handle_get_config(message.id) case "update-config": - config = RTVIConfig.model_validate(message.data) - await self._handle_update_config(message.id, config) + update_config = RTVIUpdateConfig.model_validate(message.data) + await self._handle_update_config(message.id, update_config) case "action": action = RTVIActionRun.model_validate(message.data) await self._handle_action(message.id, action) @@ -545,17 +550,14 @@ class RTVIProcessor(FrameProcessor): await handler(self, service.name, option) self._update_config_option(service.name, option) - async def _update_config(self, data: RTVIConfig): + async def _update_config(self, data: RTVIConfig, interrupt: bool): + if interrupt: + await self.interrupt_bot() for service_config in data.config: await self._update_service_config(service_config) - async def _handle_update_config(self, request_id: str, data: RTVIConfig): - # NOTE(aleix): The bot might be talking while we receive a new - # config. Let's interrupt it for now and update the config. Another - # solution is to wait until the bot stops speaking and then apply the - # config, but this definitely is more complicated to achieve. - await self.interrupt_bot() - await self._update_config(data) + async def _handle_update_config(self, request_id: str, data: RTVIUpdateConfig): + await self._update_config(RTVIConfig(config=data.config), data.interrupt) await self._handle_get_config(request_id) async def _handle_function_call_result(self, data): @@ -583,7 +585,7 @@ class RTVIProcessor(FrameProcessor): async def _maybe_send_bot_ready(self): if self._pipeline_started and self._client_ready: await self._send_bot_ready() - await self._update_config(self._config) + await self._update_config(self._config, False) async def _send_bot_ready(self): if not self._params.send_bot_ready: diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index d4a3c021e..c5cb7cc0d 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -18,8 +18,12 @@ from pipecat.frames.frames import ( ErrorFrame, Frame, LLMFullResponseEndFrame, + STTLanguageUpdateFrame, + STTModelUpdateFrame, StartFrame, StartInterruptionFrame, + TTSLanguageUpdateFrame, + TTSModelUpdateFrame, TTSSpeakFrame, TTSStartedFrame, TTSStoppedFrame, @@ -30,6 +34,7 @@ from pipecat.frames.frames import ( ) from pipecat.processors.async_frame_processor import AsyncFrameProcessor from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.transcriptions.language import Language from pipecat.utils.audio import calculate_audio_volume from pipecat.utils.string import match_endofsentence from pipecat.utils.utils import exp_smoothing @@ -173,10 +178,18 @@ class TTSService(AIService): self._stop_frame_queue: asyncio.Queue = asyncio.Queue() self._current_sentence: str = "" + @abstractmethod + async def set_model(self, model: str): + pass + @abstractmethod async def set_voice(self, voice: str): pass + @abstractmethod + async def set_language(self, language: Language): + pass + # Converts the text to audio. @abstractmethod async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: @@ -233,8 +246,12 @@ class TTSService(AIService): await self.push_frame(frame, direction) elif isinstance(frame, TTSSpeakFrame): await self._push_tts_frames(frame.text, False) + elif isinstance(frame, TTSModelUpdateFrame): + await self.set_model(frame.model) elif isinstance(frame, TTSVoiceUpdateFrame): await self.set_voice(frame.voice) + elif isinstance(frame, TTSLanguageUpdateFrame): + await self.set_language(frame.language) else: await self.push_frame(frame, direction) @@ -289,6 +306,47 @@ class TTSService(AIService): class STTService(AIService): """STTService is a base class for speech-to-text services.""" + def __init__(self, **kwargs): + super().__init__(**kwargs) + + @abstractmethod + async def set_model(self, model: str): + pass + + @abstractmethod + async def set_language(self, language: Language): + pass + + @abstractmethod + async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: + """Returns transcript as a string""" + pass + + async def process_audio_frame(self, frame: AudioRawFrame): + await self.process_generator(self.run_stt(frame.audio)) + + async def process_frame(self, frame: Frame, direction: FrameDirection): + """Processes a frame of audio data, either buffering or transcribing it.""" + await super().process_frame(frame, direction) + + if isinstance(frame, AudioRawFrame): + # In this service we accumulate audio internally and at the end we + # push a TextFrame. We don't really want to push audio frames down. + await self.process_audio_frame(frame) + elif isinstance(frame, STTModelUpdateFrame): + await self.set_model(frame.model) + elif isinstance(frame, STTLanguageUpdateFrame): + await self.set_language(frame.language) + else: + await self.push_frame(frame, direction) + + +class SegmentedSTTService(STTService): + """SegmentedSTTService is an STTService that will detect speech and will run + speech-to-text on speech segments only, instead of a continous stream. + + """ + def __init__(self, *, min_volume: float = 0.6, @@ -309,24 +367,7 @@ class STTService(AIService): self._smoothing_factor = 0.2 self._prev_volume = 0 - @abstractmethod - async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: - """Returns transcript as a string""" - pass - - def _new_wave(self): - content = io.BytesIO() - ww = wave.open(content, "wb") - ww.setsampwidth(2) - ww.setnchannels(self._num_channels) - ww.setframerate(self._sample_rate) - return (content, ww) - - def _get_smoothed_volume(self, frame: AudioRawFrame) -> float: - volume = calculate_audio_volume(frame.audio, frame.sample_rate) - return exp_smoothing(volume, self._prev_volume, self._smoothing_factor) - - async def _append_audio(self, frame: AudioRawFrame): + async def process_audio_frame(self, frame: AudioRawFrame): # Try to filter out empty background noise volume = self._get_smoothed_volume(frame) if volume >= self._min_volume: @@ -346,9 +387,7 @@ class STTService(AIService): self._silence_num_frames = 0 self._wave.close() self._content.seek(0) - await self.start_processing_metrics() await self.process_generator(self.run_stt(self._content.read())) - await self.stop_processing_metrics() (self._content, self._wave) = self._new_wave() async def stop(self, frame: EndFrame): @@ -357,16 +396,17 @@ class STTService(AIService): async def cancel(self, frame: CancelFrame): self._wave.close() - async def process_frame(self, frame: Frame, direction: FrameDirection): - """Processes a frame of audio data, either buffering or transcribing it.""" - await super().process_frame(frame, direction) + def _new_wave(self): + content = io.BytesIO() + ww = wave.open(content, "wb") + ww.setsampwidth(2) + ww.setnchannels(self._num_channels) + ww.setframerate(self._sample_rate) + return (content, ww) - if isinstance(frame, AudioRawFrame): - # In this service we accumulate audio internally and at the end we - # push a TextFrame. We don't really want to push audio frames down. - await self._append_audio(frame) - else: - await self.push_frame(frame, direction) + def _get_smoothed_volume(self, frame: AudioRawFrame) -> float: + volume = calculate_audio_volume(frame.audio, frame.sample_rate) + return exp_smoothing(volume, self._prev_volume, self._smoothing_factor) class ImageGenService(AIService): diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 735d12b5b..e3541ccea 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -10,9 +10,8 @@ import base64 import asyncio import time -from typing import AsyncGenerator +from typing import AsyncGenerator, Mapping -from pipecat.processors.frame_processor import FrameDirection from pipecat.frames.frames import ( CancelFrame, ErrorFrame, @@ -26,6 +25,8 @@ from pipecat.frames.frames import ( TextFrame, LLMFullResponseEndFrame ) +from pipecat.processors.frame_processor import FrameDirection +from pipecat.transcriptions.language import Language from pipecat.services.ai_services import TTSService from loguru import logger @@ -40,6 +41,25 @@ except ModuleNotFoundError as e: raise Exception(f"Missing module: {e}") +def language_to_cartesia_language(language: Language) -> str | None: + match language: + case Language.DE: + return "de" + case Language.EN: + return "en" + case Language.ES: + return "es" + case Language.FR: + return "fr" + case Language.JA: + return "ja" + case Language.PT: + return "pt" + case Language.ZH: + return "zh" + return None + + class CartesiaTTSService(TTSService): def __init__( @@ -90,10 +110,18 @@ class CartesiaTTSService(TTSService): def can_generate_metrics(self) -> bool: return True + async def set_model(self, model: str): + logger.debug(f"Switching TTS model to: [{model}]") + self._model_id = model + async def set_voice(self, voice: str): logger.debug(f"Switching TTS voice to: [{voice}]") self._voice_id = voice + async def set_language(self, language: Language): + logger.debug(f"Switching TTS language to: [{language}]") + self._language = language_to_cartesia_language(language) + async def start(self, frame: StartFrame): await super().start(frame) await self._connect() diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py index 035fdd25c..d899d4bdb 100644 --- a/src/pipecat/services/deepgram.py +++ b/src/pipecat/services/deepgram.py @@ -16,12 +16,11 @@ from pipecat.frames.frames import ( Frame, InterimTranscriptionFrame, StartFrame, - SystemFrame, TTSStartedFrame, TTSStoppedFrame, TranscriptionFrame) -from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.ai_services import AsyncAIService, TTSService +from pipecat.services.ai_services import STTService, TTSService +from pipecat.transcriptions.language import Language from pipecat.utils.time import time_now_iso8601 from loguru import logger @@ -30,10 +29,12 @@ from loguru import logger # See .env.example for Deepgram configuration needed try: from deepgram import ( + AsyncListenWebSocketClient, DeepgramClient, DeepgramClientOptions, LiveTranscriptionEvents, LiveOptions, + LiveResultResponse ) except ModuleNotFoundError as e: logger.error(f"Exception: {e}") @@ -107,7 +108,7 @@ class DeepgramTTSService(TTSService): logger.exception(f"{self} exception: {e}") -class DeepgramSTTService(AsyncAIService): +class DeepgramSTTService(STTService): def __init__(self, *, api_key: str, @@ -120,6 +121,8 @@ class DeepgramSTTService(AsyncAIService): channels=1, interim_results=True, smart_format=True, + punctuate=True, + profanity_filter=True, ), **kwargs): super().__init__(**kwargs) @@ -128,40 +131,62 @@ class DeepgramSTTService(AsyncAIService): self._client = DeepgramClient( api_key, config=DeepgramClientOptions(url=url, options={"keepalive": "true"})) - self._connection = self._client.listen.asynclive.v("1") + self._connection: AsyncListenWebSocketClient = self._client.listen.asyncwebsocket.v("1") self._connection.on(LiveTranscriptionEvents.Transcript, self._on_message) - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) + async def set_model(self, model: str): + logger.debug(f"Switching STT model to: [{model}]") + self._live_options.model = model + await self._disconnect() + await self._connect() - if isinstance(frame, SystemFrame): - await self.push_frame(frame, direction) - elif isinstance(frame, AudioRawFrame): - await self._connection.send(frame.audio) - else: - await self.queue_frame(frame, direction) + async def set_language(self, language: Language): + logger.debug(f"Switching STT language to: [{language}]") + self._live_options.language = language + await self._disconnect() + await self._connect() async def start(self, frame: StartFrame): await super().start(frame) + await self._connect() + + async def stop(self, frame: EndFrame): + await super().stop(frame) + await self._disconnect() + + async def cancel(self, frame: CancelFrame): + await super().cancel(frame) + await self._disconnect() + + async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: + await self.start_processing_metrics() + await self._connection.send(audio) + yield None + await self.stop_processing_metrics() + + async def _connect(self): if await self._connection.start(self._live_options): logger.debug(f"{self}: Connected to Deepgram") else: logger.error(f"{self}: Unable to connect to Deepgram") - async def stop(self, frame: EndFrame): - await super().stop(frame) - await self._connection.finish() - - async def cancel(self, frame: CancelFrame): - await super().cancel(frame) - await self._connection.finish() + async def _disconnect(self): + if self._connection.is_connected: + await self._connection.finish() + logger.debug(f"{self}: Disconnected from Deepgram") async def _on_message(self, *args, **kwargs): - result = kwargs["result"] + result: LiveResultResponse = kwargs["result"] + if len(result.channel.alternatives) == 0: + return is_final = result.is_final transcript = result.channel.alternatives[0].transcript + language = None + if result.channel.alternatives[0].languages: + language = result.channel.alternatives[0].languages[0] + language = Language(language) if len(transcript) > 0: if is_final: - await self.queue_frame(TranscriptionFrame(transcript, "", time_now_iso8601())) + await self.push_frame(TranscriptionFrame(transcript, "", time_now_iso8601(), language)) else: - await self.queue_frame(InterimTranscriptionFrame(transcript, "", time_now_iso8601())) + await self.push_frame(InterimTranscriptionFrame(transcript, "", time_now_iso8601(), language)) diff --git a/src/pipecat/services/lmnt.py b/src/pipecat/services/lmnt.py index e6688dc78..f7afd6a41 100644 --- a/src/pipecat/services/lmnt.py +++ b/src/pipecat/services/lmnt.py @@ -4,23 +4,19 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import json -import uuid -import base64 import asyncio -import time from typing import AsyncGenerator from pipecat.processors.frame_processor import FrameDirection from pipecat.frames.frames import ( + AudioRawFrame, CancelFrame, + EndFrame, ErrorFrame, Frame, - AudioRawFrame, StartFrame, StartInterruptionFrame, - EndFrame, TTSStartedFrame, TTSStoppedFrame, ) @@ -95,7 +91,8 @@ class LmntTTSService(TTSService): async def _connect(self): try: self._speech = Speech() - self._connection = await self._speech.synthesize_streaming(self._voice_id, format="raw", sample_rate=self._output_format["sample_rate"]) + self._connection = await self._speech.synthesize_streaming( + self._voice_id, format="raw", sample_rate=self._output_format["sample_rate"]) self._receive_task = self.get_event_loop().create_task(self._receive_task_handler()) except Exception as e: logger.exception(f"{self} initialization error: {e}") diff --git a/src/pipecat/services/whisper.py b/src/pipecat/services/whisper.py index 59ce51a46..04f357a94 100644 --- a/src/pipecat/services/whisper.py +++ b/src/pipecat/services/whisper.py @@ -14,7 +14,7 @@ from typing import AsyncGenerator import numpy as np from pipecat.frames.frames import ErrorFrame, Frame, TranscriptionFrame -from pipecat.services.ai_services import STTService +from pipecat.services.ai_services import SegmentedSTTService from pipecat.utils.time import time_now_iso8601 from loguru import logger @@ -38,7 +38,7 @@ class Model(Enum): DISTIL_MEDIUM_EN = "Systran/faster-distil-whisper-medium.en" -class WhisperSTTService(STTService): +class WhisperSTTService(SegmentedSTTService): """Class to transcribe audio with a locally-downloaded Whisper model""" def __init__(self, @@ -77,6 +77,7 @@ class WhisperSTTService(STTService): yield ErrorFrame("Whisper model not available") return + await self.start_processing_metrics() await self.start_ttfb_metrics() # Divide by 32768 because we have signed 16-bit data. @@ -88,7 +89,9 @@ class WhisperSTTService(STTService): if segment.no_speech_prob < self._no_speech_prob: text += f"{segment.text} " + await self.stop_ttfb_metrics() + await self.stop_processing_metrics() + if text: - await self.stop_ttfb_metrics() logger.debug(f"Transcription: [{text}]") yield TranscriptionFrame(text, "", time_now_iso8601()) diff --git a/src/pipecat/transcriptions/language.py b/src/pipecat/transcriptions/language.py new file mode 100644 index 000000000..f9e98104b --- /dev/null +++ b/src/pipecat/transcriptions/language.py @@ -0,0 +1,64 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import sys + +from enum import Enum + +if sys.version_info < (3, 11): + class StrEnum(str, Enum): + def __new__(cls, value): + obj = str.__new__(cls, value) + obj._value_ = value + return obj +else: + from enum import StrEnum + + +class Language(StrEnum): + BG = "bg" # Bulgarian + CA = "ca" # Catalan + ZH = "zh" # Chinese simplified + ZH_TW = "zh-TW" # Chinese traditional + CS = "cs" # Czech + DA = "da" # Danish + NL = "nl" # Dutch + EN = "en" # English + EN_US = "en-US" # English (USA) + EN_AU = "en-AU" # English (Australia) + EN_GB = "en-GB" # English (Great Britain) + EN_NZ = "en-NZ" # English (New Zealand) + EN_IN = "en-IN" # English (India) + ET = "et" # Estonian + FI = "fi" # Finnish + NL_BE = "nl-BE" # Flemmish + FR = "fr" # French + FR_CA = "fr-CA" # French (Canada) + DE = "de" # German + DE_CH = "de-CH" # German (Switzerland) + EL = "el" # Greek + HI = "hi" # Hindi + HU = "hu" # Hungarian + ID = "id" # Indonesian + IT = "it" # Italian + JA = "ja" # Japanese + KO = "ko" # Korean + LV = "lv" # Latvian + LT = "lt" # Lithuanian + MS = "ms" # Malay + NO = "no" # Norwegian + PL = "pl" # Polish + PT = "pt" # Portuguese + PT_BR = "pt-BR" # Portuguese (Brazil) + RO = "ro" # Romanian + RU = "ru" # Russian + SK = "sk" # Slovak + ES = "es" # Spanish + SV = "sv" # Swedish + TH = "th" # Thai + TR = "tr" # Turkish + UK = "uk" # Ukrainian + VI = "vi" # Vietnamese diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 98bc3cc3c..bb6032fa4 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -36,6 +36,7 @@ from pipecat.frames.frames import ( UserImageRawFrame, UserImageRequestFrame) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.transcriptions.language import Language from pipecat.transports.base_input import BaseInputTransport from pipecat.transports.base_output import BaseOutputTransport from pipecat.transports.base_transport import BaseTransport, TransportParams @@ -305,7 +306,7 @@ class DailyTransportClient(EventHandler): if self._token and self._params.transcription_enabled: await self._start_transcription() - await self._callbacks.on_joined(data["participants"]["local"]) + await self._callbacks.on_joined(data) else: error_msg = f"Error joining {self._room_url}: {error}" logger.error(error_msg) @@ -864,8 +865,8 @@ class DailyTransport(BaseTransport): self._input.capture_participant_video( participant_id, framerate, video_source, color_format) - async def _on_joined(self, participant): - await self._call_event_handler("on_joined", participant) + async def _on_joined(self, data): + await self._call_event_handler("on_joined", data) async def _on_left(self): await self._call_event_handler("on_left") @@ -950,11 +951,16 @@ class DailyTransport(BaseTransport): text = message["text"] timestamp = message["timestamp"] is_final = message["rawResponse"]["is_final"] + try: + language = message["rawResponse"]["channel"]["alternatives"][0]["languages"][0] + language = Language(language) + except KeyError: + language = None if is_final: - frame = TranscriptionFrame(text, participant_id, timestamp) + frame = TranscriptionFrame(text, participant_id, timestamp, language) logger.debug(f"Transcription (from: {participant_id}): [{text}]") else: - frame = InterimTranscriptionFrame(text, participant_id, timestamp) + frame = InterimTranscriptionFrame(text, participant_id, timestamp, language) if self._input: await self._input.push_transcription_frame(frame)