microphone.py

import argparse
import asyncio
import json
import signal
import sounddevice as sd
import sys
import websockets

SIGNAL_INT = False

# Flag set by SIGINT handler; checked in the sounddevice callback to stop recording.
def signal_handler(_signum, _frame):
    global SIGNAL_INT
    SIGNAL_INT = True


ASR_SERVER_HOST = "asr.mimi.fd.ai"
ASR_SERVER_PORT = "443"
ASR_SERVER_PATH = "/v1/recognize/nict"  # when using mimi ASR (not mimi ASR powered by NICT) use "/v1/recognize/fairy"

async def recognize(token, sampling_rate, response_format, enable_progressive, enable_temporary):
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": f"audio/x-pcm;bit=16;rate={sampling_rate};channels=1",
        "x-mimi-input-language": "ja",
    }

    nict_asr_options = f"response_format={response_format}"
    if response_format == "v2":
        nict_asr_options += f";progressive={str(enable_progressive).lower()}"
        nict_asr_options += f";temporary={str(enable_temporary).lower()}"
    headers["x-mimi-nict-asr-options"] = nict_asr_options

    try:
        async with websockets.connect(
                f"wss://{ASR_SERVER_HOST}:{ASR_SERVER_PORT}{ASR_SERVER_PATH}",
                additional_headers=headers,
                ping_interval=None
        ) as ws:
            await asyncio.gather(
                send(ws, sampling_rate),
                receive(ws)
            )
    except websockets.exceptions.ConnectionClosed:
        print("connection closed from server", file=sys.stderr)


async def record(sampling_rate):
    q = asyncio.Queue()
    loop = asyncio.get_running_loop()

    def callback(indata, _frame_count, _time_info, status):
        if status:
            print(status, file=sys.stderr)
        if SIGNAL_INT:
            loop.call_soon_threadsafe(
                q.put_nowait, (indata.copy(), True))
            raise sd.CallbackStop

        loop.call_soon_threadsafe(q.put_nowait, (indata.copy(), False))

    with sd.InputStream(channels=1, dtype="int16", callback=callback, samplerate=sampling_rate):
        while True:
            audio, is_final = await q.get()
            yield audio, is_final


async def send(ws, sampling_rate):
    async for audio, is_final in record(sampling_rate):
        await ws.send(audio.tobytes())
        if is_final:
            await ws.send(json.dumps({"command": "recog-break"}))
            break


async def receive(ws):
    async for message in ws:
        print(message)
        if json.loads(message)["status"] == "recog-finished":
            print("recog-finished: received all from server.", file=sys.stderr)


if __name__ == "__main__":
    signal.signal(signal.SIGINT, signal_handler)

    parser = argparse.ArgumentParser()
    parser.add_argument('access_token', help='access token file')
    parser.add_argument('-f', '--response-format', default='v2', choices=['v1', 'v2'], help='response format (v1 or v2)')
    parser.add_argument('--progressive', action='store_true', dest='progressive',
                        help='enable progressive option')
    parser.add_argument('--temporary', action='store_true', dest='temporary',
                        help='enable temporary option')
    parser.set_defaults(progressive=False, temporary=False)
    args = parser.parse_args()

    with open(args.access_token, "r") as f:
        token = f.read().strip()

    device_info = sd.query_devices(kind="input")
    sampling_rate = int(device_info["default_samplerate"])

    print("#" * 80, file=sys.stderr)
    print("start recording ...\npress Ctrl+C to stop recording", file=sys.stderr)
    print("#" * 80, file=sys.stderr)

    try:
        asyncio.run(
            recognize(token, sampling_rate, args.response_format, args.progressive, args.temporary))
    except Exception as e:
        print(f"ERROR: {e}", file=sys.stderr)
        sys.exit(1)