add on_load and on_unload

app.py

@@ -19,11 +19,14 @@ from app.session_utils import (
     stop_file_path,
     create_stop_flag,
     clear_stop_flag,
-    reset_active_sessions,
+    reset_all_active_sessions,
+    on_load,
+    on_unload
+
 )
 
 # Reset sessions at startup
-reset_active_sessions()
+reset_all_active_sessions()
 
 EXAMPLE_FILES = ["data/bonjour.wav", "data/bonjour2.wav"]
 DEFAULT_FILE = EXAMPLE_FILES[0]
@@ -60,7 +63,7 @@ def read_and_stream_audio(filepath_to_stream: str, session_id: str, chunk_second
                 logging.info(f"[{session_id}] Stop flag detected at chunk {i}. Ending stream.")
                 clear_stop_flag(session_id)
                 break
-
+            logging.info(f"[{session_id}] Streaming chunk {i}.")
             iter_start = time.perf_counter()
 
             elapsed_s = i * chunk_seconds
@@ -75,7 +78,7 @@ def read_and_stream_audio(filepath_to_stream: str, session_id: str, chunk_second
             chunk_array = np.array(chunk.get_array_of_samples(), dtype=np.int16)
             rate = chunk.frame_rate
 
-            # ✅ Save only if transcription is active
+            #  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)
@@ -184,7 +187,10 @@ with gr.Blocks(theme=gr.themes.Soft()) as demo:
         "Each user controls their own stream. Transcription runs only during streaming."
     )
 
-    session_id = gr.State(value=generate_session_id())
+    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)
     active_filepath = gr.State(value=DEFAULT_FILE)
 
     with gr.Row(equal_height=True):
@@ -274,6 +280,8 @@ with gr.Blocks(theme=gr.themes.Soft()) as demo:
         inputs=[active_filepath, session_id, chunk_slider],
         outputs=[webrtc_stream],
         trigger=start_button.click,
+        concurrency_limit=20,
+        concurrency_id="receive"
     )
 
     start_button.click(fn=start_streaming_ui, inputs=[session_id], outputs=[
@@ -372,4 +380,5 @@ demo.css = custom_css
 # MAIN
 # --------------------------------------------------------
 if __name__ == "__main__":
-    demo.queue(max_size=50, api_open=False).launch(show_api=False, debug=True)
+
+    demo.queue(max_size=20, api_open=False).launch(show_api=False, debug=True)

app/session_utils.py

@@ -1,15 +1,74 @@
 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 = "/tmp/canary_aed_streaming"
-# TMP_DIR = "/home/sifar-dev/workspace/canary_aed_streaming/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."""
+    sid = request.session_hash  # ✅ Directly use session_hash as unique ID
+    sessions = _read_sessions()
+
+    sessions[sid] = {
+        "session_id": sid,
+        "file": "",
+        "start_time": datetime.utcnow().strftime("%H:%M:%S"),
+        "status": "active",
+    }
+
+    _write_sessions(sessions)
+    logging.info(f"[{sid}] Session registered (on_load).")
+
+    return sid, sid  # 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:
@@ -18,58 +77,108 @@ def ensure_tmp_dir():
         logging.error(f"Failed to create tmp directory {TMP_DIR}: {e}")
 
 
-def reset_active_sessions():
-    """Removes all temporary session files at startup."""
+def reset_all_active_sessions():
+    """Removes all temporary session files and folders at startup."""
     ensure_tmp_dir()
 
     try:
-        # Remove active sessions file
+        # --- 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 found to reset.")
+            logging.debug("No active sessions file found to reset.")
 
-        # Clean up old progress files
+        # --- Clean progress files ---
         for f in os.listdir(TMP_DIR):
             if f.startswith("progress_") and f.endswith(".json"):
+                path = os.path.join(TMP_DIR, f)
                 try:
-                    os.remove(os.path.join(TMP_DIR, f))
+                    os.remove(path)
                     logging.debug(f"Removed leftover progress file: {f}")
                 except Exception as e:
                     logging.warning(f"Failed to remove progress file {f}: {e}")
 
-        # Clean up old stop flag files
+        # --- Clean all flag files (stream + transcribe) ---
         for f in os.listdir(TMP_DIR):
-            if f.startswith("stream_stop_flag_") and f.endswith(".txt"):
+            if (
+                f.startswith("stream_stop_flag_")
+                or f.startswith("transcribe_stop_flag_")
+                or f.startswith("transcribe_active_")
+            ) and f.endswith(".txt"):
+                path = os.path.join(TMP_DIR, f)
                 try:
-                    os.remove(os.path.join(TMP_DIR, f))
-                    logging.debug(f"Removed leftover stop flag file: {f}")
+                    os.remove(path)
+                    logging.debug(f"Removed leftover 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"):
+                    logging.warning(f"Failed to remove flag file {f}: {e}")
+
+        # --- Clean chunk directories ---
+        for name in os.listdir(TMP_DIR):
+            path = os.path.join(TMP_DIR, name)
+            if os.path.isdir(path) and name.startswith("chunks_"):
                 try:
-                    os.remove(os.path.join(TMP_DIR, f))
-                    logging.debug(f"Removed leftover transcribe active flag file: {f}")
+                    shutil.rmtree(path)
+                    logging.debug(f"Removed leftover chunk folder: {name}")
                 except Exception as e:
-                    logging.warning(f"Failed to remove transcribe active file {f}: {e}")
-
+                    logging.warning(f"Failed to remove chunk folder {name}: {e}")
 
+        logging.info("Temporary session cleanup completed successfully.")
 
     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())