# simple_audio_receiver.py import asyncio import json import logging import websockets from aiortc import RTCPeerConnection, RTCSessionDescription, RTCConfiguration, RTCIceServer from aiortc.sdp import candidate_from_sdp try: import pyaudio PYAUDIO_AVAILABLE = True except ImportError: PYAUDIO_AVAILABLE = False print("警告：pyaudio 未安裝，將無法播放遠端音訊。") print("請執行 'pip install pyaudio' 來安裝。") # --- 日誌設定 --- logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger("simple_audio_receiver") CONFIG = {} class AudioPlayerTrack: """一個用於播放接收到的音訊的軌道""" kind = "audio" def __init__(self): if not PYAUDIO_AVAILABLE: self.stream = None self.p = None return self.p = pyaudio.PyAudio() self.stream = self.p.open(format=pyaudio.paInt16, channels=1, rate=48000, output=True, frames_per_buffer=960) def play_frame_sync(self, frame): """ 同步播放音訊幀。此方法應在 executor 中執行以避免阻塞事件迴圈。 """ if not self.stream: return try: audio_data = frame.to_ndarray() self.stream.write(audio_data.tobytes()) except Exception as e: logger.warning(f"播放音訊時發生錯誤: {e}") def stop(self): if not PYAUDIO_AVAILABLE: return if hasattr(self, 'stream') and self.stream and self.stream.is_active(): self.stream.stop_stream() self.stream.close() if hasattr(self, 'p') and self.p: self.p.terminate() class WebRTCAudioReceiver: def __init__(self, sharer_id, password, server_url, ice_servers): self.sharer_id = sharer_id self.password = password self.server_url = server_url self.ice_servers = ice_servers self.websocket = None self.peer_connections = {} self.audio_player = None self.audio_tasks = {} self.pending_ice_candidates = {} # 新增：暫存 ICE candidates self.loop = asyncio.get_event_loop() async def connect(self): logger.info("正在嘗試連線到信令伺服器 %s", self.server_url) self.websocket = await websockets.connect(self.server_url) logger.info("成功連線到信令伺服器") register_payload = { "type": "register_sharer", "id": self.sharer_id, "password": self.password, } await self.websocket.send(json.dumps(register_payload)) logger.info("已註冊為純音訊接收端，ID: %s", self.sharer_id) async def wait_for_all_connections_closed(self): """等待所有 peer connections 都關閉""" if self.peer_connections: await asyncio.gather(*(pc.close() for pc in self.peer_connections.values())) async def run(self): await self.connect() try: async for message in self.websocket: data = json.loads(message) msg_type = data.get("type") controller_id = data.get("from_id") if msg_type == "request_to_connect" and controller_id: await self.create_peer_connection(controller_id) elif msg_type == "answer_to_sharer" and controller_id: pc = self.peer_connections.get(controller_id) if pc: answer = RTCSessionDescription(sdp=data["answer"]["sdp"], type=data["answer"]["type"]) logger.info("正在設定來自 %s 的 Answer", controller_id) await pc.setRemoteDescription(answer) logger.info("Answer 設定完成。") # 處理先前已收到的 ICE candidates if controller_id in self.pending_ice_candidates: logger.info("正在處理 %s 個暫存的 ICE candidates...", len(self.pending_ice_candidates[controller_id])) for ice in self.pending_ice_candidates[controller_id]: # --- 修正：從暫存區取出時，也要建立完整的 RTCIceCandidate 物件 --- candidate = candidate_from_sdp(ice["candidate"]) candidate.sdpMid = ice["sdpMid"] candidate.sdpMLineIndex = ice["sdpMLineIndex"] await pc.addIceCandidate(candidate) # 使用完整的 candidate 物件 del self.pending_ice_candidates[controller_id] elif msg_type == "ice_to_sharer" and controller_id: pc = self.peer_connections.get(controller_id) # 如果 remoteDescription 還沒設定，就先暫存起來 if pc and pc.remoteDescription is None: logger.info("遠端描述尚未設定，暫存 ICE candidate。") self.pending_ice_candidates.setdefault(controller_id, []).append(data["ice"]) elif pc and data.get("ice"): # --- 修正：建立完整的 RTCIceCandidate 物件 --- candidate = candidate_from_sdp(data["ice"]["candidate"]) candidate.sdpMid = data["ice"]["sdpMid"] candidate.sdpMLineIndex = data["ice"]["sdpMLineIndex"] await pc.addIceCandidate(candidate) except websockets.exceptions.ConnectionClosed as e: logger.warning("與信令伺服器的連線已關閉: %s", e) finally: logger.info("信令連線已關閉，正在等待所有 WebRTC 連線結束...") await self.wait_for_all_connections_closed() async def create_peer_connection(self, controller_id): logger.info("收到來自 %s 的連線請求，正在建立 WebRTC 連線...", controller_id) ice_servers_obj = [RTCIceServer(**server) for server in self.ice_servers] configuration = RTCConfiguration(iceServers=ice_servers_obj) pc = RTCPeerConnection(configuration=configuration) self.pending_ice_candidates[controller_id] = [] # 為新連線初始化暫存列表 self.peer_connections[controller_id] = pc @pc.on("connectionstatechange") async def on_connectionstatechange(): logger.info("WebRTC 連線狀態 (%s): %s", controller_id, pc.connectionState) if pc.connectionState in ("failed", "closed", "disconnected"): await self.cleanup_peer_connection(controller_id) @pc.on("track") def on_track(track): logger.info(f"收到軌道: {track.kind}") if track.kind == "audio": if not self.audio_player: self.audio_player = AudioPlayerTrack() async def play_audio_task(): try: while True: frame = await track.recv() # *** 關鍵：將阻塞的播放操作放到執行緒池中 *** await self.loop.run_in_executor(None, self.audio_player.play_frame_sync, frame) except asyncio.CancelledError: logger.info("音訊播放任務已取消。") except Exception as e: logger.error(f"音訊播放迴圈出錯: {e}", exc_info=True) self.audio_tasks[controller_id] = asyncio.create_task(play_audio_task()) # *** 關鍵：這裡不再 addTrack，因為我們不分享任何東西 *** # pc.addTrack(...) offer = await pc.createOffer() await pc.setLocalDescription(offer) offer_payload = { "type": "offer_to_controller", "from_id": self.sharer_id, "target_id": controller_id, "offer": {"sdp": pc.localDescription.sdp, "type": pc.localDescription.type}, } await self.websocket.send(json.dumps(offer_payload)) logger.info("已發送 Offer 給 %s (純接收模式)", controller_id) async def cleanup_peer_connection(self, controller_id): pc = self.peer_connections.pop(controller_id, None) if pc and pc.connectionState != "closed": await pc.close() logger.info("已關閉與 %s 的 WebRTC 連線", controller_id) task = self.audio_tasks.pop(controller_id, None) if task and not task.done(): task.cancel() self.pending_ice_candidates.pop(controller_id, None) # 清理暫存 if not self.peer_connections and self.audio_player: self.audio_player.stop() self.audio_player = None def load_config(): global CONFIG try: with open("config.json", 'r') as f: CONFIG = json.load(f) except (FileNotFoundError, json.JSONDecodeError) as e: logger.error(f"錯誤：無法載入 config.json。 {e}") exit(1) def build_connection_configs(): conn_config = CONFIG.get("connection", {}) is_secure = conn_config.get("secure", False) host = conn_config.get("host", "localhost") ws_protocol = "wss" if is_secure else "ws" signaling_port = conn_config.get("signaling_port", 6759) signaling_url = f"{ws_protocol}://{host}:{signaling_port}" turn_creds = CONFIG.get("turn_credentials", {}) turn_port = conn_config.get("turn_port_secure", 5349) if is_secure else conn_config.get("turn_port_insecure", 3478) turn_protocol = "turns" if is_secure else "turn" ice_servers = [ {"urls": f"stun:{host}:{turn_port}"}, { "urls": f"{turn_protocol}:{host}:{turn_port}?transport=tcp", "username": turn_creds.get("username"), "credential": turn_creds.get("credential") } ] return signaling_url, ice_servers async def main(): load_config() signaling_url, ice_servers = build_connection_configs() sharer_info = CONFIG.get("sharer_info", {}) receiver = WebRTCAudioReceiver( sharer_id=sharer_info.get("id", "default-sharer"), # 保持與設定檔中的 ID 一致 password=sharer_info.get("password", "password"), server_url=signaling_url, ice_servers=ice_servers ) await receiver.run() if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: logger.info("程式已手動中斷。")