deepgram.clients.live.v1.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
 4
 5from ...listen.v1 import ListenWebSocketClient as LiveClientLatest
 6from ...listen.v1 import AsyncListenWebSocketClient as AsyncLiveClientLatest
 7from ...listen.v1 import LiveOptions as LiveOptionsLatest
 8from ...listen.v1 import (
 9    OpenResponse as OpenResponseLatest,
10    LiveResultResponse as LiveResultResponseLatest,
11    ListenWSMetadataResponse as ListenWSMetadataResponseLatest,
12    SpeechStartedResponse as SpeechStartedResponseLatest,
13    UtteranceEndResponse as UtteranceEndResponseLatest,
14    CloseResponse as CloseResponseLatest,
15    ErrorResponse as ErrorResponseLatest,
16    UnhandledResponse as UnhandledResponseLatest,
17)
18
19# The vX/client.py points to the current supported version in the SDK.
20# Older versions are supported in the SDK for backwards compatibility.
21
22
23# input
24LiveOptions = LiveOptionsLatest
25OpenResponse = OpenResponseLatest
26LiveResultResponse = LiveResultResponseLatest
27ListenWSMetadataResponse = ListenWSMetadataResponseLatest
28MetadataResponse = ListenWSMetadataResponseLatest
29SpeechStartedResponse = SpeechStartedResponseLatest
30UtteranceEndResponse = UtteranceEndResponseLatest
31CloseResponse = CloseResponseLatest
32ErrorResponse = ErrorResponseLatest
33UnhandledResponse = UnhandledResponseLatest
34
35
36# clients
37LiveClient = LiveClientLatest
38AsyncLiveClient = AsyncLiveClientLatest
@dataclass
class LiveOptions(deepgram.clients.common.v1.shared_response.BaseResponse):
 17@dataclass
 18class LiveOptions(BaseResponse):  # pylint: disable=too-many-instance-attributes
 19    """
 20    Live Transcription Options for the Deepgram Platform.
 21
 22    Please see the documentation for more information on each option:
 23    https://developers.deepgram.com/reference/streaming
 24    """
 25
 26    alternatives: Optional[int] = field(
 27        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
 28    )
 29    callback: Optional[str] = field(
 30        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
 31    )
 32    callback_method: Optional[str] = field(
 33        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
 34    )
 35    channels: Optional[int] = field(
 36        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
 37    )
 38    diarize: Optional[bool] = field(
 39        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
 40    )
 41    diarize_version: Optional[str] = field(
 42        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
 43    )
 44    dictation: Optional[bool] = field(
 45        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
 46    )
 47    encoding: Optional[str] = field(
 48        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
 49    )
 50    # pylint: disable=W0511
 51    # TODO: endpointing's current type previous was `Optional[str]` which is incorrect
 52    # for backward compatibility we are keeping it as `Optional[Union[str, bool, int]]`
 53    # since it gets translated to a string to be placed as a query parameter, will keep `str` for now
 54    # but will change this to `Optional[Union[bool, int]]` in a future release
 55    endpointing: Optional[Union[str, bool, int]] = field(
 56        default=None,
 57        metadata=dataclass_config(exclude=lambda f: f is None),
 58    )
 59    # pylint: enable=W0511
 60    extra: Optional[Union[List[str], str]] = field(
 61        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
 62    )
 63    filler_words: Optional[bool] = field(
 64        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
 65    )
 66    interim_results: Optional[bool] = field(
 67        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
 68    )
 69    keywords: Optional[Union[List[str], str]] = field(
 70        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
 71    )
 72    keyterm: Optional[List[str]] = field(
 73        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
 74    )
 75    language: Optional[str] = field(
 76        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
 77    )
 78    model: Optional[str] = field(
 79        default="None", metadata=dataclass_config(exclude=lambda f: f is None)
 80    )
 81    multichannel: Optional[bool] = field(
 82        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
 83    )
 84    no_delay: Optional[bool] = field(
 85        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
 86    )
 87    numerals: Optional[bool] = field(
 88        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
 89    )
 90    punctuate: Optional[bool] = field(
 91        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
 92    )
 93    profanity_filter: Optional[bool] = field(
 94        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
 95    )
 96    redact: Optional[Union[List[str], bool, str]] = field(
 97        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
 98    )
 99    replace: Optional[Union[List[str], str]] = field(
100        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
101    )
102    sample_rate: Optional[int] = field(
103        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
104    )
105    search: Optional[Union[List[str], str]] = field(
106        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
107    )
108    smart_format: Optional[bool] = field(
109        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
110    )
111    tag: Optional[List[str]] = field(
112        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
113    )
114    tier: Optional[str] = field(
115        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
116    )
117    utterance_end_ms: Optional[str] = field(
118        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
119    )
120    vad_events: Optional[bool] = field(
121        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
122    )
123    version: Optional[str] = field(
124        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
125    )
126
127    def check(self):
128        """
129        Check the options for any deprecated or soon-to-be-deprecated options.
130        """
131        logger = verboselogs.VerboseLogger(__name__)
132        logger.addHandler(logging.StreamHandler())
133        prev = logger.level
134        logger.setLevel(verboselogs.ERROR)
135
136        if self.tier:
137            logger.warning(
138                "WARNING: Tier is deprecated. Will be removed in a future version."
139            )
140
141        if isinstance(self.endpointing, str):
142            logger.warning(
143                "WARNING: endpointing's current type previous was `Optional[str]` which is incorrect"
144                " for backward compatibility we are keeping it as `Optional[Union[str, bool, int]]`"
145                " since it gets translated to a string to be placed as a query parameter, will keep `str` for now"
146                " but will change this to `Optional[Union[bool, int]]` in a future release"
147            )
148
149        logger.setLevel(prev)
150
151        return True

Live Transcription Options for the Deepgram Platform.

Please see the documentation for more information on each option: https://developers.deepgram.com/reference/streaming

LiveOptions( alternatives: Optional[int] = None, callback: Optional[str] = None, callback_method: Optional[str] = None, channels: Optional[int] = None, diarize: Optional[bool] = None, diarize_version: Optional[str] = None, dictation: Optional[bool] = None, encoding: Optional[str] = None, endpointing: Union[str, bool, int, NoneType] = None, extra: Union[List[str], str, NoneType] = None, filler_words: Optional[bool] = None, interim_results: Optional[bool] = None, keywords: Union[List[str], str, NoneType] = None, keyterm: Optional[List[str]] = None, language: Optional[str] = None, model: Optional[str] = 'None', multichannel: Optional[bool] = None, no_delay: Optional[bool] = None, numerals: Optional[bool] = None, punctuate: Optional[bool] = None, profanity_filter: Optional[bool] = None, redact: Union[List[str], bool, str, NoneType] = None, replace: Union[List[str], str, NoneType] = None, sample_rate: Optional[int] = None, search: Union[List[str], str, NoneType] = None, smart_format: Optional[bool] = None, tag: Optional[List[str]] = None, tier: Optional[str] = None, utterance_end_ms: Optional[str] = None, vad_events: Optional[bool] = None, version: Optional[str] = None)
alternatives: Optional[int] = None
callback: Optional[str] = None
callback_method: Optional[str] = None
channels: Optional[int] = None
diarize: Optional[bool] = None
diarize_version: Optional[str] = None
dictation: Optional[bool] = None
encoding: Optional[str] = None
endpointing: Union[str, bool, int, NoneType] = None
extra: Union[List[str], str, NoneType] = None
filler_words: Optional[bool] = None
interim_results: Optional[bool] = None
keywords: Union[List[str], str, NoneType] = None
keyterm: Optional[List[str]] = None
language: Optional[str] = None
model: Optional[str] = 'None'
multichannel: Optional[bool] = None
no_delay: Optional[bool] = None
numerals: Optional[bool] = None
punctuate: Optional[bool] = None
profanity_filter: Optional[bool] = None
redact: Union[List[str], bool, str, NoneType] = None
replace: Union[List[str], str, NoneType] = None
sample_rate: Optional[int] = None
search: Union[List[str], str, NoneType] = None
smart_format: Optional[bool] = None
tag: Optional[List[str]] = None
tier: Optional[str] = None
utterance_end_ms: Optional[str] = None
vad_events: Optional[bool] = None
version: Optional[str] = None
def check(self):
127    def check(self):
128        """
129        Check the options for any deprecated or soon-to-be-deprecated options.
130        """
131        logger = verboselogs.VerboseLogger(__name__)
132        logger.addHandler(logging.StreamHandler())
133        prev = logger.level
134        logger.setLevel(verboselogs.ERROR)
135
136        if self.tier:
137            logger.warning(
138                "WARNING: Tier is deprecated. Will be removed in a future version."
139            )
140
141        if isinstance(self.endpointing, str):
142            logger.warning(
143                "WARNING: endpointing's current type previous was `Optional[str]` which is incorrect"
144                " for backward compatibility we are keeping it as `Optional[Union[str, bool, int]]`"
145                " since it gets translated to a string to be placed as a query parameter, will keep `str` for now"
146                " but will change this to `Optional[Union[bool, int]]` in a future release"
147            )
148
149        logger.setLevel(prev)
150
151        return True

Check the options for any deprecated or soon-to-be-deprecated options.

@dataclass
class OpenResponse(deepgram.clients.common.v1.shared_response.BaseResponse):
17@dataclass
18class OpenResponse(BaseResponse):
19    """
20    Open Message from the Deepgram Platform
21    """
22
23    type: str = ""

Open Message from the Deepgram Platform

OpenResponse(type: str = '')
type: str = ''
@dataclass
class LiveResultResponse(deepgram.clients.common.v1.shared_response.BaseResponse):
122@dataclass
123class LiveResultResponse(BaseResponse):  # pylint: disable=too-many-instance-attributes
124    """
125    Result Message from the Deepgram Platform
126    """
127
128    channel: ListenWSChannel
129    metadata: Metadata
130    type: str = ""
131    channel_index: List[int] = field(default_factory=list)
132    duration: float = 0
133    start: float = 0
134    is_final: bool = False
135    from_finalize: Optional[bool] = field(
136        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
137    )
138    speech_final: bool = False
139
140    def __getitem__(self, key):
141        _dict = self.to_dict()
142        if "channel" in _dict:
143            _dict["channel"] = [
144                ListenWSChannel.from_dict(channel) for channel in _dict["channel"]
145            ]
146        if "metadata" in _dict:
147            _dict["metadata"] = [
148                Metadata.from_dict(metadata) for metadata in _dict["metadata"]
149            ]
150        return _dict[key]

Result Message from the Deepgram Platform

LiveResultResponse( channel: deepgram.clients.listen.v1.websocket.response.ListenWSChannel, metadata: deepgram.clients.listen.v1.websocket.response.Metadata, type: str = '', channel_index: List[int] = <factory>, duration: float = 0, start: float = 0, is_final: bool = False, from_finalize: Optional[bool] = None, speech_final: bool = False)
type: str = ''
channel_index: List[int]
duration: float = 0
start: float = 0
is_final: bool = False
from_finalize: Optional[bool] = None
speech_final: bool = False
ListenWSMetadataResponse = <class 'MetadataResponse'>
@dataclass
class MetadataResponse(deepgram.clients.common.v1.shared_response.BaseResponse):
156@dataclass
157class MetadataResponse(BaseResponse):  # pylint: disable=too-many-instance-attributes
158    """
159    Metadata Message from the Deepgram Platform
160    """
161
162    type: str = ""
163    transaction_key: str = ""
164    request_id: str = ""
165    sha256: str = ""
166    created: str = ""
167    duration: float = 0
168    channels: int = 0
169    models: Optional[List[str]] = field(
170        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
171    )
172    model_info: Optional[Dict[str, ModelInfo]] = field(
173        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
174    )
175    extra: Optional[Dict] = field(
176        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
177    )
178
179    def __getitem__(self, key):
180        _dict = self.to_dict()
181        if "models" in _dict:
182            _dict["models"] = [str(models) for models in _dict["models"]]
183        if "model_info" in _dict:
184            _dict["model_info"] = [
185                ModelInfo.from_dict(model_info)
186                for _, model_info in _dict["model_info"].items()
187            ]
188        if "extra" in _dict:
189            _dict["extra"] = [str(extra) for _, extra in _dict["extra"].items()]
190        return _dict[key]

Metadata Message from the Deepgram Platform

MetadataResponse( type: str = '', transaction_key: str = '', request_id: str = '', sha256: str = '', created: str = '', duration: float = 0, channels: int = 0, models: Optional[List[str]] = None, model_info: Optional[Dict[str, deepgram.clients.common.ModelInfo]] = None, extra: Optional[Dict] = None)
type: str = ''
transaction_key: str = ''
request_id: str = ''
sha256: str = ''
created: str = ''
duration: float = 0
channels: int = 0
models: Optional[List[str]] = None
model_info: Optional[Dict[str, deepgram.clients.common.ModelInfo]] = None
extra: Optional[Dict] = None
@dataclass
class SpeechStartedResponse(deepgram.clients.common.v1.shared_response.BaseResponse):
196@dataclass
197class SpeechStartedResponse(BaseResponse):
198    """
199    SpeechStartedResponse Message from the Deepgram Platform
200    """
201
202    type: str = ""
203    channel: List[int] = field(default_factory=list)
204    timestamp: float = 0

SpeechStartedResponse Message from the Deepgram Platform

SpeechStartedResponse(type: str = '', channel: List[int] = <factory>, timestamp: float = 0)
type: str = ''
channel: List[int]
timestamp: float = 0
@dataclass
class UtteranceEndResponse(deepgram.clients.common.v1.shared_response.BaseResponse):
210@dataclass
211class UtteranceEndResponse(BaseResponse):
212    """
213    UtteranceEnd Message from the Deepgram Platform
214    """
215
216    type: str = ""
217    channel: List[int] = field(default_factory=list)
218    last_word_end: float = 0

UtteranceEnd Message from the Deepgram Platform

UtteranceEndResponse( type: str = '', channel: List[int] = <factory>, last_word_end: float = 0)
type: str = ''
channel: List[int]
last_word_end: float = 0
@dataclass
class CloseResponse(deepgram.clients.common.v1.shared_response.BaseResponse):
29@dataclass
30class CloseResponse(BaseResponse):
31    """
32    Close Message from the Deepgram Platform
33    """
34
35    type: str = ""

Close Message from the Deepgram Platform

CloseResponse(type: str = '')
type: str = ''
@dataclass
class ErrorResponse(deepgram.clients.common.v1.shared_response.BaseResponse):
41@dataclass
42class ErrorResponse(BaseResponse):
43    """
44    Error Message from the Deepgram Platform
45    """
46
47    description: str = ""
48    message: str = ""
49    type: str = ""
50    variant: Optional[str] = field(
51        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
52    )

Error Message from the Deepgram Platform

ErrorResponse( description: str = '', message: str = '', type: str = '', variant: Optional[str] = None)
description: str = ''
message: str = ''
type: str = ''
variant: Optional[str] = None
@dataclass
class UnhandledResponse(deepgram.clients.common.v1.shared_response.BaseResponse):
58@dataclass
59class UnhandledResponse(BaseResponse):
60    """
61    Unhandled Message from the Deepgram Platform
62    """
63
64    type: str = ""
65    raw: str = ""

Unhandled Message from the Deepgram Platform

UnhandledResponse(type: str = '', raw: str = '')
type: str = ''
raw: str = ''