clean new app

new_app.py

@@ -6,12 +6,12 @@ from fastrtc.webrtc import WebRTC
 from fastrtc.utils import AdditionalOutputs
 from pydub import AudioSegment
 import time
-import os 
+import os
 from gradio.utils import get_space
 
-from app.logger_config import logger as logging
 from app.utils import (
-    generate_coturn_config
+    generate_coturn_config,
+    raise_function
 )
 from app.new_session_utils import (
     on_load,
@@ -19,137 +19,161 @@ from app.new_session_utils import (
     get_active_sessions,
     reset_all_active_sessions,
 )
+
+# --------------------------------------------------------
+# Initialization
+# --------------------------------------------------------
 reset_all_active_sessions()
 EXAMPLE_FILES = ["data/bonjour.wav", "data/bonjour2.wav"]
-DEFAULT_FILE = EXAMPLE_FILES[0] 
+DEFAULT_FILE = EXAMPLE_FILES[0]
+
 
+# --------------------------------------------------------
+# Utility functions
+# --------------------------------------------------------
 def _is_stop_requested(stop_streaming_flags: dict) -> bool:
+    """Check if the stop signal was requested."""
     if not isinstance(stop_streaming_flags, dict):
         return False
     return bool(stop_streaming_flags.get("stop", False))
 
-def read_and_stream_audio(filepath_to_stream: str, session_id: str, stop_streaming_flags: dict):
+
+def handle_stream_error(session_id: str, error: Exception | str, stop_streaming_flags: dict | None = None):
     """
-    Un générateur synchrone qui lit un fichier audio (via filepath_to_stream)
-    et le streame chunk par chunk d'1 seconde.
+    Handle streaming errors:
+    - Log the error
+    - Send structured info to client
+    - Reset stop flag
     """
+    if isinstance(error, Exception):
+        msg = f"{type(error).__name__}: {str(error)}"
+    else:
+        msg = str(error)
 
-    if not session_id:
-        logging.warning( "Aucun session_id fourni, arrêt du stream par sécurité.")
-        return
+    logging.error(f"[{session_id}] Streaming error: {msg}", exc_info=isinstance(error, Exception))
 
     if isinstance(stop_streaming_flags, dict):
         stop_streaming_flags["stop"] = False
-    else:
-        logging.warning(f" [{session_id}] Stop stop_streaming_flags non initialisés, le stream continuera sans contrôle d'arrêt.")
 
-    if not filepath_to_stream or not os.path.exists(filepath_to_stream):
-        logging.error(f"[{session_id}] 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"[{session_id}] Utilisation du fichier par défaut : {DEFAULT_FILE}")
-            filepath_to_stream = DEFAULT_FILE
-        else:
-            logging.error(f"[{session_id}] Fichier par défaut non trouvé. Arrêt du stream.")
-            return
+    yield (None, AdditionalOutputs({"error": True, "message": msg}))
+    yield (None, AdditionalOutputs("STREAM_DONE"))
 
-    logging.info(f"[{session_id}] Préparation du segment audio depuis : {filepath_to_stream}")
+
+def read_and_stream_audio(filepath_to_stream: str, session_id: str, stop_streaming_flags: dict):
+    """
+    Read an audio file and stream it chunk by chunk (1s per chunk).
+    Handles errors safely and reports structured messages to the client.
+    """
+    if not session_id:
+        yield from handle_stream_error("unknown", "No session_id provided.", stop_streaming_flags)
+        return
+
+    if not filepath_to_stream or not os.path.exists(filepath_to_stream):
+        yield from handle_stream_error(session_id, f"Audio file not found: {filepath_to_stream}", stop_streaming_flags)
+        return
 
     try:
         segment = AudioSegment.from_file(filepath_to_stream)
-        chunk_duree_ms = 1000
-        total_chunks = len(segment) // chunk_duree_ms + 1
-        logging.info(f"[{session_id}] Début du streaming en chunks de {chunk_duree_ms}ms...")
-
-        for i, chunk in enumerate(segment[::chunk_duree_ms]):
-            iter_start_time = time.perf_counter()
-            logging.info(f"Envoi du chunk {i+1}...")
+        chunk_duration_ms = 1000
+        total_chunks = len(segment) // chunk_duration_ms + 1
+        logging.info(f"[{session_id}] Starting audio streaming ({total_chunks} chunks).")
 
+        for i, chunk in enumerate(segment[::chunk_duration_ms]):
             if _is_stop_requested(stop_streaming_flags):
-                logging.info(f"[{session_id}]Signal d'arrêt reçu, arrêt de la boucle.")
+                logging.info(f"[{session_id}] Stop signal received. Terminating stream.")
                 break
 
-            output_chunk = (
-                chunk.frame_rate,
-                np.array(chunk.get_array_of_samples()).reshape(1, -1),
-            )
-                # Calcul du pourcentage de progression
+            frame_rate = chunk.frame_rate
+            samples = np.array(chunk.get_array_of_samples()).reshape(1, -1)
             progress = round(((i + 1) / total_chunks) * 100, 2)
-            logging.debug(f"[{session_id}] Progression: {progress}%")
-
-            # Envoi du chunk et de la progression numérique
-            yield (output_chunk, AdditionalOutputs(progress))
-
-            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"[{session_id}]Temps de traitement: {processing_duration_ms:.2f}ms, Sommeil: {sleep_duration:.2f}s")
-
-            elapsed = 0.0
-            interval = 0.05
-            while elapsed < sleep_duration:
-                if _is_stop_requested(stop_streaming_flags):
-                    logging.info(f"[{session_id}]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 _is_stop_requested(stop_streaming_flags):
-                break
 
-        logging.info(f"[{session_id}]Streaming terminé.")
+            yield ((frame_rate, samples), AdditionalOutputs(progress))
+            logging.debug(f"[{session_id}] Sent chunk {i+1}/{total_chunks} ({progress}%).")
+
+            time.sleep(0.9)
+            raise_function()  # Optional injected test exception
+
+        logging.info(f"[{session_id}] Audio streaming completed successfully.")
 
     except asyncio.CancelledError:
-        logging.info(f"[{session_id}]Stream arrêté par l'utilisateur (CancelledError).")
-        raise
-    except FileNotFoundError:
-        logging.error(f"[{session_id}] Erreur critique : Fichier non trouvé : {filepath_to_stream}")
+        yield from handle_stream_error(session_id, "Streaming cancelled by user.", stop_streaming_flags)
+    except FileNotFoundError as e:
+        yield from handle_stream_error(session_id, e, stop_streaming_flags)
     except Exception as e:
-        logging.error(f"[{session_id}] Erreur pendant le stream: {e}", exc_info=True)
-        raise
+        yield from handle_stream_error(session_id, e, stop_streaming_flags)
     finally:
         if isinstance(stop_streaming_flags, dict):
             stop_streaming_flags["stop"] = False
-        logging.info(f"[{session_id}]Signal d'arrêt nettoyé.")
-        yield (None, AdditionalOutputs({"progress": 100}))  
-
+        logging.info(f"[{session_id}] Stop flag reset.")
+        yield (None, AdditionalOutputs("STREAM_DONE"))
 
 
 def stop_streaming(session_id: str, stop_streaming_flags: dict):
-    """Active le signal d'arrêt pour le générateur."""
-    logging.info("Bouton Stop cliqué: envoi du signal d'arrêt.")
+    """Trigger the stop flag for active streaming."""
+    logging.info(f"[{session_id}] Stop button clicked — sending stop signal.")
     if not isinstance(stop_streaming_flags, dict):
         stop_streaming_flags = {"stop": True}
     else:
         stop_streaming_flags["stop"] = True
     return stop_streaming_flags
 
-def handle_additional_outputs(status_slider,progress_value):
-    """Met à jour le slider selon la valeur reçue et gère sa visibilité."""
-    logging.debug(f"📡 Additional output received: {progress_value}")
+
+def handle_additional_outputs(start_button, stop_button, main_audio, status_slider, progress_value):
+    """
+    Update UI elements based on streaming progress or errors.
+    Controls button states, audio visibility, and progress slider.
+    """
+    logging.debug(f"Additional output received: {progress_value}")
+
+    # Handle structured error message
+    if isinstance(progress_value, dict) and progress_value.get("error"):
+        msg = progress_value.get("message", "Unknown error.")
+        logging.error(f"[stream_ui] Client-side error: {msg}")
+        return (
+            gr.update(interactive=True),   # start_button enabled
+            gr.update(interactive=False),  # stop_button disabled
+            gr.update(visible=True),       # audio re-shown
+            gr.update(visible=False, value=0),  # slider hidden
+        )
 
     try:
         progress = float(progress_value)
     except (ValueError, TypeError):
         progress = 0
-    # status_slider =  gr.update(interactive=True,visible=True,  value=max(0, min(progress, 100)))
-    # return status_slider
+
+    # --- Stream not started ---
+    if progress <= 0:
+        return (
+            gr.update(interactive=True),   # start_button enabled
+            gr.update(interactive=False),  # stop_button disabled
+            gr.update(visible=True),       # audio visible
+            gr.update(visible=False, value=0),  # slider hidden
+        )
+
+    # --- Stream finished ---
     if progress >= 100:
-        return gr.update(visible=False, value=100)
-    elif progress <= 0:
-        return gr.update(visible=False, value=0)
-    else:
-        return gr.update(visible=True, value=progress)
+        return (
+            gr.update(interactive=True),   # start_button re-enabled
+            gr.update(interactive=False),  # stop_button disabled
+            gr.update(visible=True),       # audio visible
+            gr.update(visible=False, value=100), # slider hidden
+        )
+
+    # --- Stream in progress ---
+    return (
+        gr.update(interactive=False),      # start_button disabled
+        gr.update(interactive=True),       # stop_button enabled
+        gr.update(visible=False),          # hide audio
+        gr.update(visible=True, value=progress), # show progress
+    )
 
-# --- Interface Gradio ---
 
+# --------------------------------------------------------
+# Gradio Interface
+# --------------------------------------------------------
 with gr.Blocks(theme=gr.themes.Soft()) as demo:
 
-    session_hash = gr.State() 
+    session_hash = gr.State()
     session_hash_box = gr.Textbox(label="Session ID", interactive=False)
     demo.load(fn=on_load, inputs=None, outputs=[session_hash, session_hash_box])
     demo.unload(on_unload)
@@ -157,135 +181,80 @@ with gr.Blocks(theme=gr.themes.Soft()) as demo:
     stop_streaming_flags = gr.State(value={"stop": False})
 
     gr.Markdown(
-        "## 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."
+        "## WebRTC Audio Streamer (Server → Client)\n"
+        "Upload or record an audio file, then click **Start** to listen to the streamed audio."
     )
 
-    # 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="Source Audio",
-                    sources=["upload", "microphone"], # Combine les deux sources
+                    label="Audio File",
+                    sources=["upload", "microphone"],
                     type="filepath",
-                    value=DEFAULT_FILE, # Défaut au premier exemple
+                    value=DEFAULT_FILE,
+                )
+                status_slider = gr.Slider(
+                    0, 100, value=0, label="Streaming Progress", interactive=False, visible=False
                 )
-                status_slider = gr.Slider(0, 100, value=0, label="Progression du streaming", interactive=False, visible=False )
 
         with gr.Column():
             webrtc_stream = WebRTC(
-                label="Stream Audio",
+                label="Live",
                 mode="receive",
                 modality="audio",
                 rtc_configuration=generate_coturn_config(),
-                visible=True, # Caché par défaut
-                height = 200,
+                visible=True,
+                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."""
+        """Update active audio path or reset to default if empty."""
         if filepath is None:
-            logging.info("Audio effacé, retour au fichier d'exemple par défaut.")
+            logging.info("[ui] Audio cleared — reverting to default example file.")
             new_path = DEFAULT_FILE
         else:
-            logging.info(f"Nouvelle source audio sélectionnée : {filepath}")
+            logging.info(f"[ui] New audio source selected: {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]
-    )
-
+    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])
 
-    # Fonctions pour mettre à jour l'état de l'interface
-    def start_streaming_ui(session_id: str, flags: dict):
-        logging.info("UI : Démarrage du streaming. Désactivation des contrôles.")
-        if not isinstance(flags, dict):
-            flags = {"stop": False}
-        else:
-            flags["stop"] = False
-        return (
-            gr.Button(interactive=False),
-            gr.Button(interactive=True),
-            gr.Audio(visible=False),
-            flags,
-        )
-
-    def stop_streaming_ui(flags: dict):
-        logging.info("UI : Arrêt du streaming. Réactivation des contrôles.")
-        return (
-            gr.Button(interactive=True),
-            gr.Button(interactive=False),
-            gr.Audio(
-                label="Source Audio",
-                sources=["upload", "microphone"],
-                type="filepath",
-                value=active_filepath.value,
-                visible=True,
-            ),
-        )
-
-
-    ui_components = [
-        start_button, stop_button, 
-        main_audio, 
-    ]
+    ui_components = [start_button, stop_button, main_audio, status_slider]
 
     stream_event = webrtc_stream.stream(
         fn=read_and_stream_audio,
-        inputs=[active_filepath, session_hash, stop_streaming_flags],  
+        inputs=[active_filepath, session_hash, stop_streaming_flags],
         outputs=[webrtc_stream],
         trigger=start_button.click,
-        concurrency_id="audio_stream", 
-        concurrency_limit=10
+        concurrency_id="audio_stream",
+        concurrency_limit=10,
     )
+
     webrtc_stream.on_additional_outputs(
         fn=handle_additional_outputs,
-        inputs=[status_slider],
-        outputs=[status_slider],
+        inputs=ui_components,
+        outputs=ui_components,
         concurrency_id="additional_outputs_audio_stream",
-        concurrency_limit=10
-    )
-    # Mettre à jour l'interface au clic sur START
-    start_button.click(
-        fn=start_streaming_ui,
-        inputs=[session_hash, stop_streaming_flags],
-        outputs=ui_components + [stop_streaming_flags]
+        concurrency_limit=10,
     )
 
-    # Correction : S'assurer que le stream est bien annulé
+    start_button.click(fn=None, inputs=None, outputs=None)
+
     stop_button.click(
-        fn=stop_streaming, 
+        fn=stop_streaming,
         inputs=[session_hash, stop_streaming_flags],
-        outputs=[stop_streaming_flags], 
-    ).then(
-        fn=stop_streaming_ui, # ENSUITE, mettre à jour l'interface
-        inputs=[stop_streaming_flags],
-        outputs=ui_components
+        outputs=[stop_streaming_flags],
     )
-    # --- Active sessions ---
+
     with gr.Accordion("📊 Active Sessions", open=False):
         sessions_table = gr.DataFrame(
             headers=["session_hash", "file", "start_time", "status"],
@@ -297,9 +266,8 @@ with gr.Blocks(theme=gr.themes.Soft()) as demo:
     gr.Timer(3.0).tick(fn=get_active_sessions, outputs=sessions_table)
 
 
-
 # --------------------------------------------------------
-# CSS
+# Custom CSS
 # --------------------------------------------------------
 custom_css = """
 #column_source {
@@ -317,9 +285,10 @@ custom_css = """
 }
 """
 demo.css = custom_css
+
+
 # --------------------------------------------------------
-# MAIN
+# Main
 # --------------------------------------------------------
-
 if __name__ == "__main__":
     demo.queue(max_size=10, api_open=False).launch(show_api=False, debug=True)