Update app.py

Add buffer 10 min

app.py

@@ -1,21 +1,57 @@
+bash -lc cat > /mnt/data/app.py <<'PY'
+import atexit
+import csv
+import io
+import os
+import re
+import shutil
+import tempfile
+import threading
+import time
+import unicodedata
+import uuid
+import wave
+from datetime import datetime
+
 import gradio as gr
-import os, uuid, librosa, threading, time, tempfile, wave, io, csv
-import unicodedata, re
+import librosa
 import numpy as np
-from huggingface_hub import HfApi, CommitOperationAdd
+from huggingface_hub import CommitOperationAdd, HfApi, hf_hub_download
 from openwakeword.model import Model
 
-# --- CẤU HÌNH ---
+# ============================================================
+# CẤU HÌNH
+# ============================================================
 DATASET_ID = "lumiwakeword/lumioiv1"
 HF_TOKEN = os.getenv("HF_TOKEN")
 MODEL_PATH = "loo_mee_oy_v2.onnx"
 
-api = HfApi()
+# Buffer / batch upload
+BUFFER_DIR = "buffer"
+AUDIO_BUFFER_DIR = os.path.join(BUFFER_DIR, "audios")
+PENDING_QUEUE_FILE = os.path.join(BUFFER_DIR, "pending_queue.csv")
+FLUSH_INTERVAL_SEC = 600          # 10 phút
+MAX_PENDING_BEFORE_FLUSH = 100    # flush sớm nếu đủ nhiều mẫu
+
+# Metadata trên Hub: giữ nguyên schema cũ để tương thích pipeline hiện tại
+HUB_METADATA_HEADER = ["file_name"]
+
+api = HfApi(token=HF_TOKEN)
+
+os.makedirs(AUDIO_BUFFER_DIR, exist_ok=True)
+
+buffer_lock = threading.Lock()
+flush_in_progress = threading.Event()
+shutdown_event = threading.Event()
 
 # --- KHỞI TẠO MODEL ---
 oww_model = None
 model_ready = threading.Event()
 
+
+# ============================================================
+# MODEL LOADER
+# ============================================================
 def load_model():
     global oww_model
     try:
@@ -24,7 +60,9 @@ def load_model():
         with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
             tmp_path = f.name
         with wave.open(tmp_path, "w") as wf:
-            wf.setnchannels(1); wf.setsampwidth(2); wf.setframerate(16000)
+            wf.setnchannels(1)
+            wf.setsampwidth(2)
+            wf.setframerate(16000)
             wf.writeframes((dummy_wav * 32767).astype(np.int16).tobytes())
         librosa.load(tmp_path, sr=16000)
         os.unlink(tmp_path)
@@ -39,7 +77,8 @@ def load_model():
             t_iter = time.perf_counter()
             m.predict(np.zeros(1280, dtype=np.int16))
             elapsed_ms = (time.perf_counter() - t_iter) * 1000
-            if i < 3: continue
+            if i < 3:
+                continue
             if elapsed_ms < 5.0:
                 stable_count += 1
                 if stable_count >= 5:
@@ -55,10 +94,240 @@ def load_model():
     finally:
         model_ready.set()
 
+
 threading.Thread(target=load_model, daemon=True).start()
 
 
-# --- PYTHON LOGIC ---
+# ============================================================
+# BUFFER / QUEUE HELPERS
+# ============================================================
+def ensure_pending_queue():
+    if not os.path.exists(PENDING_QUEUE_FILE):
+        with open(PENDING_QUEUE_FILE, "w", newline="", encoding="utf-8") as f:
+            writer = csv.writer(f)
+            writer.writerow(["local_path", "repo_path", "speaker", "score", "created_at"])
+
+
+def read_pending_queue():
+    ensure_pending_queue()
+    with open(PENDING_QUEUE_FILE, "r", encoding="utf-8") as f:
+        rows = list(csv.reader(f))
+    if not rows:
+        return ["local_path", "repo_path", "speaker", "score", "created_at"], []
+    return rows[0], rows[1:]
+
+
+def write_pending_queue(header, rows):
+    with open(PENDING_QUEUE_FILE, "w", newline="", encoding="utf-8") as f:
+        writer = csv.writer(f)
+        writer.writerow(header)
+        writer.writerows(rows)
+
+
+def get_pending_count():
+    with buffer_lock:
+        _, rows = read_pending_queue()
+        return len(rows)
+
+
+def fetch_existing_metadata_rows():
+    try:
+        existing = hf_hub_download(
+            repo_id=DATASET_ID,
+            filename="metadata.csv",
+            repo_type="dataset",
+            token=HF_TOKEN,
+        )
+        with open(existing, "r", encoding="utf-8") as f:
+            rows = list(csv.reader(f))
+        if not rows:
+            return [HUB_METADATA_HEADER]
+        # Nếu file cũ chỉ có 1 cột hoặc header khác, vẫn cố chuẩn hóa tối thiểu.
+        header = rows[0] if rows[0] else HUB_METADATA_HEADER
+        if len(header) == 1:
+            normalized = [HUB_METADATA_HEADER]
+            for row in rows[1:]:
+                if row:
+                    normalized.append([row[0]])
+            return normalized
+        return rows
+    except Exception:
+        return [HUB_METADATA_HEADER]
+
+
+def enqueue_sample_local(audio_path, score, speaker_name):
+    ensure_pending_queue()
+
+    spk = slugify(speaker_name) if speaker_name else "unknown"
+    created_at = datetime.utcnow().replace(microsecond=0).isoformat() + "Z"
+    ext = ".wav"
+    filename = f"lumi_{spk}_{uuid.uuid4().hex[:8]}{ext}"
+    local_buffer_path = os.path.join(AUDIO_BUFFER_DIR, filename)
+    repo_audio_path = f"audios/{filename}"
+
+    # Chuẩn hóa thành WAV 16k mono để dataset đồng nhất.
+    y, _ = librosa.load(audio_path, sr=16000, mono=True)
+    y = np.clip(y, -1.0, 1.0)
+    audio_int16 = (y * 32767).astype(np.int16)
+
+    with wave.open(local_buffer_path, "wb") as wf:
+        wf.setnchannels(1)
+        wf.setsampwidth(2)
+        wf.setframerate(16000)
+        wf.writeframes(audio_int16.tobytes())
+
+    with buffer_lock:
+        header, rows = read_pending_queue()
+        rows.append([local_buffer_path, repo_audio_path, spk, f"{score:.4f}", created_at])
+        write_pending_queue(header, rows)
+
+    return filename, local_buffer_path, repo_audio_path
+
+
+def flush_buffer_to_hf():
+    if flush_in_progress.is_set():
+        return False, "flush đang chạy"
+
+    if not HF_TOKEN:
+        return False, "HF_TOKEN chưa được cấu hình"
+
+    flush_in_progress.set()
+    try:
+        with buffer_lock:
+            header, pending_rows = read_pending_queue()
+            if not pending_rows:
+                return True, "không có mẫu chờ"
+            snapshot_rows = list(pending_rows)
+
+        valid_rows = []
+        missing_rows = []
+        for row in snapshot_rows:
+            if not row or len(row) < 2:
+                continue
+            local_path = row[0]
+            if os.path.exists(local_path):
+                valid_rows.append(row)
+            else:
+                missing_rows.append(row)
+
+        if not valid_rows and missing_rows:
+            with buffer_lock:
+                cur_header, cur_rows = read_pending_queue()
+                missing_keys = {(r[0], r[1]) for r in missing_rows if len(r) >= 2}
+                remaining_rows = [
+                    r for r in cur_rows
+                    if len(r) >= 2 and (r[0], r[1]) not in missing_keys
+                ]
+                write_pending_queue(cur_header, remaining_rows)
+            return False, "có bản ghi pending bị mất file local"
+
+        if not valid_rows:
+            return True, "không có file hợp lệ để flush"
+
+        existing_rows = fetch_existing_metadata_rows()
+        existing_data = existing_rows[1:] if len(existing_rows) > 1 else []
+        existing_set = {row[0] for row in existing_data if row}
+
+        operations = []
+        new_metadata_rows = []
+        for row in valid_rows:
+            local_path, repo_path = row[0], row[1]
+            operations.append(
+                CommitOperationAdd(
+                    path_in_repo=repo_path,
+                    path_or_fileobj=local_path,
+                )
+            )
+            if repo_path not in existing_set:
+                new_metadata_rows.append([repo_path])
+
+        merged_rows = [HUB_METADATA_HEADER] + existing_data + new_metadata_rows
+        buf = io.StringIO()
+        csv.writer(buf).writerows(merged_rows)
+        csv_bytes = buf.getvalue().encode("utf-8")
+
+        operations.append(
+            CommitOperationAdd(
+                path_in_repo="metadata.csv",
+                path_or_fileobj=io.BytesIO(csv_bytes),
+            )
+        )
+
+        commit_msg = (
+            f"batch upload {len(valid_rows)} samples @ "
+            f"{datetime.utcnow().replace(microsecond=0).isoformat()}Z"
+        )
+
+        api.create_commit(
+            repo_id=DATASET_ID,
+            repo_type="dataset",
+            operations=operations,
+            commit_message=commit_msg,
+            token=HF_TOKEN,
+        )
+
+        # Chỉ xóa queue/file local sau khi commit thành công.
+        with buffer_lock:
+            cur_header, cur_rows = read_pending_queue()
+            flushed_keys = {(r[0], r[1]) for r in valid_rows if len(r) >= 2}
+            missing_keys = {(r[0], r[1]) for r in missing_rows if len(r) >= 2}
+
+            remaining_rows = [
+                r for r in cur_rows
+                if len(r) >= 2 and (r[0], r[1]) not in flushed_keys and (r[0], r[1]) not in missing_keys
+            ]
+            write_pending_queue(cur_header, remaining_rows)
+
+        for row in valid_rows:
+            local_path = row[0]
+            try:
+                if os.path.exists(local_path):
+                    os.remove(local_path)
+            except Exception as rm_err:
+                print(f"⚠️ Không xóa được file buffer {local_path}: {rm_err}")
+
+        print(f"✅ Flush thành công {len(valid_rows)} mẫu lên Hugging Face")
+        return True, f"đã flush {len(valid_rows)} mẫu"
+
+    except Exception as e:
+        print(f"❌ Flush lỗi: {e}")
+        return False, str(e)
+    finally:
+        flush_in_progress.clear()
+
+
+def flush_worker():
+    while not shutdown_event.is_set():
+        try:
+            shutdown_event.wait(FLUSH_INTERVAL_SEC)
+            if shutdown_event.is_set():
+                break
+            ok, msg = flush_buffer_to_hf()
+            print(f"[flush_worker] ok={ok} msg={msg}")
+        except Exception as e:
+            print(f"❌ Flush worker error: {e}")
+
+
+threading.Thread(target=flush_worker, daemon=True).start()
+
+
+def shutdown_cleanup():
+    shutdown_event.set()
+    try:
+        pending = get_pending_count()
+        if pending > 0:
+            ok, msg = flush_buffer_to_hf()
+            print(f"[shutdown_flush] ok={ok} msg={msg}")
+    except Exception as e:
+        print(f"⚠️ shutdown cleanup error: {e}")
+
+
+atexit.register(shutdown_cleanup)
+
+
+# ============================================================
+# PYTHON LOGIC
+# ============================================================
 def verify_audio(audio_path):
     if not model_ready.wait(timeout=30):
         return "⏳ Model đang khởi động, thử lại sau vài giây...", gr.update(interactive=False), 0.0
@@ -95,51 +364,20 @@ def upload_final(audio_path, score, speaker_name):
     if not audio_path:
         return "❌ Không có file để gửi.", gr.update(interactive=False), None
     try:
-        spk = slugify(speaker_name) if speaker_name else 'unknown'
-        filename = f"lumi_{spk}_{score:.2f}_{uuid.uuid4().hex[:4]}.wav"
-        audio_repo_path = f"audios/{filename}"
-
-        try:
-            existing = api.hf_hub_download(
-                repo_id=DATASET_ID,
-                filename="metadata.csv",
-                repo_type="dataset",
-                token=HF_TOKEN
-            )
-            with open(existing, "r", encoding="utf-8") as f:
-                rows = list(csv.reader(f))
-        except Exception:
-            rows = [["file_name"]]
+        _, _, _ = enqueue_sample_local(audio_path, score, speaker_name)
+        pending_count = get_pending_count()
 
-        rows.append([audio_repo_path])
+        # Flush sớm nếu hàng đợi đạt ngưỡng.
+        if pending_count >= MAX_PENDING_BEFORE_FLUSH and not flush_in_progress.is_set():
+            threading.Thread(target=flush_buffer_to_hf, daemon=True).start()
 
-        buf = io.StringIO()
-        csv.writer(buf).writerows(rows)
-        csv_bytes = buf.getvalue().encode("utf-8")
-
-        operations = [
-            CommitOperationAdd(
-                path_in_repo=audio_repo_path,
-                path_or_fileobj=audio_path,
-            ),
-            CommitOperationAdd(
-                path_in_repo="metadata.csv",
-                path_or_fileobj=io.BytesIO(csv_bytes),
-            ),
-        ]
-
-        api.create_commit(
-            repo_id=DATASET_ID,
-            repo_type="dataset",
-            operations=operations,
-            commit_message=f"add {filename}",
-            token=HF_TOKEN,
+        return (
+            f"🎉 Đã nhận mẫu! Đang chờ đồng bộ. Số mẫu trong hàng đợi: {pending_count}",
+            gr.update(interactive=False),
+            None,
         )
-
-        return "🎉 Gửi thành công! Bạn có thể thu mẫu tiếp theo.", gr.update(interactive=False), None
-
     except Exception as e:
-        return f"❌ Lỗi gửi: {e}", gr.update(interactive=True), audio_path
+        return f"❌ Lỗi lưu buffer: {e}", gr.update(interactive=True), audio_path
 
 
 def slugify(text):
@@ -155,7 +393,7 @@ def reset_ui():
         None,
         'Sẵn sàng — Nhấn mic và đọc "Lumi ơi"',
         gr.update(interactive=False),
-        0.0
+        0.0,
     )
 
 
@@ -198,7 +436,6 @@ body, .gradio-container {
     margin: 0 auto !important;
     padding: clamp(16px,4vw,40px) clamp(12px,4vw,36px) 60px !important;
 }
-/* ── HEADER ── */
 .lumi-header {
     display: flex; align-items: center; gap: var(--sp-md);
     background: linear-gradient(135deg, #eef3ff 0%, #f4f6fb 100%);
@@ -226,7 +463,6 @@ body, .gradio-container {
 .lumi-header-text p {
     font-size: var(--fs-sm); color: var(--text3); margin-top: 2px;
 }
-/* ── PROGRESS ── */
 .lumi-progress {
     display: flex; align-items: center;
     margin-bottom: var(--sp-md);
@@ -245,7 +481,6 @@ body, .gradio-container {
     height: 1px; background: var(--border);
     width: clamp(14px,3vw,28px); flex-shrink: 0; margin: 0 var(--sp-xs);
 }
-/* ── GUIDE ── */
 .lumi-guide {
     background: var(--surface); border: 1px solid var(--border);
     border-radius: clamp(10px,2vw,14px);
@@ -278,7 +513,6 @@ body, .gradio-container {
     color: var(--accent2); font-weight: 600;
     background: rgba(59,126,244,0.08); border-radius: 4px; padding: 0 4px;
 }
-/* ── TIPS ── */
 .lumi-tips {
     margin-top: var(--sp-sm); padding-top: var(--sp-sm);
     border-top: 1px solid var(--border);
@@ -293,7 +527,6 @@ body, .gradio-container {
 }
 .lumi-tip b { color: var(--text); font-weight: 600; }
 .tip-icon { font-size: var(--fs-base); flex-shrink: 0; }
-/* ── RECORDER CARD ── */
 .lumi-recorder-card {
     background: linear-gradient(145deg, #1e2d5a 0%, #162447 55%, #1a3060 100%);
     border: 1px solid rgba(99,140,255,0.25);
@@ -322,7 +555,6 @@ body, .gradio-container {
     align-items: center; gap: var(--sp-md);
     position: relative; z-index: 1;
 }
-/* Mic button */
 #lumi-mic-btn {
     width: clamp(60px,13vw,88px); height: clamp(60px,13vw,88px);
     border-radius: 50%;
@@ -350,19 +582,16 @@ body, .gradio-container {
     0%,100% { box-shadow: 0 0 0 0    rgba(248,113,113,0.4); }
     50%      { box-shadow: 0 0 0 14px rgba(248,113,113,0);   }
 }
-/* Rec label */
 #lumi-rec-label {
     font-size: var(--fs-sm); color: rgba(255,255,255,0.45);
     font-weight: 500; letter-spacing: 0.3px; text-align: center;
 }
 #lumi-rec-label.recording { color: #fca5a5; }
-/* Canvas */
 #lumi-canvas {
     width: 100%; height: clamp(44px,7vw,68px);
     border-radius: 8px; background: rgba(0,0,0,0.22);
     border: 1px solid rgba(255,255,255,0.07); display: block;
 }
-/* Processing bar */
 #lumi-proc-bar { display: none; width: 100%; }
 #lumi-proc-bar .p-label {
     font-size: var(--fs-xs); font-family: var(--mono);
@@ -387,7 +616,6 @@ body, .gradio-container {
     50% {width:50%; margin-left:25%;}
     100%{width:0%;  margin-left:100%;}
 }
-/* Playback */
 #lumi-audio-playback { width: 100%; display: none; flex-direction: column; gap: 5px; }
 #lumi-audio-playback .play-label {
     font-size: var(--fs-xs); font-family: var(--mono); color: rgba(255,255,255,0.35);
@@ -395,7 +623,6 @@ body, .gradio-container {
 #lumi-audio-playback audio {
     width: 100%; height: 30px; border-radius: 6px; outline: none; accent-color: #638cff;
 }
-/* ── STATUS ── */
 .lumi-status {
     background: var(--surface) !important; border: 1px solid var(--border) !important;
     border-radius: 10px !important;
@@ -408,7 +635,6 @@ body, .gradio-container {
     font-size: var(--fs-base) !important; color: var(--text2) !important;
     font-family: var(--mono) !important;
 }
-/* ── BUTTONS ── */
 .lumi-btn-row {
     display: grid; grid-template-columns: 1fr 1fr;
     gap: var(--sp-xs); margin-bottom: var(--sp-xs);
@@ -453,7 +679,6 @@ button.lb-send:disabled {
     position: absolute; width: 1px; height: 1px;
     overflow: hidden; opacity: 0; pointer-events: none;
 }
-/* ── SPEAKER INPUT ── */
 .lumi-speaker-input { margin-bottom: var(--sp-xs) !important; }
 .lumi-speaker-input input {
     background: var(--surface) !important;
@@ -578,7 +803,6 @@ RECORDER_HTML = """
         }
         attempt(10);
     }
-    /* ── WAVEFORM ── */
     function drawWave() {
         if (!analyser) return;
         animId = requestAnimationFrame(drawWave);
@@ -671,7 +895,6 @@ with gr.Blocks(theme=gr.themes.Base(), css=CSS, title="Lumi Voice Collector") as
     </div>
     """)
 
-    # Card 1 — Hướng dẫn các bước
     gr.HTML("""
     <div class="lumi-guide">
         <div class="lumi-guide-icon">📋</div>
@@ -703,7 +926,6 @@ with gr.Blocks(theme=gr.themes.Base(), css=CSS, title="Lumi Voice Collector") as
     </div>
     """)
 
-    # Card 2 — Khoảng cách & Góc thu âm
     gr.HTML("""
     <div class="lumi-guide">
         <div class="lumi-guide-icon">📐</div>
@@ -745,31 +967,39 @@ with gr.Blocks(theme=gr.themes.Base(), css=CSS, title="Lumi Voice Collector") as
         placeholder="Nhập tên của bạn...",
         label="Tên người thu âm",
         max_lines=1,
-        elem_classes="lumi-speaker-input"
+        elem_classes="lumi-speaker-input",
     )
 
     status_txt = gr.Label(
         value='Sẵn sàng — Nhấn mic và đọc "Lumi ơi"',
         elem_classes="lumi-status",
-        show_label=False
+        show_label=False,
     )
 
     with gr.Row(elem_classes="lumi-btn-row"):
         btn_check = gr.Button("Kiểm tra mẫu", variant="secondary", elem_classes="lb-check")
-        btn_reset = gr.Button("Ghi lại",       variant="stop",      elem_classes="lb-reset")
+        btn_reset = gr.Button("Ghi lại", variant="stop", elem_classes="lb-reset")
 
     btn_send = gr.Button(
         "Gửi lên hệ thống →",
         variant="primary",
         interactive=False,
-        elem_classes="lb-send"
+        elem_classes="lb-send",
     )
 
     score_state = gr.State(0.0)
 
-    btn_check.click(fn=verify_audio,  inputs=audio_inst, outputs=[status_txt, btn_send, score_state])
-    btn_send.click( fn=upload_final,  inputs=[audio_inst, score_state, speaker_input], outputs=[status_txt, btn_send, audio_inst])
-    btn_reset.click(fn=reset_ui,      outputs=[audio_inst, status_txt, btn_send, score_state])
+    btn_check.click(fn=verify_audio, inputs=audio_inst, outputs=[status_txt, btn_send, score_state])
+    btn_send.click(fn=upload_final, inputs=[audio_inst, score_state, speaker_input], outputs=[status_txt, btn_send, audio_inst])
+    btn_reset.click(fn=reset_ui, outputs=[audio_inst, status_txt, btn_send, score_state])
     audio_inst.change(lambda: gr.update(interactive=False), None, btn_send)
 
-demo.launch()
+
+if __name__ == "__main__":
+    print("🚀 Starting Lumi Wake Word Collector")
+    print(f"   DATASET_ID={DATASET_ID}")
+    print(f"   FLUSH_INTERVAL_SEC={FLUSH_INTERVAL_SEC}")
+    print(f"   MAX_PENDING_BEFORE_FLUSH={MAX_PENDING_BEFORE_FLUSH}")
+    print(f"   HF_TOKEN={'OK' if HF_TOKEN else 'MISSING'}")
+    demo.launch()
+PY