Blame for app.py

#!/usr/bin/env python3
"""
JARVIS — HuggingFace Space App
Gradio chat interface for JARVIS personal AI assistant.
Runs entirely locally on HuggingFace Spaces (CPU or GPU) with 16 GB RAM.

Launch:  python app.py
Or via HuggingFace Spaces (auto-detected from README.md sdk: gradio).
"""

import os
import io
import json
import threading
import gradio as gr

# ── Init subsystems ──────────────────────────────────────────────────────────
import memory
import thinking
import brain

memory.init_db()
thinking.start_thinking_engine()

# Warm-up model in background so UI loads instantly
_warmup_thread = brain.warmup()

# ── TTS helper ────────────────────────────────────────────────────────────────
def _speak(text: str):
    """Return audio bytes for Gradio Audio component, or None."""
    try:
        from voice_tts import tts_bytes
        return tts_bytes(text)
    except Exception:
        return None


# ── Core chat logic ────────────────────────────────────────────────────────────
def chat(user_message: str, history: list[list[str]], audio_reply: bool):
    """
    Main chat handler — called by Gradio.
    history: [[user_msg, bot_msg], ...]
    Returns: (updated_history, status_text, audio_or_None)
    """
    if not user_message.strip():
        return history, "Say something, sir.", None

    lower = user_message.lower().strip()

    # ── Built-in commands ──────────────────────────────────────────────────
    if lower in ("status", "jarvis status"):
        facts = memory.get_all_facts()
        reply = (
            f"Status: Online. "
            f"Memory: {len(facts)} facts. "
            f"Model: {brain.HF_MODEL}. "
            f"TTS: {'ON' if audio_reply else 'OFF'}."
        )
        history.append([user_message, reply])
        audio = _speak(reply) if audio_reply else None
        return history, "Ready.", audio

    if lower in ("anticipate", "what are you thinking"):
        ants = thinking.get_anticipations()
        reply = "Thinking ahead: " + (", ".join(ants[:3]) if ants else "No predictions yet.")
        history.append([user_message, reply])
        audio = _speak(reply) if audio_reply else None
        return history, "Ready.", audio

    if lower.startswith("remember ") and " is " in lower:
        try:
            part = user_message[9:]  # strip "remember "
            key, val = part.split(" is ", 1)
            memory.save_fact(key.strip().lower().replace(" ", "_"), val.strip())
            reply = f"Remembered: {key.strip()} is {val.strip()}."
        except Exception as e:
            reply = f"Couldn't remember that: {e}"
        history.append([user_message, reply])
        audio = _speak(reply) if audio_reply else None
        return history, "Memory updated.", audio

    # ── Normal brain query ─────────────────────────────────────────────────
    hist_flat = [
        {"role": "user",   "message": pair[0]}
        for pair in history
        if pair[0]
    ] + [
        {"role": "jarvis", "message": pair[1]}
        for pair in history
        if len(pair) > 1 and pair[1]
    ]
    # interleave properly
    hist_pairs = []
    for pair in history:
        if pair[0]:
            hist_pairs.append({"role": "user",   "message": pair[0]})
        if len(pair) > 1 and pair[1]:
            hist_pairs.append({"role": "jarvis", "message": pair[1]})

    reply = brain.ask(user_message, hist_pairs)
    history.append([user_message, reply])
    audio = _speak(reply) if audio_reply else None
    return history, "Ready.", audio


def remember_fact(key: str, value: str):
    """Directly save a fact via the sidebar form."""
    if not key.strip() or not value.strip():
        return "Please fill in both fields."
    memory.save_fact(key.strip().lower().replace(" ", "_"), value.strip())
    return f"Saved: {key.strip()} = {value.strip()}"


def load_facts():
    facts = memory.get_all_facts()
    if not facts:
        return "No facts stored yet."
    return "\n".join(f"**{k}**: {v}" for k, v in facts.items())


def load_anticipations():
    ants = thinking.get_anticipations()
    if not ants:
        return "No predictions yet. JARVIS thinks every hour."
    return "\n".join(f"• {a}" for a in ants[:5])


# ── Build Gradio UI ────────────────────────────────────────────────────────────
def build_ui() -> gr.Blocks:
    css = """
    #jarvis-header { text-align: center; margin-bottom: 8px; }
    #jarvis-header h1 { font-size: 2.4em; font-weight: 800; letter-spacing: 2px; }
    #jarvis-header p  { color: #888; margin-top: -8px; }
    .gradio-container { max-width: 900px !important; margin: auto; }
    """

    with gr.Blocks(title="JARVIS – Personal AI", css=css, theme=gr.themes.Soft()) as demo:

        # ── Header ──────────────────────────────────────────────────────────
        with gr.Row(elem_id="jarvis-header"):
            gr.HTML(
                "<h1>🤖 J.A.R.V.I.S</h1>"
                "<p>Just A Rather Very Intelligent System — Your Personal AI Agent</p>"
                "<p style='font-size:0.8em;color:#555'>"
                f"Powered by: <code>{brain.HF_MODEL}</code> · 16 GB RAM optimised · Local inference"
                "</p>"
            )

        with gr.Row():
            # ── Main chat column ─────────────────────────────────────────────
            with gr.Column(scale=3):
                chatbot = gr.Chatbot(
                    label="JARVIS",
                    height=480,
                    avatar_images=(
                        None,  # user avatar
                        "https://upload.wikimedia.org/wikipedia/en/thumb/6/68/Iron_Man_in_Marvel_Cinematic_Universe.jpg/250px-Iron_Man_in_Marvel_Cinematic_Universe.jpg",
                    ),
                    bubble_full_width=False,
                )
                with gr.Row():
                    msg_box = gr.Textbox(
                        placeholder="Talk to JARVIS... (try: 'status', 'remember name is Tony', or any question)",
                        show_label=False,
                        scale=5,
                        container=False,
                    )
                    send_btn = gr.Button("Send", variant="primary", scale=1)

                audio_out  = gr.Audio(label="JARVIS Voice", autoplay=True, visible=True)
                audio_on   = gr.Checkbox(label="Voice replies (TTS)", value=True)
                status_bar = gr.Textbox(label="Status", value="Ready.", interactive=False)

            # ── Sidebar ──────────────────────────────────────────────────────
            with gr.Column(scale=1):
                gr.Markdown("### 📂 Memory")
                with gr.Group():
                    fact_key = gr.Textbox(placeholder="fact key (e.g. name)", label="Key", lines=1)
                    fact_val = gr.Textbox(placeholder="value", label="Value", lines=1)
                    save_btn = gr.Button("Save Fact", variant="secondary")
                    save_status = gr.Textbox(label="", interactive=False, lines=1)

                gr.Markdown("### 🧠 Stored Facts")
                facts_display = gr.Markdown(value=load_facts, every=15)

                gr.Markdown("### 💡 JARVIS Thinks Ahead")
                ants_display = gr.Markdown(value=load_anticipations, every=60)

        # ── Events ───────────────────────────────────────────────────────────
        def handle_submit(user_msg, history, audio_on_val):
            return chat(user_msg, history, audio_on_val)

        # Send on Enter or button
        msg_box.submit(
            handle_submit,
            inputs=[msg_box, chatbot, audio_on],
            outputs=[chatbot, status_bar, audio_out],
        ).then(lambda: "", outputs=msg_box)

        send_btn.click(
            handle_submit,
            inputs=[msg_box, chatbot, audio_on],
            outputs=[chatbot, status_bar, audio_out],
        ).then(lambda: "", outputs=msg_box)

        # Save fact
        save_btn.click(
            remember_fact,
            inputs=[fact_key, fact_val],
            outputs=save_status,
        )

        # Refresh facts panel when a fact is saved
        save_btn.click(load_facts, outputs=facts_display)

    return demo


# ── Entry point ────────────────────────────────────────────────────────────────
if __name__ == "__main__":
    demo = build_ui()
    demo.launch(
        server_name="0.0.0.0",
        server_port=int(os.environ.get("PORT", 7860)),
        share=False,          # set True for a public tunnel link
        show_error=True,
    )