SmartTurn: some linting cleanup

src/pipecat/audio/turn/base_smart_turn.py

@@ -46,7 +46,7 @@ class BaseSmartTurn(BaseTurnAnalyzer):
         self._audio_buffer = []
         self._speech_triggered = False
         self._silence_ms = 0
-        self._speech_start_time = None
+        self._speech_start_time = 0
 
     @property
     def speech_triggered(self) -> bool:
@@ -64,7 +64,7 @@ class BaseSmartTurn(BaseTurnAnalyzer):
             # Reset silence tracking on speech
             self._silence_ms = 0
             self._speech_triggered = True
-            if self._speech_start_time is None:
+            if self._speech_start_time == 0:
                 self._speech_start_time = time.time()
         else:
             if self._speech_triggered:
@@ -102,7 +102,7 @@ class BaseSmartTurn(BaseTurnAnalyzer):
         # If the state is still incomplete, keep the _speech_triggered as True
         self._speech_triggered = turn_state == EndOfTurnState.INCOMPLETE
         self._audio_buffer = []
-        self._speech_start_time = None
+        self._speech_start_time = 0
         self._silence_ms = 0
 
     async def _process_speech_segment(
@@ -179,11 +179,11 @@ class BaseSmartTurn(BaseTurnAnalyzer):
         return state, result_data
 
     @abstractmethod
-    async def _predict_endpoint(self, buffer: np.ndarray) -> Dict[str, Any]:
+    async def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, Any]:
         """Abstract method to predict if a turn has ended based on audio.
 
         Args:
-            buffer: Float32 numpy array of audio samples at 16kHz.
+            audio_array: Float32 numpy array of audio samples at 16kHz.
 
         Returns:
             Dictionary with:

src/pipecat/audio/turn/local_smart_turn.py

@@ -5,17 +5,16 @@
 #
 
 
-import os
-from typing import Dict
+from typing import Any, Dict
 
 import numpy as np
-import torch
 from loguru import logger
 
 from pipecat.audio.turn.base_smart_turn import BaseSmartTurn
 
 try:
     import coremltools as ct
+    import torch
     from transformers import AutoFeatureExtractor
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
@@ -41,7 +40,7 @@ class LocalCoreMLSmartTurnAnalyzer(BaseSmartTurn):
         self._turn_model = ct.models.MLModel(core_ml_model_path)
         logger.debug("Loaded Local Smart Turn")
 
-    async def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, any]:
+    async def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, Any]:
         inputs = self._turn_processor(
             audio_array,
             sampling_rate=16000,

src/pipecat/audio/turn/smart_turn.py

@@ -7,7 +7,7 @@
 
 import asyncio
 import io
-from typing import Dict
+from typing import Any, Dict
 
 import aiohttp
 import numpy as np
@@ -19,13 +19,9 @@ from pipecat.audio.turn.base_smart_turn import BaseSmartTurn, SmartTurnTimeoutEx
 class SmartTurnAnalyzer(BaseSmartTurn):
     def __init__(self, url: str, aiohttp_session: aiohttp.ClientSession, **kwargs):
         super().__init__(**kwargs)
-        self.remote_smart_turn_url = url
+        self._url = url
         self._aiohttp_session = aiohttp_session
 
-        if not self.remote_smart_turn_url:
-            logger.error("remote_smart_turn_url is not set.")
-            raise Exception("remote_smart_turn_url must be provided.")
-
     def _serialize_array(self, audio_array: np.ndarray) -> bytes:
         logger.trace("Serializing NumPy array to bytes...")
         buffer = io.BytesIO()
@@ -34,16 +30,14 @@ class SmartTurnAnalyzer(BaseSmartTurn):
         logger.trace(f"Serialized size: {len(serialized_bytes)} bytes")
         return serialized_bytes
 
-    async def _send_raw_request(self, data_bytes: bytes):
+    async def _send_raw_request(self, data_bytes: bytes) -> Dict[str, Any]:
         headers = {"Content-Type": "application/octet-stream"}
-        logger.trace(
-            f"Sending {len(data_bytes)} bytes as raw body to {self.remote_smart_turn_url}..."
-        )
+        logger.trace(f"Sending {len(data_bytes)} bytes as raw body to {self._url}...")
         try:
             timeout = aiohttp.ClientTimeout(total=self._params.stop_secs)
 
             async with self._aiohttp_session.post(
-                self.remote_smart_turn_url, data=data_bytes, headers=headers, timeout=timeout
+                self._url, data=data_bytes, headers=headers, timeout=timeout
             ) as response:
                 logger.trace("\n--- Response ---")
                 logger.trace(f"Status Code: {response.status}")
@@ -73,6 +67,6 @@ class SmartTurnAnalyzer(BaseSmartTurn):
             logger.error(f"Failed to send raw request to Daily Smart Turn: {e}")
             raise Exception("Failed to send raw request to Daily Smart Turn.")
 
-    async def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, any]:
+    async def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, Any]:
         serialized_array = self._serialize_array(audio_array)
         return await self._send_raw_request(serialized_array)