lid_stream_file.py

import argparse
import asyncio
import json
import sys
import websockets

SAMPLES_PER_CHUNK = 1024
BYTES_PER_SAMPLE = 2  # 2 bytes per sample (16-bit PCM)
SERVER_PORT = "443"
LID_SERVER_HOST = "lid.mimi.fd.ai"
LID_SERVER_PATH = "/v1/identify"

async def stream(token, file_data, args):
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "audio/x-pcm;bit=16;rate=16000;channels=1",
        "x-mimi-lid-options": args.lid_options
    }

    try:
        async with websockets.connect(
            f"wss://{LID_SERVER_HOST}:{SERVER_PORT}{LID_SERVER_PATH}",
            additional_headers=headers,
            ping_interval=None,
        ) as ws:
            sent_size = 0
            while sent_size < len(file_data):
                await ws.send(file_data[sent_size:sent_size + SAMPLES_PER_CHUNK * BYTES_PER_SAMPLE])
                sent_size += SAMPLES_PER_CHUNK * BYTES_PER_SAMPLE
            await ws.send(json.dumps({"command": "recog-break"}))

            while True:
                resp = await ws.recv()
                print(resp)
                if json.loads(resp)["status"] == "recog-finished":
                    print("recog-finished: received all from server.", file=sys.stderr)

    except websockets.exceptions.ConnectionClosed:
        print("connection closed from server", file=sys.stderr)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='Send audio to mimi LID WebSocket services')
    parser.add_argument('access_token', help='access token file')
    parser.add_argument('audio', help='audio file (RAW format)')
    parser.add_argument('--lid-options', default='lang=ja|en|zh|ko',
                        help='range of languages for lid service (default: lang=ja|en|zh|ko)')
    args = parser.parse_args()

    with open(args.access_token) as f:
        token = f.read().strip()
    with open(args.audio, 'rb') as f:
        file_data = f.read()

    print("start mimi LID ...", file=sys.stderr)
    try:
        asyncio.run(stream(token, file_data, args))
    except Exception as e:
        print(f"ERROR: {e}", file=sys.stderr)
        sys.exit(1)