add progress_stream

app.py

@@ -6,10 +6,9 @@ from fastrtc.webrtc import WebRTC
 from pydub import AudioSegment
 import time
 import os
+import json
 import spaces
 from app.utils import generate_coturn_config
-
-# ✅ Import des fonctions externalisées
 from app.session_utils import (
     generate_session_id,
     register_session,
@@ -18,20 +17,19 @@ from app.session_utils import (
     stop_file_path,
     create_stop_flag,
     clear_stop_flag,
+    reset_active_sessions,
 )
 
-# --- Constants ---
+# Reset active sessions at startup
+reset_active_sessions()
+
 EXAMPLE_FILES = ["data/bonjour.wav", "data/bonjour2.wav"]
 DEFAULT_FILE = EXAMPLE_FILES[0]
 
 
-# --- Streaming logic ---
 @spaces.GPU
 def read_and_stream_audio(filepath_to_stream: str, session_id: str):
-    """
-    Stream audio chunks for a specific session.
-    Stops when the corresponding stop flag file exists.
-    """
+    """Stream audio chunks for a specific session."""
     stop_file = stop_file_path(session_id)
     logging.debug(f"[{session_id}] read_and_stream_audio() started with file: {filepath_to_stream}")
 
@@ -41,28 +39,31 @@ def read_and_stream_audio(filepath_to_stream: str, session_id: str):
             filepath_to_stream = DEFAULT_FILE
             logging.warning(f"[{session_id}] Using default file: {DEFAULT_FILE}")
         else:
-            logging.error(f"[{session_id}] Default file missing. Abort stream.")
+            logging.error(f"[{session_id}] Default file missing. Aborting.")
             return
 
-    # Clear any leftover stop flag
     clear_stop_flag(session_id)
-
-    # Register session
     register_session(session_id, filepath_to_stream)
+    progress_path = f"/tmp/progress_{session_id}.json"
 
     try:
         segment = AudioSegment.from_file(filepath_to_stream)
         chunk_ms = 1000
-        logging.info(f"[{session_id}] Starting streaming ({chunk_ms}ms chunks)...")
+        total_chunks = len(segment) // chunk_ms + 1
+        logging.info(f"[{session_id}] Streaming {total_chunks} chunks...")
 
-        for i, chunk in enumerate(segment[::chunk_ms]):
+        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+1}. Ending stream.")
+                logging.info(f"[{session_id}] Stop flag detected at chunk {i}. Stopping.")
                 clear_stop_flag(session_id)
                 break
 
             iter_start = time.perf_counter()
-            logging.debug(f"[{session_id}] Sending chunk {i+1}...")
+            logging.debug(f"[{session_id}] Sending chunk {i}/{total_chunks}...")
+
+            progress_data = {"value": i / total_chunks, "text": f"Streaming... ({i}/{total_chunks})"}
+            with open(progress_path, "w") as f:
+                json.dump(progress_data, f)
 
             output_chunk = (
                 chunk.frame_rate,
@@ -71,81 +72,104 @@ def read_and_stream_audio(filepath_to_stream: str, session_id: str):
             yield output_chunk
 
             process_ms = (time.perf_counter() - iter_start) * 1000
-            sleep_s = max((chunk_ms / 1000.0) - (process_ms / 1000.0) - 0.1, 0.01)
-            logging.debug(f"[{session_id}] Chunk {i+1} processed in {process_ms:.2f}ms, sleep={sleep_s:.2f}s")
-            time.sleep(sleep_s)
+            time.sleep(max((chunk_ms / 1000.0) - (process_ms / 1000.0) - 0.1, 0.01))
+
+        with open(progress_path, "w") as f:
+            json.dump({"value": 1.0, "text": "Streaming completed ✅"}, f)
 
-        logging.info(f"[{session_id}] Stream finished normally.")
+        logging.info(f"[{session_id}] Stream completed successfully.")
 
     except asyncio.CancelledError:
         logging.info(f"[{session_id}] Stream cancelled by user.")
         raise
     except Exception as e:
-        logging.error(f"[{session_id}] Error during streaming: {e}", exc_info=True)
+        logging.error(f"[{session_id}] Stream error: {e}", exc_info=True)
         raise
     finally:
         unregister_session(session_id)
         clear_stop_flag(session_id)
-        logging.debug(f"[{session_id}] Exiting read_and_stream_audio().")
+        if os.path.exists(progress_path):
+            os.remove(progress_path)
+        logging.debug(f"[{session_id}] Stream closed.")
 
 
-# --- Stop Streaming ---
 def stop_streaming(session_id: str):
     create_stop_flag(session_id)
     logging.info(f"[{session_id}] Stop button clicked → stop flag created.")
     return None
 
 
-# --- Gradio UI ---
+def get_session_progress(session_id: str):
+    """Read progress data from /tmp/progress_<session_id>.json."""
+    progress_path = f"/tmp/progress_{session_id}.json"
+    if not os.path.exists(progress_path):
+        return 0.0, ""
+    try:
+        with open(progress_path, "r") as f:
+            data = json.load(f)
+        return data.get("value", 0.0), data.get("text", "")
+    except Exception as e:
+        logging.error(f"[{session_id}] Progress read error: {e}")
+        return 0.0, ""
+
+
 with gr.Blocks(theme=gr.themes.Soft()) as demo:
-    gr.Markdown(
-        "## Application 'Streamer' WebRTC (multi-utilisateur)\n"
-        "Chaque utilisateur contrôle son propre flux audio, avec suivi des sessions actives."
-    )
+    gr.Markdown("## 🎧 WebRTC Audio Streamer (Multi-user)\nEach user controls their own audio stream with live progress.")
 
-    # State
     session_id = gr.State(value=generate_session_id)
     active_filepath = gr.State(value=DEFAULT_FILE)
 
-    with gr.Row():
-        with gr.Column():
-            main_audio = gr.Audio(
-                label="Source Audio",
-                sources=["upload", "microphone"],
-                type="filepath",
-                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,
+                )
+
+                progress_bar = gr.Slider(
+                    label="Streaming Progress",
+                    minimum=0,
+                    maximum=1,
+                    value=0,
+                    step=0.01,
+                    interactive=False,
+                    visible=False,
+                )
+
+                progress_text = gr.Textbox(label="Stream Status", interactive=False, visible=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)
+
         with gr.Column():
             webrtc_stream = WebRTC(
-                label="Flux Audio",
+                label="Audio Stream",
                 mode="receive",
                 modality="audio",
                 rtc_configuration=generate_coturn_config(),
                 visible=True,
-                height=200,
             )
 
-    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()
-
-    # --- Audio selection ---
     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])
 
-    # --- UI state updates ---
     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),
             main_audio: gr.Audio(visible=False),
+            progress_bar: gr.Slider(value=0, visible=True),
+            progress_text: gr.Textbox(value="Streaming started...", visible=False),
         }
 
     def stop_streaming_ui(session_id):
@@ -154,17 +178,18 @@ with gr.Blocks(theme=gr.themes.Soft()) as demo:
             start_button: gr.Button(interactive=True),
             stop_button: gr.Button(interactive=False),
             main_audio: gr.Audio(
-                label="Source Audio",
+                label="Audio Source",
                 sources=["upload", "microphone"],
                 type="filepath",
                 value=DEFAULT_FILE,
                 visible=True,
             ),
+            progress_bar: gr.Slider(value=0, visible=False),
+            progress_text: gr.Textbox(value="", visible=False),
         }
 
-    ui_components = [start_button, stop_button, main_audio]
+    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],
@@ -174,37 +199,42 @@ with gr.Blocks(theme=gr.themes.Soft()) as demo:
         concurrency_limit=20,
     )
 
-    # --- Buttons ---
-    start_button.click(
-        fn=start_streaming_ui,
-        inputs=[session_id],
-        outputs=ui_components,
-    )
-
-    stop_button.click(
-        fn=stop_streaming,
-        inputs=[session_id],
-        outputs=[webrtc_stream],
-    ).then(
-        fn=stop_streaming_ui,
-        inputs=[session_id],
-        outputs=ui_components,
+    start_button.click(fn=start_streaming_ui, inputs=[session_id], outputs=ui_components)
+    stop_button.click(fn=stop_streaming, inputs=[session_id], outputs=[webrtc_stream]).then(
+        fn=stop_streaming_ui, inputs=[session_id], outputs=ui_components
     )
 
-    # --- Sessions Table ---
-    with gr.Accordion("📊 Sessions actives", open=False):
+    with gr.Accordion("📊 Active Sessions", open=False):
         sessions_table = gr.DataFrame(
             headers=["session_id", "file", "start_time", "status"],
             interactive=False,
             wrap=True,
-            label="Utilisateurs connectés",
+            label="Connected Users",
             max_height=200,
         )
 
-    # ✅ Timer pour mise à jour automatique du tableau
     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])
+
+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
 
 if __name__ == "__main__":
     demo.queue(max_size=50, api_open=False).launch(show_api=False, debug=True)

app/session_utils.py

@@ -7,17 +7,27 @@ from app.logger_config import logger as logging
 ACTIVE_SESSIONS_FILE = "/tmp/active_sessions.json"
 
 
-# --- ID Management ---
+def reset_active_sessions():
+    """Removes or clears the active sessions file at startup."""
+    try:
+        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 found to reset.")
+    except Exception as e:
+        logging.error(f"Error resetting active sessions: {e}")
+
+
 def generate_session_id() -> str:
-    """Génère un identifiant unique de session."""
+    """Generates a unique session ID."""
     sid = str(uuid.uuid4())
     logging.debug(f"[{sid}] New session created.")
     return sid
 
 
-# --- Active Session Registry ---
 def register_session(session_id: str, filepath: str):
-    """Ajoute une session au registre."""
+    """Registers a new session."""
     data = {}
     if os.path.exists(ACTIVE_SESSIONS_FILE):
         with open(ACTIVE_SESSIONS_FILE, "r") as f:
@@ -36,11 +46,11 @@ def register_session(session_id: str, filepath: str):
     with open(ACTIVE_SESSIONS_FILE, "w") as f:
         json.dump(data, f)
 
-    logging.debug(f"[{session_id}] Registered session in active_sessions.json.")
+    logging.debug(f"[{session_id}] Session registered in active_sessions.json.")
 
 
 def unregister_session(session_id: str):
-    """Retire une session du registre."""
+    """Removes a session from the registry."""
     if not os.path.exists(ACTIVE_SESSIONS_FILE):
         return
 
@@ -51,13 +61,13 @@ def unregister_session(session_id: str):
             data.pop(session_id)
             with open(ACTIVE_SESSIONS_FILE, "w") as f:
                 json.dump(data, f)
-            logging.debug(f"[{session_id}] Unregistered session.")
+            logging.debug(f"[{session_id}] Session unregistered.")
     except Exception as e:
         logging.error(f"[{session_id}] Error unregistering session: {e}")
 
 
 def get_active_sessions():
-    """Retourne les sessions actives sous forme de tableau pour le DataFrame."""
+    """Returns active sessions as a list of rows for the DataFrame."""
     if not os.path.exists(ACTIVE_SESSIONS_FILE):
         return []
 
@@ -65,29 +75,28 @@ def get_active_sessions():
         with open(ACTIVE_SESSIONS_FILE, "r") as f:
             data = json.load(f)
 
-        rows = []
-        for session in data.values():
-            rows.append([
-                session.get("session_id", ""),
-                session.get("file", ""),
-                session.get("start_time", ""),
-                session.get("status", ""),
-            ])
+        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"Erreur lecture sessions actives: {e}")
+        logging.error(f"Error reading active sessions: {e}")
         return []
 
 
-# --- Stop Flag Management ---
 def stop_file_path(session_id: str) -> str:
-    """Retourne le chemin du fichier stop d'une session."""
+    """Returns the stop-flag file path for a given session."""
     return f"/tmp/stream_stop_flag_{session_id}.txt"
 
 
 def create_stop_flag(session_id: str):
-    """Crée le fichier de stop pour cette session."""
+    """Creates a stop-flag file for this session."""
     path = stop_file_path(session_id)
     with open(path, "w") as f:
         f.write("1")
@@ -95,7 +104,7 @@ def create_stop_flag(session_id: str):
 
 
 def clear_stop_flag(session_id: str):
-    """Supprime le fichier de stop s'il existe."""
+    """Deletes the stop-flag file if it exists."""
     path = stop_file_path(session_id)
     if os.path.exists(path):
         os.remove(path)