deepgram.clients.analyze.client

25class UrlSource(TypedDict):
26    """
27    Represents a data source for specifying the location of a file via a URL.
28
29    This class is used to specify a hosted file URL, typically pointing to an
30    externally hosted file, such as an audio file hosted on a server or the internet.
31
32    Attributes:
33        url (str): The URL pointing to the hosted file.
34    """
35
36    url: str

53class TextSource(TypedDict):
54    """
55    Represents a data source for reading binary data from a text-like source.
56
57    This class is used to specify a source of text data that can be read from.
58
59    Attributes:
60        text (str): A string for reading text data.
61    """
62
63    text: str

39class BufferSource(TypedDict):
40    """
41    Represents a data source for handling raw binary data.
42
43    This class is used to specify raw binary data, such as audio data in its
44    binary form, which can be captured from a microphone or generated synthetically.
45
46    Attributes:
47        buffer (bytes): The binary data.
48    """
49
50    buffer: bytes

11class StreamSource(TypedDict):
12    """
13    Represents a data source for reading binary data from a stream-like source.
14
15    This class is used to specify a source of binary data that can be read from
16    a stream, such as an audio file in .wav format.
17
18    Attributes:
19        stream (BufferedReader): A BufferedReader object for reading binary data.
20    """
21
22    stream: BufferedReader

23@dataclass
24class AnalyzeOptions(BaseResponse):  # pylint: disable=too-many-instance-attributes
25    """
26    Contains all the options for the AnalyzeOptions.
27
28    Reference:
29    https://developers.deepgram.com/reference/text-intelligence-apis
30    """
31
32    callback: Optional[str] = field(
33        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
34    )
35    callback_method: Optional[str] = field(
36        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
37    )
38    custom_intent: Optional[Union[List[str], str]] = field(
39        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
40    )
41    custom_intent_mode: Optional[str] = field(
42        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
43    )
44    custom_topic: Optional[Union[List[str], str]] = field(
45        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
46    )
47    custom_topic_mode: Optional[str] = field(
48        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
49    )
50    intents: Optional[bool] = field(
51        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
52    )
53    language: Optional[str] = field(
54        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
55    )
56    sentiment: Optional[bool] = field(
57        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
58    )
59    summarize: Optional[bool] = field(
60        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
61    )
62    topics: Optional[bool] = field(
63        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
64    )
65
66    def check(self):
67        """
68        Check the options for the AnalyzeOptions.
69        """
70        logger = verboselogs.VerboseLogger(__name__)
71        logger.addHandler(logging.StreamHandler())
72        prev = logger.level
73        logger.setLevel(verboselogs.ERROR)
74
75        # no op at the moment
76
77        logger.setLevel(prev)
78
79        return True

122@dataclass
123class AnalyzeResponse(BaseResponse):
124    """
125    Analyze Response
126    """
127
128    metadata: Optional[Metadata] = field(
129        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
130    )
131    results: Optional[Results] = field(
132        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
133    )
134
135    def __getitem__(self, key):
136        _dict = self.to_dict()
137        if "metadata" in _dict:
138            _dict["metadata"] = Metadata.from_dict(_dict["metadata"])
139        if "results" in _dict:
140            _dict["results"] = Results.from_dict(_dict["results"])
141        return _dict[key]

31@dataclass
32class AsyncAnalyzeResponse(BaseResponse):
33    """
34    Async Analyze Response
35    """
36
37    request_id: str = ""

62@dataclass
63class Average(BaseResponse):
64    """
65    Average
66    """
67
68    sentiment: Sentiment
69    sentiment_score: float = 0
70
71    def __getitem__(self, key):
72        _dict = self.to_dict()
73        if "sentiment" in _dict:
74            _dict["sentiment"] = Sentiment.from_dict(_dict["sentiment"])
75        return _dict[key]

88@dataclass
89class Intent(BaseResponse):
90    """
91    Intent
92    """
93
94    intent: str = ""
95    confidence_score: float = 0

168@dataclass
169class Intents(BaseResponse):
170    """
171    Intents
172    """
173
174    segments: List[Segment] = field(default_factory=list)
175
176    def __getitem__(self, key):
177        _dict = self.to_dict()
178        if "segments" in _dict:
179            _dict["segments"] = [
180                Segment.from_dict(segments) for segments in _dict["segments"]
181            ]
182        return _dict[key]

18@dataclass
19class IntentsInfo(BaseResponse):
20    """
21    Intents Info
22    """
23
24    model_uuid: str = ""
25    input_tokens: int = 0
26    output_tokens: int = 0

 98@dataclass
 99class Segment(BaseResponse):
100    """
101    Segment
102    """
103
104    text: str = ""
105    start_word: int = 0
106    end_word: int = 0
107    sentiment: Optional[Sentiment] = field(
108        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
109    )
110    sentiment_score: Optional[float] = field(
111        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
112    )
113    intents: Optional[List[Intent]] = field(
114        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
115    )
116    topics: Optional[List[Topic]] = field(
117        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
118    )
119
120    def __getitem__(self, key):
121        _dict = self.to_dict()
122        if "sentiment" in _dict:
123            _dict["sentiment"] = Sentiment.from_dict(_dict["sentiment"])
124        if "intents" in _dict:
125            _dict["intents"] = Intent.from_dict(_dict["intents"])
126        if "topics" in _dict:
127            _dict["topics"] = Topic.from_dict(_dict["topics"])
128        return _dict[key]

29@dataclass
30class SentimentInfo(BaseResponse):
31    """
32    Sentiment Info
33    """
34
35    model_uuid: str = ""
36    input_tokens: int = 0
37    output_tokens: int = 0

11class Sentiment(StrEnum):
12    """
13    Sentiment values.
14    """
15
16    UNKNOWN: str = ""
17    NEGATIVE: str = "negative"
18    NEUTRAL: str = "neutral"
19    POSITIVE: str = "positive"

131@dataclass
132class Sentiments(BaseResponse):
133    """
134    Sentiments
135    """
136
137    average: Average
138    segments: List[Segment] = field(default_factory=list)
139
140    def __getitem__(self, key):
141        _dict = self.to_dict()
142        if "segments" in _dict:
143            _dict["segments"] = [
144                Segment.from_dict(segments) for segments in _dict["segments"]
145            ]
146        if "average" in _dict:
147            _dict["average"] = Average.from_dict(_dict["average"])
148        return _dict[key]

40@dataclass
41class SummaryInfo(BaseResponse):
42    """
43    Summary Info
44    """
45
46    model_uuid: str = ""
47    input_tokens: int = 0
48    output_tokens: int = 0

78@dataclass
79class Topic(BaseResponse):
80    """
81    Topic
82    """
83
84    topic: str = ""
85    confidence_score: float = 0

151@dataclass
152class Topics(BaseResponse):
153    """
154    Topics
155    """
156
157    segments: List[Segment] = field(default_factory=list)
158
159    def __getitem__(self, key):
160        _dict = self.to_dict()
161        if "segments" in _dict:
162            _dict["segments"] = [
163                Segment.from_dict(segments) for segments in _dict["segments"]
164            ]
165        return _dict[key]

51@dataclass
52class TopicsInfo(BaseResponse):
53    """
54    Topics Info
55    """
56
57    model_uuid: str = ""
58    input_tokens: int = 0
59    output_tokens: int = 0

 25class AnalyzeClient(AbstractSyncRestClient):
 26    """
 27    A client class for handling text data.
 28    Provides methods for transcribing text from URLs, files, etc.
 29    """
 30
 31    _logger: verboselogs.VerboseLogger
 32    _config: DeepgramClientOptions
 33
 34    def __init__(self, config: DeepgramClientOptions):
 35        self._logger = verboselogs.VerboseLogger(__name__)
 36        self._logger.addHandler(logging.StreamHandler())
 37        self._logger.setLevel(config.verbose)
 38        self._config = config
 39        super().__init__(config)
 40
 41    # pylint: disable=too-many-positional-arguments
 42
 43    def analyze_url(
 44        self,
 45        source: UrlSource,
 46        options: Optional[Union[AnalyzeOptions, Dict]] = None,
 47        addons: Optional[Dict] = None,
 48        headers: Optional[Dict] = None,
 49        timeout: Optional[httpx.Timeout] = None,
 50        endpoint: str = "v1/read",
 51        **kwargs,
 52    ) -> Union[AnalyzeResponse, AsyncAnalyzeResponse]:
 53        """
 54        Analyze text from a URL source.
 55
 56        Args:
 57            source (UrlSource): The URL source of the text to ingest.
 58            options (AnalyzeOptions): Additional options for the ingest (default is None).
 59            endpoint (str): The API endpoint for the ingest (default is "v1/read").
 60
 61        Returns:
 62            AnalyzeResponse: An object containing the result.
 63
 64        Raises:
 65            DeepgramTypeError: Raised for known API errors.
 66        """
 67        self._logger.debug("AnalyzeClient.analyze_url ENTER")
 68
 69        if (
 70            isinstance(options, dict)
 71            and "callback" in options
 72            and options["callback"] is not None
 73        ) or (isinstance(options, AnalyzeOptions) and options.callback is not None):
 74            self._logger.debug("AnalyzeClient.analyze_url LEAVE")
 75            return self.analyze_url_callback(
 76                source,
 77                callback=options["callback"],
 78                options=options,
 79                addons=addons,
 80                headers=headers,
 81                timeout=timeout,
 82                endpoint=endpoint,
 83                **kwargs,
 84            )
 85
 86        url = f"{self._config.url}/{endpoint}"
 87        if is_url_source(source):
 88            body = source
 89        else:
 90            self._logger.error("Unknown transcription source type")
 91            self._logger.debug("AnalyzeClient.analyze_url LEAVE")
 92            raise DeepgramTypeError("Unknown transcription source type")
 93
 94        if isinstance(options, AnalyzeOptions) and not options.check():
 95            self._logger.error("options.check failed")
 96            self._logger.debug("AnalyzeClient.analyze_url LEAVE")
 97            raise DeepgramError("Fatal transcription options error")
 98
 99        self._logger.info("url: %s", url)
100        self._logger.info("source: %s", source)
101        if isinstance(options, AnalyzeOptions):
102            self._logger.info("AnalyzeOptions switching class -> dict")
103            options = options.to_dict()
104        self._logger.info("options: %s", options)
105        self._logger.info("addons: %s", addons)
106        self._logger.info("headers: %s", headers)
107        result = self.post(
108            url,
109            options=options,
110            addons=addons,
111            headers=headers,
112            json=body,
113            timeout=timeout,
114            **kwargs,
115        )
116        self._logger.info("json: %s", result)
117        res = AnalyzeResponse.from_json(result)
118        self._logger.verbose("result: %s", res)
119        self._logger.notice("analyze_url succeeded")
120        self._logger.debug("AnalyzeClient.analyze_url LEAVE")
121        return res
122
123    def analyze_url_callback(
124        self,
125        source: UrlSource,
126        callback: str,
127        options: Optional[Union[AnalyzeOptions, Dict]] = None,
128        addons: Optional[Dict] = None,
129        headers: Optional[Dict] = None,
130        timeout: Optional[httpx.Timeout] = None,
131        endpoint: str = "v1/read",
132        **kwargs,
133    ) -> AsyncAnalyzeResponse:
134        """
135        Transcribes audio from a URL source and sends the result to a callback URL.
136
137        Args:
138            source (UrlSource): The URL source of the audio to transcribe.
139            callback (str): The callback URL where the transcription results will be sent.
140            options (AnalyzeOptions): Additional options for the transcription (default is None).
141            endpoint (str): The API endpoint for the transcription (default is "v1/read").
142
143        Returns:
144            AsyncAnalyzeResponse: An object containing the request_id or an error message.
145
146        Raises:
147            DeepgramTypeError: Raised for known API errors.
148        """
149        self._logger.debug("AnalyzeClient.analyze_url_callback ENTER")
150
151        url = f"{self._config.url}/{endpoint}"
152        if options is None:
153            options = {}
154        if isinstance(options, AnalyzeOptions):
155            options.callback = callback
156        else:
157            options["callback"] = callback
158        if is_url_source(source):
159            body = source
160        else:
161            self._logger.error("Unknown transcription source type")
162            self._logger.debug("AnalyzeClient.analyze_url_callback LEAVE")
163            raise DeepgramTypeError("Unknown transcription source type")
164
165        if isinstance(options, AnalyzeOptions) and not options.check():
166            self._logger.error("options.check failed")
167            self._logger.debug("AnalyzeClient.analyze_url_callback LEAVE")
168            raise DeepgramError("Fatal transcription options error")
169
170        self._logger.info("url: %s", url)
171        self._logger.info("source: %s", source)
172        if isinstance(options, AnalyzeOptions):
173            self._logger.info("AnalyzeOptions switching class -> dict")
174            options = options.to_dict()
175        self._logger.info("options: %s", options)
176        self._logger.info("addons: %s", addons)
177        self._logger.info("headers: %s", headers)
178        result = self.post(
179            url,
180            options=options,
181            addons=addons,
182            headers=headers,
183            json=body,
184            timeout=timeout,
185            **kwargs,
186        )
187        self._logger.info("json: %s", result)
188        res = AsyncAnalyzeResponse.from_json(result)
189        self._logger.verbose("result: %s", res)
190        self._logger.notice("analyze_url_callback succeeded")
191        self._logger.debug("AnalyzeClient.analyze_url_callback LEAVE")
192        return res
193
194    def analyze_text(
195        self,
196        source: FileSource,
197        options: Optional[Union[AnalyzeOptions, Dict]] = None,
198        addons: Optional[Dict] = None,
199        headers: Optional[Dict] = None,
200        timeout: Optional[httpx.Timeout] = None,
201        endpoint: str = "v1/read",
202        **kwargs,
203    ) -> Union[AnalyzeResponse, AsyncAnalyzeResponse]:
204        """
205        Analyze text from a local file source.
206
207        Args:
208            source (TextSource): The local file source of the text to ingest.
209            options (AnalyzeOptions): Additional options for the ingest (default is None).
210            endpoint (str): The API endpoint for the transcription (default is "v1/read").
211
212        Returns:
213            AnalyzeResponse: An object containing the transcription result or an error message.
214
215        Raises:
216            DeepgramTypeError: Raised for known API errors.
217        """
218        self._logger.debug("AnalyzeClient.analyze_text ENTER")
219
220        if (
221            isinstance(options, dict)
222            and "callback" in options
223            and options["callback"] is not None
224        ) or (isinstance(options, AnalyzeOptions) and options.callback is not None):
225            self._logger.debug("AnalyzeClient.analyze_text LEAVE")
226            return self.analyze_text_callback(
227                source,
228                callback=options["callback"],
229                options=options,
230                addons=addons,
231                headers=headers,
232                timeout=timeout,
233                endpoint=endpoint,
234                **kwargs,
235            )
236
237        url = f"{self._config.url}/{endpoint}"
238        if is_buffer_source(source):
239            body = source["buffer"]  # type: ignore
240        elif is_readstream_source(source):
241            body = source["stream"]  # type: ignore
242        else:
243            self._logger.error("Unknown transcription source type")
244            self._logger.debug("AnalyzeClient.analyze_text LEAVE")
245            raise DeepgramTypeError("Unknown transcription source type")
246
247        if isinstance(options, AnalyzeOptions) and not options.check():
248            self._logger.error("options.check failed")
249            self._logger.debug("AnalyzeClient.analyze_text LEAVE")
250            raise DeepgramError("Fatal transcription options error")
251
252        self._logger.info("url: %s", url)
253        if isinstance(options, AnalyzeOptions):
254            self._logger.info("AnalyzeOptions switching class -> dict")
255            options = options.to_dict()
256        self._logger.info("options: %s", options)
257        self._logger.info("addons: %s", addons)
258        self._logger.info("headers: %s", headers)
259        result = self.post(
260            url,
261            options=options,
262            addons=addons,
263            headers=headers,
264            content=body,
265            timeout=timeout,
266            **kwargs,
267        )
268        self._logger.info("json: %s", result)
269        res = AnalyzeResponse.from_json(result)
270        self._logger.verbose("result: %s", res)
271        self._logger.notice("analyze_text succeeded")
272        self._logger.debug("AnalyzeClient.analyze_text LEAVE")
273        return res
274
275    def analyze_text_callback(
276        self,
277        source: FileSource,
278        callback: str,
279        options: Optional[Union[AnalyzeOptions, Dict]] = None,
280        addons: Optional[Dict] = None,
281        headers: Optional[Dict] = None,
282        timeout: Optional[httpx.Timeout] = None,
283        endpoint: str = "v1/read",
284        **kwargs,
285    ) -> AsyncAnalyzeResponse:
286        """
287        Transcribes audio from a local file source and sends the result to a callback URL.
288
289        Args:
290            source (TextSource): The local file source of the audio to transcribe.
291            callback (str): The callback URL where the transcription results will be sent.
292            options (AnalyzeOptions): Additional options for the transcription (default is None).
293            endpoint (str): The API endpoint for the transcription (default is "v1/read").
294
295        Returns:
296            AsyncAnalyzeResponse: An object containing the request_id or an error message.
297
298        Raises:
299            DeepgramTypeError: Raised for known API errors.
300        """
301        self._logger.debug("AnalyzeClient.analyze_file_callback ENTER")
302
303        url = f"{self._config.url}/{endpoint}"
304        if options is None:
305            options = {}
306        if isinstance(options, AnalyzeOptions):
307            options.callback = callback
308        else:
309            options["callback"] = callback
310        if is_buffer_source(source):
311            body = source["buffer"]  # type: ignore
312        elif is_readstream_source(source):
313            body = source["stream"]  # type: ignore
314        else:
315            self._logger.error("Unknown transcription source type")
316            self._logger.debug("AnalyzeClient.analyze_file_callback LEAVE")
317            raise DeepgramTypeError("Unknown transcription source type")
318
319        if isinstance(options, AnalyzeOptions) and not options.check():
320            self._logger.error("options.check failed")
321            self._logger.debug("AnalyzeClient.analyze_file_callback LEAVE")
322            raise DeepgramError("Fatal transcription options error")
323
324        self._logger.info("url: %s", url)
325        if isinstance(options, AnalyzeOptions):
326            self._logger.info("AnalyzeOptions switching class -> dict")
327            options = options.to_dict()
328        self._logger.info("options: %s", options)
329        self._logger.info("addons: %s", addons)
330        self._logger.info("headers: %s", headers)
331        result = self.post(
332            url,
333            options=options,
334            addons=addons,
335            headers=headers,
336            json=body,
337            timeout=timeout,
338            **kwargs,
339        )
340        self._logger.info("json: %s", result)
341        res = AsyncAnalyzeResponse.from_json(result)
342        self._logger.verbose("result: %s", res)
343        self._logger.notice("analyze_file_callback succeeded")
344        self._logger.debug("AnalyzeClient.analyze_file_callback LEAVE")
345        return res
346
347    # pylint: enable=too-many-positional-arguments