impl on_additional_outputs

app.py

@@ -5,380 +5,218 @@ import asyncio
 from fastrtc.webrtc import WebRTC
 from pydub import AudioSegment
 import time
-import os
-import json
-import spaces
-from app.utils import generate_coturn_config
-
-from app.session_utils import (
-    TMP_DIR,
-    generate_session_id,
-    register_session,
-    unregister_session,
-    get_active_sessions,
-    stop_file_path,
-    create_stop_flag,
-    clear_stop_flag,
-    reset_all_active_sessions,
-    on_load,
-    on_unload
+import os 
+from gradio.utils import get_space
 
+from app.logger_config import logger as logging
+from app.utils import (
+    generate_coturn_config
 )
 
-# Reset sessions at startup
-reset_all_active_sessions()
-
 EXAMPLE_FILES = ["data/bonjour.wav", "data/bonjour2.wav"]
-DEFAULT_FILE = EXAMPLE_FILES[0]
-
-
-# --------------------------------------------------------
-# 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}")
-
+DEFAULT_FILE = EXAMPLE_FILES[0] 
+# Utilisé pour signaler l'arrêt du streaming à l'intérieur du générateur
+stop_stream_state = gr.State(value=False)
+
+def read_and_stream_audio(filepath_to_stream: str):
+    """
+    Un générateur synchrone qui lit un fichier audio (via filepath_to_stream)
+    et le streame chunk par chunk d'1 seconde.
+    """
+    
     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}")
-        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")    
-    chunk_dir = os.path.join(TMP_DIR, f"chunks_{session_id}")
-    os.makedirs(chunk_dir, exist_ok=True)
-
+        logging.error(f"Fichier audio non trouvé ou non spécifié : {filepath_to_stream}")
+        # Tenter d'utiliser le fichier par défaut en cas de problème
+        if os.path.exists(DEFAULT_FILE):
+            logging.warning(f"Utilisation du fichier par défaut : {DEFAULT_FILE}")
+            filepath_to_stream = DEFAULT_FILE
+        else:
+            logging.error("Fichier par défaut non trouvé. Arrêt du stream.")
+            return
+
+    logging.info(f"Préparation du segment audio depuis : {filepath_to_stream}")
+    # Réinitialiser le signal d'arrêt à chaque lancement
+    stop_stream_state.value = False
+    
     try:
         segment = AudioSegment.from_file(filepath_to_stream)
-        chunk_ms = int(chunk_seconds * 1000)
-        total_chunks = len(segment) // chunk_ms + 1
+        chunk_duree_ms = 1000
+        logging.info(f"Début du streaming en chunks de {chunk_duree_ms}ms...")
 
-        logging.info(f"[{session_id}] Streaming {total_chunks} chunks ({chunk_seconds:.2f}s each)...")
+        for i, chunk in enumerate(segment[::chunk_duree_ms]):
+            iter_start_time = time.perf_counter()
+            logging.info(f"Envoi du chunk {i+1}...")
 
-        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}. Ending stream.")
-                clear_stop_flag(session_id)
+            if stop_stream_state.value:
+                logging.info("Signal d'arrêt reçu, arrêt de la boucle.")
                 break
-            logging.info(f"[{session_id}] Streaming chunk {i}.")
-            iter_start = time.perf_counter()
 
-            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}"
-
-            percent = round((i / total_chunks) * 100, 2)
-            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
-
-            #  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)")
-
-            # Stream audio to client
-            yield (rate, chunk_array.reshape(1, -1))
-
-            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}] Streaming completed successfully.")
-
-    except Exception as e:
-        logging.error(f"[{session_id}] Stream error: {e}", exc_info=True)
-    finally:
-        unregister_session(session_id)
-        clear_stop_flag(session_id)
-        if os.path.exists(progress_path):
-            os.remove(progress_path)
-
-
-# --------------------------------------------------------
-# 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
+            output_chunk = (
+                chunk.frame_rate,
+                np.array(chunk.get_array_of_samples()).reshape(1, -1),
+            )
 
-            time.sleep(0.25)
+            yield output_chunk
+
+            iter_end_time = time.perf_counter()
+            processing_duration_ms = (iter_end_time - iter_start_time) * 1000
+            
+            sleep_duration = (chunk_duree_ms / 1000.0) - (processing_duration_ms / 1000.0) - 0.1
+            if sleep_duration < 0:
+                sleep_duration = 0.01 # Éviter un temps de sommeil négatif
+                
+            logging.debug(f"Temps de traitement: {processing_duration_ms:.2f}ms, Sommeil: {sleep_duration:.2f}s")
+
+            elapsed = 0.0
+            interval = 0.05
+            while elapsed < sleep_duration:
+                if stop_stream_state.value:
+                    logging.info("Signal d'arrêt reçu pendant l'attente.")
+                    break
+                wait_chunk = min(interval, sleep_duration - elapsed)
+                time.sleep(wait_chunk)
+                elapsed += wait_chunk
+            if stop_stream_state.value:
+                break
 
-        logging.info(f"[{session_id}] Transcription loop ended (flag removed).")
+        logging.info("Streaming terminé.")
 
+    except asyncio.CancelledError:
+        logging.info("Stream arrêté par l'utilisateur (CancelledError).")
+        raise
+    except FileNotFoundError:
+        logging.error(f"Erreur critique : Fichier non trouvé : {filepath_to_stream}")
     except Exception as e:
-        logging.error(f"[{session_id}] Transcription error: {e}", exc_info=True)
+        logging.error(f"Erreur pendant le stream: {e}", exc_info=True)
+        raise
     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_stream_state.value = False
+        logging.info("Signal d'arrêt nettoyé.")
 
 
-# --------------------------------------------------------
-# STOP STREAMING
-# --------------------------------------------------------
-def stop_streaming(session_id: str):
-    create_stop_flag(session_id)
-    logging.info(f"[{session_id}] Stop button clicked → stop flag created.")
+def stop_streaming():
+    """Active le signal d'arrêt pour le générateur."""
+    logging.info("Bouton Stop cliqué: envoi du signal d'arrêt.")
+    stop_stream_state.value = True
     return None
 
+# --- Interface Gradio ---
 
-def get_session_progress(session_id: str):
-    """Read streaming progress and return slider position + elapsed time."""
-    progress_path = os.path.join(TMP_DIR, f"progress_{session_id}.json")
-    if not os.path.exists(progress_path):
-        return 0.0, "00:00:00"
-    try:
-        with open(progress_path, "r") as f:
-            data = json.load(f)
-        value = data.get("value", 0.0)
-        elapsed = data.get("elapsed", "00:00:00")
-        return value, elapsed
-    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 stream. Transcription runs only during streaming."
+        "## Application 'Streamer' WebRTC (Serveur -> Client)\n"
+        "Utilisez l'exemple fourni, uploadez un fichier ou enregistrez depuis votre micro, "
+        "puis cliquez sur 'Start' pour écouter le stream."
     )
 
-    session_id = gr.State() 
-    sid_box = gr.Textbox(label="Session ID", interactive=False)
-    demo.load(fn=on_load, inputs=None, outputs=[session_id, sid_box])
-    demo.unload(on_unload)
+    # 1. État pour stocker le chemin du fichier à lire
     active_filepath = gr.State(value=DEFAULT_FILE)
 
-    with gr.Row(equal_height=True):
-        with gr.Column(elem_id="column_source", scale=1):
-            with gr.Group(elem_id="centered_content"):
-                main_audio = gr.Audio(
-                    label="Audio Source",
-                    sources=["upload", "microphone"],
-                    type="filepath",
-                    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,
-                    maximum=100,
-                    value=0,
-                    step=0.1,
-                    interactive=False,
-                    visible=False,
-                )
-
-                progress_text = gr.Textbox(
-                    label="Elapsed Time (hh:mm:ss)",
-                    interactive=False,
-                    visible=False,
-                )
-
-                with gr.Row():
-                    start_button = gr.Button("▶️ Start Streaming", variant="primary")
-                    stop_button = gr.Button("⏹️ Stop Streaming", variant="stop", interactive=False)
-
+    with gr.Row():
+        with gr.Column():
+            main_audio = gr.Audio(
+                label="Source Audio",
+                sources=["upload", "microphone"], # Combine les deux sources
+                type="filepath",
+                value=DEFAULT_FILE, # Défaut au premier exemple
+            )
         with gr.Column():
             webrtc_stream = WebRTC(
-                label="Audio Stream",
+                label="Stream Audio",
                 mode="receive",
                 modality="audio",
                 rtc_configuration=generate_coturn_config(),
-                visible=True,
+                visible=True, # Caché par défaut
+                height = 200,
             )
+    # 4. Boutons de contrôle
+    with gr.Row():
+        with gr.Column():
+            start_button = gr.Button("Start Streaming", variant="primary")
+            stop_button = gr.Button("Stop Streaming", variant="stop", interactive=False)
+        with gr.Column():
+            gr.Text()
+
+    def set_new_file(filepath):
+        """Met à jour l'état avec le nouveau chemin, ou revient au défaut si None."""
+        if filepath is None:
+            logging.info("Audio effacé, retour au fichier d'exemple par défaut.")
+            new_path = DEFAULT_FILE
+        else:
+            logging.info(f"Nouvelle source audio sélectionnée : {filepath}")
+            new_path = filepath
+        # Retourne la valeur à mettre dans le gr.State
+        return new_path
+
+    # Mettre à jour le chemin si l'utilisateur upload, efface, ou change le fichier
+    main_audio.change(
+        fn=set_new_file, 
+        inputs=[main_audio], 
+        outputs=[active_filepath]
+    )
+    
+    # Mettre à jour le chemin si l'utilisateur termine un enregistrement
+    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):
+    # Fonctions pour mettre à jour l'état de l'interface
+    def start_streaming_ui():
+        logging.info("UI : Démarrage du streaming. Désactivation des contrôles.")
         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),
+            main_audio: gr.Audio(visible=False), 
         }
 
-    def stop_streaming_ui(session_id):
-        logging.debug(f"[{session_id}] UI: Stop clicked → restoring controls.")
-        
+    def stop_streaming_ui():
+        logging.info("UI : Arrêt du streaming. Réactivation des contrôles.")
         return {
             start_button: gr.Button(interactive=True),
             stop_button: gr.Button(interactive=False),
-            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),
-        }
-
-    # --- Streaming event ---
-    webrtc_stream.stream(
+            main_audio: gr.Audio(
+                label="Source Audio",
+                sources=["upload", "microphone"], # Combine les deux sources
+                type="filepath",
+                value=active_filepath.value,
+                visible=True
+        ), 
+            }
+
+
+    ui_components = [
+        start_button, stop_button, 
+        main_audio, 
+    ]
+
+    stream_event = webrtc_stream.stream(
         fn=read_and_stream_audio,
-        inputs=[active_filepath, session_id, chunk_slider],
+        inputs=[active_filepath],  
         outputs=[webrtc_stream],
         trigger=start_button.click,
-        concurrency_limit=20,
-        concurrency_id="receive"
-    )
-
-    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=[start_transcribe, stop_transcribe, progress_text],
+        concurrency_id="audio_stream", # ID de concurrence
+        concurrency_limit=10
     )
 
-    # 🔥 Actual transcription loop launch
-    start_transcribe.click(
-        fn=transcribe,
-        inputs=[session_id],
-        outputs=None,
+    # Mettre à jour l'interface au clic sur START
+    start_button.click(
+        fn=start_streaming_ui,
+        outputs=ui_components
     )
 
-    stop_transcribe.click(
-        fn=stop_transcribe_ui,
-        inputs=[session_id],
-        outputs=[start_transcribe, stop_transcribe, progress_text],
+    # Correction : S'assurer que le stream est bien annulé
+    stop_button.click(
+        fn=stop_streaming, 
+        outputs=[webrtc_stream], 
+    ).then(
+        fn=stop_streaming_ui, # ENSUITE, mettre à jour l'interface
+        inputs=None,
+        outputs=ui_components
     )
 
-    # --- Active sessions ---
-    with gr.Accordion("📊 Active Sessions", open=False):
-        sessions_table = gr.DataFrame(
-            headers=["session_id", "file", "start_time", "status"],
-            interactive=False,
-            wrap=True,
-            max_height=200,
-        )
-
-    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;
-    flex-direction: column;
-    justify-content: center;
-    align-items: center;
-    gap: 1rem;
-    margin-top: auto;
-    margin-bottom: auto;
-}
-#column_source .gr-row {
-    padding-top: 12px;
-    padding-bottom: 12px;
-}
-"""
-demo.css = custom_css
-
-
-# --------------------------------------------------------
-# MAIN
-# --------------------------------------------------------
 if __name__ == "__main__":
-
-    demo.queue(max_size=20, api_open=False).launch(show_api=False, debug=True)
+    demo.queue(max_size=10, api_open=False).launch(show_api=False, debug=True)