ask.py

import time
import numpy as np
import requests
import os
from reachy_mini import ReachyMini
from reachy_mini.utils import create_head_pose

SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
RECORD_DIR = os.path.join(SCRIPT_DIR, "record")
os.makedirs(RECORD_DIR, exist_ok=True)

def get_server_ip():
    ip = os.getenv("server_ip")
    if ip:
        return ip
    try:
        env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), '.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('\'"')
                        if key == "server_ip":
                            return val
    except Exception:
        pass
    return "10.112.5.79"


# ----------------- 加入 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
# ---------------------------------------------------------------------

def transcribe_audio(file_path=None):
    if file_path is None:
        file_path = os.path.join(RECORD_DIR, "ask.mp3")
    server_ip = get_server_ip()
    url = f"http://{server_ip}:4002/v1/audio/transcriptions"
    try:
        with open(file_path, "rb") as f:
            files = {"file": f}
            response = requests.post(url, files=files)
            response.raise_for_status()
            result = response.json()
            return result.get("text", "")
    except Exception as e:
        print(f"Error during audio transcription: {e}")
        return None

def text_to_speech(text, output_file=None):
    if output_file is None:
        output_file = os.path.join(RECORD_DIR, "tmp.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 get_chat_completion(prompt):
    server_ip = get_server_ip()
    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": 128,
        "temperature": 0.7
    }
    try:
        response = requests.post(url, headers=headers, json=payload)
        response.raise_for_status()
        result = response.json()
        return result["choices"][0]["message"]["content"]
    except Exception as e:
        print(f"Error during chat completion: {e}")
        return None

def ask_gesture(mini, count=3):
    for _ in range(count):
        mini.goto_target(
            head=create_head_pose(y=0.0, roll=15, pitch=-20, mm=True),
            antennas=np.deg2rad([-30, -30]),
            body_yaw=np.deg2rad(0),
            duration=1.0,
            method="minjerk"
        )
        time.sleep(0.1)

        mini.goto_target(
            head=create_head_pose(y=0.0, roll=-15, pitch=-20, mm=True),
            antennas=np.deg2rad([30, 30]),
            body_yaw=np.deg2rad(0),
            duration=1.0,
            method="minjerk"
        )
        time.sleep(0.1)

def run_ask_workflow(mini):
    print("Connected successfully!")
    
    # Enable motors (Torque ON) so the robot can move
    print("Enabling motors (Torque ON)...")
    mini.enable_motors()
    time.sleep(0.5)
    
    print("Moving robot up (goto_target)...")
    mini.goto_target(
        head=create_head_pose(z=10, mm=True),    # Up 5mm
        antennas=np.deg2rad([0, 0]),           # Antennas out
        body_yaw=np.deg2rad(0),                 # Turn body 30 degrees
        duration=2.0,                            # Slow movement
        method="minjerk"                         # Smooth movement
    )
    time.sleep(0.5)
    

    # Transcribe audio file and print result
    transcribed_text = transcribe_audio(os.path.join(RECORD_DIR, "latest_voice.mp3"))
    print("=== Transcription Result ===")
    print(transcribed_text)
    print("============================")
    
    if transcribed_text:
        response_text = get_chat_completion(transcribed_text)
        print("=== LLM Response ===")
        print(response_text)
        print("====================")
        
        # Convert the resulting text to speech and play it
        if response_text:
            response_wav = os.path.join(RECORD_DIR, "ask_response.wav")
            if text_to_speech(response_text, response_wav):
                print('Playing...')
                mini.media.play_sound(response_wav)
                ask_gesture(mini, count=3)
                mini.media.stop_playing()
        else:
            print("No response text returned from reasoning model, skipping text-to-speech.")

    print("Moving robot down (goto_target)...")
    mini.goto_target(
        head=create_head_pose(z=-15, mm=True),    # Back down to 0mm
        antennas=np.deg2rad([0, 0]),             # Antennas back to 0
        body_yaw=np.deg2rad(0),                  # Turn body back to 0
        duration=2.0,                            # Slow movement
        method="minjerk"                         # Smooth movement
    )
    print("Down movement completed!")
    time.sleep(0.5)
    
    # Disable motors (Torque OFF) to protect motors and make it compliant
    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

    try:
        if shared_mini is not None:
            print("Reusing existing ReachyMini connection from RobotManager.")
            apply_safe_goto(shared_mini)
            run_ask_workflow(shared_mini)
        else:
            # Standard standalone execution
            with ReachyMini(connection_mode="network", host="localhost") as mini:
                apply_safe_goto(mini)
                run_ask_workflow(mini)
    finally:
        # 刪除暫存的 latest_voice.mp3 檔案
        voice_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), "record", "latest_voice.mp3")
        if os.path.exists(voice_file):
            try:
                os.remove(voice_file)
                print(f"🗑️ 已刪除暫存的語音檔: {voice_file}")
            except Exception as e:
                print(f"⚠️ 無法刪除暫存語音檔: {e}")

if __name__ == "__main__":
    main()