app.py

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)