add new_app with better stop_streaming_flags

app/new_session_utils.py

@@ -0,0 +1,237 @@
+import os
+import json
+import uuid
+import shutil
+from datetime import datetime
+from app.logger_config import logger as logging
+import gradio as gr
+# TMP_DIR = "/tmp/canary_aed_streaming"
+
+TMP_DIR = os.getenv("TMP_DIR", "/tmp/canary_aed_streaming")
+ACTIVE_SESSIONS_FILE = os.path.join(TMP_DIR, "active_sessions.json")
+
+
+# ---------------------------
+# Helper to manage the JSON
+# ---------------------------
+def _read_sessions():
+    if not os.path.exists(ACTIVE_SESSIONS_FILE):
+        return {}
+    try:
+        with open(ACTIVE_SESSIONS_FILE, "r") as f:
+            return json.load(f)
+    except Exception:
+        return {}
+
+
+def _write_sessions(data):
+    os.makedirs(os.path.dirname(ACTIVE_SESSIONS_FILE), exist_ok=True)
+    with open(ACTIVE_SESSIONS_FILE, "w") as f:
+        json.dump(data, f, indent=2)
+
+
+# ---------------------------
+# LOAD
+# ---------------------------
+def on_load(request: gr.Request):
+    """Called when a new visitor opens the app."""
+    session_hash = request.session_hash  # ✅ Directly use session_hash as unique ID
+    sessions = _read_sessions()
+
+    sessions[session_hash] = {
+        "session_hash": session_hash,
+        "file": "",
+        "start_time": datetime.utcnow().strftime("%H:%M:%S"),
+        "status": "active",
+    }
+
+    _write_sessions(sessions)
+    logging.info(f"[{session_hash}] Session registered (on_load).")
+
+    return session_hash, session_hash  # can be used as gr.State + display
+
+
+# ---------------------------
+# UNLOAD
+# ---------------------------
+def on_unload(request: gr.Request):
+    """Called when the visitor closes or refreshes the app."""
+    sid = request.session_hash
+    sessions = _read_sessions()
+
+    if sid in sessions:
+        create_stop_flag(sid)
+        sessions.pop(sid)
+        _write_sessions(sessions)
+        remove_session_data(sid)
+        unregister_session(sid)
+        logging.info(f"[{sid}] Session removed (on_unload).")
+    else:
+        logging.info(f"[{sid}] No active session found to remove.")
+
+def ensure_tmp_dir():
+    """Ensures the base temporary directory exists."""
+    try:
+        os.makedirs(TMP_DIR, exist_ok=True)
+    except Exception as e:
+        logging.error(f"Failed to create tmp directory {TMP_DIR}: {e}")
+
+
+def reset_all_active_sessions():
+    """Removes all temporary session files and folders at startup."""
+    ensure_tmp_dir()
+
+    try:
+        # --- Remove active sessions file ---
+        if os.path.exists(ACTIVE_SESSIONS_FILE):
+            os.remove(ACTIVE_SESSIONS_FILE)
+            logging.info("Active sessions file reset at startup.")
+        else:
+            logging.debug("No active sessions file found to reset.")
+       
+    except Exception as e:
+        logging.error(f"Error resetting active sessions: {e}")
+
+def remove_session_data(session_id: str):
+    """Removes all temporary files and data related to a specific session."""
+    if not session_id:
+        logging.warning("reset_session() called without a valid session_id.")
+        return
+
+    try:
+        # --- Remove session from active_sessions.json ---
+        if os.path.exists(ACTIVE_SESSIONS_FILE):
+            try:
+                with open(ACTIVE_SESSIONS_FILE, "r") as f:
+                    data = json.load(f)
+                if session_id in data:
+                    data.pop(session_id)
+                    with open(ACTIVE_SESSIONS_FILE, "w") as f:
+                        json.dump(data, f, indent=2)
+                    logging.debug(f"[{session_id}] Removed from active_sessions.json.")
+            except Exception as e:
+                logging.warning(f"[{session_id}] Failed to update active_sessions.json: {e}")
+
+        # --- Define all possible session file patterns ---
+        files_to_remove = [
+            f"progress_{session_id}.json",
+            # f"stream_stop_flag_{session_id}.txt",
+            f"transcribe_stop_flag_{session_id}.txt",
+            f"transcribe_active_{session_id}.txt",
+        ]
+
+        # --- Remove all temporary files ---
+        for fname in files_to_remove:
+            path = os.path.join(TMP_DIR, fname)
+            if os.path.exists(path):
+                try:
+                    os.remove(path)
+                    logging.debug(f"[{session_id}] Removed file: {fname}")
+                except Exception as e:
+                    logging.warning(f"[{session_id}] Failed to remove file {fname}: {e}")
+
+        # --- Remove chunk folder if exists ---
+        chunk_dir = os.path.join(TMP_DIR, f"chunks_{session_id}")
+        if os.path.isdir(chunk_dir):
+            try:
+                shutil.rmtree(chunk_dir)
+                logging.debug(f"[{session_id}] Removed chunk folder: chunks_{session_id}")
+            except Exception as e:
+                logging.warning(f"[{session_id}] Failed to remove chunk folder: {e}")
+
+        logging.info(f"[{session_id}] Session fully reset.")
+
+    except Exception as e:
+        logging.error(f"[{session_id}] Error during reset_session: {e}")
+def generate_session_id() -> str:
+    """Generates a unique session ID."""
+    sid = str(uuid.uuid4())
+    logging.debug(f"[{sid}] New session created.")
+    return sid
+
+
+def register_session(session_id: str, filepath: str):
+    """Registers a new session."""
+    ensure_tmp_dir()
+    data = {}
+    if os.path.exists(ACTIVE_SESSIONS_FILE):
+        with open(ACTIVE_SESSIONS_FILE, "r") as f:
+            try:
+                data = json.load(f)
+            except Exception:
+                data = {}
+
+    data[session_id] = {
+        "session_id": session_id,
+        "file": filepath,
+        "start_time": datetime.utcnow().strftime("%H:%M:%S"),
+        "status": "active",
+    }
+
+    with open(ACTIVE_SESSIONS_FILE, "w") as f:
+        json.dump(data, f)
+
+    logging.debug(f"[{session_id}] Session registered in active_sessions.json.")
+
+
+def unregister_session(session_id: str):
+    """Removes a session from the registry."""
+    if not os.path.exists(ACTIVE_SESSIONS_FILE):
+        return
+
+    try:
+        with open(ACTIVE_SESSIONS_FILE, "r") as f:
+            data = json.load(f)
+        if session_id in data:
+            data.pop(session_id)
+            with open(ACTIVE_SESSIONS_FILE, "w") as f:
+                json.dump(data, f)
+            logging.debug(f"[{session_id}] Session unregistered.")
+    except Exception as e:
+        logging.error(f"[{session_id}] Error unregistering session: {e}")
+
+
+def get_active_sessions():
+    """Returns active sessions as a list of rows for the DataFrame."""
+    if not os.path.exists(ACTIVE_SESSIONS_FILE):
+        return []
+
+    try:
+        with open(ACTIVE_SESSIONS_FILE, "r") as f:
+            data = json.load(f)
+
+        rows = [
+            [
+                s.get("session_id", ""),
+                s.get("file", ""),
+                s.get("start_time", ""),
+                s.get("status", ""),
+            ]
+            for s in data.values()
+        ]
+        return rows
+    except Exception as e:
+        logging.error(f"Error reading active sessions: {e}")
+        return []
+
+
+def stop_file_path(session_id: str) -> str:
+    """Returns the stop-flag file path for a given session."""
+    ensure_tmp_dir()
+    return os.path.join(TMP_DIR, f"stream_stop_flag_{session_id}.txt")
+
+
+def create_stop_flag(session_id: str):
+    """Creates a stop-flag file for this session."""
+    path = stop_file_path(session_id)
+    with open(path, "w") as f:
+        f.write("1")
+    logging.info(f"[{session_id}] Stop flag file created at {path}.")
+
+
+def clear_stop_flag(session_id: str):
+    """Deletes the stop-flag file if it exists."""
+    path = stop_file_path(session_id)
+    if os.path.exists(path):
+        os.remove(path)
+        logging.debug(f"[{session_id}] Stop flag cleared.")

new_app.py

@@ -0,0 +1,270 @@
+from app.logger_config import logger as logging
+import numpy as np
+import gradio as gr
+import asyncio
+from fastrtc.webrtc import WebRTC
+from pydub import AudioSegment
+import time
+import os 
+from gradio.utils import get_space
+
+from app.logger_config import logger as logging
+from app.utils import (
+    generate_coturn_config
+)
+from app.session_utils import (
+    on_load,
+    on_unload,
+    get_active_sessions,
+    reset_all_active_sessions,
+)
+reset_all_active_sessions()
+EXAMPLE_FILES = ["data/bonjour.wav", "data/bonjour2.wav"]
+DEFAULT_FILE = EXAMPLE_FILES[0] 
+
+def _is_stop_requested(stop_streaming_flags: dict) -> bool:
+    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):
+    """
+    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 session_id:
+        logging.warning( "Aucun session_id fourni, arrêt du stream par sécurité.")
+        return
+
+    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
+
+    logging.info(f"[{session_id}] Préparation du segment audio depuis : {filepath_to_stream}")
+
+    try:
+        segment = AudioSegment.from_file(filepath_to_stream)
+        chunk_duree_ms = 1000
+        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}...")
+
+            if _is_stop_requested(stop_streaming_flags):
+                logging.info(f"[{session_id}]Signal d'arrêt reçu, arrêt de la boucle.")
+                break
+
+            output_chunk = (
+                chunk.frame_rate,
+                np.array(chunk.get_array_of_samples()).reshape(1, -1),
+            )
+
+            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"[{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é.")
+
+    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}")
+    except Exception as e:
+        logging.error(f"[{session_id}] Erreur pendant le stream: {e}", exc_info=True)
+        raise
+    finally:
+        if isinstance(stop_streaming_flags, dict):
+            stop_streaming_flags["stop"] = False
+        logging.info(f"[{session_id}]Signal d'arrêt nettoyé.")
+
+
+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.")
+    if not isinstance(stop_streaming_flags, dict):
+        stop_streaming_flags = {"stop": True}
+    else:
+        stop_streaming_flags["stop"] = True
+    return stop_streaming_flags
+
+# --- Interface Gradio ---
+
+with gr.Blocks(theme=gr.themes.Soft()) as demo:
+
+    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)
+
+    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."
+    )
+
+    # 1. État pour stocker le chemin du fichier à lire
+    active_filepath = gr.State(value=DEFAULT_FILE)
+
+    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="Stream Audio",
+                mode="receive",
+                modality="audio",
+                rtc_configuration=generate_coturn_config(),
+                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]
+    )
+
+
+    # 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, 
+    ]
+
+    stream_event = webrtc_stream.stream(
+        fn=read_and_stream_audio,
+        inputs=[active_filepath, session_hash, stop_streaming_flags],  
+        outputs=[webrtc_stream],
+        trigger=start_button.click,
+        concurrency_id="audio_stream", # ID de concurrence
+        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]
+    )
+
+    # Correction : S'assurer que le stream est bien annulé
+    stop_button.click(
+        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
+    )
+    # --- Active sessions ---
+    with gr.Accordion("📊 Active Sessions", open=False):
+        sessions_table = gr.DataFrame(
+            headers=["session_hash", "file", "start_time", "status"],
+            interactive=False,
+            wrap=True,
+            max_height=200,
+        )
+
+    gr.Timer(3.0).tick(fn=get_active_sessions, outputs=sessions_table)
+
+if __name__ == "__main__":
+    demo.queue(max_size=10, api_open=False).launch(show_api=False, debug=True)