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  # 2 bytes for 16-bit
        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-bit
                    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("音頻播放器已啟動")
            except Exception as e:
                print(f"啟動音頻播放器失敗: {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("音頻播放器已停止")

    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"關閉音頻流時出錯: {e}")

        try:
            if self.pyaudio_instance:
                self.pyaudio_instance.terminate()
                self.pyaudio_instance = None
        except Exception as e:
            print(f"終止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.1s 容錯）。
            if self._response_done and queue_size == 0:
                min_wait = max(self._last_chunk_duration + 0.1, 0.5)  # 至少等待 0.5s
                if time_since_last_play >= min_wait:
                    return True

            # 2. 備用：如果長時間沒有新資料且播放隊列為空白
            #    當伺服器沒有明確發出 `response.done` 時，此邏輯作為保障
            if not self._waiting_for_response and queue_size == 0 and time_since_last_data > 1.0:
                print("\n(逾時未收到新音頻，判定播放結束)")
                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"寫入音頻流時出錯: {e}")
                            break

                # 標記該資料區塊已處理完成
                self.audio_queue.task_done()

            except queue.Empty:
                # 隊列為空白時繼續等待
                continue
            except Exception as e:
                print(f"播放音頻時出錯: {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"錄音流讀取錯誤，可能已關閉: {e}")
                break

    def start(self):
        """開始錄音"""
        if self._is_recording:
            print("錄音已在進行中。")
            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("麥克風錄音已開始...")
        except Exception as e:
            print(f"啟動麥克風失敗: {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("麥克風錄音已停止。")
        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"關閉音頻流時出錯: {e}")

        if self.pyaudio_instance:
            try:
                self.pyaudio_instance.terminate()
            except Exception as e:
                print(f"終止 PyAudio 執行個體時出錯: {e}")

        self.stream = None
        self.pyaudio_instance = None

async def interactive_test():
    """
    互動式測試指令碼：允許多輪連續對話，每輪可以發送音頻和圖片。
    """
    # ------------------- 1. 初始化和串連 (一次性) -------------------
    # 新加坡和北京地區的API Key不同。擷取API Key：https://www.alibabacloud.com/help/zh/model-studio/get-api-key
    api_key = os.environ.get("DASHSCOPE_API_KEY")
    if not api_key:
        print("請設定DASHSCOPE_API_KEY環境變數")
        return

    print("--- 即時多輪音視訊交談用戶端 ---")
    print("正在初始化音頻播放器和用戶端...")

    audio_player = AudioPlayer()
    audio_player.start()

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

    transcription_done = threading.Event()

    def on_transcription_completed(transcript):
        transcription_done.set()

    def on_response_done(event):
        print("\n(收到響應結束標記)")
        audio_player.stop_receiving_data()

    realtime_client = OmniRealtimeClient(
        # 以下為新加坡地區URL，調用時請將{WorkspaceId}替換為真實的業務空間ID，各地區的URL不同。
        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"助手回複: {text}", end="", flush=True),
        on_audio_delta=on_audio_received,
        on_input_transcript=on_transcription_completed,
        turn_detection_mode=TurnDetectionMode.MANUAL,
        extra_event_handlers={"response.done": on_response_done}
    )

    message_handler_task = None
    try:
        await realtime_client.connect()
        print("已串連到伺服器。輸入 'q' 或 'quit' 可隨時退出程式。")
        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_counter} 輪對話 ---")
            audio_player.prepare_for_next_turn()

            recorded_audio = None
            image_paths = []

            # --- 擷取使用者輸入：從麥克風錄音 ---
            loop = asyncio.get_event_loop()
            recorder = MicrophoneRecorder(sample_rate=16000)  # 推薦使用16k採樣率進行語音辨識

            print("準備錄音。按 Enter 鍵開始錄音 (或輸入 'q' 退出)...")
            user_input = await loop.run_in_executor(None, input)
            if user_input.strip().lower() in ['q', 'quit']:
                print("使用者請求退出...")
                return

            try:
                recorder.start()
            except Exception:
                print("無法啟動錄音，請檢查您的麥克風許可權和裝置。跳過本輪。")
                continue

            print("錄音中... 再次按 Enter 鍵停止錄音。")
            await loop.run_in_executor(None, input)

            recorded_audio = recorder.stop()

            if not recorded_audio or len(recorded_audio) == 0:
                print("未錄製到有效音頻，請重新開始本輪對話。")
                continue

            # --- 擷取圖片輸入 (可選) ---
            # 以下圖片輸入功能已被注釋，暫時禁用。若需啟用請取消下方代碼注釋。
            # print("\n請逐行輸入【圖片檔案】的絕對路徑 (可選)。完成後，輸入 's' 或按 Enter 發送請求。")
            # while True:
            #     path = input("圖片路徑: ").strip()
            #     if path.lower() == 's' or path == '':
            #         break
            #     if path.lower() in ['q', 'quit']:
            #         print("使用者請求退出...")
            #         return
            #
            #     if not os.path.isabs(path):
            #         print("錯誤: 請輸入絕對路徑。")
            #         continue
            #     if not os.path.exists(path):
            #         print(f"錯誤: 檔案不存在 -> {path}")
            #         continue
            #     image_paths.append(path)
            #     print(f"已添加圖片: {os.path.basename(path)}")

            # --- 3. 發送資料並擷取響應 ---
            print("\n--- 輸入確認 ---")
            print(f"待處理音頻: 1個 (來自麥克風), 圖片: {len(image_paths)}個")
            print("------------------")

            # 3.1 發送錄製的音頻
            try:
                print(f"發送麥克風錄音 ({len(recorded_audio)}位元組)")
                await realtime_client.stream_audio(recorded_audio)
                await asyncio.sleep(0.1)
            except Exception as e:
                print(f"發送麥克風錄音失敗: {e}")
                continue

            # 3.2 發送所有圖片檔案
            # 以下圖片發送代碼已被注釋，暫時禁用。
            # for i, path in enumerate(image_paths):
            #     try:
            #         with open(path, "rb") as f:
            #             data = f.read()
            #         print(f"發送圖片 {i+1}: {os.path.basename(path)} ({len(data)}位元組)")
            #         await realtime_client.append_image(data)
            #         await asyncio.sleep(0.1)
            #     except Exception as e:
            #         print(f"發送圖片 {os.path.basename(path)} 失敗: {e}")

            # 3.3 提交並等待響應
            print("錄音結束，等待模型回複...")
            await realtime_client.commit_audio_buffer()
            # 等待轉錄完成後再觸發模型回複，避免輸出交錯
            await asyncio.to_thread(transcription_done.wait, 10)
            transcription_done.clear()
            await realtime_client.create_response()

            print("等待並播放伺服器響應音頻...")
            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"\n等待逾時 ({max_wait_time}秒), 進入下一輪。")
                    break
                await asyncio.sleep(0.2)

            print("\n本輪音頻播放完成！")
            turn_counter += 1

    except (asyncio.CancelledError, KeyboardInterrupt):
        print("\n程式被中斷。")
    except Exception as e:
        print(f"發生未處理的錯誤: {e}")
    finally:
        # ------------------- 4. 清理資源 -------------------
        print("\n正在關閉串連並清理資源...")
        if message_handler_task and not message_handler_task.done():
            message_handler_task.cancel()

        if 'realtime_client' in locals() and realtime_client.ws:
            await realtime_client.close()
            print("串連已關閉。")

        audio_player.stop()
        print("程式退出。")

if __name__ == "__main__":
    try:
        asyncio.run(interactive_test())
    except KeyboardInterrupt:
        print("\n程式被使用者強制退出。")