Jarvis / app.py

Upload 11 files

6645e8f verified 6 days ago

9.77 kB

#!/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,
    )

	#!/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,
	)