app.py

#!/usr/bin/env python3
# app.py - Front-end dashboard for Multimodular v7 (multimodal)
# Place this file alongside your multimodular module (compact or expanded).

import os, time, sys, json, pathlib

# ---- Config: brain module names to try ----
CANDIDATE_MODULES = [
    "multimodular_modul_v7",           # compact name used earlier
    "multimodular_modul_v7_expanded",  # expanded package name used earlier
    "multimodular_modul version 7.0",  # fallback if you saved exact name (unlikely)
]

# ---- Boot splash ----
def boot_splash():
    os.system("cls" if os.name == "nt" else "clear")
    logo = r"""
 ██████╗██╗  ██╗██████╗ 
██╔════╝██║  ██║██╔══██╗
██║     ███████║██████╔╝
██║     ██╔══██║██╔═══╝ 
╚██████╗██║  ██║██║     
 ╚═════╝╚═╝  ╚═╝╚═╝     
 Close-to-Human Brain v7.0
"""
    print(logo)
    print("Initializing Universal Brain...")
    steps = [
        "Loading Core Modules",
        "Starting Local DB",
        "Bringing up CTB pipeline",
        "Starting Global Sync (if configured)",
        "Activating Creative Skill Vault",
        "Launching Dashboard"
    ]
    for s in steps:
        print(" →", s + "...")
        time.sleep(0.6)
    print("\n✅ Ready!\n")
    time.sleep(0.3)

# ---- Adaptive loader for your brain module ----
def load_brain():
    for name in CANDIDATE_MODULES:
        try:
            mod = __import__(name)
            agent = None
            # common exported instances/names:
            if hasattr(mod, "AGENT"):
                agent = getattr(mod, "AGENT")
            elif hasattr(mod, "agent"):
                agent = getattr(mod, "agent")
            else:
                # try to instantiate a class if present
                cls_names = ["SuperAgentV7", "SuperAgent", "MultimodalBrain", "Agent", "Brain"]
                for cls in cls_names:
                    if hasattr(mod, cls):
                        try:
                            agent = getattr(mod, cls)()
                            break
                        except Exception:
                            agent = None
                # as last resort, if module defines functions, return module as agent
                if agent is None:
                    agent = mod
            print(f"[INFO] Loaded brain module: {name}")
            return agent
        except Exception:
            continue
    print("[WARN] Could not auto-import expected brain module names.")
    print("Place your multimodular module in the same folder and name it one of:", ", ".join(CANDIDATE_MODULES))
    return None

# ---- Helpers: flexible invocation for common brain actions ----
def brain_call(agent, fn_names, *args, **kwargs):
    """Try to call first available function name on agent; return (ok, result)."""
    if agent is None:
        return False, "Brain not loaded"
    for fn in fn_names:
        if callable(getattr(agent, fn, None)):
            try:
                return True, getattr(agent, fn)(*args, **kwargs)
            except Exception as e:
                return False, f"error calling {fn}: {e}"
    # If agent itself exposes a 'ctb_handle' as attribute inside (e.g., agent.chb.ctb_handle)
    try:
        # try nested common path: agent.chb.ctb_handle
        chb = getattr(agent, "chb", None)
        if chb:
            for fn in fn_names:
                if callable(getattr(chb, fn, None)):
                    try:
                        return True, getattr(chb, fn)(*args, **kwargs)
                    except Exception as e:
                        return False, f"error calling chb.{fn}: {e}"
    except Exception:
        pass
    return False, f"none of {fn_names} found on agent"

# ---- UI functions ----
menus = {
    "1": "💬 Chat with AI (All Features in One Chat)",
    "2": "🔎 Search Knowledge Base",
    "3": "📤 Upload Media for Learning",
    "4": "💾 Backup / Restore Brain (download backup)",
    "5": "🎨 View Creative Skill Vault (top skills)",
    "6": "🔁 Global Brain Sync Status",
    "7": "🛠 Developer API Options",
    "8": "📴 Offline Mode / Toggle",
    "9": "❌ Exit"
}

def show_menu():
    print("=== CHB v7.0 Main Menu ===")
    for k in sorted(menus.keys(), key=int):
        print(f"[{k}] {menus[k]}")

# ---- Media helpers (simple) ----
def read_file_as_payload(path):
    p = pathlib.Path(path)
    if not p.exists():
        return None, f"file not found: {path}"
    # minimal payload: path & size
    try:
        meta = {"path": str(p.resolve()), "size": p.stat().st_size}
        return {"path": str(p.resolve()), "meta": meta}, None
    except Exception as e:
        return None, f"read error: {e}"

# ---- Menu 1: Multimodal chat loop ----
def multimodal_chat(agent):
    print("\n=== Multimodal AI Chat ===")
    print("Type naturally. Special commands:")
    print("  /upload <path>   - attach a file (image, video, audio)")
    print("  /search <query>  - run user-device search (plan + return style)")
    print("  /skills <tag>    - show top creative skills for tag")
    print("  /backup          - create a new backup and show path")
    print("  /help            - show this help")
    print("  /exit            - return to main menu\n")
    while True:
        try:
            user = input("You: ").strip()
        except (KeyboardInterrupt, EOFError):
            print("\nReturning to main menu.")
            return
        if not user:
            continue
        if user.lower() in ("/exit", "exit", "quit"):
            print("Returning to main menu.\n")
            return
        if user.startswith("/upload "):
            path = user[len("/upload "):].strip().strip('"').strip("'")
            payload, err = read_file_as_payload(path)
            if err:
                print("Error:", err); continue
            # Build a simple plan_results-like structure and submit to brain
            # plan_results should include images/videos/audios lists if agent expects that shape
            plan_results = {}
            suffix = pathlib.Path(path).suffix.lower()
            if suffix in (".png", ".jpg", ".jpeg", ".webp", ".bmp"):
                plan_results["images"] = [{"path": payload["path"], "quality_score": 0.9, "caption": "", "tags": []}]
            elif suffix in (".mp4", ".mov", ".mkv", ".webm"):
                plan_results["videos"] = [{"path": payload["path"], "quality_score": 0.8, "caption": "", "tags": []}]
            elif suffix in (".mp3", ".wav", ".m4a", ".ogg"):
                plan_results["audios"] = [{"path": payload["path"], "quality_score": 0.8, "caption": "", "tags": []}]
            else:
                plan_results["files"] = [{"path": payload["path"], "meta": payload["meta"]}]
            ok, res = brain_call(agent, ["submit_plan_results", "handle_plan_results", "submit_results", "submit_plan"], plan_id="upload_"+str(int(time.time())), results=plan_results)
            if ok:
                print("AI: (processed upload) ->", res)
            else:
                print("AI: upload processed locally, but brain call failed:", res)
            continue
        if user.startswith("/search "):
            q = user[len("/search "):].strip()
            ok, plan = brain_call(agent, ["plan_search", "plan"], q)
            if ok:
                print("AI: Generated search plan. (Run this plan on client and submit results.)")
                print(json.dumps(plan, indent=2) if isinstance(plan, dict) else plan)
            else:
                print("AI: search plan generation failed:", plan)
            continue
        if user.startswith("/skills "):
            tag = user[len("/skills "):].strip()
            ok, skills = brain_call(agent, ["top_skills", "top_skill", "top_by_tag"], tag, 5)
            if ok:
                print("Top skills for", tag, ":", skills)
            else:
                print("Could not fetch skills:", skills)
            continue
        if user.strip() == "/backup":
            ok, path = brain_call(agent, ["download_latest_backup", "latest_backup", "get_latest_backup"])
            if ok and path:
                print("Latest backup path:", path)
            else:
                # try to create a new backup if method available
                ok2, created = brain_call(agent, ["backup_create", "create_backup", "create_backup_zip"])
                if ok2:
                    print("Created backup:", created)
                else:
                    print("Backup not available:", path or created)
            continue
        if user.strip() == "/help":
            print("Commands: /upload, /search, /skills, /backup, /exit")
            continue

        # Regular freeform input: call ctb_handle if present, else agent.chat or agent.chat()
        # Prefer 'ctb_handle' (Close-to-Human Brain multimodal pipeline), fall back to 'chat' or 'plan_search'
        ok, resp = brain_call(agent, ["ctb_handle", "handle_input", "chat", "chat_message", "chat_query"], input_data=user)
        if not ok:
            # try more permissive call signatures
            try:
                # some agents expect chat(text)
                resp = agent.chat(user)
                print("AI:", resp)
            except Exception as e:
                print("AI call failed:", resp)
        else:
            print("AI:", resp)

# ---- Menus 2..9 simple wrappers that call brain functions if present ----
def menu_search_kb(agent):
    q = input("Enter search query: ").strip()
    if not q: return
    ok, res = brain_call(agent, ["search_facts", "facts_search", "query_facts"], q)
    if ok:
        print("Results:", res)
    else:
        print("Search failed:", res)

def menu_upload_media(agent):
    path = input("Path to media file: ").strip()
    if not path: return
    payload, err = read_file_as_payload(path)
    if err:
        print("Error:", err); return
    # submit via same upload command as chat
    plan_results = {}
    suffix = pathlib.Path(path).suffix.lower()
    if suffix in (".png", ".jpg", ".jpeg", ".webp", ".bmp"):
        plan_results["images"] = [{"path": payload["path"], "quality_score": 0.9}]
    elif suffix in (".mp4", ".mov", ".mkv"):
        plan_results["videos"] = [{"path": payload["path"], "quality_score": 0.8}]
    elif suffix in (".mp3", ".wav"):
        plan_results["audios"] = [{"path": payload["path"], "quality_score": 0.8}]
    else:
        plan_results["files"] = [{"path": payload["path"], "meta": payload["meta"]}]
    ok, res = brain_call(agent, ["submit_plan_results", "handle_plan_results"], plan_id="manual_upload_"+str(int(time.time())), results=plan_results)
    if ok:
        print("Upload processed:", res)
    else:
        print("Upload failed:", res)

def menu_backup_download(agent):
    ok, p = brain_call(agent, ["download_latest_backup", "latest_backup", "get_latest_backup"])
    if ok and p:
        print("Latest backup:", p)
    else:
        print("No backup available or call failed:", p)

def menu_view_vault(agent):
    tag = input("Enter skill tag (or blank to list all): ").strip()
    if tag:
        ok, s = brain_call(agent, ["top_skills", "top_by_tag"], tag, 10)
    else:
        ok, s = brain_call(agent, ["list_skills", "get_skills"], )
    if ok:
        print("Skills:", s)
    else:
        print("Failed to retrieve skills:", s)

def menu_sync_status(agent):
    ok, st = brain_call(agent, ["global_sync_status", "sync_status", "get_sync_status"])
    if ok:
        print("Global Sync Status:", st)
    else:
        print("Global sync status not available:", st)

def menu_dev_api(agent):
    print("Developer API options:")
    print(" 1) Add/Integrate module from file")
    print(" 2) List modules")
    choice = input("choice: ").strip()
    if choice == "1":
        path = input("Path to module (py or base64-wasm): ").strip()
        payload, err = read_file_as_payload(path)
        if err:
            print("Error:", err); return
        code = ""
        try:
            code = open(payload["path"], "rb").read().decode("utf-8")
        except Exception:
            import base64
            code = base64.b64encode(open(payload["path"], "rb").read()).decode()
        name = input("Module name (short): ").strip() or f"mod_{int(time.time())}"
        ok, res = brain_call(agent, ["add_module", "integrate_module"], name, code, None)
        print("Result:", res)
    elif choice == "2":
        ok, res = brain_call(agent, ["list_modules", "get_modules"])
        print("Modules:", res if ok else "failed:"+str(res))
    else:
        print("cancel")

def menu_offline_toggle(agent):
    ok, st = brain_call(agent, ["toggle_offline", "set_offline", "offline_toggle"])
    if ok:
        print("Offline toggled:", st)
    else:
        print("Offline toggle not available; try starting/stopping network in your environment.")

# ---- Main loop ----
def main():
    boot_splash()
    agent = load_brain()
    if agent is None:
        print("Brain not loaded. You can still use app UI, but brain-dependent actions will fail.")
    while True:
        show_menu()
        choice = input("Select: ").strip()
        if choice == "1":
            multimodal_chat(agent)
        elif choice == "2":
            menu_search_kb(agent)
        elif choice == "3":
            menu_upload_media(agent)
        elif choice == "4":
            menu_backup_download(agent)
        elif choice == "5":
            menu_view_vault(agent)
        elif choice == "6":
            menu_sync_status(agent)
        elif choice == "7":
            menu_dev_api(agent)
        elif choice == "8":
            menu_offline_toggle(agent)
        elif choice == "9":
            print("Goodbye.")
            break
        else:
            print("Unknown option; try again.\n")

if __name__ == "__main__":
    main()