Implement wait_for_complete() for Speaker class

Author: davidvonthenen

deepgram/audio/__init__.py

@@ -18,4 +18,5 @@
  CHANNELS as OUTPUT_CHANNELS,
  RATE as OUTPUT_RATE,
  CHUNK as OUTPUT_CHUNK,
+ PLAYBACK_DELTA as OUTPUT_PLAYBACK_DELTA,
 )

deepgram/audio/speaker/__init__.py

@@ -4,4 +4,4 @@
 
 from .speaker import Speaker
 from .errors import DeepgramSpeakerError
-from .constants import LOGGING, CHANNELS, RATE, CHUNK
+from .constants import LOGGING, CHANNELS, RATE, CHUNK, PLAYBACK_DELTA

deepgram/audio/speaker/constants.py

@@ -4,9 +4,12 @@
 
 from ...utils import verboselogs
 
-# Constants for microphone
+# Constants for speaker
 LOGGING = verboselogs.WARNING
 TIMEOUT = 0.050
 CHANNELS = 1
 RATE = 16000
 CHUNK = 8194
+
+# Constants for speaker
+PLAYBACK_DELTA = 2000

deepgram/audio/speaker/speaker.py

@@ -8,15 +8,18 @@
 import threading
 from typing import Optional, Callable, Union, TYPE_CHECKING
 import logging
+from datetime import datetime
 
 import websockets
 
 from ...utils import verboselogs
-from .constants import LOGGING, CHANNELS, RATE, CHUNK, TIMEOUT
+from .constants import LOGGING, CHANNELS, RATE, CHUNK, TIMEOUT, PLAYBACK_DELTA
 
 if TYPE_CHECKING:
  import pyaudio
 
+HALF_SECOND = 0.5
+
 
 class Speaker: # pylint: disable=too-many-instance-attributes
  """
@@ -33,6 +36,11 @@ class Speaker: # pylint: disable=too-many-instance-attributes
  _channels: int
  _output_device_index: Optional[int] = None
 
+ # last time we received audio
+ _last_datagram: datetime = datetime.now()
+ _last_play_delta_in_ms: int
+ _lock_wait: threading.Lock
+
  _queue: queue.Queue
  _exit: threading.Event
 
@@ -56,6 +64,7 @@ def __init__(
  rate: int = RATE,
  chunk: int = CHUNK,
  channels: int = CHANNELS,
+ last_play_delta_in_ms: int = PLAYBACK_DELTA,
  output_device_index: Optional[int] = None,
  ): # pylint: disable=too-many-positional-arguments
  # dynamic import of pyaudio as not to force the requirements on the SDK (and users)
@@ -68,11 +77,15 @@ def __init__(
  self._exit = threading.Event()
  self._queue = queue.Queue()
 
+ self._last_datagram = datetime.now()
+ self._lock_wait = threading.Lock()
+
  self._audio = pyaudio.PyAudio()
  self._chunk = chunk
  self._rate = rate
  self._format = pyaudio.paInt16
  self._channels = channels
+ self._last_play_delta_in_ms = last_play_delta_in_ms
  self._output_device_index = output_device_index
 
  self._push_callback_org = push_callback
@@ -192,6 +205,42 @@ def start(self, active_loop: Optional[asyncio.AbstractEventLoop] = None) -> bool
 
  return True
 
+ def wait_for_complete(self):
+ """
+ This method will block until the speak is done playing sound.
+ """
+ self._logger.debug("Speaker.wait_for_complete ENTER")
+
+ delta_in_ms = float(self._last_play_delta_in_ms)
+ self._logger.debug("Last Play delta: %f", delta_in_ms)
+
+ # set to now
+ with self._lock_wait:
+ self._last_datagram = datetime.now()
+
+ while True:
+ # sleep for a bit
+ self._exit.wait(HALF_SECOND)
+
+ # check if we should exit
+ if self._exit.is_set():
+ self._logger.debug("Exiting wait_for_complete _exit is set")
+ break
+
+ # check the time
+ with self._lock_wait:
+ delta = datetime.now() - self._last_datagram
+ diff_in_ms = delta.total_seconds() * 1000
+ if diff_in_ms < delta_in_ms:
+ self._logger.debug("LastPlay delta is less than threshold")
+ continue
+
+ # if we get here, we are done playing audio
+ self._logger.debug("LastPlay delta is greater than threshold. Exit wait!")
+ break
+
+ self._logger.debug("Speaker.wait_for_complete LEAVE")
+
  def _start_receiver(self):
  # Check if the socket is an asyncio WebSocket
  if inspect.iscoroutinefunction(self._pull_callback_org):
@@ -315,6 +364,8 @@ def _play(self, audio_out, stream, stop):
  while not stop.is_set():
  try:
  data = audio_out.get(True, TIMEOUT)
+ with self._lock_wait:
+ self._last_datagram = datetime.now()
  stream.write(data)
  except queue.Empty:
  pass

deepgram/clients/speak/v1/websocket/async_client.py

@@ -27,7 +27,7 @@
 )
 from .options import SpeakWSOptions
 
-from .....audio.speaker import Speaker, RATE, CHANNELS
+from .....audio.speaker import Speaker, RATE, CHANNELS, PLAYBACK_DELTA
 
 ONE_SECOND = 1
 HALF_SECOND = 0.5
@@ -93,6 +93,11 @@ def __init__(self, config: DeepgramClientOptions):
  channels = self._config.options.get("speaker_playback_channels")
  if channels is None:
  channels = CHANNELS
+ playback_delta_in_ms = self._config.options.get(
+ "speaker_playback_delta_in_ms"
+ )
+ if playback_delta_in_ms is None:
+ playback_delta_in_ms = PLAYBACK_DELTA
  device_index = self._config.options.get("speaker_playback_device_index")
 
  self._logger.debug("rate: %s", rate)
@@ -103,13 +108,15 @@ def __init__(self, config: DeepgramClientOptions):
  self._speaker = Speaker(
  rate=rate,
  channels=channels,
+ last_play_delta_in_ms=playback_delta_in_ms,
  verbose=self._config.verbose,
  output_device_index=device_index,
  )
  else:
  self._speaker = Speaker(
  rate=rate,
  channels=channels,
+ last_play_delta_in_ms=playback_delta_in_ms,
  verbose=self._config.verbose,
  )
 
@@ -590,6 +597,21 @@ async def clear(self) -> bool:
 
  return True
 
+ async def wait_for_complete(self):
+ """
+ This method will block until the speak is done playing sound.
+ """
+ self._logger.spam("AsyncSpeakWebSocketClient.wait_for_complete ENTER")
+
+ if self._speaker is None:
+ self._logger.error("speaker is None. Return immediately")
+ return
+
+ loop = asyncio.get_event_loop()
+ await loop.run_in_executor(None, self._speaker.wait_for_complete)
+ self._logger.notice("wait_for_complete succeeded")
+ self._logger.spam("AsyncSpeakWebSocketClient.wait_for_complete LEAVE")
+
  async def _close_message(self) -> bool:
  return await self.send_control(SpeakWebSocketMessage.Close)
 

deepgram/clients/speak/v1/websocket/client.py

@@ -27,7 +27,7 @@
 )
 from .options import SpeakWSOptions
 
-from .....audio.speaker import Speaker, RATE, CHANNELS
+from .....audio.speaker import Speaker, RATE, CHANNELS, PLAYBACK_DELTA
 
 ONE_SECOND = 1
 HALF_SECOND = 0.5
@@ -96,6 +96,11 @@ def __init__(self, config: DeepgramClientOptions):
  channels = self._config.options.get("speaker_playback_channels")
  if channels is None:
  channels = CHANNELS
+ playback_delta_in_ms = self._config.options.get(
+ "speaker_playback_delta_in_ms"
+ )
+ if playback_delta_in_ms is None:
+ playback_delta_in_ms = PLAYBACK_DELTA
  device_index = self._config.options.get("speaker_playback_device_index")
 
  self._logger.debug("rate: %s", rate)
@@ -106,13 +111,15 @@ def __init__(self, config: DeepgramClientOptions):
  self._speaker = Speaker(
  rate=rate,
  channels=channels,
+ last_play_delta_in_ms=playback_delta_in_ms,
  verbose=self._config.verbose,
  output_device_index=device_index,
  )
  else:
  self._speaker = Speaker(
  rate=rate,
  channels=channels,
+ last_play_delta_in_ms=playback_delta_in_ms,
  verbose=self._config.verbose,
  )
 
@@ -589,6 +596,20 @@ def clear(self) -> bool:
 
  return True
 
+ def wait_for_complete(self):
+ """
+ This method will block until the speak is done playing sound.
+ """
+ self._logger.spam("SpeakWebSocketClient.wait_for_complete ENTER")
+
+ if self._speaker is None:
+ self._logger.error("speaker is None. Return immediately")
+ return
+
+ self._speaker.wait_for_complete()
+ self._logger.notice("wait_for_complete succeeded")
+ self._logger.spam("SpeakWebSocketClient.wait_for_complete LEAVE")
+
  def _close_message(self) -> bool:
  return self.send_control(SpeakWebSocketMessage.Close)
 

examples/text-to-speech/websocket/async_complete/main.py

@@ -32,8 +32,11 @@ async def main():
 
  # example of setting up a client config. logging values: WARNING, VERBOSE, DEBUG, SPAM
  config: DeepgramClientOptions = DeepgramClientOptions(
- options={"auto_flush_speak_delta": "500", "speaker_playback": "true"},
- # verbose=verboselogs.SPAM,
+ options={
+ # "auto_flush_speak_delta": "500",
+ "speaker_playback": "true"
+ },
+ verbose=verboselogs.SPAM,
  )
  deepgram: DeepgramClient = DeepgramClient("", config)
 
@@ -99,11 +102,12 @@ async def on_unhandled(self, unhandled, **kwargs):
 
  # send the text to Deepgram
  await dg_connection.send_text(TTS_TEXT)
+
  # if auto_flush_speak_delta is not used, you must flush the connection by calling flush()
  await dg_connection.flush()
 
  # Indicate that we've finished
- await asyncio.sleep(7)
+ await dg_connection.wait_for_complete()
 
  # Close the connection
  await dg_connection.finish()

examples/text-to-speech/websocket/complete/main.py

@@ -93,11 +93,13 @@ def on_unhandled(self, unhandled, **kwargs):
 
  # send the text to Deepgram
  dg_connection.send_text(TTS_TEXT)
+
  # if auto_flush_speak_delta is not used, you must flush the connection by calling flush()
  dg_connection.flush()
 
  # Indicate that we've finished
- time.sleep(5)
+ dg_connection.wait_for_complete()
+
  print("\n\nPress Enter to stop...\n\n")
  input()