Aller au contenu

Audio Module

The audio module provides functionality for audio transcription and text to speed functionnality.

AudioAPI

Source code in forrasdk/api/audio.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
class AudioAPI:
    def __init__(self, base_url: str, headers: dict, retry_strategy: Callable) -> None:
        self._base_url = base_url
        self._headers = headers
        self._retry_strategy = retry_strategy

    def text_to_speech(
        self,
        text: str,
        model: Optional[str] = None,
        voice: Optional[str] = None,
        api_args: Optional[dict] = None,
    ) -> bytes:
        """
        Convert input text to speech audio using a specified model and voice.

        Args:
            text (str): The text to be converted to speech.
            model (Optional[str], optional): The speech synthesis model to use. Defaults to None.
            voice (Optional[str], optional): The voice profile to use for speech. Defaults to None.
            api_args (Optional[dict], optional): Additional API arguments for customization. Defaults to None.

        Returns:
            bytes: The audio content generated from the text.
        """
        json_payload = {
            "text": text,
            **({"model": model} if model is not None else {}),
            **({"voice": voice} if voice is not None else {}),
            **({"api_args": api_args} if api_args is not None else {}),
        }

        response = requests.post(
            url=f"{self._base_url}/api/audio/speech",
            headers=self._headers,
            json=json_payload,
        )
        try:
            response.raise_for_status()
        except requests.exceptions.HTTPError as e:
            error_message = "HTTP Error occurred"
            if e.response is not None:
                try:
                    error_details = e.response.json()
                    error_message = f"Error: {error_details}"
                except (ValueError, AttributeError):
                    # Handle cases where response is not valid JSON
                    error_message = f"Error: HTTP {e.response.status_code} - {e.response.text or 'No response text'}"
            else:
                # Handle cases where e.response is None
                error_message = f"Error: {str(e)}"

            raise Exception(error_message) from e

        return response.content

    @overload
    def transcribe_file(
        self,
        file_path: str,
        conversation_id: Optional[str] = None,
        model: Optional[str] = None,
        execute_async: Literal[True] = True,
        model_args: Optional[dict] = None,
    ) -> AsyncJobResponse: ...

    @overload
    def transcribe_file(
        self,
        file_path: str,
        conversation_id: Optional[str] = None,
        model: Optional[str] = None,
        execute_async: Literal[False] = False,
        model_args: Optional[dict] = None,
    ) -> AudioTranscriptionResponse: ...

    def transcribe_file(
        self,
        file_path: str,
        conversation_id: Optional[str] = None,
        model: Optional[str] = None,
        execute_async: bool = True,
        model_args: Optional[dict] = None,
    ) -> Union[AudioTranscriptionResponse, AsyncJobResponse]:
        """
        Transcribe an audio file using the Scout API.

        Args:
            file_path (str): The local path to the audio file to transcribe.
            conversation_id (Optional[str]): The conversation ID to associate with the transcription.
            model (Optional[str]): The transcription model to use.
            execute_async (bool): Whether to execute the transcription asynchronously. Defaults to True.
            model_args (Optional[dict]): Additional model arguments.

        Returns:
            Union[AudioTranscriptionResponse, AsyncJobResponse]: The transcription response from the API.
            Returns AsyncJobResponse when execute_async=True, AudioTranscriptionResponse when execute_async=False.

        Raises:
            FileNotFoundError: If the specified audio file doesn't exist.
            Exception: If there's an error during the transcription process.
        """
        if not os.path.exists(file_path):
            raise FileNotFoundError(f"Audio file not found: {file_path}")

        filename = os.path.basename(file_path)

        # Read file and encode to base64
        with open(file_path, "rb") as f:
            file_bytes = f.read()
            base64_encoded = base64.b64encode(file_bytes).decode("utf-8")

        # Create the request data with base64 encoded audio
        audio_object = AudioObject(filename=filename, base64=base64_encoded)

        return self._send_transcription_request(
            audio_object=audio_object,
            conversation_id=conversation_id,
            model=model,
            execute_async=execute_async,
            model_args=model_args,
        )

    @overload
    def transcribe_url(
        self,
        protected_url: str,
        filename: Optional[str] = None,
        conversation_id: Optional[str] = None,
        model: Optional[str] = None,
        execute_async: Literal[True] = True,
        model_args: Optional[dict] = None,
    ) -> AsyncJobResponse: ...

    @overload
    def transcribe_url(
        self,
        protected_url: str,
        filename: Optional[str] = None,
        conversation_id: Optional[str] = None,
        model: Optional[str] = None,
        execute_async: Literal[False] = False,
        model_args: Optional[dict] = None,
    ) -> AudioTranscriptionResponse: ...

    def transcribe_url(
        self,
        protected_url: str,
        filename: Optional[str] = None,
        conversation_id: Optional[str] = None,
        model: Optional[str] = None,
        execute_async: bool = True,
        model_args: Optional[dict] = None,
    ) -> Union[AudioTranscriptionResponse, AsyncJobResponse]:
        """
        Transcribe an audio file from a protected URL using the Scout API.

        Args:
            protected_url (str): The protected URL path to the audio file.
            filename (Optional[str]): The filename to use. If None, extracted from URL.
            conversation_id (Optional[str]): The conversation ID to associate with the transcription.
            model (Optional[str]): The transcription model to use.
            execute_async (bool): Whether to execute the transcription asynchronously. Defaults to True.
            model_args (Optional[dict]): Additional model arguments.

        Returns:
            Union[AudioTranscriptionResponse, AsyncJobResponse]: The transcription response from the API.
            Returns AsyncJobResponse when execute_async=True, AudioTranscriptionResponse when execute_async=False.

        Raises:
            Exception: If there's an error during the transcription process.
        """
        # Extract filename from URL if not provided
        if filename is None:
            filename = protected_url.split("/")[-1] if "/" in protected_url else "audio"

        # Create the request data with protected URL
        audio_object = AudioObject(filename=filename, protected_url=protected_url)

        return self._send_transcription_request(
            audio_object=audio_object,
            conversation_id=conversation_id,
            model=model,
            execute_async=execute_async,
            model_args=model_args,
        )

    def _send_transcription_request(
        self,
        audio_object: AudioObject,
        conversation_id: Optional[str] = None,
        model: Optional[str] = None,
        execute_async: bool = True,
        model_args: Optional[dict] = None,
    ) -> Union[AudioTranscriptionResponse, AsyncJobResponse]:
        """
        Internal method to send transcription request to the API.
        """
        request_data = AudioTranscriptionRequest(
            conversation_id=conversation_id,
            audio=audio_object,
            model=model,
            execute_async=execute_async,
            model_args=model_args,
        )

        # Send as JSON
        json_payload = request_data.model_dump(exclude_none=True)

        response = requests.post(
            url=f"{self._base_url}/api/audio/transcription",
            headers=self._headers,
            json=json_payload,
        )

        try:
            response.raise_for_status()
        except requests.exceptions.HTTPError as e:
            error_message = "HTTP Error occurred"
            if e.response is not None:
                try:
                    error_details = e.response.json()
                    error_message = f"Error: {error_details}"
                except (ValueError, AttributeError):
                    error_message = f"Error: HTTP {e.response.status_code} - {e.response.text or 'No response text'}"
            else:
                error_message = f"Error: {str(e)}"
            raise Exception(error_message) from e

        response_data = response.json()

        if execute_async:
            return AsyncJobResponse.model_validate(response_data)
        else:
            return AudioTranscriptionResponse.model_validate(response_data)

text_to_speech(text, model=None, voice=None, api_args=None)

Convert input text to speech audio using a specified model and voice.

Parameters:

Name Type Description Default
text str

The text to be converted to speech.

 required
model Optional[str]

The speech synthesis model to use. Defaults to None.

 None
voice Optional[str]

The voice profile to use for speech. Defaults to None.

 None
api_args Optional[dict]

Additional API arguments for customization. Defaults to None.

 None

Returns:

Name Type Description
bytes bytes

The audio content generated from the text.
Source code in forrasdk/api/audio.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def text_to_speech(
    self,
    text: str,
    model: Optional[str] = None,
    voice: Optional[str] = None,
    api_args: Optional[dict] = None,
) -> bytes:
    """
    Convert input text to speech audio using a specified model and voice.

    Args:
        text (str): The text to be converted to speech.
        model (Optional[str], optional): The speech synthesis model to use. Defaults to None.
        voice (Optional[str], optional): The voice profile to use for speech. Defaults to None.
        api_args (Optional[dict], optional): Additional API arguments for customization. Defaults to None.

    Returns:
        bytes: The audio content generated from the text.
    """
    json_payload = {
        "text": text,
        **({"model": model} if model is not None else {}),
        **({"voice": voice} if voice is not None else {}),
        **({"api_args": api_args} if api_args is not None else {}),
    }

    response = requests.post(
        url=f"{self._base_url}/api/audio/speech",
        headers=self._headers,
        json=json_payload,
    )
    try:
        response.raise_for_status()
    except requests.exceptions.HTTPError as e:
        error_message = "HTTP Error occurred"
        if e.response is not None:
            try:
                error_details = e.response.json()
                error_message = f"Error: {error_details}"
            except (ValueError, AttributeError):
                # Handle cases where response is not valid JSON
                error_message = f"Error: HTTP {e.response.status_code} - {e.response.text or 'No response text'}"
        else:
            # Handle cases where e.response is None
            error_message = f"Error: {str(e)}"

        raise Exception(error_message) from e

    return response.content

transcribe_file(file_path, conversation_id=None, model=None, execute_async=True, model_args=None) 

transcribe_file(file_path: str, conversation_id: Optional[str] = None, model: Optional[str] = None, execute_async: Literal[True] = True, model_args: Optional[dict] = None) -> AsyncJobResponse

transcribe_file(file_path: str, conversation_id: Optional[str] = None, model: Optional[str] = None, execute_async: Literal[False] = False, model_args: Optional[dict] = None) -> AudioTranscriptionResponse

Transcribe an audio file using the Scout API.

Parameters:

Name Type Description Default
file_path str

The local path to the audio file to transcribe.

 required
conversation_id Optional[str]

The conversation ID to associate with the transcription.

 None
model Optional[str]

The transcription model to use.

 None
execute_async bool

Whether to execute the transcription asynchronously. Defaults to True.

 True
model_args Optional[dict]

Additional model arguments.

 None

Returns:

Type Description
Union[AudioTranscriptionResponse, AsyncJobResponse]

Union[AudioTranscriptionResponse, AsyncJobResponse]: The transcription response from the API.
Union[AudioTranscriptionResponse, AsyncJobResponse]

Returns AsyncJobResponse when execute_async=True, AudioTranscriptionResponse when execute_async=False.

Raises:

Type Description
FileNotFoundError

If the specified audio file doesn't exist.
Exception

If there's an error during the transcription process.
Source code in forrasdk/api/audio.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def transcribe_file(
    self,
    file_path: str,
    conversation_id: Optional[str] = None,
    model: Optional[str] = None,
    execute_async: bool = True,
    model_args: Optional[dict] = None,
) -> Union[AudioTranscriptionResponse, AsyncJobResponse]:
    """
    Transcribe an audio file using the Scout API.

    Args:
        file_path (str): The local path to the audio file to transcribe.
        conversation_id (Optional[str]): The conversation ID to associate with the transcription.
        model (Optional[str]): The transcription model to use.
        execute_async (bool): Whether to execute the transcription asynchronously. Defaults to True.
        model_args (Optional[dict]): Additional model arguments.

    Returns:
        Union[AudioTranscriptionResponse, AsyncJobResponse]: The transcription response from the API.
        Returns AsyncJobResponse when execute_async=True, AudioTranscriptionResponse when execute_async=False.

    Raises:
        FileNotFoundError: If the specified audio file doesn't exist.
        Exception: If there's an error during the transcription process.
    """
    if not os.path.exists(file_path):
        raise FileNotFoundError(f"Audio file not found: {file_path}")

    filename = os.path.basename(file_path)

    # Read file and encode to base64
    with open(file_path, "rb") as f:
        file_bytes = f.read()
        base64_encoded = base64.b64encode(file_bytes).decode("utf-8")

    # Create the request data with base64 encoded audio
    audio_object = AudioObject(filename=filename, base64=base64_encoded)

    return self._send_transcription_request(
        audio_object=audio_object,
        conversation_id=conversation_id,
        model=model,
        execute_async=execute_async,
        model_args=model_args,
    )

transcribe_url(protected_url, filename=None, conversation_id=None, model=None, execute_async=True, model_args=None) 

transcribe_url(protected_url: str, filename: Optional[str] = None, conversation_id: Optional[str] = None, model: Optional[str] = None, execute_async: Literal[True] = True, model_args: Optional[dict] = None) -> AsyncJobResponse

transcribe_url(protected_url: str, filename: Optional[str] = None, conversation_id: Optional[str] = None, model: Optional[str] = None, execute_async: Literal[False] = False, model_args: Optional[dict] = None) -> AudioTranscriptionResponse

Transcribe an audio file from a protected URL using the Scout API.

Parameters:

Name Type Description Default
protected_url str

The protected URL path to the audio file.

 required
filename Optional[str]

The filename to use. If None, extracted from URL.

 None
conversation_id Optional[str]

The conversation ID to associate with the transcription.

 None
model Optional[str]

The transcription model to use.

 None
execute_async bool

Whether to execute the transcription asynchronously. Defaults to True.

 True
model_args Optional[dict]

Additional model arguments.

 None

Returns:

Type Description
Union[AudioTranscriptionResponse, AsyncJobResponse]

Union[AudioTranscriptionResponse, AsyncJobResponse]: The transcription response from the API.
Union[AudioTranscriptionResponse, AsyncJobResponse]

Returns AsyncJobResponse when execute_async=True, AudioTranscriptionResponse when execute_async=False.

Raises:

Type Description
Exception

If there's an error during the transcription process.
Source code in forrasdk/api/audio.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def transcribe_url(
    self,
    protected_url: str,
    filename: Optional[str] = None,
    conversation_id: Optional[str] = None,
    model: Optional[str] = None,
    execute_async: bool = True,
    model_args: Optional[dict] = None,
) -> Union[AudioTranscriptionResponse, AsyncJobResponse]:
    """
    Transcribe an audio file from a protected URL using the Scout API.

    Args:
        protected_url (str): The protected URL path to the audio file.
        filename (Optional[str]): The filename to use. If None, extracted from URL.
        conversation_id (Optional[str]): The conversation ID to associate with the transcription.
        model (Optional[str]): The transcription model to use.
        execute_async (bool): Whether to execute the transcription asynchronously. Defaults to True.
        model_args (Optional[dict]): Additional model arguments.

    Returns:
        Union[AudioTranscriptionResponse, AsyncJobResponse]: The transcription response from the API.
        Returns AsyncJobResponse when execute_async=True, AudioTranscriptionResponse when execute_async=False.

    Raises:
        Exception: If there's an error during the transcription process.
    """
    # Extract filename from URL if not provided
    if filename is None:
        filename = protected_url.split("/")[-1] if "/" in protected_url else "audio"

    # Create the request data with protected URL
    audio_object = AudioObject(filename=filename, protected_url=protected_url)

    return self._send_transcription_request(
        audio_object=audio_object,
        conversation_id=conversation_id,
        model=model,
        execute_async=execute_async,
        model_args=model_args,
    )