import os import urllib.request import json import ssl import time import numpy as np import requests import subprocess import platform from reachy_mini import ReachyMini from reachy_mini.utils import create_head_pose from get_location import run_shortcuts_cli, get_mac_gps SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) RECORD_DIR = os.path.join(SCRIPT_DIR, "record") os.makedirs(RECORD_DIR, exist_ok=True) # ----------------- 加入 Monkey Patch 解決遠端連線問題 ----------------- from reachy_mini.media.media_manager import MediaManager if not hasattr(MediaManager, "_is_patched"): original_media_manager_init = MediaManager.__init__ def mocked_media_manager_init(self, *args, **kwargs): # 如果傳入的信令伺服器不是 localhost，強制改為 localhost 以便走 SSH Tunnel if kwargs.get("signalling_host") not in (None, "localhost", "127.0.0.1"): kwargs["signalling_host"] = "localhost" original_media_manager_init(self, *args, **kwargs) # 替換原有的初始化函數 MediaManager.__init__ = mocked_media_manager_init MediaManager._is_patched = True # --------------------------------------------------------------------- # 1. 讀取 .env 檔案中的金鑰 def load_env_variables(): # 優先嘗試使用 python-dotenv try: from dotenv import load_dotenv load_dotenv('.env') except ImportError: pass # 手動備用解析 .env，確保在任何環境下都能正常讀取 env_path = '.env' if os.path.exists(env_path): with open(env_path, 'r', encoding='utf-8') as f: for line in f: line = line.strip() if line and not line.startswith('#') and '=' in line: parts = line.split('=', 1) key = parts[0].strip() val = parts[1].strip().strip('\'"') os.environ[key] = val load_env_variables() def get_chat_completion(prompt: str) -> str: """使用 requests 呼叫 MiniCPM5-1B API 進行內容生成""" server_ip = os.getenv("server_ip", "10.112.5.79") url = f"http://{server_ip}:4001/v1/chat/completions" headers = { "Content-Type": "application/json" } payload = { "model": "openbmb/MiniCPM5-1B", "messages": [{"role": "user", "content": prompt}], "max_tokens": 256, "temperature": 0.7 } try: response = requests.post(url, headers=headers, json=payload, timeout=30) response.raise_for_status() result = response.json() return result["choices"][0]["message"]["content"] except Exception as e: return f"❌ 呼叫 MiniCPM 失敗：{e}" def text_to_speech(text, output_file=None): if output_file is None: output_file = os.path.join(RECORD_DIR, "test.wav") url = "http://localhost:4003/v1/audio/speech" headers = { "Content-Type": "application/json" } payload = { "text": text } print(f"Converting text to speech via {url}...") try: response = requests.post(url, headers=headers, json=payload, timeout=30) response.raise_for_status() with open(output_file, "wb") as f: f.write(response.content) print(f"Audio saved to {output_file}") return True except Exception as e: print(f"Error in TTS request: {e}") return False def wiggle_antennas(mini, count=3): for _ in range(count): mini.goto_target( antennas=np.deg2rad([10, 10]), duration=3.0, method="minjerk" ) mini.goto_target( antennas=np.deg2rad([-10, -10]), duration=3.0, method="minjerk" ) def run_location_workflow(mini): print("Connected successfully!") # 啟用馬達 (Torque ON) print("Enabling motors (Torque ON)...") mini.enable_motors() time.sleep(0.5) print("=" * 60) print("📍 正在獲取您的電腦 GPS 位置資訊...") print("=" * 60) # 取得位置 raw_location = run_shortcuts_cli() gps_coords = get_mac_gps() if not raw_location and not gps_coords: location_info = "（無法取得定位，請確認定位服務與 'GetMyLocation' 捷徑是否已設定）" else: location_info = f"原始定位輸出:\n{raw_location}\n\n經緯度資料:\n{gps_coords}" print("\n" + "-" * 60) user_question = "請問我的位置在哪裡" print(f"💬 使用者提問：'{user_question}'") print("-" * 60) # 構建給 LLM 的 prompt，把位置資訊塞進 prompt 中 prompt = f""" 使用者提問：'{user_question}'。 以下是系統為您偵測到的目前電腦位置與 GPS 資訊： {location_info} 請整合上述的位置資訊，用自然流暢且親切的中文回覆使用者，請用4句簡短回覆。 """ print("⏳ 正在將位置資訊傳送給 MiniCPM 進行整合分析...") reply = get_chat_completion(prompt) print("\n🤖 MiniCPM 回覆：") print(reply) print("=" * 60) # 將回覆轉換成語音並播放 if reply: location_wav = os.path.join(RECORD_DIR, "location.wav") if text_to_speech(reply, location_wav): print('Playing...') mini.media.play_sound(location_wav) # wiggle_antennas(mini, count=3) mini.goto_target( head=create_head_pose(y=10.0, mm=True), # Up 5mm antennas=np.deg2rad([0, 0]), # Antennas out body_yaw=np.deg2rad(0), # Turn body 30 degrees duration=3, # Slow movement method="minjerk" # Smooth movement ) time.sleep(0.1) mini.goto_target( head=create_head_pose(y=-5.0, mm=True), # Up 5mm antennas=np.deg2rad([0, 0]), # Antennas out body_yaw=np.deg2rad(0), # Turn body 30 degrees duration=3, # Slow movement method="minjerk" # Smooth movement ) time.sleep(0.1) mini.media.stop_playing() else: print("No text result returned from MiniCPM, skipping text-to-speech.") # 關閉馬達 (Torque OFF) print("Disabling motors (Torque OFF - making robot limp)...") mini.disable_motors() print("Done!") def apply_safe_goto(mini): original_goto = mini.goto_target if not hasattr(original_goto, "_is_safe"): def safe_goto(*args, **kwargs): try: return original_goto(*args, **kwargs) except Exception as e: if "timeout" in str(e).lower() or "TimeoutError" in type(e).__name__: print(f"⚠️ [網路延遲] 動作發送超時但可能已在執行中，繼續下一個動作... ({e})") else: raise e safe_goto._is_safe = True mini.goto_target = safe_goto def main(): print("Connecting to Reachy Mini...") shared_mini = None try: from app import robot_manager if robot_manager and robot_manager.mini is not None: shared_mini = robot_manager.mini except Exception: pass if shared_mini is not None: print("Reusing existing ReachyMini connection from RobotManager.") apply_safe_goto(shared_mini) run_location_workflow(shared_mini) else: # Standard standalone execution with ReachyMini(connection_mode="network", host="localhost") as mini: apply_safe_goto(mini) run_location_workflow(mini) if __name__ == "__main__": main()