location.py

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()