カテゴリー: raspbery pi

import gpiod
import time

CHIP = 'gpiochip4'  # Raspberry Pi 5では'gpiochip4'を使用
PIN = 18  

chip = gpiod.Chip(CHIP)
line = chip.get_line(PIN)
line.request(consumer="Servo", type=gpiod.LINE_REQ_DIR_OUT)

def set_servo_angle(angle):
    duty_cycle = (angle / 18) + 2.5
    pulse_width = duty_cycle / 100 * 20000  # 20ms周期

    line.set_value(1)
    time.sleep(pulse_width / 1000000)
    line.set_value(0)
    time.sleep((20000 - pulse_width) / 1000000)

try:
    while True:
        for angle in range(80,141, 5):
            for _ in range(10):  # 各角度でn回パルスを送信
                set_servo_angle(angle)
            time.sleep(0.1)
except KeyboardInterrupt:
    pass
finally:
    line.release()

設定（数字が大きい方を優先する、デフォルトは0らしい）
$ sudo nmcli connection modify "preconfigured" connection.autoconnect-priority 100

確認方法
$ nmcli connection show "preconfigured" | grep connection.autoconnect-priority

connection.autoconnect-priority:        100

$ nmcli connection show "他のssid” | grep connection.autoconnect-priority

connection.autoconnect-priority:        0

import vosk
import pyaudio
import json
import numpy as np
import sounddevice as sd
import queue
import threading
import time
import os
from dotenv import load_dotenv
import google.generativeai as genai
import subprocess

class VoskSpeechRecognizer:
    def __init__(self, model_path='./vrecog/vosk-model-ja-0.22'):
        # モデルの初期化
        vosk.SetLogLevel(-1)
        self.model = vosk.Model(model_path)
        self.recognizer = vosk.KaldiRecognizer(self.model, 16000)
        
        # キュー設定
        self.audio_queue = queue.Queue()
        self.stop_event = threading.Event()
        
        # マイク設定
        self.sample_rate = 16000
        self.channels = 1
        
        # スレッド準備
        self.recording_thread = threading.Thread(target=self._record_audio)
        self.recognition_thread = threading.Thread(target=self._recognize_audio)
        
    def _record_audio(self):
        """
        連続的な音声録音スレッド
        """
        with sd.InputStream(
            samplerate=self.sample_rate, 
            channels=self.channels,
            dtype='int16',
            callback=self._audio_callback
        ):
            while not self.stop_event.is_set():
                sd.sleep(100)
    
    def _audio_callback(self, indata, frames, time, status):
        """
        音声入力のコールバック関数
        """
        if status:
            print(status)
        self.audio_queue.put(indata.copy())
    
    def _recognize_audio(self):
        """
        連続的な音声認識スレッド
        """
        while not self.stop_event.is_set():
            try:
                audio_chunk = self.audio_queue.get(timeout=0.5)
                if self.recognizer.AcceptWaveform(audio_chunk.tobytes()):
                    result = json.loads(self.recognizer.Result())
                    text = result.get('text', '').strip()
                    if text:
                        print(f"認識結果: {text}")
                        response_text = query_gemini(text)  # Gemini APIに問い合わせる
                        jtalk(response_text)  # 結果を音声合成して再生する
            except queue.Empty:
                continue
    
    def start_recognition(self):
        """
        音声認識の開始
        """
        self.stop_event.clear()
        self.recording_thread.start()
        self.recognition_thread.start()
    
    def stop_recognition(self):
        """
        音声認識の停止
        """
        self.stop_event.set()
        self.recording_thread.join()
        self.recognition_thread.join()

def query_gemini(prompt):
    """
    Gemini APIに問い合わせて応答を取得する関数。
    """
    try:
        response = model.generate_content(prompt)
        print(f"Gemini応答: {response.text}")
        return response.text.strip()
    except Exception as e:
        print(f"Gemini APIエラー: {e}")
        return "エラーが発生しました。もう一度試してください。"

def jtalk(text):
    """
    Open JTalkでテキストを音声合成し再生する関数。
    """
    open_jtalk = ['/usr/bin/open_jtalk']
    mech = ['-x', '/var/lib/mecab/dic/open-jtalk/naist-jdic']
    htsvoice = ['-m', '/usr/share/hts-voice/nitech-jp-atr503-m001/nitech_jp_atr503_m001.htsvoice']
    speed = ['-r', '1.0']
    outwav = ['-ow', 'out.wav']
    cmd = open_jtalk + mech + htsvoice + speed + outwav
    
    try:
        proc = subprocess.Popen(cmd, stdin=subprocess.PIPE)
        proc.stdin.write(text.encode('utf-8'))
        proc.stdin.close()
        proc.wait()
        
        # 音声ファイルを再生する場合
        subprocess.call(['aplay', 'out.wav'])
    except Exception as e:
        print(f"音声合成エラー: {e}")

def main():
    # 環境変数からGoogle APIキーを読み込む
    load_dotenv()
    GOOGLE_API_KEY = os.getenv('GOOGLE_API_KEY')
    
    if not GOOGLE_API_KEY:
        print("Google APIキーが設定されていません。")
        return
    
    # Google Gemini APIの設定
    genai.configure(api_key=GOOGLE_API_KEY, transport="rest")
    
    global model  # グローバル変数としてモデルを定義（query_geminiで使用）
    model = genai.GenerativeModel("gemini-1.5-flash")
    
    recognizer = VoskSpeechRecognizer()
    
    try:
        print("音声認識を開始します。Ctrl+Cで終了できます。")
        recognizer.start_recognition()
        
        # 無限ループを防ぐ（Ctrl+Cで停止可能）
        while True:
            time.sleep(1)
    
    except KeyboardInterrupt:
        print("\n音声認識を終了します...")
    finally:
        recognizer.stop_recognition()

if __name__ == "__main__":
    main()