AITuber開発日誌 2-1 声のテスト！

import aiohttp
import json
from typing import Optional, Dict, Any

class VoiceVoxClient:
    """
    VOICEVOXエンジンと通信するクライアントクラス
    """
    def __init__(self, base_url: str = "http://127.0.0.1:50021", speaker_id: int = 1):
        self.base_url = base_url.rstrip("/")
        self.speaker_id = speaker_id

    async def create_audio_query(self, text: str, speaker: Optional[int] = None) -> Optional[Dict[str, Any]]:
        """音声合成用のクエリを作成します (/audio_query)"""
        speaker_id = speaker if speaker is not None else self.speaker_id
        params = {"text": text, "speaker": speaker_id}
        
        async with aiohttp.ClientSession() as session:
            try:
                async with session.post(f"{self.base_url}/audio_query", params=params) as response:
                    response.raise_for_status()
                    return await response.json()
            except Exception as e:
                print(f"Error creating audio query: {e}")
                return None

    async def synthesis(self, query: Dict[str, Any], speaker: Optional[int] = None) -> Optional[bytes]:
        """音声を合成してバイナリデータを返します (/synthesis)"""
        speaker_id = speaker if speaker is not None else self.speaker_id
        params = {"speaker": speaker_id}
        
        async with aiohttp.ClientSession() as session:
            try:
                headers = {"Content-Type": "application/json"}
                async with session.post(
                    f"{self.base_url}/synthesis", 
                    params=params, 
                    json=query,
                    headers=headers
                ) as response:
                    response.raise_for_status()
                    return await response.read()
            except Exception as e:
                print(f"Error synthesizing audio: {e}")
                return None

    async def generate_voice(self, text: str) -> Optional[bytes]:
        """テキストから音声を生成する一連の流れを実行します"""
        query = await self.create_audio_query(text)
        if query is None:
            return None
        return await self.synthesis(query)

import asyncio
import os
from modules.expression.voicevox import VoiceVoxClient
from dotenv import load_dotenv

# .envファイルを読み込む
load_dotenv()

async def main():
    print("🤖 音声合成テストを開始します...")
    
    # 環境変数から設定を読み込む（なければデフォルト値）
    voicevox_url = os.getenv("VOICEVOX_URL", "http://127.0.0.1:50021")
    speaker_id = int(os.getenv("VOICEVOX_SPEAKER_ID", "1"))
    
    # クライアントの初期化
    client = VoiceVoxClient(base_url=voicevox_url, speaker_id=speaker_id)
    
    text = "こんにちは。聞こえていますか？これはテストです。"
    print(f"生成するテキスト: 「{text}」")
    
    # 音声生成
    audio_data = await client.generate_voice(text)
    
    if audio_data:
        # ファイルに保存
        filename = "test_output.wav"
        with open(filename, "wb") as f:
            f.write(audio_data)
        print(f"✅ 成功！ '{filename}' が作成されました。")
    else:
        print("❌ 失敗しました。VOICEVOXが起動しているか確認してください。")

if __name__ == "__main__":
    asyncio.run(main())