add lgic TRANSCRIPTION

app.py

@@ -29,87 +29,131 @@ EXAMPLE_FILES = ["data/bonjour.wav", "data/bonjour2.wav"]
 DEFAULT_FILE = EXAMPLE_FILES[0]
 
 
-@spaces.GPU
-def read_and_stream_audio(filepath_to_stream: str, session_id: str):
-    """Stream audio chunks for a specific session."""
-    stop_file = stop_file_path(session_id)
+# --------------------------------------------------------
+# STREAMING
+# --------------------------------------------------------
+def read_and_stream_audio(filepath_to_stream: str, session_id: str, chunk_seconds: float):
+    """Stream audio chunks and save .npz files only when transcription is active."""
+    stop_file = os.path.join(TMP_DIR, f"stream_stop_flag_{session_id}.txt")
+    transcribe_flag = os.path.join(TMP_DIR, f"transcribe_active_{session_id}.txt")
+
     logging.debug(f"[{session_id}] read_and_stream_audio() started with file: {filepath_to_stream}")
 
     if not filepath_to_stream or not os.path.exists(filepath_to_stream):
         logging.error(f"[{session_id}] Audio file not found: {filepath_to_stream}")
-        if os.path.exists(DEFAULT_FILE):
-            filepath_to_stream = DEFAULT_FILE
-            logging.warning(f"[{session_id}] Using default file: {DEFAULT_FILE}")
-        else:
-            logging.error(f"[{session_id}] Default file missing. Aborting.")
-            return
-
+        return
     clear_stop_flag(session_id)
     register_session(session_id, filepath_to_stream)
-    progress_path = os.path.join(TMP_DIR, f"progress_{session_id}.json")
+    progress_path = os.path.join(TMP_DIR, f"progress_{session_id}.json")    
+    chunk_dir = os.path.join(TMP_DIR, f"chunks_{session_id}")
+    os.makedirs(chunk_dir, exist_ok=True)
 
     try:
         segment = AudioSegment.from_file(filepath_to_stream)
-        chunk_ms = 1000
+        chunk_ms = int(chunk_seconds * 1000)
         total_chunks = len(segment) // chunk_ms + 1
-        logging.info(f"[{session_id}] Streaming {total_chunks} chunks...")
+
+        logging.info(f"[{session_id}] Streaming {total_chunks} chunks ({chunk_seconds:.2f}s each)...")
 
         for i, chunk in enumerate(segment[::chunk_ms], start=1):
             if os.path.exists(stop_file):
-                logging.info(f"[{session_id}] Stop flag detected at chunk {i}. Stopping.")
+                logging.info(f"[{session_id}] Stop flag detected at chunk {i}. Ending stream.")
                 clear_stop_flag(session_id)
                 break
 
             iter_start = time.perf_counter()
-            logging.debug(f"[{session_id}] Sending chunk {i}/{total_chunks}...")
 
-            # Compute elapsed time (hh:mm:ss)
-            elapsed_s = i * (chunk_ms / 1000)
+            elapsed_s = i * chunk_seconds
             hours, remainder = divmod(int(elapsed_s), 3600)
             minutes, seconds = divmod(remainder, 60)
             elapsed_str = f"{hours:02d}:{minutes:02d}:{seconds:02d}"
 
-            # Compute percentage
             percent = round((i / total_chunks) * 100, 2)
-
-            # Save progress info
-            progress_data = {
-                "value": percent,
-                "elapsed": elapsed_str,
-                "text": f"Streaming... {elapsed_str} ({percent}%)"
-            }
+            progress_data = {"value": percent, "elapsed": elapsed_str}
             with open(progress_path, "w") as f:
                 json.dump(progress_data, f)
+            chunk_array = np.array(chunk.get_array_of_samples(), dtype=np.int16)
+            rate = chunk.frame_rate
 
-            # Stream chunk
-            output_chunk = (
-                chunk.frame_rate,
-                np.array(chunk.get_array_of_samples()).reshape(1, -1),
-            )
-            yield output_chunk
+            # ✅ Save only if transcription is active
+            if os.path.exists(transcribe_flag):
+                npz_path = os.path.join(chunk_dir, f"chunk_{i:05d}.npz")
+                np.savez_compressed(npz_path, data=chunk_array, rate=rate)
+                logging.debug(f"[{session_id}] Saved chunk {i}/{total_chunks} (transcribe active)")
 
-            process_ms = (time.perf_counter() - iter_start) * 1000
-            time.sleep(max((chunk_ms / 1000.0) - (process_ms / 1000.0) - 0.1, 0.01))
+            # Stream audio to client
+            yield (rate, chunk_array.reshape(1, -1))
 
-        with open(progress_path, "w") as f:
-            json.dump({"value": 100.0, "elapsed": elapsed_str, "text": "Streaming completed ✅"}, f)
+            process_ms = (time.perf_counter() - iter_start) * 1000
+            time.sleep(max(chunk_seconds - (process_ms / 1000.0) - 0.1, 0.01))
 
-        logging.info(f"[{session_id}] Stream completed successfully.")
+        logging.info(f"[{session_id}] Streaming completed successfully.")
 
-    except asyncio.CancelledError:
-        logging.info(f"[{session_id}] Stream cancelled by user.")
-        raise
     except Exception as e:
         logging.error(f"[{session_id}] Stream error: {e}", exc_info=True)
-        raise
     finally:
         unregister_session(session_id)
         clear_stop_flag(session_id)
         if os.path.exists(progress_path):
             os.remove(progress_path)
-        logging.debug(f"[{session_id}] Stream closed.")
 
 
+# --------------------------------------------------------
+# TRANSCRIPTION
+# --------------------------------------------------------
+@spaces.GPU
+def transcribe(session_id: str):
+    """Continuously read and delete .npz chunks while transcription is active."""
+    active_flag = os.path.join(TMP_DIR, f"transcribe_active_{session_id}.txt")
+    chunk_dir = os.path.join(TMP_DIR, f"chunks_{session_id}")
+
+    if not os.path.exists(chunk_dir):
+        logging.warning(f"[{session_id}] No chunk directory found for transcription.")
+        return
+
+    logging.info(f"[{session_id}] Transcription loop started.")
+    try:
+        while os.path.exists(active_flag):
+            files = sorted(f for f in os.listdir(chunk_dir) if f.endswith(".npz"))
+            if not files:
+                time.sleep(0.25)
+                continue
+
+            for fname in files:
+                fpath = os.path.join(chunk_dir, fname)
+                try:
+                    npz = np.load(fpath)
+                    samples = npz["data"]
+                    rate = int(npz["rate"])
+
+                    text = f"Transcribed {fname}: {len(samples)} samples @ {rate}Hz"
+                    logging.debug(f"[{session_id}] {text}")
+
+                    os.remove(fpath)
+                    logging.debug(f"[{session_id}] Deleted processed chunk: {fname}")
+                except Exception as e:
+                    logging.error(f"[{session_id}] Error processing {fname}: {e}")
+                    continue
+
+            time.sleep(0.25)
+
+        logging.info(f"[{session_id}] Transcription loop ended (flag removed).")
+
+    except Exception as e:
+        logging.error(f"[{session_id}] Transcription error: {e}", exc_info=True)
+    finally:
+        try:
+            if os.path.exists(chunk_dir) and not os.listdir(chunk_dir):
+                os.rmdir(chunk_dir)
+                logging.debug(f"[{session_id}] Cleaned up empty chunk dir.")
+        except Exception as e:
+            logging.error(f"[{session_id}] Cleanup error: {e}")
+        logging.info(f"[{session_id}] Exiting transcription loop.")
+
+
+# --------------------------------------------------------
+# STOP STREAMING
+# --------------------------------------------------------
 def stop_streaming(session_id: str):
     create_stop_flag(session_id)
     logging.info(f"[{session_id}] Stop button clicked → stop flag created.")
@@ -127,18 +171,20 @@ def get_session_progress(session_id: str):
         value = data.get("value", 0.0)
         elapsed = data.get("elapsed", "00:00:00")
         return value, elapsed
-    except Exception as e:
-        logging.error(f"[{session_id}] Progress read error: {e}")
+    except Exception:
         return 0.0, "00:00:00"
 
 
+# --------------------------------------------------------
+# UI
+# --------------------------------------------------------
 with gr.Blocks(theme=gr.themes.Soft()) as demo:
     gr.Markdown(
         "## 🎧 WebRTC Audio Streamer (Multi-user)\n"
-        "Each user controls their own audio stream with elapsed time and percentage progress."
+        "Each user controls their own stream. Transcription runs only during streaming."
     )
 
-    session_id = gr.State(value=generate_session_id)
+    session_id = gr.State(value=generate_session_id())
     active_filepath = gr.State(value=DEFAULT_FILE)
 
     with gr.Row(equal_height=True):
@@ -151,6 +197,15 @@ with gr.Blocks(theme=gr.themes.Soft()) as demo:
                     value=DEFAULT_FILE,
                 )
 
+                chunk_slider = gr.Slider(
+                    label="Chunk Duration (seconds)",
+                    minimum=0.5,
+                    maximum=5.0,
+                    value=1.0,
+                    step=0.5,
+                    interactive=True,
+                )
+
                 progress_bar = gr.Slider(
                     label="Streaming Progress (%)",
                     minimum=0,
@@ -165,14 +220,11 @@ with gr.Blocks(theme=gr.themes.Soft()) as demo:
                     label="Elapsed Time (hh:mm:ss)",
                     interactive=False,
                     visible=False,
-                    show_label=False
                 )
 
                 with gr.Row():
-                    with gr.Column(scale=1, min_width=0):
-                        start_button = gr.Button("▶️ Start Streaming", variant="primary")
-                    with gr.Column(scale=1, min_width=0):
-                        stop_button = gr.Button("⏹️ Stop Streaming", variant="stop", interactive=False)
+                    start_button = gr.Button("▶️ Start Streaming", variant="primary")
+                    stop_button = gr.Button("⏹️ Stop Streaming", variant="stop", interactive=False)
 
         with gr.Column():
             webrtc_stream = WebRTC(
@@ -183,80 +235,121 @@ with gr.Blocks(theme=gr.themes.Soft()) as demo:
                 visible=True,
             )
 
-    def set_new_file(filepath):
-        return filepath if filepath else DEFAULT_FILE
-
-    main_audio.change(fn=set_new_file, inputs=[main_audio], outputs=[active_filepath])
-    main_audio.stop_recording(fn=set_new_file, inputs=[main_audio], outputs=[active_filepath])
+    # --- Transcription Controls ---
+    with gr.Row(equal_height=True):
+         with gr.Column():
+            start_transcribe = gr.Button("🎙️ Start Transcribe", interactive=False)
+            stop_transcribe = gr.Button("🛑 Stop Transcribe", interactive=False)
 
+    # --- UI Logic ---
     def start_streaming_ui(session_id):
-        logging.debug(f"[{session_id}] UI: Start clicked → disabling controls.")
         return {
             start_button: gr.Button(interactive=False),
             stop_button: gr.Button(interactive=True),
+            start_transcribe: gr.Button(interactive=True),
+            stop_transcribe: gr.Button(interactive=False),
+            chunk_slider: gr.Slider(interactive=False),
             main_audio: gr.Audio(visible=False),
             progress_bar: gr.Slider(value=0, visible=True),
             progress_text: gr.Textbox(value="00:00:00", visible=True),
         }
 
     def stop_streaming_ui(session_id):
-        logging.debug(f"[{session_id}] UI: Stop clicked or finished → restoring controls.")
+        logging.debug(f"[{session_id}] UI: Stop clicked → restoring controls.")
+        
         return {
             start_button: gr.Button(interactive=True),
             stop_button: gr.Button(interactive=False),
-            main_audio: gr.Audio(
-                label="Audio Source",
-                sources=["upload", "microphone"],
-                type="filepath",
-                value=DEFAULT_FILE,
-                visible=True,
-            ),
+            start_transcribe: gr.Button(interactive=False),
+            stop_transcribe: gr.Button(interactive=False),
+            chunk_slider: gr.Slider(interactive=True),
+            main_audio: gr.Audio(visible=True),
             progress_bar: gr.Slider(value=0, visible=False),
             progress_text: gr.Textbox(value="00:00:00", visible=False),
         }
 
-    ui_components = [start_button, stop_button, main_audio, progress_bar, progress_text]
-
+    # --- Streaming event ---
     webrtc_stream.stream(
         fn=read_and_stream_audio,
-        inputs=[active_filepath, session_id],
+        inputs=[active_filepath, session_id, chunk_slider],
         outputs=[webrtc_stream],
         trigger=start_button.click,
-        concurrency_id="audio_stream",
-        concurrency_limit=20,
     )
 
-    start_button.click(
-        fn=start_streaming_ui,
+    start_button.click(fn=start_streaming_ui, inputs=[session_id], outputs=[
+        start_button, stop_button, start_transcribe, stop_transcribe,
+        chunk_slider, main_audio, progress_bar, progress_text,
+    ])
+
+    stop_button.click(fn=stop_streaming, inputs=[session_id], outputs=[webrtc_stream]).then(
+        fn=stop_streaming_ui,
+        inputs=[session_id],
+        outputs=[
+            start_button, stop_button, start_transcribe, stop_transcribe,
+            chunk_slider, main_audio, progress_bar, progress_text,
+        ],
+    )
+
+    # --- Transcription control logic ---
+    def start_transcribe_ui(session_id: str):
+        """Create transcription flag and update UI."""
+        start_flag = os.path.join(TMP_DIR, f"transcribe_active_{session_id}.txt")
+        with open(start_flag, "w") as f:
+            f.write("1")
+        logging.info(f"[{session_id}] Transcription started.")
+        return {
+            start_transcribe: gr.Button(interactive=False),
+            stop_transcribe: gr.Button(interactive=True),
+            progress_text: gr.Textbox(value="🎙️ Transcription started..."),
+        }
+
+    def stop_transcribe_ui(session_id: str):
+        """Stop transcription by removing flag and update UI."""
+        flag_path = os.path.join(TMP_DIR, f"transcribe_active_{session_id}.txt")
+        if os.path.exists(flag_path):
+            os.remove(flag_path)
+        logging.info(f"[{session_id}] Transcription stopped.")
+        return {
+            start_transcribe: gr.Button(interactive=True),
+            stop_transcribe: gr.Button(interactive=False),
+            progress_text: gr.Textbox(value="🛑 Transcription stopped."),
+        }
+
+    # --- UI binding ---
+    start_transcribe.click(
+        fn=start_transcribe_ui,
         inputs=[session_id],
-        outputs=ui_components,
+        outputs=[start_transcribe, stop_transcribe, progress_text],
     )
 
-    stop_button.click(
-        fn=stop_streaming,
+    # 🔥 Actual transcription loop launch
+    start_transcribe.click(
+        fn=transcribe,
         inputs=[session_id],
-        outputs=[webrtc_stream],
-    ).then(
-        fn=stop_streaming_ui,
+        outputs=None,
+    )
+
+    stop_transcribe.click(
+        fn=stop_transcribe_ui,
         inputs=[session_id],
-        outputs=ui_components,
+        outputs=[start_transcribe, stop_transcribe, progress_text],
     )
 
+    # --- Active sessions ---
     with gr.Accordion("📊 Active Sessions", open=False):
         sessions_table = gr.DataFrame(
             headers=["session_id", "file", "start_time", "status"],
             interactive=False,
             wrap=True,
-            label="Connected Users",
             max_height=200,
         )
 
-    timer = gr.Timer(3.0)
-    timer.tick(fn=get_active_sessions, outputs=sessions_table)
-
-    progress_timer = gr.Timer(1.0)
-    progress_timer.tick(fn=get_session_progress, inputs=[session_id], outputs=[progress_bar, progress_text])
+    gr.Timer(3.0).tick(fn=get_active_sessions, outputs=sessions_table)
+    gr.Timer(1.0).tick(fn=get_session_progress, inputs=[session_id], outputs=[progress_bar, progress_text])
 
+# --------------------------------------------------------
+# CSS
+# --------------------------------------------------------
 custom_css = """
 #column_source {
     display: flex;
@@ -274,5 +367,9 @@ custom_css = """
 """
 demo.css = custom_css
 
+
+# --------------------------------------------------------
+# MAIN
+# --------------------------------------------------------
 if __name__ == "__main__":
     demo.queue(max_size=50, api_open=False).launch(show_api=False, debug=True)

app/session_utils.py

@@ -5,6 +5,8 @@ from datetime import datetime
 from app.logger_config import logger as logging
 
 TMP_DIR = "/tmp/canary_aed_streaming"
+# TMP_DIR = "/home/sifar-dev/workspace/canary_aed_streaming/tmp/canary_aed_streaming"
+
 ACTIVE_SESSIONS_FILE = os.path.join(TMP_DIR, "active_sessions.json")
 
 
@@ -45,6 +47,24 @@ def reset_active_sessions():
                     logging.debug(f"Removed leftover stop flag file: {f}")
                 except Exception as e:
                     logging.warning(f"Failed to remove stop flag file {f}: {e}")
+        # Clean up old transcribe_stop_flag
+        for f in os.listdir(TMP_DIR):
+            if f.startswith("transcribe_stop_flag_") and f.endswith(".txt"):
+                try:
+                    os.remove(os.path.join(TMP_DIR, f))
+                    logging.debug(f"Removed leftover transcribe_stop_flag flag file: {f}")
+                except Exception as e:
+                    logging.warning(f"Failed to remove transcribe_stop_flag  file {f}: {e}")
+        # Clean up old transcribe_active_flag
+        for f in os.listdir(TMP_DIR):
+            if f.startswith("transcribe_active_") and f.endswith(".txt"):
+                try:
+                    os.remove(os.path.join(TMP_DIR, f))
+                    logging.debug(f"Removed leftover transcribe active flag file: {f}")
+                except Exception as e:
+                    logging.warning(f"Failed to remove transcribe active file {f}: {e}")
+
+
 
     except Exception as e:
         logging.error(f"Error resetting active sessions: {e}")