import asyncio
import sys
import time
from typing import Dict, Optional, Tuple, Any
from pathlib import Path
if sys.platform == 'win32':
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
import torch
import gradio as gr
import numpy as np
# Import backend ML & DSP components
from backend import (
SR,
TARGET_LEN,
run_prediction,
_to_mono_resampled,
DEVICE,
)
def get_empty_prediction_html(message: str = "Chưa có âm thanh được xử lý. Hãy tải lên file hoặc nói vào microphone.") -> str:
return f"""
🎙️
{message}
Đang sẵn sàng phân tích mẫu âm thanh 1 giây của bạn.
"""
def get_countdown_html(message: str, seconds: float) -> str:
return f"""
{message}
Sẵn sàng sau: {seconds:.1f} giây
"""
def format_prediction_as_html(pred_label: str, pred_conf: float, model_name: str, top_k_results: Dict[str, float]) -> str:
emojis = {
"five": "5️⃣", "zero": "0️⃣", "yes": "✅", "seven": "7️⃣", "no": "❌", "nine": "9️⃣",
"down": "⬇️", "one": "1️⃣", "go": "🚀", "two": "2️⃣", "stop": "🛑", "six": "6️⃣",
"unknown": "❓", "silence": "🔇"
}
emoji = emojis.get(pred_label.lower(), "🎙️")
# Alternative predictions list HTML
alt_html = ""
for label, prob in top_k_results.items():
pct = prob * 100
# Determine bar colors class based on probability
if prob > 0.7:
bar_class = "bar-high"
elif prob > 0.3:
bar_class = "bar-med"
else:
bar_class = "bar-low"
alt_html += f"""
{emojis.get(label.lower(), "🔸")} {label}
{pct:.1f}%
"""
conf_pct = pred_conf * 100
if pred_conf > 0.7:
conf_class = "conf-high"
elif pred_conf > 0.3:
conf_class = "conf-medium"
else:
conf_class = "conf-low"
html = f"""
"""
return html
IS_MODEL_WARMED_UP = False
def predict_from_audio(audio_input: Optional[Tuple[int, np.ndarray]], model_name: str, top_k: int = 5):
global IS_MODEL_WARMED_UP
if audio_input is None:
yield get_empty_prediction_html("Chưa nhận được tín hiệu âm thanh.")
return
sample_rate, audio = audio_input
if audio is None or len(audio) == 0:
yield get_empty_prediction_html("File âm thanh trống hoặc không hợp lệ.")
return
if not IS_MODEL_WARMED_UP:
# Show a 1.5 seconds countdown sequence only on the very first click
for remaining in [1.5, 1.0, 0.5]:
yield get_countdown_html("Đang kết nối thiết bị và khởi động mô hình AI...", remaining)
time.sleep(0.5)
# Warmup models globally
try:
from backend import MODELS
dummy_input = torch.zeros((1, 1, 40, 100), device=DEVICE)
for name, m in MODELS.items():
m(dummy_input)
except Exception as e:
print(f"Warmup error: {e}")
IS_MODEL_WARMED_UP = True
pred_label, pred_conf, top_k_results = run_prediction(audio_input, model_name, top_k)
yield format_prediction_as_html(pred_label, pred_conf, model_name, top_k_results)
# ============================================================
# Streaming / near-real-time mode
# ============================================================
def init_state() -> Dict[str, Any]:
return {
"buffer": np.zeros((0,), dtype=np.float32),
"sample_rate": np.array([SR], dtype=np.int32),
"warmup_steps": 0
}
def stream_predict(
audio_chunk: Optional[Tuple[int, np.ndarray]],
state: Optional[Dict[str, Any]],
model_name: str,
top_k: int,
) -> Tuple[Dict[str, Any], str]:
if state is None:
state = init_state()
if audio_chunk is None:
state["warmup_steps"] = 0
return state, get_empty_prediction_html("Đang chờ stream microphone bắt đầu...")
sr, chunk = audio_chunk
chunk = _to_mono_resampled(chunk, sr)
prev = state.get("buffer", np.zeros((0,), dtype=np.float32))
merged = np.concatenate([prev, chunk], axis=0)
if merged.shape[0] > TARGET_LEN:
merged = merged[-TARGET_LEN:]
state["buffer"] = merged
state["sample_rate"] = np.array([SR], dtype=np.int32)
# Implement a 3-step warm-up countdown (1.5s total) to let the microphone stream synchronize
warmup_steps = state.get("warmup_steps", 0)
if warmup_steps < 3:
state["warmup_steps"] = warmup_steps + 1
remaining = (3 - warmup_steps) * 0.5
return state, get_countdown_html("Đang đồng bộ mic và khởi động mô hình AI...", remaining)
if merged.shape[0] < int(0.3 * TARGET_LEN):
return state, get_empty_prediction_html("Đang lắng nghe... cần thêm dữ liệu âm thanh")
pred_label, pred_conf, top_k_results = run_prediction((SR, merged), model_name, top_k)
html_output = format_prediction_as_html(pred_label, pred_conf, model_name, top_k_results)
return state, html_output
# ============================================================
# Load custom CSS from file
# ============================================================
with open(Path(__file__).resolve().parent / "style.css", "r", encoding="utf-8") as f:
CUSTOM_CSS = f.read()
# Custom theme
THEME = gr.themes.Default(
primary_hue="violet",
secondary_hue="indigo",
neutral_hue="slate",
).set(
body_background_fill="radial-gradient(circle at top, #1e1b4b 0%, #0f172a 50%, #020617 100%)",
block_background_fill="rgba(15, 23, 42, 0.45)",
block_border_width="1px",
block_border_color="rgba(255, 255, 255, 0.08)",
block_radius="16px",
button_primary_background_fill="linear-gradient(135deg, #7c3aed, #4f46e5)",
button_primary_text_color="#ffffff",
)
# Custom HTML Header Banner
HEADER_HTML = """
"""
INFO_HTML = """
Pipeline xử lý kỹ thuật:
1. Đầu vào âm thanh (tối đa 1 giây, 16kHz) -> Log-Mel Spectrogram (40 bins, hop 10ms, win 25ms).
2. Các từ khóa hỗ trợ:
five
zero
yes
seven
no
nine
down
one
go
two
stop
six
unknown
silence.
"""
FOOTER_HTML = """
"""
# Build Gradio Blocks App
with gr.Blocks(title="Speech Commands Realtime Demo") as demo:
# Header Section
gr.HTML(HEADER_HTML)
# Configuration Panel
with gr.Row(elem_classes=["glass-panel"]):
with gr.Column(scale=2):
model_name = gr.Radio(
choices=["Baseline CNN2D", "CRNN + SpecAugment"],
value="CRNN + SpecAugment",
label="Chọn Mô Hình Dự Đoán",
)
with gr.Column(scale=1):
top_k = gr.Slider(
minimum=1,
maximum=10,
value=5,
step=1,
label="Số lượng Top-k hiển thị"
)
# General Information Panel (Pipeline & Keywords)
gr.HTML(INFO_HTML)
# Main Tabs
with gr.Tabs(elem_classes=["tabs"]):
# Tab 1: Single Audio File / Recording
with gr.Tab("Nhận diện Audio (Tải lên / Ghi âm)", elem_id="tab-single"):
with gr.Row():
with gr.Column(scale=1, elem_classes=["glass-panel"]):
gr.Markdown("### 🎙️ Nhập Audio Đầu Vào")
audio = gr.Audio(
sources=["microphone", "upload"],
type="numpy",
label="Ghi âm hoặc tải file âm thanh (độ dài gợi ý ~1 giây)",
elem_classes=["gradio-audio"],
buttons=["download"]
)
predict_btn = gr.Button("🔮 Bắt Đầu Phân Tích", elem_classes=["primary-btn"])
with gr.Column(scale=1):
gr.Markdown("### 📊 Kết Quả Phân Tích")
output_html = gr.HTML(value=get_empty_prediction_html(), elem_id="single-output-html")
predict_btn.click(
fn=predict_from_audio,
inputs=[audio, model_name, top_k],
outputs=[output_html],
)
# Tab 2: Microphone Realtime Streaming
with gr.Tab("Streaming Microphone (Real-time)", elem_id="tab-stream"):
with gr.Row():
with gr.Column(scale=1, elem_classes=["glass-panel"]):
gr.Markdown("### ⚡ Stream Âm Thanh Trực Tiếp")
gr.Markdown(
"Nhấn **Start Recording** trên microphone bên dưới để bắt đầu. "
"Hệ thống sẽ duy trì buffer 1 giây gần nhất và tự động cập nhật kết quả liên tục."
)
stream_audio = gr.Audio(
sources=["microphone"],
type="numpy",
streaming=True,
label="Microphone stream input",
elem_classes=["gradio-audio"],
buttons=[]
)
with gr.Column(scale=1):
gr.Markdown("### 📊 Kết Quả Thời Gian Thực")
stream_output_html = gr.HTML(value=get_empty_prediction_html("Đang chờ stream microphone bắt đầu..."), elem_id="stream-output-html")
stream_state = gr.State(value=init_state())
stream_audio.stream(
fn=stream_predict,
inputs=[stream_audio, stream_state, model_name, top_k],
outputs=[stream_state, stream_output_html],
stream_every=0.5,
)
# Footer Section
gr.HTML(FOOTER_HTML)
if __name__ == "__main__":
demo.launch(server_name="0.0.0.0", server_port=7860, theme=THEME, css=CUSTOM_CSS)