From a3ac0d84e8293a6fae86614b66b6c9372a80fe72 Mon Sep 17 00:00:00 2001 From: Moishe Lettvin Date: Wed, 17 Jan 2024 13:50:55 -0500 Subject: [PATCH 1/5] working on making services more consistent/terse/easy --- src/dailyai/services/ai_services.py | 101 ++++++++++++------ src/dailyai/tests/test_ai_services.py | 129 +++++++++++++++++++++++ src/dailyai/tests/test_asyncprocessor.py | 9 +- 3 files changed, 203 insertions(+), 36 deletions(-) create mode 100644 src/dailyai/tests/test_ai_services.py diff --git a/src/dailyai/services/ai_services.py b/src/dailyai/services/ai_services.py index 69988a3de..0a82fd821 100644 --- a/src/dailyai/services/ai_services.py +++ b/src/dailyai/services/ai_services.py @@ -5,58 +5,95 @@ import re from dailyai.queue_frame import QueueFrame, FrameType from abc import abstractmethod -from typing import AsyncGenerator +from typing import AsyncGenerator, Iterable from dataclasses import dataclass +from typing import AsyncGenerator + +from collections.abc import Iterable, AsyncIterable class AIService: def __init__( - self, - input_queue: asyncio.Queue[QueueFrame] | None = None, - output_queue: asyncio.Queue[QueueFrame] | None = None, + self ): self.logger = logging.getLogger("dailyai") - self.input_queue: asyncio.Queue[QueueFrame] | None = input_queue - self.output_queue: asyncio.Queue[QueueFrame] | None = output_queue def stop(self): pass - async def run(self) -> None: - if self.input_queue is None or self.output_queue is None: - raise Exception("Input and output queues must be set before using the run method.") + def allowed_input_frame_types(self) -> set[FrameType]: + return set() - while True: - frame = await self.input_queue.get() - self.logger.debug(f"{self.__class__.__name__} got frame:", frame.frame_type) - if frame.frame_type == FrameType.END_STREAM: - self.input_queue.task_done() - await self.output_queue.put(QueueFrame(FrameType.END_STREAM, None)) - break + def possible_output_frame_types(self) -> set[FrameType]: + return set() - output_frame = await self.process_frame(frame) - if output_frame: - await self.output_queue.put(output_frame) - self.input_queue.task_done() + async def run( + self, + requested_frame_types:set[FrameType], + frames:Iterable[QueueFrame] | AsyncIterable[QueueFrame] + ) -> AsyncGenerator[QueueFrame, None]: + if self.possible_output_frame_types().intersection(requested_frame_types) == set(): + raise Exception(f"Requested frame types {requested_frame_types} are not supported by this service.") + + if isinstance(frames, AsyncIterable): + async for frame in frames: + output_frame: QueueFrame | None = await self.process_frame(requested_frame_types, frame) + if output_frame: + yield output_frame + elif isinstance(frames, Iterable): + for frame in frames: + output_frame = await self.process_frame(requested_frame_types, frame) + if output_frame: + yield output_frame + else: + raise Exception("Frames must be an iterable or async iterable") @abstractmethod - async def process_frame(self, frame) -> QueueFrame | None: + async def process_frame(self, requested_frame_types:set[FrameType], frame:QueueFrame) -> QueueFrame | None: pass +class SentenceAggregator(AIService): + def __init__(self, **kwargs): + super().__init__(**kwargs) + self.current_sentence = "" + + def allowed_input_frame_types(self) -> set[FrameType]: + return set([FrameType.TEXT_CHUNK, FrameType.SENTENCE]) + + def possible_output_frame_types(self) -> set[FrameType]: + return set([FrameType.SENTENCE]) + + async def process_frame(self, requested_frame_types: set[FrameType], frame: QueueFrame) -> QueueFrame | None: + if not FrameType.SENTENCE in requested_frame_types: + return None + + if frame.frame_type == FrameType.TEXT_CHUNK: + if type(frame.frame_data) != str: + raise Exception("Sentence aggregator requires a string for the data field") + + self.current_sentence += frame.frame_data + if self.current_sentence.endswith((".", "?", "!")): + sentence = self.current_sentence + self.current_sentence = "" + return QueueFrame(FrameType.SENTENCE, sentence) + return None + elif frame.frame_type == FrameType.END_STREAM: + if self.current_sentence: + return QueueFrame(FrameType.SENTENCE, self.current_sentence) + else: + return None + elif frame.frame_type == FrameType.SENTENCE: + return frame + else: + return None + class LLMService(AIService): - # Generate a set of responses to a prompt. Yields a list of responses. - @abstractmethod - async def run_llm_async(self, messages) -> AsyncGenerator[str, None]: - # Adding a yield here lets the linter know what this method actually does - yield "" + def allowed_input_frame_types(self) -> set[FrameType]: + return set([FrameType.LLM_MESSAGE, FrameType.SENTENCE, FrameType.TRANSCRIPTION]) - # Generate a responses to a prompt. Returns the response - @abstractmethod - async def run_llm( - self, messages - ) -> str or None: - pass + def allowed_output_frame_types(self) -> set[FrameType]: + return set([FrameType.SENTENCE, FrameType.SENTENCE, FrameType.TEXT_CHUNK]) async def run_llm_async_sentences(self, messages) -> AsyncGenerator[str, None]: current_text = "" diff --git a/src/dailyai/tests/test_ai_services.py b/src/dailyai/tests/test_ai_services.py new file mode 100644 index 000000000..6467442a1 --- /dev/null +++ b/src/dailyai/tests/test_ai_services.py @@ -0,0 +1,129 @@ +from re import A +import unittest + +from typing import AsyncGenerator, Generator + +from dailyai.services.ai_services import AIService, SentenceAggregator +from dailyai.queue_frame import QueueFrame, FrameType + +class SimpleAIService(AIService): + def allowed_input_frame_types(self) -> set[FrameType]: + return set([FrameType.TEXT_CHUNK]) + + def possible_output_frame_types(self) -> set[FrameType]: + return set([FrameType.TEXT_CHUNK]) + + async def process_frame(self, requested_frame_types: set[FrameType], frame: QueueFrame) -> QueueFrame | None: + return frame + +class TestBaseAIService(unittest.IsolatedAsyncioTestCase): + async def test_async_input(self): + service = SimpleAIService() + + input_frames = [ + QueueFrame(FrameType.TEXT_CHUNK, "hello"), + QueueFrame(FrameType.END_STREAM, None), + ] + async def iterate_frames() -> AsyncGenerator[QueueFrame, None]: + for frame in input_frames: + yield frame + + output_frames = [] + async for frame in service.run(set([FrameType.TEXT_CHUNK]), iterate_frames()): + output_frames.append(frame) + + self.assertEqual(input_frames, output_frames) + + async def test_nonasync_input(self): + service = SimpleAIService() + + input_frames = [ + QueueFrame(FrameType.TEXT_CHUNK, "hello"), + QueueFrame(FrameType.END_STREAM, None), + ] + + def iterate_frames() -> Generator[QueueFrame, None, None]: + for frame in input_frames: + yield frame + + output_frames = [] + async for frame in service.run(set([FrameType.TEXT_CHUNK]), iterate_frames()): + output_frames.append(frame) + + self.assertEqual(input_frames, output_frames) + + +class TestSentenceAggregator(unittest.IsolatedAsyncioTestCase): + async def test_clause(self) -> None: + input_frames = [ + QueueFrame(FrameType.TEXT_CHUNK, "hello"), + QueueFrame(FrameType.END_STREAM, None), + ] + + service = SentenceAggregator() + output_frames = [] + async for frame in service.run(set([FrameType.SENTENCE]), input_frames): + output_frames.append(frame) + + self.assertEqual(1, len(output_frames)) + self.assertEqual(QueueFrame(FrameType.SENTENCE, "hello"), output_frames[0]) + + async def test_sentence(self) -> None: + input_frames = [ + QueueFrame(FrameType.TEXT_CHUNK, "hello, "), + QueueFrame(FrameType.TEXT_CHUNK, "world."), + QueueFrame(FrameType.END_STREAM, None), + ] + + service = SentenceAggregator() + output_frames = [] + async for frame in service.run(set([FrameType.SENTENCE]), input_frames): + output_frames.append(frame) + + self.assertEqual(1, len(output_frames)) + self.assertEqual(QueueFrame(FrameType.SENTENCE, "hello, world."), output_frames[0]) + + async def test_sentence_and_clause(self) -> None: + input_frames = [ + QueueFrame(FrameType.TEXT_CHUNK, "hello, "), + QueueFrame(FrameType.TEXT_CHUNK, "world."), + QueueFrame(FrameType.TEXT_CHUNK, " How are"), + QueueFrame(FrameType.END_STREAM, None), + ] + + service = SentenceAggregator() + output_frames = [] + async for frame in service.run(set([FrameType.SENTENCE]), input_frames): + output_frames.append(frame) + + self.assertEqual(2, len(output_frames)) + self.assertEqual( + QueueFrame(FrameType.SENTENCE, "hello, world."), output_frames[0] + ) + self.assertEqual( + QueueFrame(FrameType.SENTENCE, " How are"), output_frames[1] + ) + + async def test_two_sentences(self) -> None: + input_frames = [ + QueueFrame(FrameType.TEXT_CHUNK, "hello, "), + QueueFrame(FrameType.TEXT_CHUNK, "world."), + QueueFrame(FrameType.TEXT_CHUNK, " How are"), + QueueFrame(FrameType.TEXT_CHUNK, " you doing?"), + QueueFrame(FrameType.END_STREAM, None), + ] + + service = SentenceAggregator() + output_frames = [] + async for frame in service.run(set([FrameType.SENTENCE]), input_frames): + output_frames.append(frame) + + self.assertEqual(2, len(output_frames)) + self.assertEqual( + QueueFrame(FrameType.SENTENCE, "hello, world."), output_frames[0] + ) + self.assertEqual(QueueFrame(FrameType.SENTENCE, " How are you doing?"), output_frames[1]) + + +if __name__ == "__main__": + unittest.main() diff --git a/src/dailyai/tests/test_asyncprocessor.py b/src/dailyai/tests/test_asyncprocessor.py index 9c2222791..fcb2781e4 100644 --- a/src/dailyai/tests/test_asyncprocessor.py +++ b/src/dailyai/tests/test_asyncprocessor.py @@ -18,7 +18,7 @@ from dailyai.services.ai_services import ( LLMService, TTSService, ) - +""" class MockTTSService(TTSService): def run_tts(self, sentence): for word in sentence.split(' '): @@ -73,7 +73,7 @@ class TestResponse(unittest.TestCase): while expected_words: actual_word:QueueFrame = output_queue.get() word = expected_words.pop(0) - self.assertEqual(actual_word.frame_type, FrameType.AUDIO) + self.assertEqual(actual_word.frame_type, FrameType.AUDIO_FRAME) self.assertEqual(actual_word.frame_data, bytes(word, "utf-8")) output_queue.task_done() @@ -128,10 +128,10 @@ class TestResponse(unittest.TestCase): while expected_words and not stop_processing_output_queue.is_set(): try: actual_word:QueueFrame = output_queue.get_nowait() - if actual_word.frame_type == FrameType.AUDIO: + if actual_word.frame_type == FrameType.AUDIO_FRAME: time.sleep(0.1) word = expected_words.pop(0) - self.assertEqual(actual_word.frame_type, FrameType.AUDIO) + self.assertEqual(actual_word.frame_type, FrameType.AUDIO_FRAME) self.assertEqual(actual_word.frame_data, bytes(word, "utf-8")) output_queue.task_done() except Empty: @@ -177,3 +177,4 @@ class TestResponse(unittest.TestCase): if __name__ == '__main__': unittest.main() +""" From 13f2f792afeb2b4fc6630d6a5ef3deb029264333 Mon Sep 17 00:00:00 2001 From: Moishe Lettvin Date: Wed, 17 Jan 2024 18:42:08 -0500 Subject: [PATCH 2/5] refactor party tonight --- src/dailyai/services/ai_services.py | 157 ++++++++++++------ src/dailyai/services/azure_ai_services.py | 25 +-- .../services/daily_transport_service.py | 3 + src/dailyai/services/elevenlabs_ai_service.py | 4 +- src/dailyai/services/open_ai_services.py | 8 +- .../theoretical-to-real/01-say-one-thing.py | 9 +- .../02-llm-say-one-thing.py | 27 ++- .../theoretical-to-real/03-still-frame.py | 11 +- .../04-utterance-and-speech.py | 54 +++--- .../05-sync-speech-and-text.py | 7 +- 10 files changed, 187 insertions(+), 118 deletions(-) diff --git a/src/dailyai/services/ai_services.py b/src/dailyai/services/ai_services.py index 0a82fd821..7652c7d09 100644 --- a/src/dailyai/services/ai_services.py +++ b/src/dailyai/services/ai_services.py @@ -2,6 +2,8 @@ import asyncio import logging import re +from httpx import request + from dailyai.queue_frame import QueueFrame, FrameType from abc import abstractmethod @@ -13,9 +15,7 @@ from collections.abc import Iterable, AsyncIterable class AIService: - def __init__( - self - ): + def __init__(self): self.logger = logging.getLogger("dailyai") def stop(self): @@ -27,30 +27,60 @@ class AIService: def possible_output_frame_types(self) -> set[FrameType]: return set() + async def run_to_queue(self, queue: asyncio.Queue, frames, add_end_of_stream=False) -> None: + async for frame in self.run(frames): + print("got frame", frame.frame_type) + await queue.put(frame) + + if add_end_of_stream: + await queue.put(QueueFrame(FrameType.END_STREAM, None)) + async def run( - self, - requested_frame_types:set[FrameType], - frames:Iterable[QueueFrame] | AsyncIterable[QueueFrame] - ) -> AsyncGenerator[QueueFrame, None]: - if self.possible_output_frame_types().intersection(requested_frame_types) == set(): + self, + frames: Iterable[QueueFrame] + | AsyncIterable[QueueFrame] + | asyncio.Queue[QueueFrame], + requested_frame_types: set[FrameType] | None=None, + ) -> AsyncGenerator[QueueFrame, None]: + if requested_frame_types and self.possible_output_frame_types().intersection(requested_frame_types) == set(): raise Exception(f"Requested frame types {requested_frame_types} are not supported by this service.") + if not requested_frame_types: + requested_frame_types = self.possible_output_frame_types() + + print("running", self.__class__.__name__, "with frame types", requested_frame_types) + if isinstance(frames, AsyncIterable): async for frame in frames: - output_frame: QueueFrame | None = await self.process_frame(requested_frame_types, frame) - if output_frame: + async for output_frame in self.process_frame(requested_frame_types, frame): + print( + "yielding frame", self.__class__.__name__, output_frame.frame_type + ) yield output_frame elif isinstance(frames, Iterable): for frame in frames: - output_frame = await self.process_frame(requested_frame_types, frame) - if output_frame: + async for output_frame in self.process_frame(requested_frame_types, frame): + print( + "yielding frame", self.__class__.__name__, output_frame.frame_type + ) yield output_frame + elif isinstance(frames, asyncio.Queue): + while True: + frame = await frames.get() + async for output_frame in self.process_frame(requested_frame_types, frame): + print( + "yielding frame", self.__class__.__name__, output_frame.frame_type + ) + yield output_frame + if frame.frame_type == FrameType.END_STREAM: + break else: raise Exception("Frames must be an iterable or async iterable") @abstractmethod - async def process_frame(self, requested_frame_types:set[FrameType], frame:QueueFrame) -> QueueFrame | None: - pass + async def process_frame(self, requested_frame_types:set[FrameType], frame:QueueFrame) -> AsyncGenerator[QueueFrame, None]: + # Yield something so the linter can deduce what should happen here. + yield QueueFrame(FrameType.END_STREAM, None) class SentenceAggregator(AIService): def __init__(self, **kwargs): @@ -63,29 +93,26 @@ class SentenceAggregator(AIService): def possible_output_frame_types(self) -> set[FrameType]: return set([FrameType.SENTENCE]) - async def process_frame(self, requested_frame_types: set[FrameType], frame: QueueFrame) -> QueueFrame | None: + async def process_frame(self, requested_frame_types: set[FrameType], frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]: if not FrameType.SENTENCE in requested_frame_types: - return None + return if frame.frame_type == FrameType.TEXT_CHUNK: if type(frame.frame_data) != str: - raise Exception("Sentence aggregator requires a string for the data field") + raise Exception( + "Sentence aggregator requires a string for the data field" + ) self.current_sentence += frame.frame_data if self.current_sentence.endswith((".", "?", "!")): sentence = self.current_sentence self.current_sentence = "" - return QueueFrame(FrameType.SENTENCE, sentence) - return None + yield QueueFrame(FrameType.SENTENCE, sentence) elif frame.frame_type == FrameType.END_STREAM: if self.current_sentence: - return QueueFrame(FrameType.SENTENCE, self.current_sentence) - else: - return None + yield QueueFrame(FrameType.SENTENCE, self.current_sentence) elif frame.frame_type == FrameType.SENTENCE: - return frame - else: - return None + yield frame class LLMService(AIService): @@ -93,30 +120,29 @@ class LLMService(AIService): return set([FrameType.LLM_MESSAGE, FrameType.SENTENCE, FrameType.TRANSCRIPTION]) def allowed_output_frame_types(self) -> set[FrameType]: - return set([FrameType.SENTENCE, FrameType.SENTENCE, FrameType.TEXT_CHUNK]) + return set([FrameType.SENTENCE, FrameType.TEXT_CHUNK]) - async def run_llm_async_sentences(self, messages) -> AsyncGenerator[str, None]: - current_text = "" - async for text in self.run_llm_async(messages): - current_text += text - if re.match(r"^.*[.!?]$", text): - yield current_text - current_text = "" + @abstractmethod + async def run_llm_async(self, messages) -> AsyncGenerator[str, None]: + yield "" - if current_text: - yield current_text - - async def process_frame(self, frame:QueueFrame) -> QueueFrame | None: - if not self.output_queue: - raise Exception("Output queue must be set before using the run method.") + @abstractmethod + async def run_llm(self, messages) -> str: + pass + async def process_frame(self, requested_frame_types: set[FrameType], frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]: if frame.frame_type == FrameType.LLM_MESSAGE: if type(frame.frame_data) != list: raise Exception("LLM service requires a dict for the data field") messages: list[dict[str, str]] = frame.frame_data - async for message in self.run_llm_async_sentences(messages): - await self.output_queue.put(QueueFrame(FrameType.SENTENCE, message)) + if FrameType.SENTENCE in requested_frame_types: + yield QueueFrame(FrameType.SENTENCE, await self.run_llm(messages)) + else: + async for text_chunk in self.run_llm_async(messages): + yield QueueFrame(FrameType.TEXT_CHUNK, text_chunk) + + # TODO: handle other frame types! Need to aggregate into messages class TTSService(AIService): @@ -124,6 +150,12 @@ class TTSService(AIService): def get_mic_sample_rate(self): return 16000 + def allowed_input_frame_types(self) -> set[FrameType]: + return set([FrameType.SENTENCE, FrameType.TRANSCRIPTION, FrameType.TEXT_CHUNK]) + + def possible_output_frame_types(self) -> set[FrameType]: + return set([FrameType.AUDIO]) + # Converts the sentence to audio. Yields a list of audio frames that can # be sent to the microphone device @abstractmethod @@ -131,25 +163,48 @@ class TTSService(AIService): # yield empty bytes here, so linting can infer what this method does yield bytes() - async def process_frame(self, frame:QueueFrame) -> QueueFrame | None: - if not self.output_queue: - raise Exception("Output queue must be set before using the run method.") + async def process_frame(self, requested_frame_types: set[FrameType], frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]: + if not FrameType.AUDIO in requested_frame_types: + return - if frame.frame_type == FrameType.SENTENCE: - if type(frame.frame_data) != str: - raise Exception("TTS service requires a string for the data field") + if type(frame.frame_data) != str: + raise Exception("TTS service requires a string for the data field") - text = frame.frame_data - async for audio in self.run_tts(text): - await self.output_queue.put(QueueFrame(FrameType.AUDIO, audio)) + async for audio_chunk in self.run_tts(frame.frame_data): + yield QueueFrame(FrameType.AUDIO, audio_chunk) + + # Convenience function to send the audio for a sentence to the given queue + async def say(self, sentence, queue: asyncio.Queue): + async for audio_chunk in self.run_tts(sentence): + await queue.put(QueueFrame(FrameType.AUDIO, audio_chunk)) class ImageGenService(AIService): + def __init__(self, image_size, **kwargs): + super().__init__(**kwargs) + self.image_size = image_size + + def allowed_input_frame_types(self) -> set[FrameType]: + return set([FrameType.SENTENCE, FrameType.TRANSCRIPTION, FrameType.TEXT_CHUNK, FrameType.IMAGE_DESCRIPTION]) + + def possible_output_frame_types(self) -> set[FrameType]: + return set([FrameType.IMAGE]) + # Renders the image. Returns an Image object. @abstractmethod - async def run_image_gen(self, sentence, size) -> tuple[str, bytes]: + async def run_image_gen(self, sentence) -> tuple[str, bytes]: pass + async def process_frame(self, requested_frame_types: set[FrameType], frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]: + if not FrameType.IMAGE in requested_frame_types: + return + + if type(frame.frame_data) != str: + raise Exception("Image service requires a string for the data field") + + (_, image_data) = await self.run_image_gen(frame.frame_data) + yield QueueFrame(FrameType.IMAGE, image_data) + @dataclass class AIServiceConfig: diff --git a/src/dailyai/services/azure_ai_services.py b/src/dailyai/services/azure_ai_services.py index 452797be0..b723e77e4 100644 --- a/src/dailyai/services/azure_ai_services.py +++ b/src/dailyai/services/azure_ai_services.py @@ -16,8 +16,8 @@ from PIL import Image from azure.cognitiveservices.speech import SpeechSynthesizer, SpeechConfig, ResultReason, CancellationReason class AzureTTSService(TTSService): - def __init__(self, input_queue=None, output_queue=None, speech_key=None, speech_region=None): - super().__init__(input_queue, output_queue) + def __init__(self, speech_key=None, speech_region=None): + super().__init__() speech_key = speech_key or os.getenv("AZURE_SPEECH_SERVICE_KEY") speech_region = speech_region or os.getenv("AZURE_SPEECH_SERVICE_REGION") @@ -35,7 +35,10 @@ class AzureTTSService(TTSService): "" \ f"{sentence}" \ " " - result = await asyncio.to_thread(self.speech_synthesizer.speak_ssml, (ssml)) + try: + result = await asyncio.to_thread(self.speech_synthesizer.speak_ssml, (ssml)) + except Exception as e: + self.logger.error("Error in azure tts", e) self.logger.info("Got azure tts result") if result.reason == ResultReason.SynthesizingAudioCompleted: self.logger.info("Returning result") @@ -48,8 +51,8 @@ class AzureTTSService(TTSService): self.logger.info("Error details: {}".format(cancellation_details.error_details)) class AzureLLMService(LLMService): - def __init__(self, input_queue=None, output_queue=None, api_key=None, azure_endpoint=None, api_version=None, model=None): - super().__init__(input_queue, output_queue) + def __init__(self, api_key=None, azure_endpoint=None, api_version=None, model=None): + super().__init__() api_key = api_key or os.getenv("AZURE_CHATGPT_KEY") azure_endpoint = azure_endpoint or os.getenv("AZURE_CHATGPT_ENDPOINT") @@ -92,14 +95,14 @@ class AzureLLMService(LLMService): class AzureImageGenServiceREST(ImageGenService): - def __init__(self, api_key=None, azure_endpoint=None, api_version=None, model=None): - super().__init__() + def __init__(self, image_size:str, api_key=None, azure_endpoint=None, api_version=None, model=None): + super().__init__(image_size=image_size) self.api_key = api_key or os.getenv("AZURE_DALLE_KEY") self.azure_endpoint = azure_endpoint or os.getenv("AZURE_DALLE_ENDPOINT") self.api_version = api_version or "2023-06-01-preview" self.model = model or os.getenv("AZURE_DALLE_DEPLOYMENT_ID") - async def run_image_gen(self, sentence, size) -> tuple[str, bytes]: + async def run_image_gen(self, sentence) -> tuple[str, bytes]: # TODO hoist the session to app-level async with aiohttp.ClientSession() as session: url = f"{self.azure_endpoint}openai/images/generations:submit?api-version={self.api_version}" @@ -107,7 +110,7 @@ class AzureImageGenServiceREST(ImageGenService): body = { # Enter your prompt text here "prompt": sentence, - "size": size, + "size": self.image_size, "n": 1, } async with session.post(url, headers=headers, json=body) as submission: @@ -153,14 +156,14 @@ class AzureImageGenService(ImageGenService): api_version=api_version, ) - async def run_image_gen(self, sentence, size) -> tuple[str, bytes]: + async def run_image_gen(self, sentence) -> tuple[str, bytes]: self.logger.info("Generating azure image", sentence) image = self.client.images.generate( model=self.model, prompt=sentence, n=1, - size=size, + size=self.image_size, ) url = image["data"][0]["url"] diff --git a/src/dailyai/services/daily_transport_service.py b/src/dailyai/services/daily_transport_service.py index 3147ecfa1..171cc36a6 100644 --- a/src/dailyai/services/daily_transport_service.py +++ b/src/dailyai/services/daily_transport_service.py @@ -206,6 +206,9 @@ class DailyTransportService(EventHandler): if frame.frame_type == FrameType.END_STREAM: break + def wait_for_send_queue_to_empty(self): + self.threadsafe_send_queue.join() + async def run(self) -> None: self.configure_daily() diff --git a/src/dailyai/services/elevenlabs_ai_service.py b/src/dailyai/services/elevenlabs_ai_service.py index 0d9ec8b54..5d6514dec 100644 --- a/src/dailyai/services/elevenlabs_ai_service.py +++ b/src/dailyai/services/elevenlabs_ai_service.py @@ -9,8 +9,8 @@ from dailyai.services.ai_services import TTSService class ElevenLabsTTSService(TTSService): - def __init__(self, input_queue=None, output_queue=None, api_key=None, voice_id=None): - super().__init__(input_queue, output_queue) + def __init__(self, api_key=None, voice_id=None): + super().__init__() self.api_key = api_key or os.getenv("ELEVENLABS_API_KEY") self.voice_id = voice_id or os.getenv("ELEVENLABS_VOICE_ID") diff --git a/src/dailyai/services/open_ai_services.py b/src/dailyai/services/open_ai_services.py index 8f2b6154a..ea6ea07ba 100644 --- a/src/dailyai/services/open_ai_services.py +++ b/src/dailyai/services/open_ai_services.py @@ -50,20 +50,20 @@ class OpenAILLMService(LLMService): return None class OpenAIImageGenService(ImageGenService): - def __init__(self, api_key=None, model=None): - super().__init__() + def __init__(self, image_size:str, api_key=None, model=None): + super().__init__(image_size=image_size) api_key = api_key or os.getenv("OPEN_AI_KEY") self.model = model or os.getenv("OPEN_AI_IMAGE_MODEL") or "dall-e-3" self.client = AsyncOpenAI(api_key=api_key) - async def run_image_gen(self, sentence, size) -> tuple[str, bytes]: + async def run_image_gen(self, sentence) -> tuple[str, bytes]: self.logger.info("Generating OpenAI image", sentence) image = await self.client.images.generate( prompt=sentence, model=self.model, n=1, - size=size + size=self.image_size ) image_url = image.data[0].url if not image_url: diff --git a/src/samples/theoretical-to-real/01-say-one-thing.py b/src/samples/theoretical-to-real/01-say-one-thing.py index e531e7e47..80ba91a32 100644 --- a/src/samples/theoretical-to-real/01-say-one-thing.py +++ b/src/samples/theoretical-to-real/01-say-one-thing.py @@ -27,21 +27,16 @@ async def main(room_url): # similarly, create a tts service tts = AzureTTSService() - # Get the generator for the audio. This will start running in the background, - # and when we ask the generator for its items, we'll get what it's generated. - audio_generator: AsyncGenerator[bytes, None] = tts.run_tts("hello world") - # Register an event handler so we can play the audio when the participant joins. @transport.event_handler("on_participant_joined") async def on_participant_joined(transport, participant): if participant["info"]["isLocal"]: return - async for audio in audio_generator: - transport.output_queue.put(QueueFrame(FrameType.AUDIO, audio)) + await tts.say("Hello there, " + participant["info"]["userName"] + "!", transport.send_queue) # wait for the output queue to be empty, then leave the meeting - transport.output_queue.join() + transport.wait_for_send_queue_to_empty() transport.stop() await transport.run() diff --git a/src/samples/theoretical-to-real/02-llm-say-one-thing.py b/src/samples/theoretical-to-real/02-llm-say-one-thing.py index b1688eace..301b2762c 100644 --- a/src/samples/theoretical-to-real/02-llm-say-one-thing.py +++ b/src/samples/theoretical-to-real/02-llm-say-one-thing.py @@ -4,6 +4,7 @@ from typing import AsyncGenerator from dailyai.queue_frame import QueueFrame, FrameType from dailyai.services.daily_transport_service import DailyTransportService +from dailyai.services.ai_services import SentenceAggregator from dailyai.services.azure_ai_services import AzureLLMService from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService @@ -17,29 +18,27 @@ async def main(room_url): ) transport.mic_enabled = True - text_to_llm_queue = asyncio.Queue() - llm_to_tts_queue = asyncio.Queue() - - tts = ElevenLabsTTSService( - llm_to_tts_queue, transport.get_async_send_queue(), voice_id="29vD33N1CtxCmqQRPOHJ" - ) - llm = AzureLLMService(text_to_llm_queue, llm_to_tts_queue) + tts = ElevenLabsTTSService(voice_id="29vD33N1CtxCmqQRPOHJ") + llm = AzureLLMService() messages = [{ "role": "system", "content": "You are an LLM in a WebRTC session, and this is a 'hello world' demo. Say hello to the world." }] - await text_to_llm_queue.put(QueueFrame(FrameType.LLM_MESSAGE, messages)) - await text_to_llm_queue.put(QueueFrame(FrameType.END_STREAM, None)) - - llm_task = asyncio.create_task(llm.run()) + tts_task = asyncio.create_task( + tts.run_to_queue( + transport.send_queue, + SentenceAggregator().run( + llm.run([QueueFrame(FrameType.LLM_MESSAGE, messages)]) + ) + ) + ) @transport.event_handler("on_first_other_participant_joined") async def on_first_other_participant_joined(transport): - await asyncio.gather(llm_task, tts.run()) + await tts_task - # wait for the output queue to be empty, then leave the meeting - transport.output_queue.join() + transport.wait_for_send_queue_to_empty() transport.stop() await transport.run() diff --git a/src/samples/theoretical-to-real/03-still-frame.py b/src/samples/theoretical-to-real/03-still-frame.py index eccb7cc83..79261214d 100644 --- a/src/samples/theoretical-to-real/03-still-frame.py +++ b/src/samples/theoretical-to-real/03-still-frame.py @@ -21,13 +21,14 @@ async def main(room_url): transport.camera_width = 1024 transport.camera_height = 1024 - imagegen = OpenAIImageGenService() - image_task = asyncio.create_task(imagegen.run_image_gen("a cat in the style of picasso", "1024x1024")) + imagegen = OpenAIImageGenService(image_size="1024x1024") + image_task = asyncio.create_task( + imagegen.run_to_queue(transport.send_queue, [QueueFrame(FrameType.IMAGE_DESCRIPTION, "a cat in the style of picasso")]) + ) @transport.event_handler("on_participant_joined") async def on_participant_joined(transport, participant): - (_, image_bytes) = await image_task - transport.output_queue.put(QueueFrame(FrameType.IMAGE, image_bytes)) + await image_task await transport.run() @@ -38,6 +39,6 @@ if __name__ == "__main__": "-u", "--url", type=str, required=True, help="URL of the Daily room to join" ) - args: argparse.Namespace = parser.parse_args() + args, unknown = parser.parse_known_args() asyncio.run(main(args.url)) diff --git a/src/samples/theoretical-to-real/04-utterance-and-speech.py b/src/samples/theoretical-to-real/04-utterance-and-speech.py index 92fcf0db4..0d72905b4 100644 --- a/src/samples/theoretical-to-real/04-utterance-and-speech.py +++ b/src/samples/theoretical-to-real/04-utterance-and-speech.py @@ -2,9 +2,11 @@ import argparse import asyncio import re +from dailyai.services.ai_services import SentenceAggregator from dailyai.services.daily_transport_service import DailyTransportService from dailyai.services.azure_ai_services import AzureLLMService, AzureTTSService from dailyai.queue_frame import QueueFrame, FrameType +from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService async def main(room_url:str): global transport @@ -22,34 +24,46 @@ async def main(room_url:str): transport.camera_enabled = False llm = AzureLLMService() - tts = AzureTTSService() + azure_tts = AzureTTSService() + elevenlabs_tts = ElevenLabsTTSService(voice_id="ErXwobaYiN019PkySvjV") + + messages = [{"role": "system", "content": "tell the user a joke about llamas"}] + + # Start a task to run the LLM to create a joke, and convert the LLM output to audio frames. This task + # will run in parallel with generating and speaking the audio for static text, so there's no delay to + # speak the LLM response. + buffer_queue = asyncio.Queue() + llm_response_task = asyncio.create_task( + elevenlabs_tts.run_to_queue( + buffer_queue, + SentenceAggregator().run( + llm.run([QueueFrame(FrameType.LLM_MESSAGE, messages)]) + ), + True, + ) + ) @transport.event_handler("on_participant_joined") async def on_joined(transport, participant): if participant["id"] == transport.my_participant_id: return - # queue two pieces of speech: one specified as a text literal, - # and one generated by an llm. We'll kick off the llm first, and let - # it generate a response while we're speaking the literal string. - # - # Note that in this case, we don't use `run_llm_async` because we're - # taking advantage of the time spent speaking the first phrase to generate - # the entire LLM response, and this happens asynchronously in a task. - llm_response_task = asyncio.create_task(llm.run_llm( - [{"role": "system", "content": "tell the user a joke about llamas"}] - )) + await azure_tts.run_to_queue( + transport.send_queue, + [QueueFrame(FrameType.SENTENCE, "My friend the LLM is now going to tell a joke about llamas.")] + ) - async for audio_chunk in tts.run_tts("My friend the LLM is now going to tell a joke about llamas."): - transport.output_queue.put(QueueFrame(FrameType.AUDIO, audio_chunk)) + async def buffer_to_send_queue(): + while True: + frame = await buffer_queue.get() + await transport.send_queue.put(frame) + buffer_queue.task_done() + if frame.frame_type == FrameType.END_STREAM: + break - llm_response = await llm_response_task - async for audio_chunk in tts.run_tts(llm_response): - transport.output_queue.put(QueueFrame(FrameType.AUDIO, audio_chunk)) + await asyncio.gather(llm_response_task, buffer_to_send_queue()) - - # wait for the output queue to be empty, then leave the meeting - transport.output_queue.join() + transport.wait_for_send_queue_to_empty() transport.stop() await transport.run() @@ -61,6 +75,6 @@ if __name__ == "__main__": "-u", "--url", type=str, required=True, help="URL of the Daily room to join" ) - args: argparse.Namespace = parser.parse_args() + args, unknown = parser.parse_known_args() asyncio.run(main(args.url)) diff --git a/src/samples/theoretical-to-real/05-sync-speech-and-text.py b/src/samples/theoretical-to-real/05-sync-speech-and-text.py index e6ffd94c0..9936dff6b 100644 --- a/src/samples/theoretical-to-real/05-sync-speech-and-text.py +++ b/src/samples/theoretical-to-real/05-sync-speech-and-text.py @@ -26,10 +26,9 @@ async def main(room_url): transport.camera_height = 1024 llm = AzureLLMService() - #tts = ElevenLabsTTSService(voice_id="ErXwobaYiN019PkySvjV") - tts = ElevenLabsTTSService() dalle = FalImageGenService() - # dalle = OpenAIImageGenService() + tts = ElevenLabsTTSService(voice_id="ErXwobaYiN019PkySvjV") + #dalle = OpenAIImageGenService(image_size="1024x1024") # Get a complete audio chunk from the given text. Splitting this into its own # coroutine lets us ensure proper ordering of the audio chunks on the output queue. @@ -61,7 +60,7 @@ async def main(room_url): tts_tasks.append(get_all_audio(sentence)) - tts_tasks.insert(0, dalle.run_image_gen(image_text, "1024x1024")) + tts_tasks.insert(0, dalle.run_image_gen(image_text)) print(f"waiting for tasks to finish for {month}") data = await asyncio.gather( From 0d21768d00270ddb6c08659606b951712a651ab7 Mon Sep 17 00:00:00 2001 From: Moishe Lettvin Date: Wed, 17 Jan 2024 18:58:03 -0500 Subject: [PATCH 3/5] Fix example 5 --- src/dailyai/services/ai_services.py | 12 -------- .../services/daily_transport_service.py | 4 +-- .../to_be_updated/huggingface_ai_service.py | 1 - .../05-sync-speech-and-text.py | 30 +++++++++++-------- 4 files changed, 19 insertions(+), 28 deletions(-) diff --git a/src/dailyai/services/ai_services.py b/src/dailyai/services/ai_services.py index 7652c7d09..4c5604c0d 100644 --- a/src/dailyai/services/ai_services.py +++ b/src/dailyai/services/ai_services.py @@ -29,7 +29,6 @@ class AIService: async def run_to_queue(self, queue: asyncio.Queue, frames, add_end_of_stream=False) -> None: async for frame in self.run(frames): - print("got frame", frame.frame_type) await queue.put(frame) if add_end_of_stream: @@ -48,29 +47,18 @@ class AIService: if not requested_frame_types: requested_frame_types = self.possible_output_frame_types() - print("running", self.__class__.__name__, "with frame types", requested_frame_types) - if isinstance(frames, AsyncIterable): async for frame in frames: async for output_frame in self.process_frame(requested_frame_types, frame): - print( - "yielding frame", self.__class__.__name__, output_frame.frame_type - ) yield output_frame elif isinstance(frames, Iterable): for frame in frames: async for output_frame in self.process_frame(requested_frame_types, frame): - print( - "yielding frame", self.__class__.__name__, output_frame.frame_type - ) yield output_frame elif isinstance(frames, asyncio.Queue): while True: frame = await frames.get() async for output_frame in self.process_frame(requested_frame_types, frame): - print( - "yielding frame", self.__class__.__name__, output_frame.frame_type - ) yield output_frame if frame.frame_type == FrameType.END_STREAM: break diff --git a/src/dailyai/services/daily_transport_service.py b/src/dailyai/services/daily_transport_service.py index 171cc36a6..be34b56a1 100644 --- a/src/dailyai/services/daily_transport_service.py +++ b/src/dailyai/services/daily_transport_service.py @@ -200,10 +200,10 @@ class DailyTransportService(EventHandler): async def marshal_frames(self): while True: - frame = await self.send_queue.get() + frame: QueueFrame | list = await self.send_queue.get() self.threadsafe_send_queue.put(frame) self.send_queue.task_done() - if frame.frame_type == FrameType.END_STREAM: + if type(frame) == QueueFrame and frame.frame_type == FrameType.END_STREAM: break def wait_for_send_queue_to_empty(self): diff --git a/src/dailyai/services/to_be_updated/huggingface_ai_service.py b/src/dailyai/services/to_be_updated/huggingface_ai_service.py index 4492cda26..86db63bf4 100644 --- a/src/dailyai/services/to_be_updated/huggingface_ai_service.py +++ b/src/dailyai/services/to_be_updated/huggingface_ai_service.py @@ -13,7 +13,6 @@ class HuggingFaceAIService(AIService): # available models at https://huggingface.co/Helsinki-NLP (**not all models use 2-character language codes**) def run_text_translation(self, sentence, source_language, target_language): translator = pipeline(f"translation", model=f"Helsinki-NLP/opus-mt-{source_language}-{target_language}") - print(translator(sentence)) return translator(sentence)[0]["translation_text"] diff --git a/src/samples/theoretical-to-real/05-sync-speech-and-text.py b/src/samples/theoretical-to-real/05-sync-speech-and-text.py index 9936dff6b..89c0230e3 100644 --- a/src/samples/theoretical-to-real/05-sync-speech-and-text.py +++ b/src/samples/theoretical-to-real/05-sync-speech-and-text.py @@ -5,6 +5,7 @@ from asyncio.queues import Queue import re from dailyai.queue_frame import QueueFrame, FrameType +from dailyai.services.ai_services import SentenceAggregator from dailyai.services.azure_ai_services import AzureLLMService from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService from dailyai.services.open_ai_services import OpenAIImageGenService @@ -31,7 +32,7 @@ async def main(room_url): #dalle = OpenAIImageGenService(image_size="1024x1024") # Get a complete audio chunk from the given text. Splitting this into its own - # coroutine lets us ensure proper ordering of the audio chunks on the output queue. + # coroutine lets us ensure proper ordering of the audio chunks on the send queue. async def get_all_audio(text): all_audio = bytearray() async for audio in tts.run_tts(text): @@ -43,14 +44,18 @@ async def main(room_url): image_text = "" tts_tasks = [] first_sentence = True - async for sentence in llm.run_llm_async_sentences( - [ - { - "role": "system", - "content": f"Describe a nature photograph suitable for use in a calendar, for the month of {month}. Include only the image description with no preamble. Limit the description to one sentence, please." - } - ] - ): + messages = [ + { + "role": "system", + "content": f"Describe a nature photograph suitable for use in a calendar, for the month of {month}. Include only the image description with no preamble. Limit the description to one sentence, please.", + } + ] + + async for frame in SentenceAggregator().run(llm.run([QueueFrame(FrameType.LLM_MESSAGE, messages)])): + if type(frame.frame_data) != str: + raise Exception("LLM service requires a string for the data field") + + sentence: str = frame.frame_data image_text += sentence if first_sentence: @@ -100,18 +105,17 @@ async def main(room_url): # likely no delay between months, but the months won't display in order. for month_data_task in asyncio.as_completed(month_tasks): data = await month_data_task - print(f"got data, queueing frames...") - transport.output_queue.put( + await transport.send_queue.put( [ QueueFrame(FrameType.IMAGE, data["image"]), QueueFrame(FrameType.AUDIO, data["audio"][0]), ] ) for audio in data["audio"][1:]: - transport.output_queue.put(QueueFrame(FrameType.AUDIO, audio)) + await transport.send_queue.put(QueueFrame(FrameType.AUDIO, audio)) # wait for the output queue to be empty, then leave the meeting - transport.output_queue.join() + transport.wait_for_send_queue_to_empty() transport.stop() month_tasks = [asyncio.create_task(get_month_data(month)) for month in months] From f9f2e2d7ea480100a5e549e53c755528530f2f7f Mon Sep 17 00:00:00 2001 From: Moishe Lettvin Date: Wed, 17 Jan 2024 19:25:01 -0500 Subject: [PATCH 4/5] stop_when_done --- src/dailyai/services/ai_services.py | 3 +-- src/dailyai/services/daily_transport_service.py | 4 ++++ src/samples/theoretical-to-real/01-say-one-thing.py | 13 +++++++------ .../theoretical-to-real/02-llm-say-one-thing.py | 3 +-- .../theoretical-to-real/04-utterance-and-speech.py | 3 +-- .../theoretical-to-real/05-sync-speech-and-text.py | 3 +-- 6 files changed, 15 insertions(+), 14 deletions(-) diff --git a/src/dailyai/services/ai_services.py b/src/dailyai/services/ai_services.py index 4c5604c0d..83c1c7099 100644 --- a/src/dailyai/services/ai_services.py +++ b/src/dailyai/services/ai_services.py @@ -163,8 +163,7 @@ class TTSService(AIService): # Convenience function to send the audio for a sentence to the given queue async def say(self, sentence, queue: asyncio.Queue): - async for audio_chunk in self.run_tts(sentence): - await queue.put(QueueFrame(FrameType.AUDIO, audio_chunk)) + await self.run_to_queue(queue, [QueueFrame(FrameType.SENTENCE, sentence)]) class ImageGenService(AIService): diff --git a/src/dailyai/services/daily_transport_service.py b/src/dailyai/services/daily_transport_service.py index be34b56a1..5349036e5 100644 --- a/src/dailyai/services/daily_transport_service.py +++ b/src/dailyai/services/daily_transport_service.py @@ -209,6 +209,10 @@ class DailyTransportService(EventHandler): def wait_for_send_queue_to_empty(self): self.threadsafe_send_queue.join() + def stop_when_done(self): + self.wait_for_send_queue_to_empty() + self.stop() + async def run(self) -> None: self.configure_daily() diff --git a/src/samples/theoretical-to-real/01-say-one-thing.py b/src/samples/theoretical-to-real/01-say-one-thing.py index 80ba91a32..9a95bbd9b 100644 --- a/src/samples/theoretical-to-real/01-say-one-thing.py +++ b/src/samples/theoretical-to-real/01-say-one-thing.py @@ -5,6 +5,7 @@ from typing import AsyncGenerator from dailyai.queue_frame import QueueFrame, FrameType from dailyai.services.daily_transport_service import DailyTransportService from dailyai.services.azure_ai_services import AzureTTSService +from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService async def main(room_url): # create a transport service object using environment variables for @@ -23,9 +24,7 @@ async def main(room_url): meeting_duration_minutes, ) transport.mic_enabled = True - - # similarly, create a tts service - tts = AzureTTSService() + tts = ElevenLabsTTSService(voice_id="ErXwobaYiN019PkySvjV") # Register an event handler so we can play the audio when the participant joins. @transport.event_handler("on_participant_joined") @@ -33,11 +32,13 @@ async def main(room_url): if participant["info"]["isLocal"]: return - await tts.say("Hello there, " + participant["info"]["userName"] + "!", transport.send_queue) + await tts.say( + "Hello there, " + participant["info"]["userName"] + "!", + transport.send_queue, + ) # wait for the output queue to be empty, then leave the meeting - transport.wait_for_send_queue_to_empty() - transport.stop() + transport.stop_when_done() await transport.run() diff --git a/src/samples/theoretical-to-real/02-llm-say-one-thing.py b/src/samples/theoretical-to-real/02-llm-say-one-thing.py index 301b2762c..425eacf2f 100644 --- a/src/samples/theoretical-to-real/02-llm-say-one-thing.py +++ b/src/samples/theoretical-to-real/02-llm-say-one-thing.py @@ -38,8 +38,7 @@ async def main(room_url): async def on_first_other_participant_joined(transport): await tts_task - transport.wait_for_send_queue_to_empty() - transport.stop() + transport.stop_when_done() await transport.run() diff --git a/src/samples/theoretical-to-real/04-utterance-and-speech.py b/src/samples/theoretical-to-real/04-utterance-and-speech.py index 0d72905b4..bf831c29b 100644 --- a/src/samples/theoretical-to-real/04-utterance-and-speech.py +++ b/src/samples/theoretical-to-real/04-utterance-and-speech.py @@ -63,8 +63,7 @@ async def main(room_url:str): await asyncio.gather(llm_response_task, buffer_to_send_queue()) - transport.wait_for_send_queue_to_empty() - transport.stop() + transport.stop_when_done() await transport.run() diff --git a/src/samples/theoretical-to-real/05-sync-speech-and-text.py b/src/samples/theoretical-to-real/05-sync-speech-and-text.py index 89c0230e3..a839d8385 100644 --- a/src/samples/theoretical-to-real/05-sync-speech-and-text.py +++ b/src/samples/theoretical-to-real/05-sync-speech-and-text.py @@ -115,8 +115,7 @@ async def main(room_url): await transport.send_queue.put(QueueFrame(FrameType.AUDIO, audio)) # wait for the output queue to be empty, then leave the meeting - transport.wait_for_send_queue_to_empty() - transport.stop() + transport.stop_when_done() month_tasks = [asyncio.create_task(get_month_data(month)) for month in months] From 0245f98eb56148d7edbf72dffadfa42e85418b89 Mon Sep 17 00:00:00 2001 From: Moishe Lettvin Date: Thu, 18 Jan 2024 11:29:13 -0500 Subject: [PATCH 5/5] update fal image_gen to use size in constructor --- src/dailyai/services/fal_ai_services.py | 12 +++++------- .../theoretical-to-real/05-sync-speech-and-text.py | 4 ++-- 2 files changed, 7 insertions(+), 9 deletions(-) diff --git a/src/dailyai/services/fal_ai_services.py b/src/dailyai/services/fal_ai_services.py index 324ff0ec4..8527cb168 100644 --- a/src/dailyai/services/fal_ai_services.py +++ b/src/dailyai/services/fal_ai_services.py @@ -9,12 +9,10 @@ from PIL import Image from dailyai.services.ai_services import LLMService, TTSService, ImageGenService # Fal expects FAL_KEY_ID and FAL_KEY_SECRET to be set in the env class FalImageGenService(ImageGenService): - def __init__(self): - super().__init__() + def __init__(self, image_size): + super().__init__(image_size) - - - async def run_image_gen(self, sentence, size) -> tuple[str, bytes]: + async def run_image_gen(self, sentence) -> tuple[str, bytes]: def get_image_url(sentence, size): print("starting fal submit...") handler = fal.apps.submit( @@ -37,7 +35,7 @@ class FalImageGenService(ImageGenService): return image_url print(f"fetching image url...") - image_url = await asyncio.to_thread(get_image_url, sentence, size) + image_url = await asyncio.to_thread(get_image_url, sentence, self.image_size) print(f"got image url, downloading image...") # Load the image from the url async with aiohttp.ClientSession() as session: @@ -48,4 +46,4 @@ class FalImageGenService(ImageGenService): image = Image.open(image_stream) return (image_url, image.tobytes()) - # return (image_url, dalle_im.tobytes()) \ No newline at end of file + # return (image_url, dalle_im.tobytes()) diff --git a/src/samples/theoretical-to-real/05-sync-speech-and-text.py b/src/samples/theoretical-to-real/05-sync-speech-and-text.py index a839d8385..b0cd1e7c7 100644 --- a/src/samples/theoretical-to-real/05-sync-speech-and-text.py +++ b/src/samples/theoretical-to-real/05-sync-speech-and-text.py @@ -27,9 +27,9 @@ async def main(room_url): transport.camera_height = 1024 llm = AzureLLMService() - dalle = FalImageGenService() + dalle = FalImageGenService(image_size="1024x1024") tts = ElevenLabsTTSService(voice_id="ErXwobaYiN019PkySvjV") - #dalle = OpenAIImageGenService(image_size="1024x1024") + # dalle = OpenAIImageGenService(image_size="1024x1024") # Get a complete audio chunk from the given text. Splitting this into its own # coroutine lets us ensure proper ordering of the audio chunks on the send queue.