Qwen-Omni-Realtime

# -- coding: utf-8 --
import os
import asyncio
import time
import threading
import queue
import pyaudio
from omni_realtime_client import OmniRealtimeClient, TurnDetectionMode

class AudioPlayer:
    """リアルタイムオーディオプレーヤークラス。"""

    def __init__(self, sample_rate=24000, channels=1, sample_width=2):
        self.sample_rate = sample_rate
        self.channels = channels
        self.sample_width = sample_width  # 16 ビットの場合は 2 バイト
        self.audio_queue = queue.Queue()
        self.is_playing = False
        self.play_thread = None
        self.pyaudio_instance = None
        self.stream = None
        self._lock = threading.Lock()  # 同期アクセスのためのロックを追加します。
        self._last_data_time = time.time()  # 最後のデータを受信した時刻を記録します。
        self._response_done = False  # 応答が完了したことを示すフラグを追加します。
        self._waiting_for_response = False  # クライアントがサーバーの応答を待っているかどうかを示すフラグ。
        # オーディオストリームへの最後の書き込み時刻と、より正確な再生終了検出のための最後のオーディオチャンクの持続時間を記録します。
        self._last_play_time = time.time()
        self._last_chunk_duration = 0.0

    def start(self):
        """オーディオプレーヤーを開始します。"""
        with self._lock:
            if self.is_playing:
                return

            self.is_playing = True

            try:
                self.pyaudio_instance = pyaudio.PyAudio()

                # オーディオ出力ストリームを作成します。
                self.stream = self.pyaudio_instance.open(
                    format=pyaudio.paInt16,  # 16 ビット
                    channels=self.channels,
                    rate=self.sample_rate,
                    output=True,
                    frames_per_buffer=1024
                )

                # 再生スレッドを開始します。
                self.play_thread = threading.Thread(target=self._play_audio)
                self.play_thread.daemon = True
                self.play_thread.start()

                print("Audio player started")
            except Exception as e:
                print(f"Failed to start audio player: {e}")
                self._cleanup_resources()
                raise

    def stop(self):
        """オーディオプレーヤーを停止します。"""
        with self._lock:
            if not self.is_playing:
                return

            self.is_playing = False

        # キューをクリアします。
        while not self.audio_queue.empty():
            try:
                self.audio_queue.get_nowait()
            except queue.Empty:
                break

        # 再生スレッドが終了するのを待ちます。デッドロックを避けるためにロックの外で待ちます。
        if self.play_thread and self.play_thread.is_alive():
            self.play_thread.join(timeout=2.0)

        # リソースをクリーンアップするために再度ロックを取得します。
        with self._lock:
            self._cleanup_resources()

        print("Audio player stopped")

    def _cleanup_resources(self):
        """オーディオリソースをクリーンアップします。これはロック内で呼び出す必要があります。"""
        try:
            # オーディオストリームを閉じます。
            if self.stream:
                if not self.stream.is_stopped():
                    self.stream.stop_stream()
                self.stream.close()
                self.stream = None
        except Exception as e:
            print(f"Error closing audio stream: {e}")

        try:
            if self.pyaudio_instance:
                self.pyaudio_instance.terminate()
                self.pyaudio_instance = None
        except Exception as e:
            print(f"Error terminating PyAudio: {e}")

    def add_audio_data(self, audio_data):
        """再生キューにオーディオデータを追加します。"""
        if self.is_playing and audio_data:
            self.audio_queue.put(audio_data)
            with self._lock:
                self._last_data_time = time.time()  # 最後のデータを受信した時刻を更新します。
                self._waiting_for_response = False  # データを受信したので、待機を解除します。

    def stop_receiving_data(self):
        """これ以上新しいオーディオデータを受信しないことをマークします。"""
        with self._lock:
            self._response_done = True
            self._waiting_for_response = False  # 応答が終了したので、待機を解除します。

    def prepare_for_next_turn(self):
        """次の会話ターンのためにプレーヤーの状態をリセットします。"""
        with self._lock:
            self._response_done = False
            self._last_data_time = time.time()
            self._last_play_time = time.time()
            self._last_chunk_duration = 0.0
            self._waiting_for_response = True  # 次の応答を待機開始します。

        # 前のターンから残っているオーディオデータをクリアします。
        while not self.audio_queue.empty():
            try:
                self.audio_queue.get_nowait()
            except queue.Empty:
                break

    def is_finished_playing(self):
        """すべてのオーディオデータが再生されたかどうかを確認します。"""
        with self._lock:
            queue_size = self.audio_queue.qsize()
            time_since_last_data = time.time() - self._last_data_time
            time_since_last_play = time.time() - self._last_play_time

            # ---------------------- スマート終了検出 ----------------------
            # 1. 推奨方法：サーバーが完了をマークし、再生キューが空の場合、
            #    最新のオーディオチャンクが再生し終わるのを待ちます (チャンク持続時間 + 0.1 秒の許容範囲)。
            if self._response_done and queue_size == 0:
                min_wait = max(self._last_chunk_duration + 0.1, 0.5)  # 少なくとも 0.5 秒待ちます。
                if time_since_last_play >= min_wait:
                    return True

            # 2. フォールバック方法：1 秒以上新しいデータが受信されず、再生キューが空の場合。
            #    このロジックは、サーバーが明示的に `response.done` を送信しない場合のセーフガードとして機能します。
            if not self._waiting_for_response and queue_size == 0 and time_since_last_data > 1.0:
                print("\n(No new audio received for a while, assuming playback is finished)")
                return True

            return False

    def _play_audio(self):
        """オーディオデータを再生するためのワーカースレッド。"""
        while True:
            # 停止すべきかどうかを確認します。
            with self._lock:
                if not self.is_playing:
                    break
                stream_ref = self.stream  # ストリームへの参照を取得します。

            try:
                # キューからオーディオデータを取得します。タイムアウトは 0.1 秒です。
                audio_data = self.audio_queue.get(timeout=0.1)

                # ステータスとストリームの有効性を再度確認します。
                with self._lock:
                    if self.is_playing and stream_ref and not stream_ref.is_stopped():
                        try:
                            # オーディオデータを再生します。
                            stream_ref.write(audio_data)
                            # 最新の再生情報を更新します。
                            self._last_play_time = time.time()
                            self._last_chunk_duration = len(audio_data) / (
                                        self.channels * self.sample_width) / self.sample_rate
                        except Exception as e:
                            print(f"Error writing to audio stream: {e}")
                            break

                # このデータブロックを処理済みとしてマークします。
                self.audio_queue.task_done()

            except queue.Empty:
                # キューが空の場合は待機を続けます。
                continue
            except Exception as e:
                print(f"Error playing audio: {e}")
                break

class MicrophoneRecorder:
    """リアルタイムマイクレコーダー。"""

    def __init__(self, sample_rate=16000, channels=1, chunk_size=3200):
        self.sample_rate = sample_rate
        self.channels = channels
        self.chunk_size = chunk_size
        self.pyaudio_instance = None
        self.stream = None
        self.frames = []
        self._is_recording = False
        self._record_thread = None

    def _recording_thread(self):
        """録音ワーカースレッド。"""
        # _is_recording が True の間、オーディオストリームからデータを継続的に読み取ります。
        while self._is_recording:
            try:
                # バッファオーバーフローによるクラッシュを避けるために exception_on_overflow=False を使用します。
                data = self.stream.read(self.chunk_size, exception_on_overflow=False)
                self.frames.append(data)
            except (IOError, OSError) as e:
                # ストリームが閉じられると、読み取り操作でエラーが発生する可能性があります。
                print(f"Error reading from recording stream, it might be closed: {e}")
                break

    def start(self):
        """録音を開始します。"""
        if self._is_recording:
            print("Recording is already in progress.")
            return

        self.frames = []
        self._is_recording = True

        try:
            self.pyaudio_instance = pyaudio.PyAudio()
            self.stream = self.pyaudio_instance.open(
                format=pyaudio.paInt16,
                channels=self.channels,
                rate=self.sample_rate,
                input=True,
                frames_per_buffer=self.chunk_size
            )

            self._record_thread = threading.Thread(target=self._recording_thread)
            self._record_thread.daemon = True
            self._record_thread.start()
            print("Microphone recording started...")
        except Exception as e:
            print(f"Failed to start microphone: {e}")
            self._is_recording = False
            self._cleanup()
            raise

    def stop(self):
        """録音を停止し、オーディオデータを返します。"""
        if not self._is_recording:
            return None

        self._is_recording = False

        # 録音スレッドが安全に終了するのを待ちます。
        if self._record_thread:
            self._record_thread.join(timeout=1.0)

        self._cleanup()

        print("Microphone recording stopped.")
        return b''.join(self.frames)

    def _cleanup(self):
        """PyAudio リソースを安全にクリーンアップします。"""
        if self.stream:
            try:
                if self.stream.is_active():
                    self.stream.stop_stream()
                self.stream.close()
            except Exception as e:
                print(f"Error closing audio stream: {e}")

        if self.pyaudio_instance:
            try:
                self.pyaudio_instance.terminate()
            except Exception as e:
                print(f"Error terminating PyAudio instance: {e}")

        self.stream = None
        self.pyaudio_instance = None

async def interactive_test():
    """
    オーディオと画像のサポートを備えたマルチターン会話のインタラクティブテスト。
    """
    # ------------------- 1. 初期化と接続 (1 回限り) -------------------
    # シンガポールと中国 (北京) リージョンの API キーは異なります。API キーを取得するには、https://www.alibabacloud.com/help/model-studio/get-api-key をご参照ください。
    api_key = os.environ.get("DASHSCOPE_API_KEY")
    if not api_key:
        print("Please set the DASHSCOPE_API_KEY environment variable.")
        return

    print("--- Real-time Multimodal Audio/Video Chat Client ---")
    print("Initializing audio player and client...")

    audio_player = AudioPlayer()
    audio_player.start()

    def on_audio_received(audio_data):
        audio_player.add_audio_data(audio_data)

    def on_response_done(event):
        print("\n(Received response end marker)")
        audio_player.stop_receiving_data()

    realtime_client = OmniRealtimeClient(
        # これはシンガポールリージョンの base_url です。中国 (北京) リージョンのモデルを使用する場合は、base_url を wss://WorkspaceId.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime に置き換えてください。
        base_url="wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime",
        api_key=api_key,
        model="qwen3.5-omni-plus-realtime",
        voice="Ethan",
        instructions="あなたはシャオユンという名のパーソナルアシスタントです。ユーザーの質問に正確かつフレンドリーに回答し、常に親切な態度で対応してください。", # モデルの役割を設定します。
        on_text_delta=lambda text: print(f"assistant: {text}", end="", flush=True),
        on_audio_delta=on_audio_received,
        turn_detection_mode=TurnDetectionMode.MANUAL,
        extra_event_handlers={"response.done": on_response_done}
    )

    message_handler_task = None
    try:
        await realtime_client.connect()
        print("Connected to the server. Enter 'q' or 'quit' to exit at any time.")
        message_handler_task = asyncio.create_task(realtime_client.handle_messages())
        await asyncio.sleep(0.5)

        turn_counter = 1
        # ------------------- 2. マルチターン会話ループ -------------------
        while True:
            print(f"\n--- Turn {turn_counter} ---")
            audio_player.prepare_for_next_turn()

            recorded_audio = None
            image_paths = []

            # --- ユーザー入力を取得：マイクから録音 ---
            loop = asyncio.get_event_loop()
            recorder = MicrophoneRecorder(sample_rate=16000)  # 音声認識には 16k のサンプルレートを推奨します。

            print("Ready to record. Press Enter to start recording (or enter 'q' to exit)...")
            user_input = await loop.run_in_executor(None, input)
            if user_input.strip().lower() in ['q', 'quit']:
                print("User requested to exit...")
                return

            try:
                recorder.start()
            except Exception:
                print("Could not start recording. Please check your microphone permissions and device. Skipping this turn.")
                continue

            print("Recording... Press Enter again to stop recording.")
            await loop.run_in_executor(None, input)

            recorded_audio = recorder.stop()

            if not recorded_audio or len(recorded_audio) == 0:
                print("No valid audio was recorded. Please start this turn again.")
                continue

            # --- 画像入力を取得 (オプション) ---
            # 画像入力機能はデフォルトで無効になっています。有効にするには、以下のコードのコメントを解除してください。
            # print("\nEnter the absolute path of an [image file] on each line (optional). When finished, enter 's' or press Enter to send the request.")
            # while True:
            #     path = input("Image path: ").strip()
            #     if path.lower() == 's' or path == '':
            #         break
            #     if path.lower() in ['q', 'quit']:
            #         print("User requested to exit...")
            #         return
            #
            #     if not os.path.isabs(path):
            #         print("Error: Please enter an absolute path.")
            #         continue
            #     if not os.path.exists(path):
            #         print(f"Error: File not found -> {path}")
            #         continue
            #     image_paths.append(path)
            #     print(f"Image added: {os.path.basename(path)}")

            # --- 3. データを送信して応答を取得 ---
            print("\n--- Input Confirmation ---")
            print(f"Audio to process: 1 (from microphone), Images: {len(image_paths)}")
            print("------------------")

            # 3.1 録音した音声を送信します。
            try:
                print(f"Sending microphone recording ({len(recorded_audio)} bytes)")
                await realtime_client.stream_audio(recorded_audio)
                await asyncio.sleep(0.1)
            except Exception as e:
                print(f"Failed to send microphone recording: {e}")
                continue

            # 3.2 すべての画像ファイルを送信します。
            # 画像入力機能はデフォルトで無効になっています。有効にするには、以下のコードのコメントを解除してください。
            # for i, path in enumerate(image_paths):
            #     try:
            #         with open(path, "rb") as f:
            #             data = f.read()
            #         print(f"Sending image {i+1}: {os.path.basename(path)} ({len(data)} bytes)")
            #         await realtime_client.append_image(data)
            #         await asyncio.sleep(0.1)
            #     except Exception as e:
            #         print(f"Failed to send image {os.path.basename(path)}: {e}")

            # 3.3 送信して応答を待ちます。
            print("Submitting all inputs, requesting server response...")
            await realtime_client.commit_audio_buffer()
            await realtime_client.create_response()

            print("Waiting for and playing server response audio...")
            start_time = time.time()
            max_wait_time = 60
            while not audio_player.is_finished_playing():
                if time.time() - start_time > max_wait_time:
                    print(f"\nWait timed out ({max_wait_time} seconds). Moving to the next turn.")
                    break
                await asyncio.sleep(0.2)

            print("\nAudio playback for this turn is complete!")
            turn_counter += 1

    except (asyncio.CancelledError, KeyboardInterrupt):
        print("\nProgram was interrupted.")
    except Exception as e:
        print(f"An unhandled error occurred: {e}")
    finally:
        # ------------------- 4. リソースのクリーンアップ -------------------
        print("\nClosing connection and cleaning up resources...")
        if message_handler_task and not message_handler_task.done():
            message_handler_task.cancel()

        if 'realtime_client' in locals() and realtime_client.ws and not realtime_client.ws.close:
            await realtime_client.close()
            print("Connection closed.")

        audio_player.stop()
        print("Program exited.")

if __name__ == "__main__":
    try:
        asyncio.run(interactive_test())
    except KeyboardInterrupt:
        print("\nProgram was forcibly exited by the user.")