hi:pls  mody it ,# -*- encoding: utf-8 -*-
import os
import ssl
import asyncio
import argparse
import json
import logging
import traceback
from multiprocessing import Process
from queue import Queue

# Additional imports
import time
import websockets

try:
    import pyaudio
except ImportError:
    pyaudio = None  # Microphone input will not be available

logging.basicConfig(level=logging.ERROR)

parser = argparse.ArgumentParser()
parser.add_argument(
    "--host", type=str, default="localhost", required=False, help="Host IP (e.g., localhost, 0.0.0.0)"
)
parser.add_argument("--port", type=int, default=10095, required=False, help="WebSocket server port")
parser.add_argument("--chunk_size", type=str, default="5,10,5", help="Chunk sizes in milliseconds")
parser.add_argument("--encoder_chunk_look_back", type=int, default=4, help="Encoder chunk look-back")
parser.add_argument("--decoder_chunk_look_back", type=int, default=0, help="Decoder chunk look-back")
parser.add_argument("--chunk_interval", type=int, default=10, help="Chunk interval in milliseconds")
parser.add_argument(
    "--hotword",
    type=str,
    default="",
    help="Hotword file path, one hotword per line (e.g., '阿里巴巴 20')",
)
parser.add_argument("--audio_in", type=str, default=None, help="Input audio file or SCP file")
parser.add_argument("--audio_fs", type=int, default=16000, help="Audio sampling rate")
parser.add_argument(
    "--send_without_sleep",
    action="store_true",
    default=True,
    help="If audio_in is set, send without sleep",
)
parser.add_argument("--thread_num", type=int, default=1, help="Number of threads")
parser.add_argument("--words_max_print", type=int, default=10000, help="Maximum words to print")
parser.add_argument("--output_dir", type=str, default=None, help="Output directory")
parser.add_argument("--ssl", type=int, default=1, help="1 for SSL connection, 0 for no SSL")
parser.add_argument("--use_itn", type=int, default=1, help="1 for using ITN, 0 for not using ITN")
parser.add_argument("--mode", type=str, default="2pass", help="Mode: offline, online, 2pass")

args = parser.parse_args()
args.chunk_size = [int(x) for x in args.chunk_size.split(",")]
print(args)

# Global variables
voices = Queue()
offline_msg_done = False
websocket = None  # Global websocket variable

# Create output directory if it doesn't exist
if args.output_dir is not None:
    if not os.path.exists(args.output_dir):
        os.makedirs(args.output_dir)

async def record_microphone():
    global voices, websocket

    if pyaudio is None:
        print("pyaudio is not installed. Microphone recording is unavailable.")
        return

    FORMAT = pyaudio.paInt16
    CHANNELS = 1
    RATE = args.audio_fs
    # Calculate chunk size
    chunk_duration_ms = 60 * args.chunk_size[1] / args.chunk_interval
    CHUNK = int(RATE * chunk_duration_ms / 1000)

    p = pyaudio.PyAudio()
    stream = p.open(
        format=FORMAT, channels=CHANNELS, rate=RATE, input=True, frames_per_buffer=CHUNK
    )

    # Prepare hotwords
    hotword_msg = prepare_hotwords(args.hotword)

    use_itn = args.use_itn == 1

    message = json.dumps(
        {
            "mode": args.mode,
            "chunk_size": args.chunk_size,
            "chunk_interval": args.chunk_interval,
            "encoder_chunk_look_back": args.encoder_chunk_look_back,
            "decoder_chunk_look_back": args.decoder_chunk_look_back,
            "wav_name": "microphone",
            "is_speaking": True,
            "hotwords": hotword_msg,
            "itn": use_itn,
        }
    )
    await websocket.send(message)

    try:
        while True:
            data = stream.read(CHUNK)
            await websocket.send(data)
            await asyncio.sleep(0.005)
    except Exception as e:
        print(f"An error occurred during microphone recording: {e}")
    finally:
        stream.stop_stream()
        stream.close()
        p.terminate()

async def record_from_scp(chunk_begin, chunk_size):
    global voices, websocket, offline_msg_done

    if args.audio_in is None:
        print("No input audio provided.")
        return

    # Read audio file paths
    if args.audio_in.endswith(".scp"):
        with open(args.audio_in, "r") as f:
            wavs = f.readlines()
    else:
        wavs = [args.audio_in]

    # Prepare hotwords
    hotword_msg = prepare_hotwords(args.hotword)

    sample_rate = args.audio_fs
    wav_format = "pcm"
    use_itn = args.use_itn == 1

    if chunk_size > 0:
        wavs = wavs[chunk_begin : chunk_begin + chunk_size]

    for wav in wavs:
        wav_splits = wav.strip().split()
        wav_name = wav_splits[0] if len(wav_splits) > 1 else "demo"
        wav_path = wav_splits[1] if len(wav_splits) > 1 else wav_splits[0]

        if not wav_path.strip():
            continue

        # Read audio data
        audio_bytes, sample_rate, wav_format = read_audio_data(wav_path, sample_rate)

        # Calculate stride
        stride = int(
            60 * args.chunk_size[1] / args.chunk_interval / 1000 * sample_rate * 2
        )
        chunk_num = (len(audio_bytes) - 1) // stride + 1

        # Send initial message
        message = json.dumps(
            {
                "mode": args.mode,
                "chunk_size": args.chunk_size,
                "chunk_interval": args.chunk_interval,
                "encoder_chunk_look_back": args.encoder_chunk_look_back,
                "decoder_chunk_look_back": args.decoder_chunk_look_back,
                "audio_fs": sample_rate,
                "wav_name": wav_name,
                "wav_format": wav_format,
                "is_speaking": True,
                "hotwords": hotword_msg,
                "itn": use_itn,
            }
        )
        await websocket.send(message)
        is_speaking = True

        for i in range(chunk_num):
            beg = i * stride
            data = audio_bytes[beg : beg + stride]
            await websocket.send(data)

            if i == chunk_num - 1:
                is_speaking = False
                message = json.dumps({"is_speaking": is_speaking})
                await websocket.send(message)

            sleep_duration = (
                0.001
                if args.mode == "offline"
                else 60 * args.chunk_size[1] / args.chunk_interval / 1000
            )
            await asyncio.sleep(sleep_duration)

        if args.mode != "offline":
            await asyncio.sleep(2)

        if args.mode == "offline":
            while not offline_msg_done:
                await asyncio.sleep(1)

async def message_handler(process_id):
    global websocket, offline_msg_done
    text_print = ""
    text_print_2pass_online = ""
    text_print_2pass_offline = ""

    # Prepare output file if needed
    ibest_writer = None
    if args.output_dir is not None:
        output_file = os.path.join(args.output_dir, f"text.{process_id}")
        ibest_writer = open(output_file, "a", encoding="utf-8")

    try:
        while True:
            msg = await websocket.recv()
            msg = json.loads(msg)
            wav_name = msg.get("wav_name", "demo")
            text = msg.get("text", "")
            timestamp = msg.get("timestamp", "")
            offline_msg_done = msg.get("is_final", False)

            # Write to output file
            if ibest_writer is not None:
                if timestamp:
                    text_write_line = f"{wav_name}\t{text}\t{timestamp}\n"
                else:
                    text_write_line = f"{wav_name}\t{text}\n"
                ibest_writer.write(text_write_line)
                ibest_writer.flush()

            # Display text based on mode
            if "mode" not in msg:
                continue

            mode = msg["mode"]
            if mode == "online":
                text_print += f"{text}"
                text_print = text_print[-args.words_max_print :]
                os.system("clear")
                print(f"\rProcess {process_id}: {text_print}")
            elif mode == "offline":
                if timestamp:
                    text_print += f"{text} timestamp: {timestamp}"
                else:
                    text_print += f"{text}"
                print(f"\rProcess {process_id}: {wav_name}: {text_print}")
                offline_msg_done = True
            else:
                if mode == "2pass-online":
                    text_print_2pass_online += f"{text}"
                    text_print = text_print_2pass_offline + text_print_2pass_online
                else:
                    text_print_2pass_online = ""
                    text_print = text_print_2pass_offline + f"{text}"
                    text_print_2pass_offline += f"{text}"
                text_print = text_print[-args.words_max_print :]
                os.system("clear")
                print(f"\rProcess {process_id}: {text_print}")
    except Exception as e:
        print(f"Exception in message handler: {e}")
        # Optional: Uncomment the following line to print the traceback
        # traceback.print_exc()

async def ws_client(process_id, chunk_begin, chunk_size):
    global websocket, offline_msg_done

    # Determine URI and SSL context
    if args.ssl == 1:
        ssl_context = ssl.SSLContext()
        ssl_context.check_hostname = False
        ssl_context.verify_mode = ssl.CERT_NONE
        uri = f"wss://{args.host}:{args.port}"
    else:
        uri = f"ws://{args.host}:{args.port}"
        ssl_context = None

    print(f"Connecting to {uri}")

    try:
        async with websockets.connect(
            uri, subprotocols=["binary"], ping_interval=None, ssl=ssl_context
        ) as websocket:
            if args.audio_in is not None:
                task_record = asyncio.create_task(record_from_scp(chunk_begin, chunk_size))
            else:
                task_record = asyncio.create_task(record_microphone())

            task_message = asyncio.create_task(message_handler(f"{process_id}"))
            await asyncio.gather(task_record, task_message)
    except Exception as e:
        print(f"Exception in ws_client: {e}")
        # Optional: Uncomment the following line to print the traceback
        # traceback.print_exc()
    finally:
        if websocket and not websocket.closed:
            await websocket.close()

def one_thread(process_id, chunk_begin, chunk_size):
    asyncio.run(ws_client(process_id, chunk_begin, chunk_size))

def prepare_hotwords(hotword_path):
    hotword_msg = ""
    if hotword_path.strip():
        fst_dict = {}
        if os.path.exists(hotword_path):
            with open(hotword_path, "r", encoding="utf-8") as f:
                for line in f:
                    words = line.strip().split()
                    if len(words) < 2:
                        print("Invalid hotword format. Each line should contain hotword and weight.")
                        continue
                    try:
                        fst_dict[" ".join(words[:-1])] = int(words[-1])
                    except ValueError:
                        print("Hotword weight must be an integer.")
                hotword_msg = json.dumps(fst_dict, ensure_ascii=False)
        else:
            hotword_msg = hotword_path
    return hotword_msg

def read_audio_data(wav_path, sample_rate):
    wav_format = "pcm"
    audio_bytes = b""
    if wav_path.endswith(".pcm"):
        with open(wav_path, "rb") as f:
            audio_bytes = f.read()
    elif wav_path.endswith(".wav"):
        import wave

        with wave.open(wav_path, "rb") as wav_file:
            params = wav_file.getparams()
            sample_rate = wav_file.getframerate()
            frames = wav_file.readframes(wav_file.getnframes())
            audio_bytes = frames
    else:
        wav_format = "others"
        with open(wav_path, "rb") as f:
            audio_bytes = f.read()
    return audio_bytes, sample_rate, wav_format

if __name__ == "__main__":
    if args.audio_in is None:
        # Microphone input
        print("Starting microphone recording...")
        process = Process(target=one_thread, args=(0, 0, 0))
        process.start()
        process.join()
        print("Microphone recording ended.")
    else:
        # Audio file input
        if args.audio_in.endswith(".scp"):
            with open(args.audio_in, "r") as f:
                wavs = f.readlines()
        else:
            wavs = [args.audio_in]

        total_files = len(wavs)
        if total_files >= args.thread_num:
            files_per_thread = total_files // args.thread_num
            extra_files = total_files % args.thread_num
        else:
            files_per_thread = 1
            extra_files = 0

        processes = []
        chunk_begin = 0

        for i in range(args.thread_num):
            current_chunk_size = files_per_thread + (1 if i < extra_files else 0)
            p = Process(target=one_thread, args=(i, chunk_begin, current_chunk_size))
            chunk_begin += current_chunk_size
            p.start()
            processes.append(p)

        for p in processes:
            p.join()

        print("Audio file processing ended.")Connecting to wss://localhost:10095
ALSA lib pcm_dmix.c:1032:(snd_pcm_dmix_open) unable to open slave
ALSA lib pcm.c:2664:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.rear
ALSA lib pcm.c:2664:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.center_lfe
ALSA lib pcm.c:2664:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.side
ALSA lib pcm_route.c:877:(find_matching_chmap) Found no matching channel map
ALSA lib pcm_oss.c:397:(_snd_pcm_oss_open) Cannot open device /dev/dsp
ALSA lib pcm_oss.c:397:(_snd_pcm_oss_open) Cannot open device /dev/dsp
ALSA lib confmisc.c:160:(snd_config_get_card) Invalid field card
ALSA lib pcm_usb_stream.c:482:(_snd_pcm_usb_stream_open) Invalid card 'card'
ALSA lib confmisc.c:160:(snd_config_get_card) Invalid field card
ALSA lib pcm_usb_stream.c:482:(_snd_pcm_usb_stream_open) Invalid card 'card'
ALSA lib pcm_dmix.c:1032:(snd_pcm_dmix_open) unable to open slave