ラズパイ５でpythonからAquesTalk Piとgemini APIを使う事ができたので、今回はこれを組み合わせて今日は何の日かを教えてくれるプログラムを組んでみました。
今回はAquesTalk Piとgemini APIをそれぞれクラス化した別ファイルにして、インポートして使用しました。
ファイル階層は以下のようにします。
python_app ├─.venv（仮想環境） ├─gemini.py ├─aquestalk.py └─wakeup_app.py
まずはgemini.pyから
import os import json import google.generativeai as genai class GeminiClient: def __init__(self, model_name="gemini-2.5-flash", history_file="gemini_history.json"): self.model_name = model_name self.history_file = history_file # 🔹 APIキー確認 api_key = os.environ.get("GOOGLE_API_KEY") if not api_key: raise ValueError("環境変数 GOOGLE_API_KEY が設定されていません！") genai.configure(api_key=api_key) # 🔹 Gemini モデル self.model = genai.GenerativeModel(model_name) # 🔹 履歴読み込み self.history = self.load_history() # 🔹 チャット開始 self.chat = self.model.start_chat(history=self.history) # ========================== # 履歴の読み込み # ========================== def load_history(self): if not os.path.exists(self.history_file): return [] try: with open(self.history_file, "r", encoding="utf-8") as f: data = f.read().strip() if not data: return [] items = json.loads(data) # 最新100件だけ返す if isinstance(items, list): return items[-100:] return [] except (json.JSONDecodeError, OSError): print("⚠ 履歴ファイルが壊れていたのでリセットしました。") return [] # ========================== # 履歴の保存 # ========================== def save_history(self): serializable = [] # chat.history をリスト化して最新100件にトリム try: history_list = list(self.chat.history) except Exception: history_list = getattr(self.chat, "history", []) or [] history_list = history_list[-100:] # 可能ならチャットオブジェクト側の履歴も更新 try: self.chat.history = history_list except Exception: pass for item in history_list: role = getattr(item, "role", None) # parts は item.parts に text が入っている parts = [] for p in getattr(item, "parts", []): if hasattr(p, "text"): parts.append({"text": p.text}) else: parts.append({"text": str(p)}) serializable.append({ "role": role, "parts": parts }) with open(self.history_file, "w", encoding="utf-8") as f: json.dump(serializable, f, ensure_ascii=False, indent=2) # ========================== # メッセージ送信 # ========================== def fetch(self, prompt): try: res = self.chat.send_message(prompt) self.save_history() return res.text except Exception as e: print("Gemini エラー:", e) return "データ取得エラーです。" # 直接実行されたときだけ動く if __name__ == "__main__": gem = GeminiClient() print("終了するには 'exit' または Ctrl+C を押してください。") try: while True: question = input("\n質問を入力してください: ").strip() if not question: print("入力が空です。") continue if question.lower() in ("exit", "quit", "q"): print("終了します。") break response = gem.fetch(question) print(response) except KeyboardInterrupt: print("\nユーザーによって中断されました。終了します。")
これは前回記事と同じ内容で、geminiとやり取りするプログラムです。
slowtech.hateblo.jp
次に、aquestalk.pyです。
import subprocess import shlex class AquesTalkPi: def __init__(self, aquestalk_path="/home/ユーザー名/Documents/aquestalkpi/AquesTalkPi", device="plughw:2,0"): """ :param aquestalk_path: AquesTalkPi の実行ファイルのパス :param device: aplay の出力先デバイス """ self.aquestalk_path = aquestalk_path self.device = device def speak(self, text, speed=100, voice_type="f1"): """ AquesTalk Pi を使用してテキストを音声合成し再生します。 :param text: 読み上げたい日本語テキスト :param speed: 話速 (50〜300) :param voice_type: 声の種類 (f1〜f8, m1〜m3 など) """ # echo から pipe で AquesTalkPi に渡し、aplay へ流す command = ( f"echo {shlex.quote(text)} | " f"{shlex.quote(self.aquestalk_path)} -s {speed} -v {voice_type} -p -f - | " f"aplay -D {self.device}" ) try: subprocess.run(command, shell=True, check=True) except subprocess.CalledProcessError as e: print(f"コマンドの実行中にエラーが発生しました: {e}") except FileNotFoundError: print(f"'{self.aquestalk_path}' または 'aplay' が見つかりません。パスを確認してください。") # ---- 使用例 ---- #from aquestalk import AquesTalkPi if __name__ == "__main__": aq = AquesTalkPi() aq.speak("こんにちわ、AquesTalk Piからの音声合成です。", speed=100, voice_type="f2") aq.speak("CPUの温度は25度です。", speed=100)
これは過去記事のAquesTalk Piでしゃべらせるコードをクラス化しました。
slowtech.hateblo.jp
そして、メインプログラムとなるwakeup_app.pyです。
import datetime from aquestalk import AquesTalkPi from gemini import GeminiClient def main(): # ★ AquesTalk 初期化 aq = AquesTalkPi() # ★ 今日の日付を作成 today = datetime.date.today() week_jp = ["月曜日", "火曜日", "水曜日", "木曜日", "金曜日", "土曜日", "日曜日"] weekday_text = week_jp[today.weekday()] date_text = f"{today.year}年{today.month}月{today.day}日、{weekday_text}です。" # ★ Gemini 初期化と質問作成 gem = GeminiClient() prompt = ( "今日は何の日かを1つだけ、日本語で短く説明してください。\n" "・『今日は○○の日です』の後に、詳細な経緯などの説明をつけてください。\n" "・推論過程や思考プロセスは書かないでください。\n" "・結論のみを自然な話し言葉で答えてください。\n" "・同じ質問を繰り返された場合でも、できるだけ毎回異なる内容で答えてください。\n" f"今日は {date_text} です。" ) answer = gem.fetch(prompt) answer = answer.replace("\n", "").replace("\r", "") # ← 改行除去 # ★ あいさつ greeting = f"おはようございます。今日は{date_text}" aq.speak(greeting, speed=100, voice_type="f1") # ★ 音声出力 aq.speak(answer, speed=100, voice_type="f1") ending = "以上、今日の情報でした。良いいちにちをお過ごしください。" aq.speak(ending, speed=100, voice_type="f1") # ★ プログラム終了 print("Wakeup App 完了。終了します。") if __name__ == "__main__": main()
これがgeminiへ問い合わせるプロンプトを作り、返答をAquesTalkPiへ送って音声出力するプログラムです。
実行するとちゃんと喋りました。
ラズパイ5でgemini に今日は何の日か問い合わせて、AquesTalk Piで喋らせる事が出来ました pic.twitter.com/wL8zDyQdmA
— slowtech (@slowtech) 2025年12月7日
geminiからの返答が改行されていると、AquesTalkがその行までしかしゃべらなかったので、改行をすべて削除しています。
AquesTalk側の問題らしいのですが、まあこの辺は触れないようにしておきます。
今度はこのプログラムをラズパイ起動時に自動実行しようと思います。
