deepgram.clients.listen.v1.websocket.client

  1# Copyright 2023-2024 Deepgram SDK contributors. All Rights Reserved.
  2# Use of this source code is governed by a MIT license that can be found in the LICENSE file.
  3# SPDX-License-Identifier: MIT
  4import json
  5import time
  6import logging
  7from typing import Dict, Union, Optional, cast, Any, Callable, Type
  8from datetime import datetime
  9import threading
 10
 11from .....utils import verboselogs
 12from .....options import DeepgramClientOptions
 13from ...enums import LiveTranscriptionEvents
 14from ....common import AbstractSyncWebSocketClient
 15from ....common import DeepgramError
 16
 17from .response import (
 18    OpenResponse,
 19    LiveResultResponse,
 20    MetadataResponse,
 21    SpeechStartedResponse,
 22    UtteranceEndResponse,
 23    CloseResponse,
 24    ErrorResponse,
 25    UnhandledResponse,
 26)
 27from .options import ListenWebSocketOptions
 28
 29ONE_SECOND = 1
 30HALF_SECOND = 0.5
 31DEEPGRAM_INTERVAL = 5
 32PING_INTERVAL = 20
 33
 34
 35class ListenWebSocketClient(
 36    AbstractSyncWebSocketClient
 37):  # pylint: disable=too-many-instance-attributes
 38    """
 39    Client for interacting with Deepgram's live transcription services over WebSockets.
 40
 41    This class provides methods to establish a WebSocket connection for live transcription and handle real-time transcription events.
 42
 43    Args:
 44        config (DeepgramClientOptions): all the options for the client.
 45        thread_cls (Type[threading.Thread]): optional thread class to use for creating threads,
 46        defaults to threading.Thread. Useful for custom thread management like ContextVar support.
 47    """
 48
 49    _logger: verboselogs.VerboseLogger
 50    _config: DeepgramClientOptions
 51    _endpoint: str
 52
 53    _lock_flush: threading.Lock
 54    _event_handlers: Dict[LiveTranscriptionEvents, list]
 55
 56    _keep_alive_thread: Union[threading.Thread, None]
 57    _flush_thread: Union[threading.Thread, None]
 58    _last_datagram: Optional[datetime] = None
 59
 60    _thread_cls: Type[threading.Thread]
 61
 62    _kwargs: Optional[Dict] = None
 63    _addons: Optional[Dict] = None
 64    _options: Optional[Dict] = None
 65    _headers: Optional[Dict] = None
 66
 67    def __init__(
 68        self,
 69        config: DeepgramClientOptions,
 70        thread_cls: Type[threading.Thread] = threading.Thread,
 71    ):
 72        if config is None:
 73            raise DeepgramError("Config is required")
 74
 75        self._logger = verboselogs.VerboseLogger(__name__)
 76        self._logger.addHandler(logging.StreamHandler())
 77        self._logger.setLevel(config.verbose)
 78
 79        self._config = config
 80        self._endpoint = "v1/listen"
 81
 82        self._flush_thread = None
 83        self._keep_alive_thread = None
 84
 85        # auto flush
 86        self._last_datagram = None
 87        self._lock_flush = threading.Lock()
 88
 89        self._thread_cls = thread_cls
 90
 91        # init handlers
 92        self._event_handlers = {
 93            event: [] for event in LiveTranscriptionEvents.__members__.values()
 94        }
 95
 96        # call the parent constructor
 97        super().__init__(
 98            config=self._config,
 99            endpoint=self._endpoint,
100            thread_cls=self._thread_cls,
101        )
102
103    # pylint: disable=too-many-statements,too-many-branches
104    def start(
105        self,
106        options: Optional[Union[ListenWebSocketOptions, Dict]] = None,
107        addons: Optional[Dict] = None,
108        headers: Optional[Dict] = None,
109        members: Optional[Dict] = None,
110        **kwargs,
111    ) -> bool:
112        """
113        Starts the WebSocket connection for live transcription.
114        """
115        self._logger.debug("ListenWebSocketClient.start ENTER")
116        self._logger.info("options: %s", options)
117        self._logger.info("addons: %s", addons)
118        self._logger.info("headers: %s", headers)
119        self._logger.info("members: %s", members)
120        self._logger.info("kwargs: %s", kwargs)
121
122        if isinstance(options, ListenWebSocketOptions) and not options.check():
123            self._logger.error("options.check failed")
124            self._logger.debug("ListenWebSocketClient.start LEAVE")
125            raise DeepgramError("Fatal transcription options error")
126
127        self._addons = addons
128        self._headers = headers
129
130        # add "members" as members of the class
131        if members is not None:
132            self.__dict__.update(members)
133
134        # set kwargs as members of the class
135        if kwargs is not None:
136            self._kwargs = kwargs
137        else:
138            self._kwargs = {}
139
140        if isinstance(options, ListenWebSocketOptions):
141            self._logger.info("ListenWebSocketOptions switching class -> dict")
142            self._options = options.to_dict()
143        elif options is not None:
144            self._options = options
145        else:
146            self._options = {}
147
148        try:
149            # call parent start
150            if (
151                super().start(
152                    self._options,
153                    self._addons,
154                    self._headers,
155                    **dict(cast(Dict[Any, Any], self._kwargs)),
156                )
157                is False
158            ):
159                self._logger.error("ListenWebSocketClient.start failed")
160                self._logger.debug("ListenWebSocketClient.start LEAVE")
161                return False
162
163            # debug the threads
164            for thread in threading.enumerate():
165                self._logger.debug("after running thread: %s", thread.name)
166            self._logger.debug("number of active threads: %s", threading.active_count())
167
168            # keepalive thread
169            if self._config.is_keep_alive_enabled():
170                self._logger.notice("keepalive is enabled")
171                self._keep_alive_thread = self._thread_cls(target=self._keep_alive)
172                self._keep_alive_thread.start()
173            else:
174                self._logger.notice("keepalive is disabled")
175
176            # flush thread
177            if self._config.is_auto_flush_reply_enabled():
178                self._logger.notice("autoflush is enabled")
179                self._flush_thread = self._thread_cls(target=self._flush)
180                self._flush_thread.start()
181            else:
182                self._logger.notice("autoflush is disabled")
183
184            # debug the threads
185            for thread in threading.enumerate():
186                self._logger.debug("after running thread: %s", thread.name)
187            self._logger.debug("number of active threads: %s", threading.active_count())
188
189            self._logger.notice("start succeeded")
190            self._logger.debug("ListenWebSocketClient.start LEAVE")
191            return True
192
193        except Exception as e:  # pylint: disable=broad-except
194            self._logger.error(
195                "WebSocketException in ListenWebSocketClient.start: %s", e
196            )
197            self._logger.debug("ListenWebSocketClient.start LEAVE")
198            if self._config.options.get("termination_exception_connect") is True:
199                raise e
200            return False
201
202    # pylint: enable=too-many-statements,too-many-branches
203
204    def on(
205        self, event: LiveTranscriptionEvents, handler: Callable
206    ) -> None:  # registers event handlers for specific events
207        """
208        Registers event handlers for specific events.
209        """
210        self._logger.info("event subscribed: %s", event)
211        if event in LiveTranscriptionEvents.__members__.values() and callable(handler):
212            self._event_handlers[event].append(handler)
213
214    def _emit(self, event: LiveTranscriptionEvents, *args, **kwargs) -> None:
215        """
216        Emits events to the registered event handlers.
217        """
218        self._logger.debug("ListenWebSocketClient._emit ENTER")
219        self._logger.debug("callback handlers for: %s", event)
220
221        # debug the threads
222        for thread in threading.enumerate():
223            self._logger.debug("after running thread: %s", thread.name)
224        self._logger.debug("number of active threads: %s", threading.active_count())
225
226        self._logger.debug("callback handlers for: %s", event)
227        for handler in self._event_handlers[event]:
228            handler(self, *args, **kwargs)
229
230        # debug the threads
231        for thread in threading.enumerate():
232            self._logger.debug("after running thread: %s", thread.name)
233        self._logger.debug("number of active threads: %s", threading.active_count())
234
235        self._logger.debug("ListenWebSocketClient._emit LEAVE")
236
237    # pylint: disable=too-many-return-statements,too-many-statements,too-many-locals,too-many-branches
238    def _process_text(self, message: str) -> None:
239        """
240        Processes messages received over the WebSocket connection.
241        """
242        self._logger.debug("ListenWebSocketClient._process_text ENTER")
243
244        try:
245            if len(message) == 0:
246                self._logger.debug("message is empty")
247                self._logger.debug("ListenWebSocketClient._process_text LEAVE")
248                return
249
250            data = json.loads(message)
251            response_type = data.get("type")
252            self._logger.debug("response_type: %s, data: %s", response_type, data)
253
254            match response_type:
255                case LiveTranscriptionEvents.Open:
256                    open_result: OpenResponse = OpenResponse.from_json(message)
257                    self._logger.verbose("OpenResponse: %s", open_result)
258                    self._emit(
259                        LiveTranscriptionEvents(LiveTranscriptionEvents.Open),
260                        open=open_result,
261                        **dict(cast(Dict[Any, Any], self._kwargs)),
262                    )
263                case LiveTranscriptionEvents.Transcript:
264                    msg_result: LiveResultResponse = LiveResultResponse.from_json(
265                        message
266                    )
267                    self._logger.verbose("LiveResultResponse: %s", msg_result)
268
269                    #  auto flush
270                    if self._config.is_inspecting_listen():
271                        inspect_res = self._inspect(msg_result)
272                        if not inspect_res:
273                            self._logger.error("inspect_res failed")
274
275                    self._emit(
276                        LiveTranscriptionEvents(LiveTranscriptionEvents.Transcript),
277                        result=msg_result,
278                        **dict(cast(Dict[Any, Any], self._kwargs)),
279                    )
280                case LiveTranscriptionEvents.Metadata:
281                    meta_result: MetadataResponse = MetadataResponse.from_json(message)
282                    self._logger.verbose("MetadataResponse: %s", meta_result)
283                    self._emit(
284                        LiveTranscriptionEvents(LiveTranscriptionEvents.Metadata),
285                        metadata=meta_result,
286                        **dict(cast(Dict[Any, Any], self._kwargs)),
287                    )
288                case LiveTranscriptionEvents.SpeechStarted:
289                    ss_result: SpeechStartedResponse = SpeechStartedResponse.from_json(
290                        message
291                    )
292                    self._logger.verbose("SpeechStartedResponse: %s", ss_result)
293                    self._emit(
294                        LiveTranscriptionEvents(LiveTranscriptionEvents.SpeechStarted),
295                        speech_started=ss_result,
296                        **dict(cast(Dict[Any, Any], self._kwargs)),
297                    )
298                case LiveTranscriptionEvents.UtteranceEnd:
299                    ue_result: UtteranceEndResponse = UtteranceEndResponse.from_json(
300                        message
301                    )
302                    self._logger.verbose("UtteranceEndResponse: %s", ue_result)
303                    self._emit(
304                        LiveTranscriptionEvents(LiveTranscriptionEvents.UtteranceEnd),
305                        utterance_end=ue_result,
306                        **dict(cast(Dict[Any, Any], self._kwargs)),
307                    )
308                case LiveTranscriptionEvents.Close:
309                    close_result: CloseResponse = CloseResponse.from_json(message)
310                    self._logger.verbose("CloseResponse: %s", close_result)
311                    self._emit(
312                        LiveTranscriptionEvents(LiveTranscriptionEvents.Close),
313                        close=close_result,
314                        **dict(cast(Dict[Any, Any], self._kwargs)),
315                    )
316                case LiveTranscriptionEvents.Error:
317                    err_error: ErrorResponse = ErrorResponse.from_json(message)
318                    self._logger.verbose("ErrorResponse: %s", err_error)
319                    self._emit(
320                        LiveTranscriptionEvents(LiveTranscriptionEvents.Error),
321                        error=err_error,
322                        **dict(cast(Dict[Any, Any], self._kwargs)),
323                    )
324                case _:
325                    self._logger.warning(
326                        "Unknown Message: response_type: %s, data: %s",
327                        response_type,
328                        data,
329                    )
330                    unhandled_error: UnhandledResponse = UnhandledResponse(
331                        type=LiveTranscriptionEvents(LiveTranscriptionEvents.Unhandled),
332                        raw=message,
333                    )
334                    self._emit(
335                        LiveTranscriptionEvents(LiveTranscriptionEvents.Unhandled),
336                        unhandled=unhandled_error,
337                        **dict(cast(Dict[Any, Any], self._kwargs)),
338                    )
339
340            self._logger.notice("_process_text Succeeded")
341            self._logger.debug("SpeakStreamClient._process_text LEAVE")
342
343        except Exception as e:  # pylint: disable=broad-except
344            self._logger.error(
345                "Exception in ListenWebSocketClient._process_text: %s", e
346            )
347            e_error: ErrorResponse = ErrorResponse(
348                "Exception in ListenWebSocketClient._process_text",
349                f"{e}",
350                "Exception",
351            )
352            self._logger.error(
353                "Exception in ListenWebSocketClient._process_text: %s", str(e)
354            )
355            self._emit(
356                LiveTranscriptionEvents(LiveTranscriptionEvents.Error),
357                e_error,
358                **dict(cast(Dict[Any, Any], self._kwargs)),
359            )
360
361            # signal exit and close
362            super()._signal_exit()
363
364            self._logger.debug("ListenWebSocketClient._process_text LEAVE")
365
366            if self._config.options.get("termination_exception") is True:
367                raise
368            return
369
370    # pylint: enable=too-many-return-statements,too-many-statements
371
372    def _process_binary(self, message: bytes) -> None:
373        raise NotImplementedError("no _process_binary method should be called")
374
375    # pylint: disable=too-many-return-statements
376    def _keep_alive(self) -> None:
377        self._logger.debug("ListenWebSocketClient._keep_alive ENTER")
378
379        counter = 0
380        while True:
381            try:
382                counter += 1
383                self._exit_event.wait(timeout=ONE_SECOND)
384
385                if self._exit_event.is_set():
386                    self._logger.notice("_keep_alive exiting gracefully")
387                    self._logger.debug("ListenWebSocketClient._keep_alive LEAVE")
388                    return
389
390                # deepgram keepalive
391                if counter % DEEPGRAM_INTERVAL == 0:
392                    self.keep_alive()
393
394            except Exception as e:  # pylint: disable=broad-except
395                self._logger.error(
396                    "Exception in ListenWebSocketClient._keep_alive: %s", e
397                )
398                e_error: ErrorResponse = ErrorResponse(
399                    "Exception in ListenWebSocketClient._keep_alive",
400                    f"{e}",
401                    "Exception",
402                )
403                self._logger.error(
404                    "Exception in ListenWebSocketClient._keep_alive: %s", str(e)
405                )
406                self._emit(
407                    LiveTranscriptionEvents(LiveTranscriptionEvents.Error),
408                    e_error,
409                    **dict(cast(Dict[Any, Any], self._kwargs)),
410                )
411
412                # signal exit and close
413                super()._signal_exit()
414
415                self._logger.debug("ListenWebSocketClient._keep_alive LEAVE")
416
417                if self._config.options.get("termination_exception") is True:
418                    raise
419                return
420
421    ## pylint: disable=too-many-return-statements,too-many-statements
422    def _flush(self) -> None:
423        self._logger.debug("ListenWebSocketClient._flush ENTER")
424
425        delta_in_ms_str = self._config.options.get("auto_flush_reply_delta")
426        if delta_in_ms_str is None:
427            self._logger.error("auto_flush_reply_delta is None")
428            self._logger.debug("ListenWebSocketClient._flush LEAVE")
429            return
430        delta_in_ms = float(delta_in_ms_str)
431
432        _flush_event = threading.Event()
433        while True:
434            try:
435                _flush_event.wait(timeout=HALF_SECOND)
436
437                if self._exit_event.is_set():
438                    self._logger.notice("_flush exiting gracefully")
439                    self._logger.debug("ListenWebSocketClient._flush LEAVE")
440                    return
441
442                with self._lock_flush:
443                    if self._last_datagram is None:
444                        self._logger.debug("AutoFlush last_datagram is None")
445                        continue
446
447                    delta = datetime.now() - self._last_datagram
448                    diff_in_ms = delta.total_seconds() * 1000
449                    self._logger.debug("AutoFlush delta: %f", diff_in_ms)
450                    if diff_in_ms < delta_in_ms:
451                        self._logger.debug("AutoFlush delta is less than threshold")
452                        continue
453
454                    with self._lock_flush:
455                        self._last_datagram = None
456                    self.finalize()
457
458            except Exception as e:  # pylint: disable=broad-except
459                self._logger.error("Exception in ListenWebSocketClient._flush: %s", e)
460                e_error: ErrorResponse = ErrorResponse(
461                    "Exception in ListenWebSocketClient._flush",
462                    f"{e}",
463                    "Exception",
464                )
465                self._logger.error(
466                    "Exception in ListenWebSocketClient._flush: %s", str(e)
467                )
468                self._emit(
469                    LiveTranscriptionEvents(LiveTranscriptionEvents.Error),
470                    e_error,
471                    **dict(cast(Dict[Any, Any], self._kwargs)),
472                )
473
474                # signal exit and close
475                super()._signal_exit()
476
477                self._logger.debug("ListenWebSocketClient._flush LEAVE")
478
479                if self._config.options.get("termination_exception") is True:
480                    raise
481                return
482
483    # pylint: enable=too-many-return-statements
484
485    def keep_alive(self) -> bool:
486        """
487        Sends a KeepAlive message
488        """
489        self._logger.spam("ListenWebSocketClient.keep_alive ENTER")
490
491        self._logger.notice("Sending KeepAlive...")
492        ret = self.send(json.dumps({"type": "KeepAlive"}))
493
494        if not ret:
495            self._logger.error("keep_alive failed")
496            self._logger.spam("ListenWebSocketClient.keep_alive LEAVE")
497            return False
498
499        self._logger.notice("keep_alive succeeded")
500        self._logger.spam("ListenWebSocketClient.keep_alive LEAVE")
501
502        return True
503
504    def finalize(self) -> bool:
505        """
506        Finalizes the Transcript connection by flushing it
507        """
508        self._logger.spam("ListenWebSocketClient.finalize ENTER")
509
510        self._logger.notice("Sending Finalize...")
511        ret = self.send(json.dumps({"type": "Finalize"}))
512
513        if not ret:
514            self._logger.error("finalize failed")
515            self._logger.spam("ListenWebSocketClient.finalize LEAVE")
516            return False
517
518        self._logger.notice("finalize succeeded")
519        self._logger.spam("ListenWebSocketClient.finalize LEAVE")
520
521        return True
522
523    def _close_message(self) -> bool:
524        return self.send(json.dumps({"type": "CloseStream"}))
525
526    # closes the WebSocket connection gracefully
527    def finish(self) -> bool:
528        """
529        Closes the WebSocket connection gracefully.
530        """
531        self._logger.spam("ListenWebSocketClient.finish ENTER")
532
533        # call parent finish
534        if super().finish() is False:
535            self._logger.error("ListenWebSocketClient.finish failed")
536
537        # debug the threads
538        for thread in threading.enumerate():
539            self._logger.debug("before running thread: %s", thread.name)
540        self._logger.debug("number of active threads: %s", threading.active_count())
541
542        # stop the threads
543        self._logger.verbose("cancelling tasks...")
544        if self._flush_thread is not None:
545            self._flush_thread.join()
546            self._flush_thread = None
547            self._logger.notice("processing _flush_thread thread joined")
548
549        if self._keep_alive_thread is not None:
550            self._keep_alive_thread.join()
551            self._keep_alive_thread = None
552            self._logger.notice("processing _keep_alive_thread thread joined")
553
554        if self._listen_thread is not None:
555            self._listen_thread.join()
556            self._listen_thread = None
557        self._logger.notice("listening thread joined")
558
559        # debug the threads
560        for thread in threading.enumerate():
561            self._logger.debug("before running thread: %s", thread.name)
562        self._logger.debug("number of active threads: %s", threading.active_count())
563
564        self._logger.notice("finish succeeded")
565        self._logger.spam("ListenWebSocketClient.finish LEAVE")
566        return True
567
568    def _inspect(self, msg_result: LiveResultResponse) -> bool:
569        # auto flush_inspect is generically used to track any messages you might want to snoop on
570        # place additional logic here to inspect messages of interest
571
572        # for auto flush functionality
573        # set the last datagram
574        sentence = msg_result.channel.alternatives[0].transcript
575        if len(sentence) == 0:
576            return True
577
578        if msg_result.is_final:
579            with self._lock_flush:
580                self._logger.debug("AutoFlush is_final received")
581                self._last_datagram = None
582        else:
583            with self._lock_flush:
584                self._last_datagram = datetime.now()
585                self._logger.debug(
586                    "AutoFlush interim received: %s",
587                    str(self._last_datagram),
588                )
589
590        return True