Gemini Multimodal Live APIを使ってみる

"""利用可能な Gemini モデルを一覧し、bidiGenerateContent 対応を確認する"""

import google.auth
import google.auth.transport.requests
import requests as http_requests
from google.oauth2 import service_account

from src.core.config import settings

SCOPES = ["https://www.googleapis.com/auth/generative-language"]


def main():
    if settings.GOOGLE_SERVICE_ACCOUNT_FILE:
        creds = service_account.Credentials.from_service_account_file(
            settings.GOOGLE_SERVICE_ACCOUNT_FILE, scopes=SCOPES
        )
    else:
        creds, _ = google.auth.default(scopes=SCOPES)

    creds.refresh(google.auth.transport.requests.Request())

    resp = http_requests.get(
        "https://generativelanguage.googleapis.com/v1beta/models",
        headers={"Authorization": f"Bearer {creds.token}"},
    )
    models = resp.json().get("models", [])

    print(f"Found {len(models)} models\n")
    for m in models:
        methods = m.get("supportedGenerationMethods", [])
        if "bidiGenerateContent" in methods:
            print(f"  ✓ {m['name']}  ({', '.join(methods)})")


if __name__ == "__main__":
    main()

"""
Voice WebSocket テスト用スクリプト

使い方:
  Firebase トークン取得:
     const { getAuth } = await import('firebase/auth')
     const token = await getAuth().currentUser.getIdToken()
     console.log(token)

  実行:
    poetry run python scripts/test_voice_ws.py <token>
"""

import asyncio
import json
import sys

import websockets


async def test(token: str):
    uri = f"ws://localhost:8000/api/v1/voice/ws?token={token}"
    print(f"Connecting to {uri[:80]}...")

    async with websockets.connect(uri) as ws:
        print("Connected!")

        # テキストメッセージを送信
        msg = {"type": "text", "data": "こんにちは。自己紹介してください。"}
        await ws.send(json.dumps(msg))
        print(f"Sent: {msg}")

        # 応答を受信（複数メッセージの可能性あり）
        print("Waiting for response...")
        audio_chunks = 0
        try:
            for i in range(30):
                response = await asyncio.wait_for(ws.recv(), timeout=10)
                data = json.loads(response)

                # 音声データが含まれているか確認
                parts = (
                    data.get("serverContent", {})
                    .get("modelTurn", {})
                    .get("parts", [])
                )
                for part in parts:
                    if "inlineData" in part:
                        audio_chunks += 1
                        mime = part["inlineData"].get("mimeType", "")
                        size = len(part["inlineData"].get("data", ""))
                        print(f"  Audio chunk [{audio_chunks}]: {mime}, {size} bytes (base64)")
                    elif "text" in part:
                        print(f"  Text: {part['text']}")

                # ターン完了チェック
                if data.get("serverContent", {}).get("turnComplete"):
                    print(f"\nTurn complete! ({audio_chunks} audio chunks received)")
                    break
        except asyncio.TimeoutError:
            print("\nNo more responses (timeout)")
        except websockets.ConnectionClosed as e:
            print(f"\nConnection closed: {e}")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: poetry run python scripts/test_voice_ws.py <firebase_token>")
        sys.exit(1)

    asyncio.run(test(sys.argv[1]))