Music Detection Streaming

import asyncio
import websockets
import json

WS_URL = "wss://modulate-developer-apis.com/api/velma-2-music-detection-streaming"
API_KEY = "YOUR_API_KEY"

async def stream_audio(file_path: str) -> None:
    url = (
        f"{WS_URL}?api_key={API_KEY}"
        f"&audio_format=s16le&sample_rate=16000&num_channels=1"
    )
    async with websockets.connect(url) as ws:
        # Send audio in chunks
        with open(file_path, "rb") as f:
            while chunk := f.read(16000):  # 0.5s of s16le/16kHz mono
                await ws.send(chunk)

        # Signal end of stream
        await ws.send("")

        # Receive results
        async for message in ws:
            msg = json.loads(message)
            if msg["type"] == "frame":
                frame = msg["frame"]
                print(
                    f"{frame['start_time_ms']}ms – {frame['end_time_ms']}ms  "
                    f"music={frame['music_prob']:.4f}  speech={frame['speech_prob']:.4f}"
                )
            elif msg["type"] == "done":
                print(f"\nDone — {msg['duration_ms']}ms, {msg['frame_count']} frames")
                print(f"music={msg['music_pct']}%  speech={msg['speech_pct']}%  label={msg['primary_label']}")
                break
            elif msg["type"] == "error":
                raise RuntimeError(f"Server error: {msg['error']}")

asyncio.run(stream_audio("/path/to/audio.raw"))