-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtest_client.py
105 lines (87 loc) · 2.86 KB
/
test_client.py
1
2
3
4
5
6
7
8
9
10
11
12
13
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
import requests
import os
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
def process_audio(audio_file_path, task="transcribe", language=None, initial_prompt=None, prefix=None):
"""
Process audio file with specified parameters for either transcription or translation
Args:
audio_file_path (str): Path to the audio file
task (str): Either "transcribe" or "translate"
language (str): Target language code (e.g., "en", "es", "fr")
initial_prompt (str): Optional initial prompt for the model
prefix (str): Optional prefix for the model
"""
# Get token from environment variable
token = os.getenv("WHISPER_API_TOKEN", "default_token_change_me")
# Read the audio file as bytes
with open(audio_file_path, 'rb') as f:
audio_bytes = f.read()
# Set up headers with authentication token
headers = {
"Authorization": f"Bearer {token}"
}
# Build query parameters
params = {}
if task:
params["task"] = task
if language:
params["language"] = language
if initial_prompt:
params["initial_prompt"] = initial_prompt
if prefix:
params["prefix"] = prefix
print(f"\nSending request with parameters:")
print(f"Task: {task}")
print(f"Language: {language}")
print(f"Initial Prompt: {initial_prompt}")
print(f"Prefix: {prefix}\n")
# Send the request
response = requests.post(
"http://127.0.0.1:8000/",
params=params,
data=audio_bytes,
headers=headers
)
if response.status_code == 401:
print("Authentication failed - check your WHISPER_API_TOKEN")
return
# Print the response
if response.status_code == 200:
print("Result:", response.json())
else:
print(f"Error: {response.status_code}", response.text)
def run_tests():
audio_file_path = "test_audio.webm"
# Test 1: Basic transcription
print("\n=== Test 1: Basic Transcription ===")
process_audio(
audio_file_path,
task="transcribe"
)
# Test 2: Translation to English
print("\n=== Test 2: Translation to English ===")
process_audio(
audio_file_path,
task="translate",
language="en"
)
# Test 3: Transcription with initial prompt
print("\n=== Test 3: Transcription with Initial Prompt ===")
process_audio(
audio_file_path,
task="transcribe",
initial_prompt="This is a test recording"
)
# Test 4: Translation to Spanish
print("\n=== Test 4: Translation to Spanish ===")
process_audio(
audio_file_path,
task="translate",
language="es"
)
if __name__ == "__main__":
run_tests()
#ray start --head --node-ip-address=0.0.0.0 --port=6379
#serve deploy app:transcriber_app