Create app.py

app.py

@@ -0,0 +1,232 @@
+import gradio as gr
+import os
+import json
+import numpy as np
+import soundfile as sf
+from pathlib import Path
+import threading
+import time
+
+# Set up Hive home
+os.environ["HIVE_HOME"] = "/data/hive_data"
+HIVE_HOME = Path(os.environ["HIVE_HOME"])
+HIVE_HOME.mkdir(parents=True, exist_ok=True)
+
+# Global state
+voice_authenticated = False
+current_user = "guest"
+hive_instance = None
+
+# Lazy initialization
+def get_hive():
+    global hive_instance
+    if hive_instance is None:
+        from hive.core.bootstrap import Bootstrap
+        config_path = HIVE_HOME / "system" / "config.json"
+        bootstrap = Bootstrap(str(config_path))
+        hive_instance = bootstrap.run()
+    return hive_instance
+
+# Authentication functions
+def login(username, password, audio):
+    global voice_authenticated, current_user
+    try:
+        hive = get_hive()
+        
+        # Check password
+        if not hive.security.auth.check_password(username, password):
+            return "❌ Invalid credentials", False, False
+        
+        # Check voice for owner
+        if username == "owner":
+            if audio is None:
+                return "🎤 Please speak to verify voice", False, False
+            audio_path = HIVE_HOME / "voice" / "verify.wav"
+            sf.write(str(audio_path), audio[1], audio[0])
+            score = hive.voice.voiceprint.verify(audio_path, username)
+            if score < 0.7:
+                return f"❌ Voice match: {score:.2f} < 0.7", False, False
+        
+        voice_authenticated = True
+        current_user = username
+        return f"✅ Logged in as {username}", True, False
+    except Exception as e:
+        return f"❌ Error: {str(e)}", False, False
+
+def logout():
+    global voice_authenticated, current_user
+    voice_authenticated = False
+    current_user = "guest"
+    return "👋 Logged out", False, True
+
+# Voice processing
+def process_voice(audio):
+    global current_user
+    if audio is None:
+        return None
+    
+    try:
+        hive = get_hive()
+        audio_path = HIVE_HOME / "voice" / "input.wav"
+        sf.write(str(audio_path), audio[1], audio[0])
+        
+        # Authenticate or identify
+        if not voice_authenticated:
+            user_id = hive.voice.voiceprint.identify(audio_path)
+            if user_id:
+                current_user = user_id
+                return f"🎙️ Welcome back, {user_id}!"
+            else:
+                return "🎙️ Unknown speaker. Say 'enroll' to register."
+        
+        # Transcribe
+        text = hive.voice.asr.transcribe(audio_path)
+        return text
+    except Exception as e:
+        return f"❌ Voice error: {str(e)}"
+
+# Chat function
+def chat(message, history, audio_input):
+    global current_user
+    
+    # Process voice if provided
+    if audio_input is not None:
+        voice_result = process_voice(audio_input)
+        if voice_result and not voice_result.startswith("🎙️"):
+            history = history + [[None, voice_result]]
+            yield history
+            return
+        
+        if voice_result and not voice_result.startswith("🎙️ Welcome"):
+            message = voice_result
+        elif voice_result and "enroll" in voice_result.lower():
+            # Start enrollment
+            hive = get_hive()
+            enroll_audio = HIVE_HOME / "voice" / "enroll.wav"
+            sf.write(str(enroll_audio), audio_input[1], audio_input[0])
+            new_user_id = f"user_{int(time.time())}"
+            hive.voice.voiceprint.enroll(enroll_audio, new_user_id)
+            current_user = new_user_id
+            history = history + [[None, f"✅ Enrolled as {new_user_id}"]]
+            yield history
+            return
+    
+    if not message or not message.strip():
+        return history
+    
+    try:
+        hive = get_hive()
+        response = hive.convo.dialogue_manager.submit_turn(
+            session_id="default",
+            user_id=current_user,
+            raw_input={"text": message}
+        )
+        
+        response_text = response.postprocessed_output.get("text", "Sorry, I couldn't process that.")
+        history = history + [[message, response_text]]
+        yield history
+    except Exception as e:
+        history = history + [[message, f"❌ Error: {str(e)}"]]
+        yield history
+
+# Admin functions
+def admin_rollback():
+    try:
+        hive = get_hive()
+        result = hive.persistence.rollback()
+        return f"✅ Rollback: {result}"
+    except Exception as e:
+        return f"❌ Rollback failed: {str(e)}"
+
+def admin_snapshot():
+    try:
+        hive = get_hive()
+        result = hive.persistence.save_snapshot()
+        return f"✅ Snapshot: {result}"
+    except Exception as e:
+        return f"❌ Snapshot failed: {str(e)}"
+
+def admin_logs():
+    try:
+        logs_path = HIVE_HOME / "admin" / "audit.jsonl"
+        if logs_path.exists():
+            return logs_path.read_text()
+        return "No logs found"
+    except Exception as e:
+        return f"❌ Error reading logs: {str(e)}"
+
+# Build UI
+with gr.Blocks(title="🐝 Hive — Local AI Tutor", css="""
+    .login-box { background: #f0f0f0; padding: 20px; border-radius: 10px; margin: 10px; }
+    .admin-tab { background: #fff3cd; }
+""") as demo:
+    
+    # State
+    logged_in = gr.State(False)
+    
+    # Login section
+    with gr.Row():
+        with gr.Column(scale=3):
+            gr.Markdown("# 🐝 Hive — Your Private AI Tutor")
+            gr.Markdown("*Voice-first, offline-capable AI assistant with self-optimizing memory*")
+        with gr.Column(scale=1):
+            with gr.Group(elem_classes=["login-box"]):
+                username = gr.Textbox(label="Username", value="owner")
+                password = gr.Textbox(label="Password", type="password", value="Fehr2008")
+                login_audio = gr.Audio(source="microphone", type="numpy", label="Verify voice (owner)")
+                login_btn = gr.Button("Login", variant="primary")
+                logout_btn = gr.Button("Logout", visible=False)
+                status = gr.Text(label="Status", interactive=False)
+    
+    # Main chat interface
+    with gr.Row():
+        with gr.Column(scale=4):
+            chatbot = gr.Chatbot(height=500, label="Conversation")
+            with gr.Row():
+                msg = gr.Textbox(label="Message", placeholder="Type or speak...")
+                audio = gr.Audio(source="microphone", type="numpy", label="🎤")
+                send = gr.Button("Send")
+        with gr.Column(scale=1, visible=False) as admin_col:
+            with gr.Tab("Admin", elem_classes=["admin-tab"]):
+                gr.Markdown("### 🔐 Admin Controls")
+                with gr.Row():
+                    rollback_btn = gr.Button("⏪ Force Rollback", variant="secondary")
+                    snapshot_btn = gr.Button("💾 Save Snapshot", variant="secondary")
+                admin_status = gr.Text(label="Result")
+                gr.Markdown("### 📋 Audit Logs")
+                logs = gr.TextArea(label="Logs", lines=20, max_lines=30)
+                refresh_logs = gr.Button("🔄 Refresh")
+    
+    # Event handlers
+    login_btn.click(
+        login,
+        inputs=[username, password, login_audio],
+        outputs=[status, logged_in, admin_col]
+    ).then(
+        lambda: (gr.update(visible=False), gr.update(visible=True)),
+        outputs=[login_btn, logout_btn]
+    )
+    
+    logout_btn.click(
+        logout,
+        outputs=[status, logged_in, admin_col]
+    ).then(
+        lambda: (gr.update(visible=True), gr.update(visible=False)),
+        outputs=[login_btn, logout_btn]
+    )
+    
+    # Chat events
+    msg.submit(chat, [msg, chatbot, audio], chatbot).then(lambda: "", outputs=msg)
+    send.click(chat, [msg, chatbot, audio], chatbot).then(lambda: "", outputs=msg)
+    audio.stop_recording(
+        chat, [msg, chatbot, audio], chatbot
+    )
+    
+    # Admin events
+    rollback_btn.click(admin_rollback, outputs=admin_status)
+    snapshot_btn.click(admin_snapshot, outputs=admin_status)
+    refresh_logs.click(admin_logs, outputs=logs)
+
+# Launch
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0", server_port=7860, share=False)