Fix Issue 264 - Using Shared WebSocket Class Implementation

Author: davidvonthenen

deepgram/audio/speaker/speaker.py

@@ -9,6 +9,8 @@
 from typing import Optional, Callable, Union, TYPE_CHECKING
 import logging
 
+import websockets
+
 from ...utils import verboselogs
 from .constants import LOGGING, CHANNELS, RATE, CHUNK, TIMEOUT
 
@@ -24,22 +26,22 @@ class Speaker: # pylint: disable=too-many-instance-attributes
  _logger: verboselogs.VerboseLogger
 
  _audio: "pyaudio.PyAudio"
- _stream: "pyaudio.Stream"
+ _stream: Optional["pyaudio.Stream"] = None
 
  _chunk: int
  _rate: int
  _channels: int
- _output_device_index: Optional[int]
+ _output_device_index: Optional[int] = None
 
  _queue: queue.Queue
  _exit: threading.Event
 
- _thread: threading.Thread
+ _thread: Optional[threading.Thread] = None
  # _asyncio_loop: asyncio.AbstractEventLoop
  # _asyncio_thread: threading.Thread
- _receiver_thread: threading.Thread
+ _receiver_thread: Optional[threading.Thread] = None
 
- _loop: asyncio.AbstractEventLoop
+ _loop: Optional[asyncio.AbstractEventLoop] = None
 
  _push_callback_org: Optional[Callable] = None
  _push_callback: Optional[Callable] = None
@@ -217,6 +219,17 @@ async def _start_asyncio_receiver(self):
  elif isinstance(message, bytes):
  self._logger.verbose("Received audio data...")
  self.add_audio_to_queue(message)
+ except websockets.exceptions.ConnectionClosedOK as e:
+ self._logger.debug("send() exiting gracefully: %d", e.code)
+ except websockets.exceptions.ConnectionClosed as e:
+ if e.code in [1000, 1001]:
+ self._logger.debug("send() exiting gracefully: %d", e.code)
+ return
+ self._logger.error("_start_asyncio_receiver - ConnectionClosed: %s", str(e))
+ except websockets.exceptions.WebSocketException as e:
+ self._logger.error(
+ "_start_asyncio_receiver- WebSocketException: %s", str(e)
+ )
  except Exception as e: # pylint: disable=broad-except
  self._logger.error("_start_asyncio_receiver exception: %s", str(e))
 
@@ -266,23 +279,26 @@ def finish(self) -> bool:
  self._logger.notice("stopping stream...")
  self._stream.stop_stream()
  self._stream.close()
- self._stream = None # type: ignore
+ self._stream = None
  self._logger.notice("stream stopped")
 
- self._thread.join()
- self._thread = None # type: ignore
+ if self._thread is not None:
+ self._logger.notice("joining thread...")
+ self._thread.join()
+ self._thread = None
+ self._logger.notice("thread stopped")
 
  # if self._asyncio_thread is not None:
  # self._logger.notice("stopping asyncio loop...")
  # self._asyncio_loop.call_soon_threadsafe(self._asyncio_loop.stop)
  # self._asyncio_thread.join()
- # self._asyncio_thread = None # type: ignore
+ # self._asyncio_thread = None
  # self._logger.notice("_asyncio_thread joined")
 
  if self._receiver_thread is not None:
  self._logger.notice("stopping asyncio loop...")
  self._receiver_thread.join()
- self._receiver_thread = None # type: ignore
+ self._receiver_thread = None
  self._logger.notice("_receiver_thread joined")
 
  self._queue = None # type: ignore

deepgram/clients/__init__.py

@@ -40,8 +40,13 @@
  UnhandledResponse,
  ErrorResponse,
 )
-from .common import DeepgramError, DeepgramTypeError
-from .errors import DeepgramModuleError, DeepgramApiError, DeepgramUnknownApiError
+from .common import (
+ DeepgramError,
+ DeepgramTypeError,
+ DeepgramApiError,
+ DeepgramUnknownApiError,
+)
+from .errors import DeepgramModuleError
 
 from .listen_router import Listen
 from .read_router import Read

deepgram/clients/analyze/v1/async_client.py

@@ -9,8 +9,8 @@
 
 from ....utils import verboselogs
 from ....options import DeepgramClientOptions
-from ...abstract_async_client import AbstractAsyncRestClient
-from ...common.v1.errors import DeepgramError, DeepgramTypeError
+from ...common import AbstractAsyncRestClient
+from ...common import DeepgramError, DeepgramTypeError
 
 from .helpers import is_buffer_source, is_readstream_source, is_url_source
 from .options import (

deepgram/clients/analyze/v1/client.py

@@ -9,8 +9,8 @@
 
 from ....utils import verboselogs
 from ....options import DeepgramClientOptions
-from ...abstract_sync_client import AbstractSyncRestClient
-from ...common.v1.errors import DeepgramError, DeepgramTypeError
+from ...common import AbstractSyncRestClient
+from ...common import DeepgramError, DeepgramTypeError
 
 from .helpers import is_buffer_source, is_readstream_source, is_url_source
 from .options import (

deepgram/clients/common/__init__.py

@@ -2,7 +2,17 @@
 # Use of this source code is governed by a MIT license that can be found in the LICENSE file.
 # SPDX-License-Identifier: MIT
 
-from .v1 import DeepgramError, DeepgramTypeError
+from .v1 import (
+ DeepgramError,
+ DeepgramTypeError,
+ DeepgramApiError,
+ DeepgramUnknownApiError,
+)
+
+from .v1 import AbstractAsyncRestClient
+from .v1 import AbstractSyncRestClient
+from .v1 import AbstractAsyncWebSocketClient
+from .v1 import AbstractSyncWebSocketClient
 
 from .v1 import (
  TextSource as TextSourceLatest,

deepgram/clients/common/v1/__init__.py

@@ -4,7 +4,16 @@
 
 from .enums import Sentiment
 
-from .errors import DeepgramError, DeepgramTypeError
+from .errors import (
+ DeepgramError,
+ DeepgramTypeError,
+ DeepgramApiError,
+ DeepgramUnknownApiError,
+)
+from .abstract_async_rest import AbstractAsyncRestClient
+from .abstract_sync_rest import AbstractSyncRestClient
+from .abstract_async_websocket import AbstractAsyncWebSocketClient
+from .abstract_sync_websocket import AbstractSyncWebSocketClient
 
 from .options import (
  TextSource,

deepgram/clients/abstract_async_client.py renamed to deepgram/clients/common/v1/abstract_async_rest.py

@@ -9,9 +9,8 @@
 import httpx
 
 from .helpers import append_query_params
-from ..options import DeepgramClientOptions
-from .errors import DeepgramApiError, DeepgramUnknownApiError
-from .common.v1.errors import DeepgramError
+from ....options import DeepgramClientOptions
+from .errors import DeepgramError, DeepgramApiError, DeepgramUnknownApiError
 
 
 class AbstractAsyncRestClient: