【しゃべるAI】おまけーAIにデバイスを操作させる

"
Raspberry Pi向け TTS-APIサーバ接続と GPIO/サーボモーター制御
"""

import os
import re
import io
import time
import tempfile
import requests
import platform
import subprocess
import threading
from pydub import AudioSegment
from pydub.playback import play
import json

# Raspberry Pi環境でのみ必要なモジュールをインポート
try:
    import RPi.GPIO as GPIO
    import pigpio
    IS_RASPBERRY_PI = True
    print("Raspberry Piモードで実行中")
except ImportError:
    IS_RASPBERRY_PI = False
    print("通常モードで実行中（GPIO機能は無効）")

# ユーザー設定 - ここで初期設定を変更できます
TTS_API_URL = "http://192.168.xxx.xxx:5000/voice"  # TTSサーバーのアドレス
MODEL_NAME = "【モデル】"                              # 使用するモデル
MAX_CHUNK_LENGTH = 100                             # テキスト分割の最大長
AUDIO_PARAMS = {
    "length": 0.86,                                # 音声の長さ調整
    "split_interval": 0,                           # スプリット間隔
    "style_weight": 1                              # スタイルの重み
}

# GPIO設定
LED_PIN = 17                        # LEDのGPIOピン番号
SERVO_PIN = 18                      # サーボモーターのGPIOピン番号

# AIレスポンスのキーワードとアクション設定
KEYWORD_ACTIONS = {
    "喜び": {"led": True, "servo": 90},      # 喜びの表現時、LEDを点灯しサーボを90度に
    "悲しみ": {"led": False, "servo": 0},    # 悲しみの表現時、LEDを消灯しサーボを0度に
    "驚き": {"led": "blink", "servo": 180},  # 驚きの表現時、LEDを点滅しサーボを180度に
    "質問": {"led": "pulse", "servo": 45},   # 質問の表現時、LEDをパルスさせサーボを45度に
}

try:
    import simpleaudio as sa
    SIMPLEAUDIO_AVAILABLE = True
except ImportError:
    SIMPLEAUDIO_AVAILABLE = False
    print("注意: simpleaudioがインストールされていません。代替の再生方法を使用します。")

# 基本設定
temp_dir = tempfile.gettempdir()
output_file = os.path.join(temp_dir, "tts_output.wav")

# GPIO初期化（Raspberry Piで実行されている場合のみ）
if IS_RASPBERRY_PI:
    # GPIO設定
    GPIO.setmode(GPIO.BCM)
    GPIO.setup(LED_PIN, GPIO.OUT)
    GPIO.output(LED_PIN, GPIO.LOW)
    
    # PWM設定（LED用）
    led_pwm = GPIO.PWM(LED_PIN, 100)  # 100Hz
    led_pwm.start(0)
    
    # pigpioサーボ用初期化
    pi = pigpio.pi()
    if not pi.connected:
        print("警告: pigpioデーモンに接続できません。サーボ制御は無効です。")
        print("pigpioを開始するには: 'sudo pigpiod'を実行してください。")
    else:
        # サーボ用のGPIOをセットアップ（パルス幅: 500-2500μs = 0-180度）
        pi.set_servo_pulsewidth(SERVO_PIN, 1500)  # 中央位置（90度）

def cleanup_gpio():
    """GPIO関連のクリーンアップ処理"""
    if IS_RASPBERRY_PI:
        try:
            led_pwm.stop()
            GPIO.cleanup()
            pi.stop()
        except:
            pass

def control_led(state, duration=None):
    """LEDを制御する関数
    
    Args:
        state: Trueで点灯、Falseで消灯、"blink"で点滅、"pulse"でパルス
        duration: 効果の持続時間（秒）、Noneの場合は永続的
    """
    if not IS_RASPBERRY_PI:
        print(f"LED制御シミュレーション: {state}")
        return
        
    if state == True:
        GPIO.output(LED_PIN, GPIO.HIGH)
        print("LEDをONにしました")
    elif state == False:
        GPIO.output(LED_PIN, GPIO.LOW)
        print("LEDをOFFにしました")
    elif state == "blink":
        def blink_thread():
            end_time = time.time() + (duration if duration else 3)
            while time.time() < end_time:
                GPIO.output(LED_PIN, GPIO.HIGH)
                time.sleep(0.2)
                GPIO.output(LED_PIN, GPIO.LOW)
                time.sleep(0.2)
            GPIO.output(LED_PIN, GPIO.LOW)
            
        print("LEDを点滅させています")
        t = threading.Thread(target=blink_thread)
        t.daemon = True
        t.start()
    elif state == "pulse":
        def pulse_thread():
            end_time = time.time() + (duration if duration else 3)
            while time.time() < end_time:
                for dc in range(0, 101, 5):
                    led_pwm.ChangeDutyCycle(dc)
                    time.sleep(0.05)
                for dc in range(100, -1, -5):
                    led_pwm.ChangeDutyCycle(dc)
                    time.sleep(0.05)
            led_pwm.ChangeDutyCycle(0)
            
        print("LEDをパルスさせています")
        t = threading.Thread(target=pulse_thread)
        t.daemon = True
        t.start()

def control_servo(angle):
    """サーボモーターを制御する関数
    
    Args:
        angle: 0-180度の角度
    """
    if not IS_RASPBERRY_PI:
        print(f"サーボモーターシミュレーション: {angle}度")
        return
        
    if not pi.connected:
        print(f"サーボ制御スキップ: pigpioが接続されていません")
        return
        
    # 角度を範囲内に制限
    angle = max(0, min(180, angle))
    
    # 角度をパルス幅に変換（500μs=0度、2500μs=180度）
    pulsewidth = 500 + (angle * 2000 / 180)
    
    pi.set_servo_pulsewidth(SERVO_PIN, pulsewidth)
    print(f"サーボを{angle}度に設定しました（パルス幅: {pulsewidth}μs）")

def analyze_text_for_actions(text):
    """テキストを分析してアクションを決定する関数
    
    Returns:
        dict: LEDとサーボモーターのアクション
    """
    text = text.lower()
    action = {"led": False, "servo": None}  # デフォルトアクション
    
    # テキスト内のキーワードを検索
    for keyword, actions in KEYWORD_ACTIONS.items():
        if keyword in text:
            action = actions
            print(f"キーワード「{keyword}」を検出: {json.dumps(actions, ensure_ascii=False)}")
            break
            
    # テキストに「?」「？」があれば質問として扱う
    if "?" in text or "？" in text:
        action = KEYWORD_ACTIONS.get("質問", action)
        print("質問を検出しました")
        
    return action

def split_text_custom(text, max_length=MAX_CHUNK_LENGTH):
    """テキストを文末や絵文字に基づいてセグメントに分割"""
    pattern = r'(?<=[。！？\U0001F300-\U0001F64F\U0001F680-\U0001F6FF\U00002728])'
    sentences = re.split(pattern, text)
    segments = []
    for sentence in sentences:
        sentence = sentence.strip()
        if not sentence:
            continue
        if len(sentence) > max_length:
            for i in range(0, len(sentence), max_length):
                segments.append(sentence[i:i+max_length])
        else:
            segments.append(sentence)
    return segments

def generate_tts_audio(text, tts_api_url=TTS_API_URL, model_name=MODEL_NAME):
    """テキストからTTS音声を生成するためにTTS APIを呼び出す"""
    if not text.strip():
        print("テキストが空です。")
        return None
        
    print(f"テキスト「{text}」を音声に変換中...")
    
    # テキストを分析してアクションを決定
    action = analyze_text_for_actions(text)
    
    # LEDとサーボを制御
    if action["led"] is not None:
        control_led(action["led"])
    if action["servo"] is not None:
        control_servo(action["servo"])
    
    segments = split_text_custom(text)
    combined_audio = None
    
    for i, chunk in enumerate(segments, 1):
        if len(chunk.strip()) <= 1:
            continue
            
        print(f"  チャンク {i}/{len(segments)}: 「{chunk}」")
        params = {
            "text": chunk,
            "model_name": model_name,
            "length": AUDIO_PARAMS["length"],
            "split_interval": AUDIO_PARAMS["split_interval"],
            "style_weight": AUDIO_PARAMS["style_weight"]
        }
        
        try:
            response = requests.get(tts_api_url, params=params)
            if response.status_code == 200:
                audio_segment = AudioSegment.from_file(io.BytesIO(response.content), format="wav")
                if combined_audio is None:
                    combined_audio = audio_segment
                else:
                    combined_audio += audio_segment
                print(f"  チャンク {i} 処理完了")
            else:
                print(f"  TTS APIエラー（ステータス: {response.status_code}）")
        except Exception as e:
            print(f"  TTS生成エラー: {e}")
    
    if combined_audio:
        print("音声生成完了")
        out_buffer = io.BytesIO()
        combined_audio.export(out_buffer, format="wav")
        return out_buffer.getvalue()
    
    print("音声生成に失敗しました")
    return None

def play_audio(audio_data, output_file="tts_output.wav", autoplay=True, delete_after=True):
    """音声データをファイルに保存し、必要に応じて自動再生"""
    if not audio_data:
        print("再生する音声データがありません")
        return False
    
    # 再生中はLEDをパルスさせる（Raspberry Piのみ）
    if IS_RASPBERRY_PI:
        control_led("pulse")
    
    # ファイルに保存
    try:
        with open(output_file, "wb") as f:
            f.write(audio_data)
        print(f"音声ファイルを保存しました: {output_file}")
    except Exception as e:
        print(f"ファイル保存エラー: {e}")
        return False
    
    # 自動再生が有効な場合
    if autoplay:
        try:
            # simpleaudioを使って再生（利用可能な場合）
            if SIMPLEAUDIO_AVAILABLE:
                try:
                    print("simpleaudioで再生中...")
                    wave_obj = sa.WaveObject.from_wave_file(output_file)
                    play_obj = wave_obj.play()
                    play_obj.wait_done()
                    print("再生完了")
                    
                    if delete_after:
                        try:
                            os.remove(output_file)
                            print("一時ファイルを削除しました")
                        except Exception as e:
                            print(f"ファイル削除エラー: {e}")
                    
                    # 再生終了後はLEDを消灯
                    if IS_RASPBERRY_PI:
                        control_led(False)
                    
                    return True
                except Exception as e:
                    print(f"simpleaudioでの再生エラー: {e}")
                    print("代替方法で再生を試みます...")
            
            # OS別の再生方法
            system = platform.system()
            print(f"プラットフォーム: {system}での再生を試みます")
            
            if system == "Windows":
                subprocess.call(["start", os.path.abspath(output_file)], shell=True)
                time.sleep(5)  # 再生が終わるまで適当に待機
                print("Windowsでの再生を試みました")
            elif system == "Darwin":  # macOS
                subprocess.call(["afplay", output_file])
                print("macOSでの再生完了")
            elif system == "Linux":
                subprocess.call(["aplay", output_file])
                print("Linuxでの再生完了")
            
            if delete_after:
                try:
                    os.remove(output_file)
                    print("一時ファイルを削除しました")
                except Exception as e:
                    print(f"ファイル削除エラー: {e}")
            
            # 再生終了後はLEDを消灯
            if IS_RASPBERRY_PI:
                control_led(False)
            
            return True
        except Exception as e:
            print(f"自動再生中にエラーが発生しました: {e}")
            # エラー時もLEDを消灯
            if IS_RASPBERRY_PI:
                control_led(False)
            return False
    
    return True

def show_current_settings():
    """現在の設定を表示"""
    print("\n=== 現在の設定 ===")
    print(f"TTSサーバーURL: {TTS_API_URL}")
    print(f"モデル名: {MODEL_NAME}")
    print(f"テキスト分割最大長: {MAX_CHUNK_LENGTH}")
    print(f"音声パラメータ:")
    print(f"  - 長さ調整: {AUDIO_PARAMS['length']}")
    print(f"  - スプリット間隔: {AUDIO_PARAMS['split_interval']}")
    print(f"  - スタイルの重み: {AUDIO_PARAMS['style_weight']}")
    
    if IS_RASPBERRY_PI:
        print("\n=== GPIO設定 ===")
        print(f"LED PIN: {LED_PIN}")
        print(f"サーボモーター PIN: {SERVO_PIN}")
        print("\n=== キーワードアクション ===")
        for keyword, action in KEYWORD_ACTIONS.items():
            print(f"「{keyword}」: LED={action['led']}, サーボ={action['servo']}度")
    
    print("===============")

def test_gpio():
    """GPIO機能のテスト"""
    if not IS_RASPBERRY_PI:
        print("Raspberry Piではないため、GPIO機能をテストできません。")
        return
        
    print("\n=== GPIO機能テスト ===")
    print("1. LEDテスト")
    
    print("LEDを点灯します...")
    control_led(True)
    time.sleep(1)
    
    print("LEDを消灯します...")
    control_led(False)
    time.sleep(1)
    
    print("LEDを点滅させます...")
    control_led("blink", 3)
    time.sleep(3.5)
    
    print("LEDをパルスさせます...")
    control_led("pulse", 3)
    time.sleep(3.5)
    
    print("2. サーボモーターテスト")
    
    print("サーボを0度に動かします...")
    control_servo(0)
    time.sleep(1)
    
    print("サーボを90度に動かします...")
    control_servo(90)
    time.sleep(1)
    
    print("サーボを180度に動かします...")
    control_servo(180)
    time.sleep(1)
    
    print("サーボを90度（中央）に戻します...")
    control_servo(90)
    
    print("GPIOテスト完了")

def update_gpio_settings():
    """GPIO設定を更新する"""
    global LED_PIN, SERVO_PIN, KEYWORD_ACTIONS
    
    if not IS_RASPBERRY_PI:
        print("Raspberry Piではないため、GPIO設定を変更できません。")
        return
        
    print("\n=== GPIO設定の変更 ===")
    
    try:
        # ピン設定の変更
        print("\n--- ピン設定 ---")
        new_led_pin = int(input(f"LED PIN [{LED_PIN}]: ").strip() or LED_PIN)
        new_servo_pin = int(input(f"サーボモーター PIN [{SERVO_PIN}]: ").strip() or SERVO_PIN)
        
        # 設定が変更された場合は、GPIOをクリーンアップして再設定
        if new_led_pin != LED_PIN or new_servo_pin != SERVO_PIN:
            LED_PIN = new_led_pin
            SERVO_PIN = new_servo_pin
            
            # クリーンアップ
            cleanup_gpio()
            
            # 再設定
            GPIO.setmode(GPIO.BCM)
            GPIO.setup(LED_PIN, GPIO.OUT)
            GPIO.output(LED_PIN, GPIO.LOW)
            
            led_pwm = GPIO.PWM(LED_PIN, 100)
            led_pwm.start(0)
            
            # pigpioサーボ用再設定
            pi = pigpio.pi()
            if pi.connected:
                pi.set_servo_pulsewidth(SERVO_PIN, 1500)
                
            print("GPIO設定を更新しました。")
        
        # キーワードアクションの変更
        print("\n--- キーワードアクション設定 ---")
        print("現在のキーワードアクション:")
        for keyword, action in KEYWORD_ACTIONS.items():
            print(f"「{keyword}」: LED={action['led']}, サーボ={action['servo']}度")
        
        update_keyword = input("\nキーワードアクションを更新しますか？ (y/n): ").strip().lower()
        if update_keyword == 'y':
            keyword = input("更新するキーワード（新規の場合は新しいキーワード）: ").strip()
            if keyword:
                led_action = input("LEDアクション (True/False/blink/pulse): ").strip()
                led_action = eval(led_action) if led_action in ["True", "False"] else led_action
                
                try:
                    servo_angle = int(input("サーボ角度 (0-180): ").strip())
                    servo_angle = max(0, min(180, servo_angle))
                    
                    KEYWORD_ACTIONS[keyword] = {"led": led_action, "servo": servo_angle}
                    print(f"キーワード「{keyword}」のアクションを更新しました。")
                except ValueError:
                    print("サーボ角度には数値を入力してください。")
    
    except ValueError as e:
        print(f"入力エラー: 数値を入力してください: {e}")

def main():
    global TTS_API_URL, MODEL_NAME, MAX_CHUNK_LENGTH, AUDIO_PARAMS
    
    print("="*50)
    print("=== Raspberry Pi TTS-API & GPIO制御テスト ===")
    print("="*50)
    print("※ 設定変更コマンド:")
    print("  - 'exit': プログラムを終了")
    print("  - 'url': TTSサーバーURLを変更")
    print("  - 'model': モデルを変更")
    print("  - 'settings': 現在の設定を表示")
    print("  - 'params': 音声パラメータを変更")
    if IS_RASPBERRY_PI:
        print("  - 'gpio': GPIO設定を変更")
        print("  - 'test': GPIO機能をテスト")
    print("="*50)
    
    show_current_settings()
    
    try:
        # メインループ
        while True:
            user_input = input("\n話させるテキスト（もしくはコマンド）: ").strip()
            
            # 終了コマンド
            if user_input.lower() == "exit":
                print("プログラムを終了します。")
                break
                
            # URL変更コマンド
            elif user_input.lower() == "url":
                new_url = input("新しいTTS-API URLを入力してください: ").strip()
                if new_url:
                    TTS_API_URL = new_url
                    print(f"TTS-API URLを変更しました: {TTS_API_URL}")
                continue
                
            # モデル変更コマンド
            elif user_input.lower() == "model":
                new_model = input("新しいモデル名を入力してください: ").strip()
                if new_model:
                    MODEL_NAME = new_model
                    print(f"モデルを変更しました: {MODEL_NAME}")
                continue
                
            # 設定表示コマンド
            elif user_input.lower() == "settings":
                show_current_settings()
                continue
                
            # パラメータ変更コマンド
            elif user_input.lower() == "params":
                print("\n=== 音声パラメータの変更 ===")
                try:
                    new_length = float(input(f"長さ調整 [{AUDIO_PARAMS['length']}]: ").strip() or str(AUDIO_PARAMS['length']))
                    new_split = float(input(f"スプリット間隔 [{AUDIO_PARAMS['split_interval']}]: ").strip() or str(AUDIO_PARAMS['split_interval']))
                    new_weight = float(input(f"スタイルの重み [{AUDIO_PARAMS['style_weight']}]: ").strip() or str(AUDIO_PARAMS['style_weight']))
                    new_chunk = int(input(f"テキスト分割最大長 [{MAX_CHUNK_LENGTH}]: ").strip() or str(MAX_CHUNK_LENGTH))
                    
                    AUDIO_PARAMS["length"] = new_length
                    AUDIO_PARAMS["split_interval"] = new_split
                    AUDIO_PARAMS["style_weight"] = new_weight
                    MAX_CHUNK_LENGTH = new_chunk
                    
                    print("パラメータを更新しました。")
                    show_current_settings()
                except ValueError as e:
                    print(f"入力エラー: 数値を入力してください: {e}")
                continue
                
            # GPIO設定変更コマンド（Raspberry Piのみ）
            elif user_input.lower() == "gpio" and IS_RASPBERRY_PI:
                update_gpio_settings()
                continue
                
            # GPIO機能テストコマンド（Raspberry Piのみ）
            elif user_input.lower() == "test" and IS_RASPBERRY_PI:
                test_gpio()
                continue
            
            # TTS処理と再生
            audio_data = generate_tts_audio(
                text=user_input,
                tts_api_url=TTS_API_URL,
                model_name=MODEL_NAME
            )
            
            play_audio(
                audio_data=audio_data,
                output_file=output_file,
                autoplay=True,
                delete_after=True
            )
    
    finally:
        # プログラム終了時にGPIOをクリーンアップ
        if IS_RASPBERRY_PI:
            cleanup_gpio()
            print("GPIO設定をクリーンアップしました。")

if __name__ == "__main__":
    main()

sudo apt-get update
sudo apt-get install python3-rpi.gpio
sudo apt-get install pigpio python3-pigpio
sudo pip3 install pydub simpleaudio requests

sudo pigpiod

python3 tts_gpio_control.py