Telemetry collector with Gradio API endpoint + periodic flush

app.py

@@ -1,8 +1,7 @@
 import json
 import os
 from datetime import datetime
-from pathlib import Path
-from threading import Lock
+from threading import Lock, Timer
 
 import gradio as gr
 from huggingface_hub import HfApi
@@ -11,7 +10,8 @@ DATASET_REPO = "bobber/routangseng-telemetry"
 HF_TOKEN = os.environ.get("HF_TOKEN")
 BUFFER = []
 BUFFER_LOCK = Lock()
-FLUSH_SIZE = 5  # flush every 5 entries
+FLUSH_INTERVAL = 60  # seconds
+FLUSH_SIZE = 10  # or flush when buffer hits this size
 
 api = HfApi(token=HF_TOKEN)
 
@@ -25,9 +25,8 @@ def flush_buffer():
         entries = BUFFER.copy()
         BUFFER = []
 
-    # Download existing data, append, re-upload
     try:
-        path = api.hf_hub_download(DATASET_REPO, "data/telemetry.jsonl", repo_type="dataset")
+        path = api.hf_hub_download(DATASET_REPO, "data/telemetry.jsonl", repo_type="dataset", token=HF_TOKEN)
         existing = open(path).read().strip()
     except Exception:
         existing = ""
@@ -40,58 +39,83 @@ def flush_buffer():
         path_in_repo="data/telemetry.jsonl",
         repo_id=DATASET_REPO,
         repo_type="dataset",
+        token=HF_TOKEN,
         commit_message=f"Add {len(entries)} telemetry entries",
     )
     return len(entries)
 
 
-def collect(data: str):
+def periodic_flush():
+    """Flush buffer periodically."""
+    try:
+        n = flush_buffer()
+        if n > 0:
+            print(f"[telemetry] Flushed {n} entries to dataset")
+    except Exception as e:
+        print(f"[telemetry] Flush error: {e}")
+    Timer(FLUSH_INTERVAL, periodic_flush).start()
+
+
+# Start periodic flush
+Timer(FLUSH_INTERVAL, periodic_flush).start()
+
+
+def collect(data: str) -> dict:
     """Receive telemetry JSON from the WebGPU Space."""
     try:
         entry = json.loads(data)
-        # Validate required fields
         if "question" not in entry or "answer" not in entry:
             return {"status": "error", "message": "missing question or answer"}
-        
+
         entry["received_at"] = datetime.utcnow().isoformat() + "Z"
-        
+
         with BUFFER_LOCK:
             BUFFER.append(entry)
             buf_size = len(BUFFER)
 
         if buf_size >= FLUSH_SIZE:
-            n = flush_buffer()
-            return {"status": "ok", "flushed": n}
-        
+            try:
+                n = flush_buffer()
+                return {"status": "ok", "flushed": n}
+            except Exception as e:
+                return {"status": "ok", "buffered": buf_size, "flush_error": str(e)}
+
         return {"status": "ok", "buffered": buf_size}
     except json.JSONDecodeError:
         return {"status": "error", "message": "invalid JSON"}
 
 
-def manual_flush():
-    """Manually flush buffer to dataset."""
-    n = flush_buffer()
-    return f"Flushed {n} entries"
+def status():
+    with BUFFER_LOCK:
+        buf = len(BUFFER)
+    return f"Buffer: {buf} entries | Dataset: {DATASET_REPO} | Token: {'set' if HF_TOKEN else 'MISSING'}"
 
 
-def status():
-    """Check buffer status."""
-    return f"Buffer: {len(BUFFER)} entries"
+def manual_flush():
+    try:
+        n = flush_buffer()
+        return f"Flushed {n} entries to {DATASET_REPO}"
+    except Exception as e:
+        return f"Flush error: {e}"
 
 
-# Gradio interface (minimal UI + API)
 with gr.Blocks(title="肉糖生 Telemetry") as demo:
-    gr.Markdown("# 肉糖生 Telemetry Collector\nReceives Q&A data from the WebGPU Space.")
-    
+    gr.Markdown("# 📊 肉糖生 Telemetry\nCollects Q&A from WebGPU Space → HuggingFace Dataset")
+
     with gr.Row():
-        status_btn = gr.Button("Check Status")
-        flush_btn = gr.Button("Flush to Dataset")
-    output = gr.Textbox(label="Status")
-    
+        status_btn = gr.Button("Status")
+        flush_btn = gr.Button("Flush Now")
+    output = gr.Textbox(label="Result")
+
     status_btn.click(fn=status, outputs=output)
     flush_btn.click(fn=manual_flush, outputs=output)
 
-# API endpoint: POST /api/collect
+    # API endpoint for the WebGPU Space
+    collect_input = gr.Textbox(visible=False)
+    collect_output = gr.JSON(visible=False)
+    collect_btn = gr.Button(visible=False)
+    collect_btn.click(fn=collect, inputs=collect_input, outputs=collect_output, api_name="collect")
+
 demo.queue()
 
 if __name__ == "__main__":