Upload 4 files

app.py

@@ -0,0 +1,419 @@
+# SPDX-License-Identifier: Apache-2.0
+
+import asyncio
+import base64
+import json
+import os
+import queue
+import threading
+import time
+
+import gradio as gr
+import numpy as np
+import websockets
+
+# Load Voxtral icon as base64
+VOXTRAL_ICON_B64 = ""
+icon_path = os.path.join(os.path.dirname(__file__), "assets", "voxtral.png")
+if os.path.exists(icon_path):
+    with open(icon_path, "rb") as f:
+        VOXTRAL_ICON_B64 = base64.b64encode(f.read()).decode("utf-8")
+
+SAMPLE_RATE = 16_000
+WARMUP_DURATION = 2.0  # seconds of silence for warmup
+WPM_WINDOW = 10  # seconds for running mean calculation
+CALIBRATION_PERIOD = 5  # seconds before showing WPM
+SESSION_TIMEOUT = 300  # 5 minutes session timeout
+
+# Global config (shared across users)
+ws_url = ""
+model = ""
+
+
+class UserSession:
+    """Per-user session state."""
+    def __init__(self):
+        self.audio_queue = queue.Queue(maxsize=100)  # Limit queue size
+        self.transcription_text = ""
+        self.is_running = False
+        self.status_message = "ready"
+        self.word_timestamps = []
+        self.current_wpm = "Calibrating..."
+        self.session_start_time = None
+
+
+# Load CSS from external file
+css_path = os.path.join(os.path.dirname(__file__), "style.css")
+with open(css_path, "r") as f:
+    CUSTOM_CSS = f.read()
+
+
+def get_header_html() -> str:
+    """Generate the header HTML with Voxtral logo."""
+    if VOXTRAL_ICON_B64:
+        logo_html = f'<img src="data:image/png;base64,{VOXTRAL_ICON_B64}" alt="Voxtral" class="header-logo" />'
+    else:
+        logo_html = ''
+    
+    return f"""
+    <div class="header-card">
+        <h1 class="header-title">{logo_html}Real-time Speech Transcription</h1>
+        <p class="header-subtitle">Powered by Voxtral Mini — Click the microphone to start recording</p>
+    </div>
+    """
+
+
+def get_status_html(status: str) -> str:
+    """Generate status badge HTML based on current status."""
+    status_configs = {
+        "ready": ("STANDBY", "status-ready", ""),
+        "connecting": ("CONNECTING", "status-connecting", "fast"),
+        "warming": ("WARMING UP", "status-warming", "fast"),
+        "listening": ("LISTENING", "status-listening", "animate"),
+        "timeout": ("TIMEOUT", "status-timeout", ""),
+        "error": ("ERROR", "status-error", ""),
+    }
+    label, css_class, dot_class = status_configs.get(status, status_configs["ready"])
+    dot_anim = f" {dot_class}" if dot_class else ""
+    
+    return f"""<div class="status-badge {css_class}"><span class="status-dot{dot_anim}"></span><span style="color: inherit !important;">{label}</span></div>"""
+
+
+def get_transcription_html(transcript: str, status: str, wpm: str = "Calibrating...") -> str:
+    """Generate the full transcription card HTML."""
+    status_badge = get_status_html(status)
+    wpm_badge = f'<div class="wpm-badge"><span style="color: #1E1E1E !important;">{wpm}</span></div>'
+    
+    if transcript:
+        cursor_html = '<span class="transcript-cursor"></span>' if status == "listening" else ""
+        content_html = f"""
+        <div class="transcript-text" style="color: #000000 !important;">
+            {transcript}{cursor_html}
+        </div>
+        """
+    elif status in ["listening", "warming", "connecting"]:
+        content_html = """
+        <div class="empty-state">
+            <div class="empty-dots">
+                <div class="empty-dot"></div>
+                <div class="empty-dot"></div>
+                <div class="empty-dot"></div>
+            </div>
+            <p class="empty-text" style="color: #555555 !important;">Listening for audio...</p>
+        </div>
+        """
+    elif status == "timeout":
+        content_html = """
+        <div class="empty-state">
+            <p class="empty-text" style="color: #B30400 !important;">Session timeout (5 minutes)</p>
+            <p class="empty-text" style="color: #555555 !important;">Click 'Clear History' and refresh to restart.</p>
+        </div>
+        """
+    else:
+        content_html = """
+        <div class="empty-state">
+            <p class="empty-text" style="color: #555555 !important;">// Awaiting audio input...</p>
+            <p class="empty-text" style="color: #555555 !important;">// Click the microphone to start.</p>
+        </div>
+        """
+    
+    # Use base64 image if available
+    if VOXTRAL_ICON_B64:
+        icon_html = f'<img src="data:image/png;base64,{VOXTRAL_ICON_B64}" alt="Voxtral" class="voxtral-icon" />'
+    else:
+        icon_html = '<span style="font-size:20px;">🎙️</span>'
+    
+    return f"""
+    <div class="transcription-card">
+        <div class="card-header">
+            <div class="card-header-left">
+                {icon_html}
+                <span class="card-title" style="color: #1E1E1E !important;">Transcription Output</span>
+            </div>
+            <div class="card-header-right">
+                {wpm_badge}
+                {status_badge}
+            </div>
+        </div>
+        <div class="card-content">
+            {content_html}
+        </div>
+        <div class="card-footer">
+            <span style="color: #555555 !important;">Voxtral Mini</span>
+            <span style="color: #555555 !important;">Real-time Audio Transcription</span>
+        </div>
+    </div>
+    """
+
+
+def calculate_wpm(session):
+    """Calculate words per minute based on running mean of last WPM_WINDOW seconds."""
+    if session.session_start_time is not None:
+        elapsed = time.time() - session.session_start_time
+        if elapsed < CALIBRATION_PERIOD:
+            return "Calibrating..."
+    
+    if len(session.word_timestamps) < 2:
+        return "0.0 WPM"
+    
+    current_time = time.time()
+    cutoff_time = current_time - WPM_WINDOW
+    session.word_timestamps = [ts for ts in session.word_timestamps if ts >= cutoff_time]
+    
+    if len(session.word_timestamps) < 2:
+        return "0.0 WPM"
+    
+    time_span = current_time - session.word_timestamps[0]
+    if time_span == 0:
+        return "0.0 WPM"
+    
+    word_count = len(session.word_timestamps)
+    wpm = (word_count / time_span) * 60
+    return f"{round(wpm, 1)} WPM"
+
+
+async def send_silence(ws, duration=2.0):
+    """Send silence to warm up the model."""
+    num_samples = int(SAMPLE_RATE * duration)
+    silence = np.zeros(num_samples, dtype=np.int16)
+    
+    chunk_size = int(SAMPLE_RATE * 0.1)
+    for i in range(0, num_samples, chunk_size):
+        chunk = silence[i:i + chunk_size]
+        b64_chunk = base64.b64encode(chunk.tobytes()).decode("utf-8")
+        await ws.send(
+            json.dumps(
+                {"type": "input_audio_buffer.append", "audio": b64_chunk}
+            )
+        )
+        await asyncio.sleep(0.05)
+
+
+async def websocket_handler(session):
+    """Connect to WebSocket and handle audio streaming + transcription."""
+    try:
+        async with websockets.connect(ws_url) as ws:
+            await ws.recv()
+            await ws.send(json.dumps({"type": "session.update", "model": model}))
+
+            session.status_message = "warming"
+            await send_silence(ws, WARMUP_DURATION)
+            await ws.send(json.dumps({"type": "input_audio_buffer.commit"}))
+            session.status_message = "listening"
+
+            async def send_audio():
+                while session.is_running:
+                    try:
+                        if session.session_start_time is not None:
+                            elapsed = time.time() - session.session_start_time
+                            if elapsed >= SESSION_TIMEOUT:
+                                print("Session timeout reached (5 minutes)")
+                                session.is_running = False
+                                session.status_message = "timeout"
+                                break
+                        
+                        chunk = await asyncio.get_event_loop().run_in_executor(
+                            None, lambda: session.audio_queue.get(timeout=0.1)
+                        )
+                        if session.is_running:
+                            await ws.send(
+                                json.dumps(
+                                    {"type": "input_audio_buffer.append", "audio": chunk}
+                                )
+                            )
+                    except queue.Empty:
+                        continue
+                    except Exception as e:
+                        print(f"Error sending audio: {e}")
+                        break
+
+            async def receive_transcription():
+                try:
+                    async for message in ws:
+                        if not session.is_running:
+                            break
+                        
+                        if session.session_start_time is not None:
+                            elapsed = time.time() - session.session_start_time
+                            if elapsed >= SESSION_TIMEOUT:
+                                session.status_message = "timeout"
+                                session.is_running = False
+                                break
+                        
+                        data = json.loads(message)
+                        if data.get("type") == "transcription.delta":
+                            delta = data["delta"]
+                            session.transcription_text += delta
+                            
+                            words = delta.split()
+                            for _ in words:
+                                session.word_timestamps.append(time.time())
+                            
+                            session.current_wpm = calculate_wpm(session)
+                except Exception as e:
+                    print(f"Error receiving transcription: {e}")
+
+            await asyncio.gather(send_audio(), receive_transcription(), return_exceptions=True)
+    except Exception as e:
+        print(f"WebSocket connection error: {e}")
+        session.status_message = "error"
+
+
+def start_websocket(session):
+    """Start WebSocket connection in background thread."""
+    session.is_running = True
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    try:
+        loop.run_until_complete(websocket_handler(session))
+    except Exception as e:
+        print(f"WebSocket error: {e}")
+    finally:
+        try:
+            loop.close()
+        except Exception:
+            pass
+
+
+def auto_start_recording(session):
+    """Automatically start the transcription service when audio begins."""
+    if not session.is_running:
+        session.transcription_text = ""
+        session.word_timestamps = []
+        session.current_wpm = "Calibrating..."
+        session.session_start_time = time.time()
+        session.status_message = "connecting"
+        thread = threading.Thread(target=start_websocket, args=(session,), daemon=True)
+        thread.start()
+    
+    return get_transcription_html(session.transcription_text, session.status_message, session.current_wpm)
+
+
+def clear_history(session):
+    """Stop the websocket connection and clear all history."""
+    session.is_running = False
+    
+    # Clear the audio queue without blocking
+    try:
+        while True:
+            session.audio_queue.get_nowait()
+    except queue.Empty:
+        pass
+    
+    session.transcription_text = ""
+    session.word_timestamps = []
+    session.current_wpm = "Calibrating..."
+    session.session_start_time = None
+    session.status_message = "ready"
+    
+    return get_transcription_html("", "ready", "Calibrating..."), None
+
+
+def process_audio(audio, session):
+    """Process incoming audio and queue for streaming."""
+    try:
+        # Quick return if audio is None
+        if audio is None:
+            wpm = session.current_wpm if session.is_running else "Calibrating..."
+            return get_transcription_html(session.transcription_text, session.status_message, wpm)
+
+        # Auto-start if not running
+        if not session.is_running and session.status_message not in ["timeout", "error"]:
+            auto_start_recording(session)
+
+        # Skip processing if session stopped
+        if not session.is_running:
+            return get_transcription_html(session.transcription_text, session.status_message, session.current_wpm)
+
+        sample_rate, audio_data = audio
+
+        # Convert to mono if stereo
+        if len(audio_data.shape) > 1:
+            audio_data = audio_data.mean(axis=1)
+
+        # Normalize to float
+        if audio_data.dtype == np.int16:
+            audio_float = audio_data.astype(np.float32) / 32767.0
+        else:
+            audio_float = audio_data.astype(np.float32)
+
+        # Resample to 16kHz if needed
+        if sample_rate != SAMPLE_RATE:
+            num_samples = int(len(audio_float) * SAMPLE_RATE / sample_rate)
+            audio_float = np.interp(
+                np.linspace(0, len(audio_float) - 1, num_samples),
+                np.arange(len(audio_float)),
+                audio_float,
+            )
+
+        # Convert to PCM16 and base64 encode
+        pcm16 = (audio_float * 32767).astype(np.int16)
+        b64_chunk = base64.b64encode(pcm16.tobytes()).decode("utf-8")
+        
+        # Non-blocking put to queue
+        try:
+            session.audio_queue.put_nowait(b64_chunk)
+        except queue.Full:
+            pass  # Skip if queue is full
+
+        return get_transcription_html(session.transcription_text, session.status_message, session.current_wpm)
+    except Exception as e:
+        print(f"Error processing audio: {e}")
+        return get_transcription_html(session.transcription_text, session.status_message, session.current_wpm)
+
+
+# Gradio interface
+with gr.Blocks(title="Voxtral Real-time Transcription") as demo:
+    # Create session state
+    session_state = gr.State(value=UserSession)
+    
+    # Header
+    gr.HTML(get_header_html())
+    
+    # Transcription output
+    transcription_display = gr.HTML(
+        value=get_transcription_html("", "ready", "Calibrating..."),
+        elem_id="transcription-output"
+    )
+    
+    # Audio input
+    audio_input = gr.Audio(
+        sources=["microphone"],
+        streaming=True,
+        type="numpy",
+        format="wav",
+        elem_id="audio-input",
+        label="Microphone Input"
+    )
+    
+    # Clear button
+    clear_btn = gr.Button(
+        "Clear History",
+        elem_classes=["clear-btn"]
+    )
+    
+    # Info text
+    gr.HTML('<p class="info-text">Click "Clear History" and refresh the page to start a new session.</p>')
+
+    # Event handlers
+    clear_btn.click(
+        clear_history,
+        inputs=[session_state],
+        outputs=[transcription_display, audio_input]
+    )
+    
+    audio_input.stream(
+        process_audio,
+        inputs=[audio_input, session_state],
+        outputs=[transcription_display],
+        show_progress="hidden",
+        concurrency_limit=None,
+    )
+
+model = os.environ.get("MODEL", "mistralai/Voxtral-Mini-4B-Realtime-2602")
+host = os.environ.get("HOST", "")
+
+ws_url = f"wss://{host}/v1/realtime"
+
+demo.queue(default_concurrency_limit=20)
+demo.launch(css=CUSTOM_CSS, theme=gr.themes.Base(), ssr_mode=False, max_threads=40)

requirements.txt

@@ -0,0 +1,3 @@
+gradio>=4.0.0
+websockets
+numpy

style.css

@@ -0,0 +1,297 @@
+@import url('https://fonts.googleapis.com/css2?family=JetBrains+Mono:wght@400;500;600&family=Inter:wght@400;500;600;700&display=swap');
+
+body, .gradio-container {
+    background-color: #FFFAEB !important;
+    background-image: 
+        linear-gradient(#E9E2CB 1px, transparent 1px), 
+        linear-gradient(90deg, #E9E2CB 1px, transparent 1px) !important;
+    background-size: 40px 40px !important;
+    font-family: 'Inter', sans-serif !important;
+}
+
+@keyframes pulse {
+    0%, 100% { opacity: 1; transform: scale(1); }
+    50% { opacity: 0.5; transform: scale(0.8); }
+}
+
+@keyframes blink {
+    0%, 100% { opacity: 1; }
+    50% { opacity: 0; }
+}
+
+@keyframes bounce {
+    0%, 100% { transform: translateY(0); }
+    50% { transform: translateY(-8px); }
+}
+
+.header-card {
+    background: linear-gradient(135deg, #FFFAEB 0%, #FFF0C3 100%);
+    border: 2px solid #E9E2CB;
+    border-top: 4px solid #FF8205;
+    padding: 2rem;
+    margin-bottom: 1.5rem;
+    box-shadow: 0 4px 24px rgba(0,0,0,0.06);
+}
+
+.header-title {
+    font-size: 2rem;
+    font-weight: 700;
+    color: #1E1E1E !important;
+    margin: 0 0 0.5rem 0;
+    letter-spacing: -0.02em;
+    display: flex;
+    align-items: center;
+    gap: 0.75rem;
+}
+
+.header-logo {
+    width: 40px;
+    height: 40px;
+    object-fit: contain;
+}
+
+.header-subtitle {
+    color: #444444 !important;
+    font-size: 1rem;
+    margin: 0;
+}
+
+.transcription-card {
+    background: #FFFAEB !important;
+    border: 2px solid #E9E2CB;
+    box-shadow: 0 8px 32px rgba(0,0,0,0.08);
+    overflow: hidden;
+    min-height: 350px;
+}
+
+.card-header {
+    background: rgba(255,255,255,0.6) !important;
+    border-bottom: 1px solid #E9E2CB;
+    padding: 0.75rem 1rem;
+    display: flex;
+    align-items: center;
+    justify-content: space-between;
+}
+
+.card-header-left {
+    display: flex;
+    align-items: center;
+    gap: 0.75rem;
+}
+
+.card-header-right {
+    display: flex;
+    align-items: center;
+    gap: 1rem;
+}
+
+.card-title {
+    font-size: 0.65rem;
+    font-weight: 700;
+    color: #1E1E1E !important;
+    text-transform: uppercase;
+    letter-spacing: 0.1em;
+    margin: 0;
+}
+
+.voxtral-icon {
+    width: 24px;
+    height: 24px;
+    object-fit: contain;
+}
+
+.wpm-badge {
+    display: inline-flex;
+    align-items: center;
+    gap: 0.4rem;
+    padding: 0.25rem 0.6rem;
+    font-size: 0.6rem;
+    font-weight: 700;
+    text-transform: uppercase;
+    letter-spacing: 0.05em;
+    border-radius: 2px;
+    background: #FFF0C3;
+    color: #1E1E1E !important;
+    border: 1px solid #E9E2CB;
+    font-family: 'JetBrains Mono', monospace;
+}
+
+.status-badge {
+    display: inline-flex;
+    align-items: center;
+    gap: 0.5rem;
+    padding: 0.25rem 0.75rem;
+    font-size: 0.6rem;
+    font-weight: 700;
+    text-transform: uppercase;
+    letter-spacing: 0.08em;
+    border-radius: 2px;
+}
+
+.status-ready {
+    background: #f5f5f5 !important;
+    color: #555555 !important;
+    border: 1px solid #E9E2CB;
+}
+
+.status-connecting, .status-warming {
+    background: rgba(255, 130, 5, 0.15) !important;
+    color: #CC6A04 !important;
+    border: 1px solid #FF8205;
+}
+
+.status-listening {
+    background: rgba(255, 130, 5, 0.2) !important;
+    color: #CC6A04 !important;
+    border: 1px solid #FF8205;
+}
+
+.status-timeout {
+    background: rgba(225, 5, 0, 0.1) !important;
+    color: #B30400 !important;
+    border: 1px solid #E10500;
+}
+
+.status-error {
+    background: rgba(225, 5, 0, 0.1) !important;
+    color: #B30400 !important;
+    border: 1px solid #E10500;
+}
+
+.status-dot {
+    width: 6px;
+    height: 6px;
+    border-radius: 50%;
+    background: currentColor !important;
+}
+
+.status-dot.animate {
+    animation: pulse 1.5s ease-in-out infinite;
+}
+
+.status-dot.fast {
+    animation: pulse 0.8s ease-in-out infinite;
+}
+
+.card-content {
+    padding: 1.5rem;
+    min-height: 250px;
+    position: relative;
+    background-color: #FFFAEB !important;
+    background-image: 
+        linear-gradient(rgba(0,0,0,0.02) 1px, transparent 1px), 
+        linear-gradient(90deg, rgba(0,0,0,0.02) 1px, transparent 1px);
+    background-size: 20px 20px;
+}
+
+.transcript-text {
+    font-family: 'JetBrains Mono', monospace !important;
+    font-size: 1.1rem !important;
+    line-height: 1.8 !important;
+    color: #000000 !important;
+    white-space: pre-wrap;
+    word-break: break-word;
+}
+
+.transcript-cursor {
+    display: inline-block;
+    width: 10px;
+    height: 20px;
+    background: #FF8205 !important;
+    margin-left: 4px;
+    vertical-align: middle;
+    animation: blink 1s step-end infinite;
+}
+
+.empty-state {
+    display: flex;
+    flex-direction: column;
+    align-items: center;
+    justify-content: center;
+    height: 200px;
+}
+
+.empty-dots {
+    display: flex;
+    gap: 6px;
+    margin-bottom: 1rem;
+}
+
+.empty-dot {
+    width: 8px;
+    height: 8px;
+    border-radius: 50%;
+    background: #FF8205 !important;
+    animation: bounce 1s ease-in-out infinite;
+}
+
+.empty-dot:nth-child(2) { animation-delay: 0.1s; }
+.empty-dot:nth-child(3) { animation-delay: 0.2s; }
+
+.empty-text {
+    font-family: 'JetBrains Mono', monospace !important;
+    font-size: 0.875rem !important;
+    color: #555555 !important;
+    font-style: italic;
+}
+
+.card-footer {
+    background: #fafafa !important;
+    border-top: 1px solid #E9E2CB;
+    padding: 0.5rem 1rem;
+    display: flex;
+    justify-content: space-between;
+    font-family: 'JetBrains Mono', monospace;
+    font-size: 0.65rem;
+    color: #555555 !important;
+}
+
+/* Audio component styling - preserve Gradio defaults */
+#audio-input {
+    margin-top: 1rem;
+}
+
+#audio-input button {
+    color: inherit !important;
+}
+
+#audio-input .controls button {
+    background: var(--button-secondary-background-fill) !important;
+    color: var(--button-secondary-text-color) !important;
+}
+
+.clear-btn {
+    background: #E10500 !important;
+    color: #FFFFFF !important;
+    border: none !important;
+    padding: 0.75rem 1.5rem !important;
+    font-weight: 600 !important;
+    text-transform: uppercase !important;
+    letter-spacing: 0.05em !important;
+    font-size: 0.75rem !important;
+    cursor: pointer !important;
+    transition: all 0.2s !important;
+    border-radius: 0 !important;
+    margin-top: 1rem !important;
+}
+
+.clear-btn:hover {
+    background: #B30400 !important;
+}
+
+footer {
+    display: none !important;
+}
+
+.gradio-container .prose {
+    max-width: none !important;
+}
+
+.info-text {
+    font-family: 'JetBrains Mono', monospace !important;
+    font-size: 0.75rem !important;
+    color: #888888 !important;
+    font-style: italic;
+    text-align: center;
+    margin-top: 1rem;
+}