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)