Spaces:
Sleeping
Sleeping
File size: 1,812 Bytes
09083dc 8b23235 09083dc | 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 | # test_audio_api.py
import asyncio
import websockets
import json
import base64
# Test locally - change to HF URL when deployed
BASE_URL = "localhost:8000"
# BASE_URL = "localhost:7860"
# BASE_URL = "arnavam-afs-backend.hf.space"
async def test_audio():
session_id = "test-session-123"
# WebSocket URI
ws_protocol = "ws" if "localhost" in BASE_URL else "wss"
uri = f"{ws_protocol}://{BASE_URL}/ws/audio/{session_id}"
print(f"Connecting to {uri}...")
try:
async with websockets.connect(uri, ping_timeout=10) as ws:
print("Connected!")
# Send audio chunk with angle
audio_bytes = b'\x00\x00' * 1000 # dummy audio data
payload = {
"audio_data": base64.b64encode(audio_bytes).decode(),
"angle": 45.5
}
print(f"Sending audio chunk with angle {payload['angle']}...")
await ws.send(json.dumps(payload))
response = await ws.recv()
print(f"Response: {response}")
# Send another chunk with different angle
payload["angle"] = 90.0
print(f"Sending audio chunk with angle {payload['angle']}...")
await ws.send(json.dumps(payload))
response = await ws.recv()
print(f"Response: {response}")
# Stop stream
print("Stopping stream...")
await ws.send(json.dumps({"command": "stop"}))
response = await ws.recv()
print(f"Response: {response}")
except websockets.exceptions.ConnectionClosedError as e:
print(f"Connection closed: {e}")
except Exception as e:
print(f"Error: {e}")
if __name__ == "__main__":
asyncio.run(test_audio())
|