abstract waveform_utils; save wave_form on_generate audiocast

app/src/utils/render_audiocast_utils.py

@@ -3,7 +3,7 @@
 from typing import TypedDict
 
 import streamlit as st
-from src.utils.waveform_utils import render_waveform
+from app.src.utils.render_waveform import render_waveform
 
 from env_var import APP_URL
 

app/src/utils/waveform_utils.py → app/src/utils/render_waveform.py

@@ -1,43 +1,16 @@
 import os
-import tempfile
-from pathlib import Path
 
 import streamlit as st
 from pydub import AudioSegment
-from seewav import visualize
 
 from services.storage import BLOB_BASE_URI, StorageManager
-
-
-def save_waveform_video_to_gcs(session_id: str, video_path: str):
-    """Ingest waveform visualization to GCS."""
-    full_path = StorageManager().upload_video_to_gcs(video_path, f"{session_id}.mp4")
-    return full_path
-
-
-def generate_waveform_video(output_path: Path, audio_path: str) -> Path:
-    """Generate waveform video from audio file using SeeWav."""
-    with tempfile.TemporaryDirectory() as temp_dir:
-        visualize(
-            audio=Path(audio_path),
-            tmp=Path(temp_dir),
-            out=output_path,
-            bars=60,
-            speed=4,
-            time=0.4,
-            # rate=60,
-            size=(120, 68),
-            fg_color=(0.0, 1.0, 0.6),  # Bright green. Try 0.2 0.2 0.2 for dark green
-            bg_color=(0.05, 0.05, 0.05),  # Near black
-        )
-        return output_path
+from utils_pkg.waveform_utils import WaveformUtils
 
 
 def render_waveform(session_id: str, audio_path: str):
     """Render waveform visualization from audio file."""
-    tmp_directory = Path("/tmp/audiora/waveforms")
-    tmp_directory.mkdir(parents=True, exist_ok=True)
-    tmp_vid_path = tmp_directory / f"{session_id}.mp4"
+    waveform_utils = WaveformUtils(session_id, audio_path)
+    tmp_vid_path = waveform_utils.get_tmp_video_path()
 
     video_path = None
     if os.path.exists(tmp_vid_path):
@@ -56,8 +29,8 @@ def render_waveform(session_id: str, audio_path: str):
     try:
         if not video_path:
             with st.spinner("Generating waveform visualization..."):
-                video_path = generate_waveform_video(tmp_vid_path, audio_path)
-                save_waveform_video_to_gcs(session_id, str(video_path))
+                video_path = waveform_utils.generate_waveform_video(tmp_vid_path)
+                waveform_utils.save_waveform_video_to_gcs(str(video_path))
 
         # st.video(str(video_path), autoplay=True)
         with open(video_path, "rb") as video_file:

server/src/utils/generate_audiocast.py

@@ -9,6 +9,7 @@
 from utils_pkg.audiocast_request import AudioScriptMaker, generate_source_content
 from utils_pkg.chat_utils import ContentCategory
 from utils_pkg.session_manager import SessionManager
+from utils_pkg.waveform_utils import WaveformUtils
 
 
 class GenerateAudioCastRequest(BaseModel):
@@ -33,12 +34,13 @@ async def generate_audiocast(
     1. Generate source content
     2. Generate audio script
     3. Generate audio
-    4. Store audio
+    4a. Store audio
+    4b. TODO: Store the audio waveform on GCS
     5. Update session
     """
     summary = request.summary
     category = request.category
-    sessionId = request.sessionId
+    session_id = request.sessionId
 
     source_content = generate_source_content(category, summary)
     if not source_content:
@@ -51,27 +53,31 @@ async def generate_audiocast(
         raise HTTPException(status_code=500, detail="Failed to generate audio script")
 
     # Generate audio
-    output_file = await AudioManager(
+    audio_path = await AudioManager(
         custom_config=AudioManagerConfig(tts_provider="elevenlabs")
     ).generate_speech(audio_script)
 
     def _run_on_background():
         # Store audio
         try:
             storage_manager = StorageManager()
-            storage_manager.upload_audio_to_gcs(output_file, sessionId)
-        except Exception as e:
-            print(f"Storage warning: {str(e)}")
+            storage_manager.upload_audio_to_gcs(audio_path, session_id)
+
+            # Update session
+            db = SessionManager(session_id)
+            db._update_source(source_content)
+            db._update_transcript(audio_script)
 
-        # Update session
-        db = SessionManager(sessionId)
-        db._update_source(source_content)
-        db._update_transcript(audio_script)
+            # Generate and save audio waveform as mp4
+            waveform_utils = WaveformUtils(session_id, audio_path)
+            waveform_utils.run_all()
+        except Exception as e:
+            print(f"Error in generate_audiocast background_tasks: {str(e)}")
 
     background_tasks.add_task(_run_on_background)
 
     return GenerateAudioCastResponse(
-        url=output_file,
+        url=audio_path,
         script=audio_script,
         source_content=source_content,
         created_at=datetime.now().strftime("%Y-%m-%d %H:%M"),

utils_pkg/waveform_utils.py

@@ -0,0 +1,59 @@
+import tempfile
+from pathlib import Path
+
+from seewav import visualize
+
+from services.storage import StorageManager
+
+
+class WaveformUtils:
+    def __init__(self, session_id: str, audio_path: str):
+        self.session_id = session_id
+        self.audio_path = audio_path
+
+    def run_all(self):
+        """
+        1. Generate a waveform video from the audio file
+        2. Upload it to Google Cloud Storage.
+        """
+        tmp_path = self.get_tmp_video_path()
+        self.generate_waveform_video(tmp_path)
+        self.save_waveform_video_to_gcs(str(tmp_path))
+
+    def get_tmp_video_path(self):
+        """
+        Get temporary video path for waveform visualization.
+        """
+        tmp_directory = Path("/tmp/audiora/waveforms")
+        tmp_directory.mkdir(parents=True, exist_ok=True)
+        tmp_vid_path = tmp_directory / f"{self.session_id}.mp4"
+
+        return tmp_vid_path
+
+    def save_waveform_video_to_gcs(self, video_path: str):
+        """Ingest waveform visualization to GCS."""
+        full_path = StorageManager().upload_video_to_gcs(
+            video_path, f"{self.session_id}.mp4"
+        )
+        return full_path
+
+    def generate_waveform_video(self, output_path: Path) -> Path:
+        """Generate waveform video from audio file using SeeWav."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            visualize(
+                audio=Path(self.audio_path),
+                tmp=Path(temp_dir),
+                out=output_path,
+                bars=60,
+                speed=4,
+                time=0.4,
+                # rate=60,
+                size=(120, 68),
+                fg_color=(
+                    0.0,
+                    1.0,
+                    0.6,
+                ),  # Bright green. Try 0.2 0.2 0.2 for dark green
+                bg_color=(0.05, 0.05, 0.05),  # Near black
+            )
+            return output_path