Merge stream_utils and utils files

app.py

@@ -13,7 +13,11 @@ import os
 from gradio.utils import get_space
 
 from app.utils import (
-    raise_function
+    raise_function,
+    generate_coturn_config,
+    read_and_stream_audio,
+    stop_streaming,
+    task
 )
 from app.session_utils import (
     on_load,
@@ -36,13 +40,6 @@ from app.ui_utils import (
     on_file_load
 )
 
-from app.stream_utils import (
-    generate_coturn_config,
-    read_and_stream_audio,
-    stop_streaming,
-    task
-)
-
 # --------------------------------------------------------
 # Initialization
 # --------------------------------------------------------
@@ -219,7 +216,6 @@ with gr.Blocks(theme=theme, css=css_style) as demo:
         # === STEP 4 ===
         with gr.Step("Task", id=3) as task_step:
             gr.Markdown("## Step 4: Start the Task")
-
             with gr.Group():
                 with gr.Column():
                     status_slider = gr.Slider(
@@ -235,8 +231,10 @@ with gr.Blocks(theme=theme, css=css_style) as demo:
                         label="Transcription / Translation Result",
                         placeholder="Waiting for output...",
                         lines=10,
+                        max_lines= 10,
                         interactive=False,
-                        visible=True
+                        visible=True,
+                        autoscroll=True
                     )
 
                     start_task_button = gr.Button("Start Task", visible=True)

app/stream_utils.py

@@ -1,231 +0,0 @@
-
-from app.logger_config import logger as logging
-from fastrtc.utils import AdditionalOutputs
-from pydub import AudioSegment
-import asyncio
-import os
-import time
-import numpy as np
-import spaces
-import hmac
-import hashlib
-import base64
-import os
-import time
-import random
-
-from app.session_utils import (
-    get_active_task_flag_file,
-    get_folder_chunks
-)
-
-# --------------------------------------------------------
-# Utility functions
-# --------------------------------------------------------
-def generate_coturn_config():
-    """
-    Génère une configuration Coturn complète avec authentification dynamique (use-auth-secret).
-    Returns:
-        dict: Objet coturn_config prêt à être utilisé côté client WebRTC.
-    """
-
-    secret_key = os.getenv("TURN_SECRET_KEY", "your_secret_key")
-    ttl = int(os.getenv("TURN_TTL", 3600))
-    turn_url = os.getenv("TURN_URL", "turn:*******")
-    turn_s_url = os.getenv("TURN_S_URL", "turns:*****")
-    user = os.getenv("TURN_USER", "client")
-
-    timestamp = int(time.time()) + ttl
-    username = f"{timestamp}:{user}"
-    password = base64.b64encode(
-        hmac.new(secret_key.encode(), username.encode(), hashlib.sha1).digest()
-    ).decode()
- 
-    coturn_config = {
-        "iceServers": [
-            {
-                "urls": [
-                    f"{turn_url}",
-                    f"{turn_s_url}",
-                ],
-                "username": username,
-                "credential": password,
-            }
-        ]
-    }
-    return coturn_config
-
-
-
-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
-    transcribe_flag = get_active_task_flag_file(session_id)
-    try:
-        segment = AudioSegment.from_file(filepath_to_stream)
-        chunk_duration_ms = 1000
-        total_chunks = len(segment) // chunk_duration_ms + 1
-        logging.info(f"[{session_id}] Starting audio streaming {filepath_to_stream} ({total_chunks} chunks).")
-
-        for i, chunk in enumerate(segment[::chunk_duration_ms]):
-         
-
-            frame_rate = chunk.frame_rate
-            samples = np.array(chunk.get_array_of_samples()).reshape(1, -1)
-            progress = round(((i + 1) / total_chunks) * 100, 2)
-            if _is_stop_requested(stop_streaming_flags):
-                logging.info(f"[{session_id}] Stop signal received. Terminating stream.")
-                yield ((frame_rate, samples), AdditionalOutputs({"stoped": True, "value": "STREAM_STOPED"} ) )
-                break
-            
-            yield ((frame_rate, samples), AdditionalOutputs({"progressed": True, "value": progress} ))
-            logging.debug(f"[{session_id}] Sent chunk {i+1}/{total_chunks} ({progress}%).")
-
-            time.sleep(1)
-            #  Save only if transcription is active
-            if os.path.exists(transcribe_flag) :
-                chunk_dir = get_folder_chunks(session_id)
-                if not os.path.exists(chunk_dir) : 
-                    os.makedirs(chunk_dir, exist_ok=True)
-                npz_path = os.path.join(chunk_dir, f"chunk_{i:05d}.npz")
-                chunk_array = np.array(chunk.get_array_of_samples(), dtype=np.int16)
-                np.savez_compressed(npz_path, data=chunk_array, rate=frame_rate)
-                logging.debug(f"[{session_id}] Saved chunk {i}/{total_chunks} (transcribe active)")
-
-            # raise_function()  # Optional injected test exception
-
-        logging.info(f"[{session_id}] Audio streaming completed successfully.")
-
-    except asyncio.CancelledError:
-        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:
-        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}] Stop flag reset.")
-
-
-
-def handle_stream_error(session_id: str, error: Exception | str, stop_streaming_flags: dict | None = None):
-    """
-    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)
-
-    logging.error(f"[{session_id}] Streaming error: {msg}", exc_info=isinstance(error, Exception))
-
-    if isinstance(stop_streaming_flags, dict):
-        stop_streaming_flags["stop"] = False
-
-    yield ((16000,np.zeros(16000, dtype=np.float32).reshape(1, -1)), AdditionalOutputs({"errored": True, "value": msg}))
-    
-
-
-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))
-
-
-@spaces.GPU
-def task(session_id: str):
-    """Continuously read and delete .npz chunks while task is active."""
-    active_flag = get_active_task_flag_file(session_id)
-    with open(active_flag, "w") as f:
-        f.write("1")
-    chunk_dir = get_folder_chunks(session_id)
-    logging.info(f"[{session_id}] task started. {chunk_dir}")
-
-
-    try:
-        logging.info(f"[{session_id}] task loop started.")
-        yield f"Task started for session {session_id}\n\n"
-        while os.path.exists(active_flag):
-            if not os.path.exists(chunk_dir):
-                logging.warning(f"[{session_id}] No chunk directory found for task.")
-                yield "No audio chunks yet... waiting for stream.\n"
-                time.sleep(0.25)
-                continue
-            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"
-                    yield f"{text}\n"
-                    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}")
-                    yield f"Error processing {fname}: {e}\n"
-                    continue
-
-            time.sleep(0.25)
-            # raise_function()
-        yield "\nTask stopped by user or stream ended.\n"
-        logging.info(f"[{session_id}] task loop ended (flag removed).")
-
-    except Exception as e:
-        logging.error(f"[{session_id}] task error: {e}", exc_info=True)
-        yield f"Unexpected error: {e}\n"
-    finally:
-        # active_flag = os.path.join(TMP_DIR, f"transcribe_active_{session_id}.txt")
-        if os.path.exists(active_flag):
-            os.remove(active_flag)
-        logging.info(f"[{session_id}] task stopped.")
-        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}")
-            yield "\nCleanup error: {e}"
-        logging.info(f"[{session_id}] Exiting task loop.")
-        yield "\nTask finished and cleaned up.\n"
-
-
-
-
-def stop_streaming(session_id: str, stop_streaming_flags: dict):
-    """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 raise_function():
-    """Raise an error randomly (1 out of 10 times)."""
-    if random.randint(1, 10) == 1:
-        raise RuntimeError("Random failure triggered!")

app/utils.py

@@ -1,6 +1,244 @@
-import torch
 from app.logger_config import logger as logging
+from fastrtc.utils import AdditionalOutputs
+from pydub import AudioSegment
+import asyncio
+import os
+import time
+import numpy as np
+import spaces
+import hmac
+import hashlib
+import base64
+import os
+import time
 import random
+
+from app.session_utils import (
+    get_active_task_flag_file,
+    get_folder_chunks
+)
+
+
+# --------------------------------------------------------
+# Utility functions
+# --------------------------------------------------------
+def generate_coturn_config():
+    """
+    Génère une configuration Coturn complète avec authentification dynamique (use-auth-secret).
+    Returns:
+        dict: Objet coturn_config prêt à être utilisé côté client WebRTC.
+    """
+
+    secret_key = os.getenv("TURN_SECRET_KEY", "your_secret_key")
+    ttl = int(os.getenv("TURN_TTL", 3600))
+    turn_url = os.getenv("TURN_URL", "turn:*******")
+    turn_s_url = os.getenv("TURN_S_URL", "turns:*****")
+    user = os.getenv("TURN_USER", "client")
+
+    timestamp = int(time.time()) + ttl
+    username = f"{timestamp}:{user}"
+    password = base64.b64encode(
+        hmac.new(secret_key.encode(), username.encode(), hashlib.sha1).digest()
+    ).decode()
+ 
+    coturn_config = {
+        "iceServers": [
+            {
+                "urls": [
+                    f"{turn_url}",
+                    f"{turn_s_url}",
+                ],
+                "username": username,
+                "credential": password,
+            }
+        ]
+    }
+    return coturn_config
+
+
+
+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
+    transcribe_flag = get_active_task_flag_file(session_id)
+    try:
+        segment = AudioSegment.from_file(filepath_to_stream)
+        chunk_duration_ms = 1000
+        total_chunks = len(segment) // chunk_duration_ms + 1
+        logging.info(f"[{session_id}] Starting audio streaming {filepath_to_stream} ({total_chunks} chunks).")
+
+        for i, chunk in enumerate(segment[::chunk_duration_ms]):
+         
+
+            frame_rate = chunk.frame_rate
+            samples = np.array(chunk.get_array_of_samples()).reshape(1, -1)
+            progress = round(((i + 1) / total_chunks) * 100, 2)
+            if _is_stop_requested(stop_streaming_flags):
+                logging.info(f"[{session_id}] Stop signal received. Terminating stream.")
+                yield ((frame_rate, samples), AdditionalOutputs({"stoped": True, "value": "STREAM_STOPED"} ) )
+                break
+            
+            yield ((frame_rate, samples), AdditionalOutputs({"progressed": True, "value": progress} ))
+            logging.debug(f"[{session_id}] Sent chunk {i+1}/{total_chunks} ({progress}%).")
+
+            time.sleep(1)
+            #  Save only if transcription is active
+            if os.path.exists(transcribe_flag) :
+                chunk_dir = get_folder_chunks(session_id)
+                if not os.path.exists(chunk_dir) : 
+                    os.makedirs(chunk_dir, exist_ok=True)
+                npz_path = os.path.join(chunk_dir, f"chunk_{i:05d}.npz")
+                chunk_array = np.array(chunk.get_array_of_samples(), dtype=np.int16)
+                np.savez_compressed(npz_path, data=chunk_array, rate=frame_rate)
+                logging.debug(f"[{session_id}] Saved chunk {i}/{total_chunks} (transcribe active)")
+
+            # raise_function()  # Optional injected test exception
+
+        logging.info(f"[{session_id}] Audio streaming completed successfully.")
+
+    except asyncio.CancelledError:
+        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:
+        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}] Stop flag reset.")
+
+
+
+def handle_stream_error(session_id: str, error: Exception | str, stop_streaming_flags: dict | None = None):
+    """
+    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)
+
+    logging.error(f"[{session_id}] Streaming error: {msg}", exc_info=isinstance(error, Exception))
+
+    if isinstance(stop_streaming_flags, dict):
+        stop_streaming_flags["stop"] = False
+
+    yield ((16000,np.zeros(16000, dtype=np.float32).reshape(1, -1)), AdditionalOutputs({"errored": True, "value": msg}))
+    
+
+
+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))
+
+# --- Decorator compatibility layer ---
+if os.environ.get("SPACE_ID", "").startswith("zero-gpu"):
+    logging.warning("Running on ZeroGPU — disabling @spaces.GPU")
+    def gpu_decorator(f): return f
+else:
+    gpu_decorator = spaces.GPU
+
+# --- Audio Stream Function ---
+@gpu_decorator
+def task(session_id: str):
+    """Continuously read and delete .npz chunks while task is active."""
+    active_flag = get_active_task_flag_file(session_id)
+    with open(active_flag, "w") as f:
+        f.write("1")
+    chunk_dir = get_folder_chunks(session_id)
+    logging.info(f"[{session_id}] task started. {chunk_dir}")
+
+
+    try:
+        logging.info(f"[{session_id}] task loop started.")
+        yield f"Task started for session {session_id}\n\n"
+        while os.path.exists(active_flag):
+            if not os.path.exists(chunk_dir):
+                logging.warning(f"[{session_id}] No chunk directory found for task.")
+                yield "No audio chunks yet... waiting for stream.\n"
+                time.sleep(0.25)
+                continue
+            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"
+                    yield f"{text}\n"
+                    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}")
+                    yield f"Error processing {fname}: {e}\n"
+                    continue
+
+            time.sleep(0.25)
+            # raise_function()
+        yield "\nTask stopped by user or stream ended.\n"
+        logging.info(f"[{session_id}] task loop ended (flag removed).")
+
+    except Exception as e:
+        logging.error(f"[{session_id}] task error: {e}", exc_info=True)
+        yield f"Unexpected error: {e}\n"
+    finally:
+        # active_flag = os.path.join(TMP_DIR, f"transcribe_active_{session_id}.txt")
+        if os.path.exists(active_flag):
+            os.remove(active_flag)
+        logging.info(f"[{session_id}] task stopped.")
+        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}")
+            yield "\nCleanup error: {e}"
+        logging.info(f"[{session_id}] Exiting task loop.")
+        yield "\nTask finished and cleaned up.\n"
+
+
+
+
+def stop_streaming(session_id: str, stop_streaming_flags: dict):
+    """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 raise_function():
+    """Raise an error randomly (1 out of 10 times)."""
+    if random.randint(1, 10) == 1:
+        raise RuntimeError("Random failure triggered!")
+
+
+
 def debug_current_device():
     """Safely logs GPU or CPU information without crashing on stateless GPU."""
     logging.debug("=== Debugging current device ===")
@@ -49,9 +287,6 @@ def get_current_device():
 
 
 
-
-
-
 def raise_function():
     """Raise an error randomly (1 out of 10 times)."""
     if random.randint(1, 50) == 1: