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
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)
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.
Inherited Members
17@dataclass 18class OpenResponse(BaseResponse): 19 """ 20 Open Message from the Deepgram Platform 21 """ 22 23 type: str = ""
Open Message from the Deepgram Platform
Inherited Members
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)
Inherited Members
ListenWSMetadataResponse =
<class 'MetadataResponse'>
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)
Inherited Members
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
Inherited Members
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
Inherited Members
29@dataclass 30class CloseResponse(BaseResponse): 31 """ 32 Close Message from the Deepgram Platform 33 """ 34 35 type: str = ""
Close Message from the Deepgram Platform
Inherited Members
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)
Inherited Members
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
Inherited Members
LiveClient =
<class 'deepgram.clients.listen.v1.websocket.client.ListenWebSocketClient'>
AsyncLiveClient =
<class 'deepgram.clients.listen.v1.websocket.async_client.AsyncListenWebSocketClient'>