services: update azure services

src/pipecat/services/azure.py

@@ -1,12 +1,20 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
 import aiohttp
 import asyncio
 import io
+
+from PIL import Image
+
 from openai import AsyncAzureOpenAI
 
-from collections.abc import AsyncGenerator
-
+from pipecat.frames.frames import AudioRawFrame, ErrorFrame, URLImageRawFrame
 from pipecat.services.ai_services import TTSService, ImageGenService
-from PIL import Image
+from pipecat.services.openai import BaseOpenAILLMService
 
 from loguru import logger
 
@@ -24,8 +32,6 @@ except ModuleNotFoundError as e:
         "In order to use Azure TTS, you need to `pip install pipecat-ai[azure]`. Also, set `AZURE_SPEECH_API_KEY` and `AZURE_SPEECH_REGION` environment variables.")
     raise Exception(f"Missing module: {e}")
 
-from pipecat.services.openai_api_llm_service import BaseOpenAILLMService
-
 
 class AzureTTSService(TTSService):
     def __init__(self, *, api_key, region, voice="en-US-SaraNeural"):
@@ -37,8 +43,9 @@ class AzureTTSService(TTSService):
         )
         self._voice = voice
 
-    async def run_tts(self, sentence) -> AsyncGenerator[bytes, None]:
-        self.logger.info("Running azure tts")
+    async def run_tts(self, text: str):
+        logger.debug(f"Transcribing text: {text}")
+
         ssml = (
             "<speak version='1.0' xml:lang='en-US' xmlns='http://www.w3.org/2001/10/synthesis' "
             "xmlns:mstts='http://www.w3.org/2001/mstts'>"
@@ -46,23 +53,19 @@ class AzureTTSService(TTSService):
             "<mstts:silence type='Sentenceboundary' value='20ms' />"
             "<mstts:express-as style='lyrical' styledegree='2' role='SeniorFemale'>"
             "<prosody rate='1.05'>"
-            f"{sentence}"
+            f"{text}"
             "</prosody></mstts:express-as></voice></speak> ")
+
         result = await asyncio.to_thread(self.speech_synthesizer.speak_ssml, (ssml))
-        self.logger.info("Got azure tts result")
+
         if result.reason == ResultReason.SynthesizingAudioCompleted:
-            self.logger.info("Returning result")
-            # azure always sends a 44-byte header. Strip it off.
-            yield result.audio_data[44:]
+            # Azure always sends a 44-byte header. Strip it off.
+            await self.push_frame(AudioRawFrame(audio=result.audio_data[44:], sample_rate=16000, num_channels=1))
         elif result.reason == ResultReason.Canceled:
             cancellation_details = result.cancellation_details
-            self.logger.info(
-                "Speech synthesis canceled: {}".format(
-                    cancellation_details.reason))
+            logger.warning(f"Speech synthesis canceled: {cancellation_details.reason}")
             if cancellation_details.reason == CancellationReason.Error:
-                self.logger.info(
-                    "Error details: {}".format(
-                        cancellation_details.error_details))
+                logger.error(f"Error details: {cancellation_details.error_details}")
 
 
 class AzureLLMService(BaseOpenAILLMService):
@@ -73,10 +76,9 @@ class AzureLLMService(BaseOpenAILLMService):
             endpoint,
             api_version="2023-12-01-preview",
             model):
+        super().__init__(api_key=api_key, model=model)
         self._endpoint = endpoint
         self._api_version = api_version
-
-        super().__init__(api_key=api_key, model=model)
         self._model: str = model
 
     def create_client(self, api_key=None, base_url=None):
@@ -108,20 +110,21 @@ class AzureImageGenServiceREST(ImageGenService):
         self._aiohttp_session = aiohttp_session
         self._image_size = image_size
 
-    async def run_image_gen(self, prompt: str) -> tuple[str, bytes, tuple[int, int]]:
+    async def run_image_gen(self, prompt: str):
         url = f"{self._azure_endpoint}openai/images/generations:submit?api-version={self._api_version}"
+
         headers = {
             "api-key": self._api_key,
             "Content-Type": "application/json"}
+
         body = {
             # Enter your prompt text here
             "prompt": prompt,
             "size": self._image_size,
             "n": 1,
         }
-        async with self._aiohttp_session.post(
-            url, headers=headers, json=body
-        ) as submission:
+
+        async with self._aiohttp_session.post(url, headers=headers, json=body) as submission:
             # We never get past this line, because this header isn't
             # defined on a 429 response, but something is eating our
             # exceptions!
@@ -132,21 +135,30 @@ class AzureImageGenServiceREST(ImageGenService):
             while status != "succeeded":
                 attempts_left -= 1
                 if attempts_left == 0:
-                    raise Exception("Image generation timed out")
+                    logger.error("Image generation timed out")
+                    await self.push_error(ErrorFrame("Image generation timed out"))
+                    return
 
                 await asyncio.sleep(1)
-                response = await self._aiohttp_session.get(
-                    operation_location, headers=headers
-                )
+
+                response = await self._aiohttp_session.get(operation_location, headers=headers)
+
                 json_response = await response.json()
                 status = json_response["status"]
 
-            image_url = (
-                json_response["result"]["data"][0]["url"] if json_response else None)
+            image_url = json_response["result"]["data"][0]["url"] if json_response else None
             if not image_url:
-                raise Exception("Image generation failed")
+                logger.error("Image generation failed")
+                await self.push_error(ErrorFrame("Image generation failed"))
+                return
+
             # Load the image from the url
             async with self._aiohttp_session.get(image_url) as response:
                 image_stream = io.BytesIO(await response.content.read())
                 image = Image.open(image_stream)
-                return (image_url, image.tobytes(), image.size)
+                frame = URLImageRawFrame(
+                    url=image_url,
+                    image=image.tobytes(),
+                    size=image.size,
+                    format=image.format)
+                await self.push_frame(frame)

src/pipecat/services/openai.py

@@ -1,3 +1,9 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
 import io
 import json
 import time

src/pipecat/transports/base_transport.py

@@ -33,8 +33,8 @@ class BaseTransport(ABC):
 
     @abstractmethod
     def input(self) -> FrameProcessor:
-        pass
+        raise NotImplementedError
 
     @abstractmethod
     def output(self) -> FrameProcessor:
-        pass
+        raise NotImplementedError