update interface for tts

Author: ks6088ts

template_langgraph/services/streamlits/pages/chat_with_tools_agent.py

@@ -1,7 +1,6 @@
 import os
 import tempfile
 from base64 import b64encode
-from datetime import datetime
 
 import streamlit as st
 from audio_recorder_streamlit import audio_recorder
@@ -14,7 +13,7 @@
     ChatWithToolsAgent,
 )
 from template_langgraph.speeches.stt import SttWrapper
-from template_langgraph.speeches.tts import synthesize_audio
+from template_langgraph.speeches.tts import TtsWrapper
 from template_langgraph.tools.common import get_default_tools
 
 
@@ -169,35 +168,25 @@ def load_stt_wrapper(model_size: str = "base"):
         with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_audio_file:
             temp_audio_file.write(audio_bytes)
             temp_audio_file_path = temp_audio_file.name
-            st.download_button(
-                label="🎧 録音データを保存",
-                data=audio_bytes,
-                file_name=f"recorded_{datetime.now().strftime('%Y%m%d_%H%M%S')}.wav",
-                mime="audio/wav",
-                use_container_width=True,
-            )
         try:
-            if input_output_mode == "音声":
-                with st.spinner("音声を認識中..."):
-                    stt_wrapper = load_stt_wrapper(selected_model)
-                    language_param = None if transcription_language == "auto" else transcription_language
-                    result = stt_wrapper.transcribe(str(temp_audio_file_path), language=language_param)
-                    transcribed_text = result.get("text", "").strip()
-                    prompt_text = transcribed_text
-
-                    if prompt_text:
-                        st.success(f"音声認識完了: {prompt_text}")
-                        prompt = prompt_text
-                    else:
-                        st.warning("音声が認識できませんでした")
+            with st.spinner("音声を認識中..."):
+                stt_wrapper = load_stt_wrapper(selected_model)
+                language_param = None if transcription_language == "auto" else transcription_language
+                transcribed_text = stt_wrapper.transcribe(str(temp_audio_file_path), language=language_param)
+                prompt_text = transcribed_text
+
+                if prompt_text:
+                    st.success(f"音声認識結果: {prompt_text}")
+                    prompt = prompt_text
+                else:
+                    st.warning("音声が認識できませんでした")
         except Exception as e:
             st.error(f"音声認識でエラーが発生しました: {e}")
             prompt_text = "音声入力でエラーが発生しました"
         finally:
             if os.path.exists(temp_audio_file_path):
                 os.unlink(temp_audio_file_path)
-
-else:
+elif input_output_mode == "テキスト":
     # 既存のテキスト入力モード
     if prompt := st.chat_input(
         accept_file="multiple",
@@ -210,6 +199,8 @@ def load_stt_wrapper(model_size: str = "base"):
         ],
     ):
         pass  # promptは既に設定済み
+else:
+    st.error("不明な入出力モードです")
 
 # 共通の入力処理ロジック
 if prompt:
@@ -290,7 +281,7 @@ def load_stt_wrapper(model_size: str = "base"):
         if input_output_mode == "音声":
             try:
                 with st.spinner("音声を生成中です..."):
-                    audio_bytes = synthesize_audio(
+                    audio_bytes = TtsWrapper().synthesize_audio(
                         text=response_content,
                         language=tts_language,
                         speed=tts_speed,

template_langgraph/speeches/stt.py

@@ -28,4 +28,4 @@ def transcribe(
             audio=audio_path,
             language=language,
         )
-        return result["text"]
+        return result.get("text", "").strip()

template_langgraph/speeches/tts.py

@@ -13,51 +13,59 @@
 )
 
 
-def synthesize_audio(
-    text: str,
-    language: str = "ja",
-    speed: float = 1.0,
-    pitch_shift: int = 0,
-    volume_db: float = 0.0,
-) -> bytes | None:
-    """Convert text to speech audio using gTTS and pydub adjustments."""
-
-    if not text.strip():
-        return None
-
-    try:
-        tts = gTTS(text=text, lang=language)
-        mp3_buffer = io.BytesIO()
-        tts.write_to_fp(mp3_buffer)
-        mp3_buffer.seek(0)
-
-        audio_segment = AudioSegment.from_file(mp3_buffer, format="mp3")
-        original_rate = audio_segment.frame_rate
-
-        if pitch_shift != 0:
-            semitone_ratio = 2.0 ** (pitch_shift / 12.0)
-            shifted = audio_segment._spawn(
-                audio_segment.raw_data,
-                overrides={"frame_rate": int(original_rate * semitone_ratio)},
-            )
-            audio_segment = shifted.set_frame_rate(original_rate)
-
-        if speed != 1.0:
-            if speed > 1.0:
-                audio_segment = speedup(audio_segment, playback_speed=float(speed))
-            else:
-                slowed_rate = max(int(original_rate * float(speed)), 1)
-                audio_segment = audio_segment._spawn(
+class TtsWrapper:
+    def __init__(self):
+        pass
+
+    def load_model(self):
+        pass
+
+    def synthesize_audio(
+        self,
+        text: str,
+        language: str = "ja",
+        speed: float = 1.0,
+        pitch_shift: int = 0,
+        volume_db: float = 0.0,
+    ) -> bytes | None:
+        """Convert text to speech audio using gTTS and pydub adjustments."""
+
+        if not text.strip():
+            return None
+
+        try:
+            tts = gTTS(text=text, lang=language)
+            mp3_buffer = io.BytesIO()
+            tts.write_to_fp(mp3_buffer)
+            mp3_buffer.seek(0)
+
+            audio_segment = AudioSegment.from_file(mp3_buffer, format="mp3")
+            original_rate = audio_segment.frame_rate
+
+            if pitch_shift != 0:
+                semitone_ratio = 2.0 ** (pitch_shift / 12.0)
+                shifted = audio_segment._spawn(
                     audio_segment.raw_data,
-                    overrides={"frame_rate": slowed_rate},
-                ).set_frame_rate(original_rate)
-
-        if volume_db != 0:
-            audio_segment += float(volume_db)
-
-        output_buffer = io.BytesIO()
-        audio_segment.export(output_buffer, format="mp3")
-        return output_buffer.getvalue()
-    except Exception as e:  # pragma: no cover
-        logger.error(f"Error in synthesize_audio: {e}")
-        return None
+                    overrides={"frame_rate": int(original_rate * semitone_ratio)},
+                )
+                audio_segment = shifted.set_frame_rate(original_rate)
+
+            if speed != 1.0:
+                if speed > 1.0:
+                    audio_segment = speedup(audio_segment, playback_speed=float(speed))
+                else:
+                    slowed_rate = max(int(original_rate * float(speed)), 1)
+                    audio_segment = audio_segment._spawn(
+                        audio_segment.raw_data,
+                        overrides={"frame_rate": slowed_rate},
+                    ).set_frame_rate(original_rate)
+
+            if volume_db != 0:
+                audio_segment += float(volume_db)
+
+            output_buffer = io.BytesIO()
+            audio_segment.export(output_buffer, format="mp3")
+            return output_buffer.getvalue()
+        except Exception as e:  # pragma: no cover
+            logger.error(f"Error in synthesize_audio: {e}")
+            return None