diff --git a/actions/browser_control.py b/actions/browser_control.py index 352927a..8f18767 100644 --- a/actions/browser_control.py +++ b/actions/browser_control.py @@ -11,22 +11,13 @@ import threading from pathlib import Path from typing import Optional -try: - from playwright.async_api import ( - async_playwright, - BrowserContext, - Page, - Playwright, - TimeoutError as PlaywrightTimeout, - ) - _PLAYWRIGHT_OK = True -except ImportError: - _PLAYWRIGHT_OK = False - async_playwright = None # type: ignore[assignment] - BrowserContext = None # type: ignore[assignment] - Page = None # type: ignore[assignment] - Playwright = None # type: ignore[assignment] - PlaywrightTimeout = Exception # type: ignore[assignment,misc] +from playwright.async_api import ( + async_playwright, + BrowserContext, + Page, + Playwright, + TimeoutError as PlaywrightTimeout, +) _OS = platform.system() # "Windows" | "Darwin" | "Linux" def _normalize_url(url: str) -> str: @@ -816,14 +807,6 @@ def browser_control( player=None, session_memory=None, ) -> str: - if not _PLAYWRIGHT_OK: - msg = ( - "Playwright is not installed.\n" - "Fix: pip install playwright && python -m playwright install chromium" - ) - _log(player, msg) - return msg - params = parameters or {} action = params.get("action", "").lower().strip() browser = params.get("browser", "").lower().strip() or None diff --git a/actions/code_helper.py b/actions/code_helper.py index 6cfe81f..38e6357 100644 --- a/actions/code_helper.py +++ b/actions/code_helper.py @@ -12,10 +12,26 @@ def get_base_dir(): return Path(__file__).resolve().parent.parent BASE_DIR = get_base_dir() +API_CONFIG_PATH = BASE_DIR / "config" / "api_keys.json" DESKTOP = Path.home() / "Desktop" MAX_BUILD_ATTEMPTS = 3 +GEMINI_MODEL = "gemini-2.5-flash" -from core.llm_client import call_llm_text as _llm + +def _get_api_key() -> str: + with open(API_CONFIG_PATH, "r", encoding="utf-8") as f: + return json.load(f)["gemini_api_key"] + + +def _get_gemini(model: str = GEMINI_MODEL): + from google import genai + _c = genai.Client(api_key=_get_api_key()) + + class _W: + def generate_content(self, contents): + return _c.models.generate_content(model=model, contents=contents) + + return _W() def _clean_code(text: str) -> str: @@ -134,28 +150,47 @@ def _detect_intent(description: str, file_path: str, code: str) -> str: return "write" def _write(description: str, language: str, output_path: str, player=None) -> tuple[str, Path]: - lang = language or "python" - system = f"You are an expert {lang} developer. Output ONLY raw code, no markdown, no backticks, no explanations." - prompt = ( - f"Write clean, working, well-commented {lang} code.\n" - f"Handle errors and edge cases. Use modern best practices.\n\n" - f"Description: {description}\n\nCode:" - ) - code = _clean_code(_llm(prompt, system=system)) - path = _resolve_save_path(output_path, lang) + lang = language or "python" + model = _get_gemini() + + prompt = f"""You are an expert {lang} developer. +Write clean, working, well-commented {lang} code for the description below. + +Rules: +- Output ONLY the code. No explanation, no markdown, no backticks. +- Add helpful inline comments. +- Handle errors and edge cases properly. +- Use modern best practices. + +Description: {description} + +Code:""" + + response = model.generate_content(prompt) + code = _clean_code(response.text) + path = _resolve_save_path(output_path, lang) _save_file(path, code) return code, path def _fix_code(code: str, error_output: str, description: str) -> str: - system = "You are an expert debugger. Return ONLY the corrected code — no explanation, no markdown, no backticks." - prompt = ( - f"Fix the code below. It failed with this error.\n\n" - f"Original goal: {description}\n\n" - f"Error:\n{error_output[:2000]}\n\n" - f"Broken code:\n{code}\n\nFixed code:" - ) - return _clean_code(_llm(prompt, system=system)) + model = _get_gemini() + prompt = f"""You are an expert debugger. +The code below failed with the following error. Fix it. +Return ONLY the corrected code — no explanation, no markdown, no backticks. + +Original goal: {description} + +Error: +{error_output[:2000]} + +Broken code: +{code} + +Fixed code:""" + + response = model.generate_content(prompt) + return _clean_code(response.text) def _run_file(path: Path, args: list, timeout: int) -> str: @@ -273,14 +308,21 @@ def _edit_action(file_path, instruction, player) -> str: if player: player.write_log("[Code] Editing file...") - system = "You are an expert code editor. Return ONLY the complete updated code — no explanation, no markdown, no backticks." - prompt = ( - f"Apply the following change to the code below.\n\n" - f"Change: {instruction}\n\n" - f"Original code:\n{content}\n\nUpdated code:" - ) + model = _get_gemini() + prompt = f"""You are an expert code editor. +Apply the following change to the code below. +Return ONLY the complete updated code — no explanation, no markdown, no backticks. + +Change: {instruction} + +Original code: +{content} + +Updated code:""" + try: - edited = _clean_code(_llm(prompt, system=system)) + response = model.generate_content(prompt) + edited = _clean_code(response.text) except Exception as e: return f"Could not edit code: {e}" @@ -300,13 +342,19 @@ def _explain_action(file_path, code, player) -> str: if player: player.write_log("[Code] Analyzing code...") - system = "You are an expert programmer. Explain code concisely in 3-6 sentences." - prompt = ( - f"Explain what this code does — what it does, how it works, important details.\n\n" - f"Code:\n{code[:4000]}\n\nExplanation:" - ) + model = _get_gemini() + prompt = f"""Explain what this code does in simple, clear language. +Focus on: what it does, how it works, and any important details. +Be concise — 3 to 6 sentences maximum. + +Code: +{code[:4000]} + +Explanation:""" + try: - return _llm(prompt, system=system) + response = model.generate_content(prompt) + return response.text.strip() except Exception as e: return f"Could not explain code: {e}" @@ -334,15 +382,26 @@ def _optimize_action(file_path, code, language, output_path, player) -> str: if player: player.write_log("[Code] Optimizing code...") - lang = language or "python" - system = f"You are an expert {lang} developer. Return ONLY the optimized code — no explanation, no markdown, no backticks." - prompt = ( - f"Optimize this {lang} code for performance, readability, and best practices. " - f"Remove dead code and unnecessary complexity.\n\n" - f"Original code:\n{code[:6000]}\n\nOptimized code:" - ) + lang = language or "python" + model = _get_gemini() + + prompt = f"""You are an expert {lang} developer and code reviewer. +Optimize the following code for: +1. Performance — eliminate unnecessary operations, use efficient data structures +2. Readability — clear variable names, proper formatting, logical structure +3. Best practices — modern {lang} patterns, error handling, type hints if applicable +4. Remove dead code, redundant comments, and unnecessary complexity + +Return ONLY the optimized code — no explanation, no markdown, no backticks. + +Original code: +{code[:6000]} + +Optimized code:""" + try: - optimized = _clean_code(_llm(prompt, system=system)) + response = model.generate_content(prompt) + optimized = _clean_code(response.text) except Exception as e: return f"Could not optimize code: {e}" @@ -387,51 +446,44 @@ def _screen_debug_action(description, file_path, player, speak=None) -> str: print(f"[Code] ⚠️ Could not read file: {err}") try: - import base64, json as _json, requests as _req - from pathlib import Path as _Path + from google import genai + from google.genai import types - cfg = {} - cfg_file = _Path(__file__).resolve().parent.parent / "config" / "api_keys.json" - try: - cfg = _json.loads(cfg_file.read_text(encoding="utf-8")) - except Exception: - pass + client = genai.Client(api_key=_get_api_key()) - ollama_url = cfg.get("llm_url", "http://localhost:11434").rstrip("/") - vision_model = cfg.get("vision_model") or cfg.get("llm_model", "llava") + image_bytes = screenshot_path.read_bytes() + image_base64 = _image_to_base64(screenshot_path) - image_bytes = screenshot_path.read_bytes() - b64 = base64.b64encode(image_bytes).decode("ascii") user_question = description or "What error or problem do you see on the screen? How can it be fixed?" - context = "" + + context = "" if file_content: - context = f"\n\nRelated file content:\n```\n{file_content[:4000]}\n```" + context = f"\n\nAdditionally, here is the related file content:\n```\n{file_content[:4000]}\n```" - analysis_prompt = ( - f"You are an expert programmer/debugger analyzing a screenshot.\n\n" - f"User question: {user_question}{context}\n\n" - f"Identify errors, explain the cause, and provide a fix. " - f"If you see code, show the corrected version." + analysis_prompt = f"""You are an expert programmer and debugger analyzing a screenshot. + +User's question: {user_question}{context} + +Please: +1. Identify any errors, exceptions, or problems visible on the screen +2. Explain what is causing the problem in simple terms +3. Provide a concrete fix or solution +4. If there's code visible, show the corrected version + +Be specific and actionable. If you see an error message, quote it exactly.""" + + contents = [ + types.Part.from_bytes(data=image_bytes, mime_type="image/png"), + analysis_prompt, + ] + + response = client.models.generate_content( + model="gemini-2.5-flash", + contents=contents, ) - resp = _req.post( - f"{ollama_url}/api/chat", - json={ - "model": vision_model, - "stream": False, - "messages": [ - { - "role": "user", - "content": analysis_prompt, - "images": [b64], - } - ], - }, - timeout=60, - ) - resp.raise_for_status() - analysis = (resp.json().get("message", {}).get("content") or "").strip() - print("[Code] ✅ Screen analysis complete") + analysis = response.text.strip() + print(f"[Code] ✅ Screen analysis complete") try: screenshot_path.unlink() @@ -439,6 +491,7 @@ def _screen_debug_action(description, file_path, player, speak=None) -> str: pass if file_path and file_content: + code_match = re.search(r"```[a-zA-Z]*\n(.*?)```", analysis, re.DOTALL) if code_match: fixed_code = code_match.group(1).strip() @@ -450,6 +503,7 @@ def _screen_debug_action(description, file_path, player, speak=None) -> str: return analysis except Exception as e: + try: screenshot_path.unlink() except Exception: diff --git a/actions/computer_control.py b/actions/computer_control.py index 92aa2d1..d1062b0 100644 --- a/actions/computer_control.py +++ b/actions/computer_control.py @@ -1,6 +1,7 @@ #computer_control.py import io import json +import platform import re import string import subprocess @@ -39,13 +40,17 @@ def _load_config() -> dict: except Exception: return {} -def _get_os() -> str: - import platform - s = platform.system().lower() - if s == "darwin": - return "mac" - return s # "windows" or "linux" +def _platform_os() -> str: + return {"Windows": "windows", "Darwin": "mac", "Linux": "linux"}.get( + platform.system(), "linux" + ) +def _get_os() -> str: + return _load_config().get("os_system", _platform_os()).lower() + + +def _get_api_key() -> str: + return _load_config().get("gemini_api_key", "") _SAFE_SCREENSHOT_ROOTS = ( Path.home(), @@ -160,7 +165,8 @@ def _smart_type(text: str, clear_first: bool = True) -> str: if len(text) > 20 and _PYPERCLIP: pyperclip.copy(text) time.sleep(0.1) - pyautogui.hotkey("ctrl", "v") + paste_key = "command" if _get_os() == "mac" else "ctrl" + pyautogui.hotkey(paste_key, "v") return f"Smart-typed (clipboard): {text[:60]}{'…' if len(text) > 60 else ''}" pyautogui.typewrite(text, interval=0.04) @@ -222,7 +228,8 @@ def _clipboard_paste(text: str) -> str: pyperclip.copy(text) time.sleep(0.1) _require_pyautogui() - pyautogui.hotkey("ctrl", "v") + paste_key = "command" if _get_os() == "mac" else "ctrl" + pyautogui.hotkey(paste_key, "v") return f"Pasted: {text[:60]}{'…' if len(text) > 60 else ''}" return "pyperclip not available" @@ -237,7 +244,8 @@ def _screenshot(save_path: str | None = None) -> str: def _clear_field() -> str: _require_pyautogui() - pyautogui.hotkey("ctrl", "a") + select_key = "command" if _get_os() == "mac" else "ctrl" + pyautogui.hotkey(select_key, "a") time.sleep(0.1) pyautogui.press("delete") return "Field cleared" @@ -298,27 +306,23 @@ def _focus_window(title: str) -> str: return f"focus_window: unknown OS '{os_name}'" def _screen_find(description: str) -> tuple[int, int] | None: + api_key = _get_api_key() + if not api_key: + print("[ComputerControl] ⚠️ No API key for screen_find") + return None + try: - import base64 - import requests as _req - import json as _json - - cfg_path = Path(__file__).resolve().parent.parent / "config" / "api_keys.json" - try: - cfg = _json.loads(cfg_path.read_text(encoding="utf-8")) - except Exception: - cfg = {} - - ollama_url = cfg.get("llm_url", "http://localhost:11434").rstrip("/") - vision_model = cfg.get("vision_model") or cfg.get("llm_model", "llava") + from google import genai + from google.genai import types as gtypes _require_pyautogui() w, h = pyautogui.size() img = pyautogui.screenshot() buf = io.BytesIO() img.save(buf, format="PNG") - b64 = base64.b64encode(buf.getvalue()).decode("ascii") + image_bytes = buf.getvalue() + client = genai.Client(api_key=api_key) prompt = ( f"This is a screenshot of a {w}×{h} pixel screen. " f"Locate the UI element described as: '{description}'. " @@ -326,18 +330,15 @@ def _screen_find(description: str) -> tuple[int, int] | None: f"If the element is not visible, reply: NOT_FOUND" ) - resp = _req.post( - f"{ollama_url}/api/chat", - json={ - "model": vision_model, - "stream": False, - "messages": [{"role": "user", "content": prompt, "images": [b64]}], - }, - timeout=30, + response = client.models.generate_content( + model="gemini-2.5-flash-lite", + contents=[ + gtypes.Part.from_bytes(data=image_bytes, mime_type="image/png"), + prompt, + ], ) - resp.raise_for_status() - text = (resp.json().get("message", {}).get("content") or "").strip() + text = (response.text or "").strip() if "NOT_FOUND" in text.upper(): return None diff --git a/actions/computer_settings.py b/actions/computer_settings.py index 12649d9..5a9d2ed 100644 --- a/actions/computer_settings.py +++ b/actions/computer_settings.py @@ -29,6 +29,11 @@ def _get_base_dir() -> Path: return Path(sys.executable).parent return Path(__file__).resolve().parent.parent +def _get_api_key() -> str: + path = _get_base_dir() / "config" / "api_keys.json" + with open(path, "r", encoding="utf-8") as f: + return json.load(f)["gemini_api_key"] + def _get_macos_wifi_interface() -> str: try: result = subprocess.run( @@ -564,24 +569,34 @@ _DANGEROUS_ACTIONS = {"restart", "shutdown"} def _detect_action(description: str) -> dict: - from core.llm_client import call_llm_text + + from google import genai as _genai + _client = _genai.Client(api_key=_get_api_key()) available = ", ".join(sorted(ACTION_MAP.keys())) + \ ", volume_set, type_text, press_key, reload_n" - system = ( - "You are an intent detector. Return ONLY a valid JSON object — " - "no explanation, no markdown." - ) - prompt = ( - f'User command (any language): "{description}"\n\n' - f"Available actions: {available}\n\n" - f'Return: {{"action": "action_name", "value": null_or_value}}\n' - f"Rules: volume_set→int 0-100; type_text→exact text; press_key→key name; reload_n→int." - ) + prompt = f"""You are an intent detector for a computer control assistant. + +The user issued a command (possibly in any language): "{description}" + +Available actions: {available} + +Return ONLY a valid JSON object: +{{"action": "action_name", "value": null_or_value}} + +Rules: +- Pick the single best matching action from the available list. +- For volume_set: value is an integer 0-100. +- For type_text: value is the exact text to type. +- For press_key: value is the key name (e.g. "f5", "tab", "enter"). +- For reload_n: value is an integer (number of times to reload). +- If no clear match, pick the closest action. +- Return ONLY the JSON, no explanation, no markdown.""" + try: - text = call_llm_text(prompt, system=system) - text = re.sub(r"```(?:json)?", "", text).strip().rstrip("`").strip() + resp = _client.models.generate_content(model="gemini-2.5-flash-lite", contents=prompt) + text = re.sub(r"```(?:json)?", "", resp.text).strip().rstrip("`").strip() return json.loads(text) except Exception as e: print(f"[Settings] Intent detection failed: {e}") diff --git a/actions/desktop.py b/actions/desktop.py index 495eeca..ee62717 100644 --- a/actions/desktop.py +++ b/actions/desktop.py @@ -23,6 +23,11 @@ def _get_base_dir() -> Path: return Path(sys.executable).parent return Path(__file__).resolve().parent.parent +def _get_api_key() -> str: + path = _get_base_dir() / "config" / "api_keys.json" + with open(path, "r", encoding="utf-8") as f: + return json.load(f)["gemini_api_key"] + def _get_desktop() -> Path: if _OS == "Linux": xdg = os.environ.get("XDG_DESKTOP_DIR", "") @@ -97,29 +102,48 @@ def _execute_generated_code(code: str, player=None) -> str: def _ask_gemini_for_desktop_action(task: str) -> str: - from core.llm_client import call_llm_text - desktop = str(_get_desktop()) + from google import genai as _genai + _client = _genai.Client(api_key=_get_api_key()) + + desktop = str(_get_desktop()) + os_specific = "" if _OS == "Windows": os_specific = "- ctypes (Windows API calls, read-only)\n- winreg (registry READ only)" + elif _OS == "Darwin": + os_specific = "- subprocess is NOT available; use pyautogui or Path only" else: os_specific = "- subprocess is NOT available; use pyautogui or Path only" - system = ( - "You are a desktop automation code generator. " - "Output ONLY raw Python code — no explanation, no markdown, no backticks." - ) - prompt = ( - f"Current OS: {_OS}\nDesktop path: {desktop}\n\n" - f"Allowed modules: pyautogui, pathlib.Path, shutil.copy2, shutil.disk_usage, time.sleep\n" - f"{os_specific}\n\n" - f"Rules: NO deletion, NO subprocess, NO exec/eval, NO imports.\n" - f"If unsafe, output exactly: UNSAFE\n\n" - f"Task: {task}" - ) + prompt = f"""You are a desktop automation assistant. +Current OS: {_OS} +Desktop path: {desktop} + +Generate safe Python code to accomplish the task below. +Allowed modules ONLY: +- pyautogui (mouse, keyboard — if needed) +- pathlib.Path (file/folder inspection only, no deletion) +- shutil.copy2, shutil.copytree, shutil.disk_usage (NO move, NO rmtree) +- os_path (os.path equivalent, read-only) +- time.sleep +{os_specific} + +Hard rules: +- NO file deletion (no unlink, no rmtree, no remove) +- NO subprocess calls +- NO exec() or eval() inside the code +- NO import statements (modules are pre-injected) +- NO file write operations except explicitly requested +- If task cannot be done safely with these tools, output exactly: UNSAFE + +Output ONLY the Python code. No explanation, no markdown, no backticks. + +Task: {task}""" + try: - code = call_llm_text(prompt, system=system).strip() + response = _client.models.generate_content(model="gemini-2.5-flash", contents=prompt) + code = response.text.strip() if code.startswith("```"): lines = code.split("\n") code = "\n".join(lines[1:-1]).strip() diff --git a/actions/dev_agent.py b/actions/dev_agent.py index 871f1b6..a93107c 100644 --- a/actions/dev_agent.py +++ b/actions/dev_agent.py @@ -13,10 +13,26 @@ def get_base_dir(): BASE_DIR = get_base_dir() +API_CONFIG_PATH = BASE_DIR / "config" / "api_keys.json" PROJECTS_DIR = Path.home() / "Desktop" / "JarvisProjects" MAX_FIX_ATTEMPTS = 5 +MODEL_PLANNER = "gemini-2.5-flash" +MODEL_WRITER = "gemini-2.5-flash" -from core.llm_client import call_llm_text as _llm +def _get_api_key() -> str: + with open(API_CONFIG_PATH, "r", encoding="utf-8") as f: + return json.load(f)["gemini_api_key"] + + +def _get_model(model_name: str): + from google import genai + _c = genai.Client(api_key=_get_api_key()) + + class _W: + def generate_content(self, contents): + return _c.models.generate_content(model=model_name, contents=contents) + + return _W() def _strip_fences(text: str) -> str: @@ -86,28 +102,26 @@ class RateLimitError(Exception): def _plan_project(description: str, language: str) -> dict: - system = ( - "You are a senior software architect. " - "Return ONLY valid JSON — no markdown, no explanation." - ) - prompt = f"""Create a minimal, complete file plan for this project. + model = _get_model(MODEL_PLANNER) + + prompt = f"""You are a senior software architect. Create a minimal, complete file plan for this project. Language: {language} Description: {description} -Return ONLY valid JSON: +Return ONLY valid JSON — no markdown, no explanation: {{ "project_name": "snake_case_name", "entry_point": "main.py", "files": [ {{ "path": "main.py", - "description": "Entry point", - "imports": ["utils.helpers"] + "description": "Entry point — what it does and which modules it imports", + "imports": ["utils.helpers", "core.engine"] }}, {{ "path": "utils/helpers.py", - "description": "Helpers", + "description": "Helper utilities — what functions it exposes", "imports": [] }} ], @@ -115,16 +129,25 @@ Return ONLY valid JSON: "dependencies": ["requests"] }} -Rules: dependency order, relative paths, entry point last, stdlib NOT in dependencies. +Critical rules: +1. List files in DEPENDENCY ORDER — files with no imports come first, entry point comes last. +2. The "imports" field must list every other project module this file imports (dot-notation, e.g. "utils.helpers"). +3. Keep it minimal — only files truly needed. +4. Entry point must be in the files list. +5. Use relative paths only (e.g. "utils/helpers.py", not absolute paths). +6. Standard library modules (os, sys, json, etc.) do NOT go in "dependencies". JSON:""" try: - raw = _strip_fences(_llm(prompt, system=system)) + response = model.generate_content(prompt) + raw = _strip_fences(response.text) return json.loads(raw) except json.JSONDecodeError as e: - raise ValueError(f"Planner returned invalid JSON: {e}") + raise ValueError(f"Planner returned invalid JSON: {e}\nRaw: {response.text[:300]}") except Exception as e: + if _is_rate_limit(e): + raise RateLimitError(str(e)) raise def _write_file( @@ -135,43 +158,80 @@ def _write_file( project_dir: Path, already_written: dict[str, str], ) -> str: - file_path = file_info["path"] - file_desc = file_info.get("description", "") + model = _get_model(MODEL_WRITER) + + file_path = file_info["path"] + file_desc = file_info.get("description", "") file_imports = file_info.get("imports", []) file_list = "\n".join( f" [{i+1}] {f['path']}: {f.get('description', '')}" for i, f in enumerate(all_files) ) - dep_ctx = "" + + dependency_context = "" for dep_dotted in file_imports: dep_path = dep_dotted.replace(".", "/") + ".py" if dep_path in already_written: - dep_ctx += f"\n\n--- {dep_path} ---\n{already_written[dep_path][:2000]}" + code_snippet = already_written[dep_path][:2000] + dependency_context += f"\n\n--- {dep_path} (you must import from this) ---\n{code_snippet}" - system = ( - f"You are a senior {language} developer. " - "Output ONLY raw code — no explanation, no markdown, no backticks." - ) - prompt = ( - f"Project goal: {project_description}\n\n" - f"Files (dependency order):\n{file_list}\n" - f"{dep_ctx}\n\n" - f"Write complete code for: {file_path}\n" - f"Purpose: {file_desc}\n" - f"{'Imports from: ' + ', '.join(file_imports) if file_imports else 'No project-internal imports.'}\n\n" - f"Rules: COMPLETE & RUNNABLE code, no placeholders, match import paths exactly.\n\n" - f"Code for {file_path}:" - ) + lang_rules = "" + if language.lower() == "python": + lang_rules = """ +Python-specific rules: +- Use type hints for all function signatures. +- Add docstrings for all public functions and classes. +- Use if __name__ == "__main__": guard in the entry point. +- For relative imports within the project, use: from utils.helpers import foo (match the project structure exactly). +- Do NOT use implicit relative imports (from . import ...) unless it's a proper package with __init__.py. +- If this is a package subdirectory, create __init__.py files where needed.""" + elif language.lower() in ("javascript", "typescript", "js", "ts"): + lang_rules = """ +JS/TS-specific rules: +- Use ES modules (import/export), not CommonJS (require). +- Add JSDoc comments for all exported functions. +- Handle promise rejections with try/catch in async functions.""" + + prompt = f"""You are a senior {language} developer writing production-quality code for a real project. + +Project goal: {project_description} + +Complete project file structure (in dependency order): +{file_list} + +{f"Dependencies this file must import from other project files:{dependency_context}" if dependency_context else ""} + +Your task: Write the complete, working code for: {file_path} +Purpose of this file: {file_desc} +{f"This file imports from: {', '.join(file_imports)}" if file_imports else "This file has no project-internal imports."} + +{lang_rules} + +General rules: +- Output ONLY raw code. Absolutely no explanation, no markdown, no triple backticks. +- Write COMPLETE, RUNNABLE code — no placeholders, no "# TODO", no "pass" stubs. +- Every import must either be from the standard library, listed dependencies, or the project files shown above. +- Match import paths EXACTLY to the file paths in the project structure (e.g. if file is "utils/helpers.py", import as "from utils.helpers import ..."). +- Use proper error handling (try/except) where I/O or network calls are made. +- The code must work correctly when the project entry point is run from the project root directory. + +Code for {file_path}:""" try: - code = _strip_fences(_llm(prompt, system=system)) + response = model.generate_content(prompt) + code = _strip_fences(response.text) + full_path = project_dir / file_path full_path.parent.mkdir(parents=True, exist_ok=True) full_path.write_text(code, encoding="utf-8") + print(f"[DevAgent] ✅ Written: {file_path} ({len(code)} chars)") return code + except Exception as e: + if _is_rate_limit(e): + raise RateLimitError(str(e)) raise def _install_dependencies(dependencies: list[str], project_dir: Path) -> str: @@ -295,10 +355,13 @@ def _fix_files( entry_point: str, ) -> dict[str, str]: + model = _get_model(MODEL_PLANNER) + error_file, error_line = _parse_traceback(error_output, list(file_codes.keys())) error_type = _classify_error(error_output) files_to_fix: list[str] = [] + if error_file: files_to_fix.append(error_file) if error_type == "import_error": @@ -312,40 +375,61 @@ def _fix_files( updated_codes: dict[str, str] = {} - system = f"You are an expert {language} debugger. Return ONLY the complete fixed code — no explanation, no markdown, no backticks." - for fix_path in files_to_fix: current_code = file_codes.get(fix_path, "") + other_ctx = "" for fp, code in file_codes.items(): if fp != fix_path and code: - other_ctx += f"\n--- {fp} ---\n{code[:1500]}\n" + snippet = code[:1500] + ("..." if len(code) > 1500 else "") + other_ctx += f"\n--- {fp} ---\n{snippet}\n" - line_hint = ( - f"\nError near line {error_line} in this file." - if error_line and fix_path == error_file else "" - ) + line_hint = f"\nError appears to be near line {error_line} in this file." if ( + error_line and fix_path == error_file + ) else "" - prompt = ( - f"Project goal: {project_description}\n\n" - f"All files:\n" - + "\n".join(f" - {f['path']}: {f.get('description', '')}" for f in all_files) - + f"\n\nOther files (context):\n{other_ctx[:3500]}\n\n" - f"File to fix: {fix_path}{line_hint}\n" - f"Error type: {error_type}\n\n" - f"Error output:\n{error_output[:2500]}\n\n" - f"Current code:\n{current_code}\n\n" - f"Fixed code for {fix_path}:" - ) + prompt = f"""You are an expert {language} debugger. Fix the broken file below. + +Project goal: {project_description} + +All project files: +{chr(10).join(f" - {f['path']}: {f.get('description', '')}" for f in all_files)} + +Other files for context (read-only — fix only the target file): +{other_ctx[:3500]} + +File to fix: {fix_path}{line_hint} +Error type: {error_type} + +Error output: +{error_output[:2500]} + +Current (broken) code: +{current_code} + +Rules: +- Output ONLY the complete fixed code. No explanation, no markdown, no backticks. +- Fix ALL errors visible in the error output. +- Keep all existing correct logic — do not remove working features. +- Ensure import paths match the actual project file structure exactly. +- Do NOT introduce new bugs or remove error handling. + +Fixed code for {fix_path}:""" try: - fixed = _strip_fences(_llm(prompt, system=system)) + response = model.generate_content(prompt) + fixed = _strip_fences(response.text) + full_path = project_dir / fix_path full_path.parent.mkdir(parents=True, exist_ok=True) full_path.write_text(fixed, encoding="utf-8") + updated_codes[fix_path] = fixed print(f"[DevAgent] 🔧 Fixed: {fix_path}") + except Exception as e: + if _is_rate_limit(e): + raise RateLimitError(str(e)) print(f"[DevAgent] ⚠️ Could not fix {fix_path}: {e}") return updated_codes diff --git a/actions/file_processor.py b/actions/file_processor.py index c0654c0..cbc2620 100644 --- a/actions/file_processor.py +++ b/actions/file_processor.py @@ -25,19 +25,21 @@ import tempfile from pathlib import Path from datetime import datetime -from core.llm_client import call_llm_text as _llm_text +def _get_api_key() -> str: + config_path = Path(__file__).resolve().parent.parent / "config" / "api_keys.json" + with open(config_path, "r", encoding="utf-8") as f: + return json.load(f)["gemini_api_key"] def _gemini_client(): - # Compatibility shim — returns an object with .generate_content() - class _Shim: - def generate_content(self, prompt: str) -> "_R": - class _R: - pass - r = _R() - r.text = _llm_text(prompt) - return r - return _Shim() + from google import genai + _c = genai.Client(api_key=_get_api_key()) + + class _W: + def generate_content(self, contents): + return _c.models.generate_content(model="gemini-2.5-flash", contents=contents) + + return _W() def _detect_type(path: Path) -> str: diff --git a/actions/flight_finder.py b/actions/flight_finder.py index c6e4b8a..270f5b1 100644 --- a/actions/flight_finder.py +++ b/actions/flight_finder.py @@ -18,6 +18,10 @@ BASE_DIR = _get_base_dir() API_CONFIG_PATH = BASE_DIR / "config" / "api_keys.json" +def _get_api_key() -> str: + with open(API_CONFIG_PATH, "r", encoding="utf-8") as f: + return json.load(f)["gemini_api_key"] + _MONTH_MAP: dict[str, int] = { "january": 1, "february": 2, "march": 3, "april": 4, @@ -58,16 +62,21 @@ def _parse_date(raw: str) -> str: return val.strftime("%Y-%m-%d") try: - from core.llm_client import call_llm_text - result = call_llm_text( - f"Today is {today.strftime('%Y-%m-%d')}. " - f"Convert this date expression to YYYY-MM-DD: '{raw}'. " - f"Return ONLY the date string, nothing else." - ).strip() + from google import genai as _genai + _client = _genai.Client(api_key=_get_api_key()) + response = _client.models.generate_content( + model="gemini-2.5-flash-lite", + contents=( + f"Today is {today.strftime('%Y-%m-%d')}. " + f"Convert this date expression to YYYY-MM-DD: '{raw}'. " + f"Return ONLY the date string, nothing else." + ) + ) + result = response.text.strip() if re.match(r"\d{4}-\d{2}-\d{2}", result): return result except Exception as e: - print(f"[FlightFinder] ⚠️ LLM date parse failed: {e}") + print(f"[FlightFinder] ⚠️ Gemini date parse failed: {e}") for month_name, month_num in _MONTH_MAP.items(): if month_name in lower: @@ -145,28 +154,36 @@ def _parse_flights_with_gemini( destination: str, date: str, ) -> list[dict]: - from core.llm_client import call_llm_text + from google import genai as _genai + from google.genai import types - system = ( - "You are a flight data extraction expert. " - "Extract flight information from raw webpage text. " - "Return ONLY valid JSON — no markdown, no explanation." - ) - prompt = ( + _client = _genai.Client(api_key=_get_api_key()) + prompt = ( f"Extract flight options from {origin} to {destination} on {date} " - f"from this Google Flights page text:\n\n{raw_text[:8000]}\n\n" + f"from this Google Flights page text:\n\n{raw_text[:12000]}\n\n" f"Return a JSON array of up to 5 flights:\n" f'[{{"airline":"...","departure":"HH:MM","arrival":"HH:MM",' f'"duration":"Xh Ym","stops":0,"price":"...","currency":"USD"}}]\n' f"If no flights found, return: []" ) + try: - text = call_llm_text(prompt, system=system) - text = re.sub(r"```(?:json)?", "", text).strip().rstrip("`").strip() - flights = json.loads(text) + response = _client.models.generate_content( + model="gemini-2.5-flash", + contents=prompt, + config=types.GenerateContentConfig( + system_instruction=( + "You are a flight data extraction expert. " + "Extract flight information from raw webpage text. " + "Return ONLY valid JSON — no markdown, no explanation." + ) + ), + ) + text = re.sub(r"```(?:json)?", "", response.text).strip().rstrip("`").strip() + flights = json.loads(text) return flights if isinstance(flights, list) else [] except Exception as e: - print(f"[FlightFinder] ⚠️ LLM parse failed: {e}") + print(f"[FlightFinder] ⚠️ Gemini parse failed: {e}") return [] def _format_spoken( diff --git a/actions/screen_processor.py b/actions/screen_processor.py index 23c388f..84db42e 100644 --- a/actions/screen_processor.py +++ b/actions/screen_processor.py @@ -1,272 +1,442 @@ -""" -MARK XL — Screen / Camera Processor -Replaces Gemini Live vision session with a direct Ollama vision-model call. -The analysis text is returned (and optionally spoken via the `speak` callback). -""" -from __future__ import annotations - -import base64 -import io -import json -import sys -from pathlib import Path -from typing import Optional, Callable - -try: - import cv2 - _CV2 = True -except ImportError: - _CV2 = False - -try: - import mss - import mss.tools - _MSS = True -except ImportError: - _MSS = False - -try: - import PIL.Image - _PIL = True -except ImportError: - _PIL = False - -import platform - -import requests - - -def _base_dir() -> Path: - if getattr(sys, "frozen", False): - return Path(sys.executable).parent - return Path(__file__).resolve().parent.parent - - -_BASE = _base_dir() -_CONFIG_PATH = _BASE / "config" / "api_keys.json" - -_IMG_MAX_W = 640 -_IMG_MAX_H = 360 -_JPEG_Q = 60 - - -def _load_config() -> dict: - try: - return json.loads(_CONFIG_PATH.read_text(encoding="utf-8")) - except Exception: - return {} - - -def _save_config_key(key: str, value) -> None: - try: - cfg = _load_config() - cfg[key] = value - _CONFIG_PATH.write_text(json.dumps(cfg, indent=4), encoding="utf-8") - except Exception as e: - print(f"[Vision] ⚠️ Could not save config key '{key}': {e}") - - -def _get_os() -> str: - s = platform.system().lower() - if s == "darwin": return "mac" - if s == "windows": return "windows" - return "linux" - - -# --------------------------------------------------------------------------- -# Image capture helpers (unchanged from original) -# --------------------------------------------------------------------------- - -def _compress(img_bytes: bytes, source_format: str = "PNG") -> tuple[bytes, str]: - if not _PIL: - return img_bytes, f"image/{source_format.lower()}" - try: - img = PIL.Image.open(io.BytesIO(img_bytes)).convert("RGB") - img.thumbnail((_IMG_MAX_W, _IMG_MAX_H), PIL.Image.BILINEAR) - buf = io.BytesIO() - img.save(buf, format="JPEG", quality=_JPEG_Q, optimize=False) - return buf.getvalue(), "image/jpeg" - except Exception as e: - print(f"[Vision] ⚠️ Image compress failed: {e}") - return img_bytes, f"image/{source_format.lower()}" - - -def _capture_screen() -> tuple[bytes, str]: - if not _MSS: - raise RuntimeError("mss is not installed. Run: pip install mss") - with mss.mss() as sct: - monitors = sct.monitors - target = monitors[1] if len(monitors) > 1 else monitors[0] - shot = sct.grab(target) - png = mss.tools.to_png(shot.rgb, shot.size) - return _compress(png, "PNG") - - -def _cv2_backend() -> int: - if not _CV2: - return 0 - os_name = _get_os() - if os_name == "windows": - return cv2.CAP_DSHOW - if os_name == "mac": - return cv2.CAP_AVFOUNDATION - return cv2.CAP_ANY - - -def _probe_camera(index: int, backend: int, warmup: int = 5) -> bool: - if not _CV2: - return False - import numpy as np - cap = cv2.VideoCapture(index, backend) - if not cap.isOpened(): - cap.release(); return False - for _ in range(warmup): - cap.read() - ret, frame = cap.read() - cap.release() - if not ret or frame is None: - return False - return bool(np.mean(frame) > 8) - - -def _detect_camera_index() -> int: - backend = _cv2_backend() - print("[Vision] 🔍 Auto-detecting camera…") - for idx in range(6): - if _probe_camera(idx, backend): - print(f"[Vision] ✅ Camera found at index {idx}") - _save_config_key("camera_index", idx) - return idx - print(f"[Vision] ⚠️ Camera index {idx}: no usable frame") - print("[Vision] ⚠️ No camera found — defaulting to index 0") - _save_config_key("camera_index", 0) - return 0 - - -def _get_camera_index() -> int: - cfg = _load_config() - if "camera_index" in cfg: - return int(cfg["camera_index"]) - return _detect_camera_index() - - -def _capture_camera() -> tuple[bytes, str]: - if not _CV2: - raise RuntimeError("OpenCV (cv2) is not installed. Run: pip install opencv-python") - import numpy as np - index = _get_camera_index() - backend = _cv2_backend() - cap = cv2.VideoCapture(index, backend) - if not cap.isOpened(): - raise RuntimeError(f"Camera index {index} could not be opened.") - for _ in range(10): - cap.read() - ret, frame = cap.read() - cap.release() - if not ret or frame is None: - raise RuntimeError("Camera returned no frame.") - if _PIL: - rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) - img = PIL.Image.fromarray(rgb) - img.thumbnail((_IMG_MAX_W, _IMG_MAX_H), PIL.Image.BILINEAR) - buf = io.BytesIO() - img.save(buf, format="JPEG", quality=_JPEG_Q) - return buf.getvalue(), "image/jpeg" - _, buf = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, _JPEG_Q]) - return buf.tobytes(), "image/jpeg" - - -# --------------------------------------------------------------------------- -# Vision analysis via Ollama -# --------------------------------------------------------------------------- - -_SYSTEM_PROMPT = ( - "You are JARVIS, an advanced AI assistant. " - "Analyze the provided image with precision and intelligence. " - "Be concise and direct — maximum two sentences unless the user's question " - "requires more detail. " - "Address the user respectfully." -) - - -def _call_vision(image_bytes: bytes, mime: str, user_text: str) -> str: - cfg = _load_config() - url = cfg.get("llm_url", "http://localhost:11434").rstrip("/") - vision_model = cfg.get("vision_model") or cfg.get("llm_model", "llava") - - b64 = base64.b64encode(image_bytes).decode("ascii") - - payload = { - "model": vision_model, - "stream": False, - "messages": [ - {"role": "system", "content": _SYSTEM_PROMPT}, - { - "role": "user", - "content": user_text, - "images": [b64], - }, - ], - } - try: - resp = requests.post(f"{url}/api/chat", json=payload, timeout=60) - resp.raise_for_status() - return (resp.json().get("message", {}).get("content") or "").strip() - except requests.exceptions.ConnectionError: - return "Cannot connect to Ollama. Make sure Ollama is running." - except Exception as e: - return f"Vision analysis failed: {e}" - - -# --------------------------------------------------------------------------- -# Public entry point -# --------------------------------------------------------------------------- - -def screen_process( - parameters: dict, - response=None, - player=None, - session_memory=None, - speak: Optional[Callable[[str], None]] = None, -) -> str: - """ - Capture screen or camera, analyse with Ollama vision model. - - Returns the analysis text (str). - Optionally speaks via `speak` callback and logs to `player`. - """ - params = parameters or {} - user_text = (params.get("text") or params.get("user_text") or "").strip() - angle = params.get("angle", "screen").lower().strip() - - if not user_text: - user_text = "What do you see? Describe briefly." - - if player: - player.write_log(f"SYS: Vision [{angle}] — {user_text[:60]}") - - # Capture - try: - if angle == "camera": - image_bytes, mime = _capture_camera() - print(f"[Vision] 📷 Camera: {len(image_bytes):,} bytes") - else: - image_bytes, mime = _capture_screen() - print(f"[Vision] 🖥️ Screen: {len(image_bytes):,} bytes") - except Exception as e: - msg = f"Capture error: {e}" - print(f"[Vision] ❌ {msg}") - if player: player.write_log(f"ERR: {msg}") - return msg - - # Analyse - analysis = _call_vision(image_bytes, mime, user_text) - print(f"[Vision] 💬 {analysis[:120]}") - - if player: - player.write_log(f"Jarvis: {analysis}") - - if speak and analysis: - speak(analysis) - - return analysis +from __future__ import annotations + +import asyncio +import base64 +import io +import json +import re +import sys +import threading +import time +from pathlib import Path +from typing import Optional + +import numpy as np +import sounddevice as sd + +try: + import cv2 + _CV2 = True +except ImportError: + _CV2 = False + +try: + import mss + import mss.tools + _MSS = True +except ImportError: + _MSS = False + +try: + import PIL.Image + _PIL = True +except ImportError: + _PIL = False + +from google import genai +from google.genai import types as gtypes + +def _base_dir() -> Path: + if getattr(sys, "frozen", False): + return Path(sys.executable).parent + return Path(__file__).resolve().parent.parent + + +_BASE = _base_dir() +_CONFIG_PATH = _BASE / "config" / "api_keys.json" + + +def _load_config() -> dict: + try: + return json.loads(_CONFIG_PATH.read_text(encoding="utf-8")) + except Exception: + return {} + + +def _save_config_key(key: str, value) -> None: + try: + cfg = _load_config() + cfg[key] = value + _CONFIG_PATH.write_text(json.dumps(cfg, indent=4), encoding="utf-8") + except Exception as e: + print(f"[Vision] ⚠️ Could not save config key '{key}': {e}") + + +def _get_api_key() -> str: + key = _load_config().get("gemini_api_key", "") + if not key: + raise RuntimeError("gemini_api_key not found in config.") + return key + + +def _get_os() -> str: + return _load_config().get("os_system", "windows").lower() + +_LIVE_MODEL = "models/gemini-2.5-flash-native-audio-preview-12-2025" +_CHANNELS = 1 +_RECEIVE_SAMPLE_RATE = 24_000 +_CHUNK_SIZE = 1_024 + +_IMG_MAX_W = 640 +_IMG_MAX_H = 360 +_JPEG_Q = 60 + +_SYSTEM_PROMPT = ( + "You are JARVIS, an advanced AI assistant. " + "Analyze the provided image with precision and intelligence. " + "Be concise and direct — maximum two sentences unless the user's question " + "requires more detail. " + "Address the user respectfully. " + "Always call the appropriate tool; never simulate results." +) + + +def _compress(img_bytes: bytes, source_format: str = "PNG") -> tuple[bytes, str]: + if not _PIL: + return img_bytes, f"image/{source_format.lower()}" + + try: + img = PIL.Image.open(io.BytesIO(img_bytes)).convert("RGB") + img.thumbnail((_IMG_MAX_W, _IMG_MAX_H), PIL.Image.BILINEAR) + buf = io.BytesIO() + img.save(buf, format="JPEG", quality=_JPEG_Q, optimize=False) + return buf.getvalue(), "image/jpeg" + except Exception as e: + print(f"[Vision] ⚠️ Image compress failed: {e}") + return img_bytes, f"image/{source_format.lower()}" + +def _capture_screen() -> tuple[bytes, str]: + + if not _MSS: + raise RuntimeError("mss is not installed. Run: pip install mss") + + with mss.mss() as sct: + monitors = sct.monitors # [0] = all combined, [1..n] = real screens + target = monitors[1] if len(monitors) > 1 else monitors[0] + shot = sct.grab(target) + png = mss.tools.to_png(shot.rgb, shot.size) + + return _compress(png, "PNG") + + +def _cv2_backend() -> int: + """Return the best OpenCV camera backend for the current OS.""" + if not _CV2: + return 0 + os_name = _get_os() + if os_name == "windows": + return cv2.CAP_DSHOW + if os_name == "mac": + return cv2.CAP_AVFOUNDATION + return cv2.CAP_ANY + + +def _probe_camera(index: int, backend: int, warmup: int = 5) -> bool: + + if not _CV2: + return False + cap = cv2.VideoCapture(index, backend) + if not cap.isOpened(): + cap.release() + return False + for _ in range(warmup): + cap.read() + ret, frame = cap.read() + cap.release() + if not ret or frame is None: + return False + return bool(np.mean(frame) > 8) + + +def _detect_camera_index() -> int: + + backend = _cv2_backend() + print("[Vision] 🔍 Auto-detecting camera...") + for idx in range(6): + if _probe_camera(idx, backend): + print(f"[Vision] ✅ Camera found at index {idx}") + _save_config_key("camera_index", idx) + return idx + print(f"[Vision] ⚠️ Camera index {idx}: no usable frame") + + print("[Vision] ⚠️ No camera found — defaulting to index 0") + _save_config_key("camera_index", 0) + return 0 + + +def _get_camera_index() -> int: + cfg = _load_config() + if "camera_index" in cfg: + return int(cfg["camera_index"]) + return _detect_camera_index() + + +def _capture_camera() -> tuple[bytes, str]: + if not _CV2: + raise RuntimeError("OpenCV (cv2) is not installed. Run: pip install opencv-python") + + index = _get_camera_index() + backend = _cv2_backend() + cap = cv2.VideoCapture(index, backend) + + if not cap.isOpened(): + raise RuntimeError(f"Camera index {index} could not be opened.") + + for _ in range(10): + cap.read() + + ret, frame = cap.read() + cap.release() + + if not ret or frame is None: + raise RuntimeError("Camera returned no frame.") + + if _PIL: + rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) + img = PIL.Image.fromarray(rgb) + img.thumbnail((_IMG_MAX_W, _IMG_MAX_H), PIL.Image.BILINEAR) + buf = io.BytesIO() + img.save(buf, format="JPEG", quality=_JPEG_Q) + return buf.getvalue(), "image/jpeg" + + _, buf = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, _JPEG_Q]) + return buf.tobytes(), "image/jpeg" + +class _VisionSession: + def __init__(self): + self._loop: Optional[asyncio.AbstractEventLoop] = None + self._thread: Optional[threading.Thread] = None + self._session = None + self._out_queue: Optional[asyncio.Queue] = None + self._audio_in: Optional[asyncio.Queue] = None + self._ready_evt: threading.Event = threading.Event() + self._player = None + self._lock: threading.Lock = threading.Lock() + + def start(self, player=None, timeout: float = 25.0) -> None: + with self._lock: + if self._thread and self._thread.is_alive(): + if player is not None: + self._player = player + return + self._player = player + self._thread = threading.Thread( + target=self._run_event_loop, + daemon=True, + name="VisionSessionThread", + ) + self._thread.start() + + if not self._ready_evt.wait(timeout=timeout): + raise RuntimeError(f"Vision session did not connect within {timeout}s.") + print("[Vision] ✅ Session ready") + + def analyze(self, image_bytes: bytes, mime_type: str, user_text: str) -> None: + if not self._loop or not self._out_queue: + print("[Vision] ⚠️ Session not started — dropping request") + return + asyncio.run_coroutine_threadsafe( + self._out_queue.put((image_bytes, mime_type, user_text)), + self._loop, + ) + + def is_ready(self) -> bool: + return self._session is not None + + def _run_event_loop(self) -> None: + self._loop = asyncio.new_event_loop() + asyncio.set_event_loop(self._loop) + self._loop.run_until_complete(self._session_loop()) + + async def _session_loop(self) -> None: + self._out_queue = asyncio.Queue(maxsize=30) + self._audio_in = asyncio.Queue() + + client = genai.Client( + api_key=_get_api_key(), + http_options={"api_version": "v1beta"}, + ) + config = gtypes.LiveConnectConfig( + response_modalities=["AUDIO"], + output_audio_transcription={}, + system_instruction=_SYSTEM_PROMPT, + speech_config=gtypes.SpeechConfig( + voice_config=gtypes.VoiceConfig( + prebuilt_voice_config=gtypes.PrebuiltVoiceConfig( + voice_name="Charon" + ) + ) + ), + ) + + backoff = 2.0 + while True: + try: + print("[Vision] 🔌 Connecting...") + async with client.aio.live.connect( + model=_LIVE_MODEL, config=config + ) as session: + self._session = session + self._ready_evt.set() + backoff = 2.0 + print("[Vision] ✅ Connected") + + async with asyncio.TaskGroup() as tg: + tg.create_task(self._send_loop()) + tg.create_task(self._recv_loop()) + tg.create_task(self._play_loop()) + + except* Exception as eg: + for exc in eg.exceptions: + print(f"[Vision] ⚠️ Session error: {exc}") + finally: + self._session = None + self._ready_evt.clear() + + print(f"[Vision] 🔄 Reconnecting in {backoff:.0f}s...") + await asyncio.sleep(backoff) + backoff = min(backoff * 1.5, 30.0) + self._ready_evt.set() + + async def _send_loop(self) -> None: + while True: + image_bytes, mime_type, user_text = await self._out_queue.get() + if not self._session: + print("[Vision] ⚠️ No session — dropping image") + continue + try: + b64 = base64.b64encode(image_bytes).decode("ascii") + await self._session.send_client_content( + turns={ + "parts": [ + {"inline_data": {"mime_type": mime_type, "data": b64}}, + {"text": user_text}, + ] + }, + turn_complete=True, + ) + print(f"[Vision] 📤 Sent {len(image_bytes):,} bytes — '{user_text[:60]}'") + except Exception as e: + print(f"[Vision] ⚠️ Send error: {e}") + + async def _recv_loop(self) -> None: + transcript: list[str] = [] + try: + async for response in self._session.receive(): + if response.data: + await self._audio_in.put(response.data) + + sc = response.server_content + if not sc: + continue + + if sc.output_transcription and sc.output_transcription.text: + chunk = sc.output_transcription.text.strip() + if chunk: + transcript.append(chunk) + + if sc.turn_complete: + if transcript and self._player: + full = re.sub(r"\s+", " ", " ".join(transcript)).strip() + if full: + self._player.write_log(f"Jarvis: {full}") + print(f"[Vision] 💬 {full}") + transcript = [] + + except Exception as e: + print(f"[Vision] ⚠️ Recv error: {e}") + raise + + async def _play_loop(self) -> None: + stream = sd.RawOutputStream( + samplerate=_RECEIVE_SAMPLE_RATE, + channels=_CHANNELS, + dtype="int16", + blocksize=_CHUNK_SIZE, + ) + stream.start() + try: + while True: + chunk = await self._audio_in.get() + await asyncio.to_thread(stream.write, chunk) + except Exception as e: + print(f"[Vision] ❌ Play error: {e}") + raise + finally: + stream.stop() + stream.close() + +_session = _VisionSession() +_session_lock = threading.Lock() +_session_up = False + + +def _ensure_session(player=None) -> None: + global _session_up + with _session_lock: + if not _session_up: + _session.start(player=player) + _session_up = True + elif player is not None: + _session._player = player + + +def screen_process( + parameters: dict, + response=None, + player=None, + session_memory=None, +) -> bool: + + params = parameters or {} + user_text = (params.get("text") or params.get("user_text") or "").strip() + angle = params.get("angle", "screen").lower().strip() + + if not user_text: + print("[Vision] ⚠️ No question provided — aborting") + return False + + print(f"[Vision] ▶ angle={angle!r} question='{user_text[:80]}'") + + try: + _ensure_session(player=player) + except Exception as e: + print(f"[Vision] ❌ Could not start session: {e}") + return False + + try: + if angle == "camera": + image_bytes, mime_type = _capture_camera() + print(f"[Vision] 📷 Camera: {len(image_bytes):,} bytes") + else: + image_bytes, mime_type = _capture_screen() + print(f"[Vision] 🖥️ Screen: {len(image_bytes):,} bytes") + except Exception as e: + print(f"[Vision] ❌ Capture error: {e}") + return False + + _session.analyze(image_bytes, mime_type, user_text) + return True + + +def warmup_session(player=None) -> None: + try: + _ensure_session(player=player) + except Exception as e: + print(f"[Vision] ⚠️ Warmup failed: {e}") + +if __name__ == "__main__": + print("[TEST] screen_processor.py") + print("=" * 52) + mode = input("angle — screen / camera (default: screen): ").strip().lower() or "screen" + q = input("Question (Enter = default): ").strip() or "What do you see? Be brief." + + t0 = time.perf_counter() + warmup_session() + print(f"Session ready in {time.perf_counter()-t0:.2f}s\n") + + t1 = time.perf_counter() + ok = screen_process({"angle": mode, "text": q}) + print(f"Queued in {time.perf_counter()-t1:.3f}s — waiting for audio...") + time.sleep(10) + print("Done." if ok else "Failed.") \ No newline at end of file diff --git a/actions/web_search.py b/actions/web_search.py index 192d698..22a9109 100644 --- a/actions/web_search.py +++ b/actions/web_search.py @@ -1,117 +1,141 @@ -# web_search.py -# Gemini grounded-search replaced with DuckDuckGo + Ollama LLM summarization. -import json -import sys -from pathlib import Path - - -def _get_base_dir() -> Path: - if getattr(sys, "frozen", False): - return Path(sys.executable).parent - return Path(__file__).resolve().parent.parent - - -BASE_DIR = _get_base_dir() - - -def _ddg_search(query: str, max_results: int = 6) -> list[dict]: - try: - from ddgs import DDGS - except ImportError: - from duckduckgo_search import DDGS - - results = [] - with DDGS() as ddgs: - for r in ddgs.text(query, max_results=max_results): - results.append({ - "title": r.get("title", ""), - "snippet": r.get("body", ""), - "url": r.get("href", ""), - }) - return results - - -def _format_ddg(query: str, results: list[dict]) -> str: - if not results: - return f"No results found for: {query}" - lines = [f"Search results for: {query}\n"] - for i, r in enumerate(results, 1): - if r.get("title"): lines.append(f"{i}. {r['title']}") - if r.get("snippet"): lines.append(f" {r['snippet']}") - if r.get("url"): lines.append(f" {r['url']}") - lines.append("") - return "\n".join(lines).strip() - - -def _llm_summarize(query: str, raw_results: str) -> str: - try: - from core.llm_client import call_llm_text - system = ( - "You are JARVIS. Summarize web search results clearly and concisely. " - "Answer the user's query directly. Be factual. Address user as 'sir'." - ) - prompt = ( - f"User question: {query}\n\n" - f"Web search results:\n{raw_results[:4000]}\n\n" - "Answer the question based on these results:" - ) - return call_llm_text(prompt, system=system) - except Exception: - return raw_results - - -def _compare(items: list[str], aspect: str) -> str: - all_results: dict[str, list] = {} - for item in items: - try: - all_results[item] = _ddg_search(f"{item} {aspect}", max_results=3) - except Exception: - all_results[item] = [] - - lines = [f"Comparison — {aspect.upper()}", "─" * 40] - for item in items: - lines.append(f"\n▸ {item}") - for r in all_results.get(item, [])[:2]: - if r.get("snippet"): - lines.append(f" • {r['snippet']}") - raw = "\n".join(lines) - return _llm_summarize(f"Compare {', '.join(items)} regarding {aspect}", raw) - - -def web_search( - parameters: dict, - response=None, - player=None, - session_memory=None, -) -> str: - params = parameters or {} - query = params.get("query", "").strip() - mode = params.get("mode", "search").lower().strip() - items = params.get("items", []) - aspect = params.get("aspect", "general").strip() or "general" - - if not query and not items: - return "Please provide a search query, sir." - - if items and mode != "compare": - mode = "compare" - - if player: - player.write_log(f"[Search] {query or ', '.join(items)}") - - print(f"[WebSearch] 🔍 Query: {query!r} Mode: {mode}") - - try: - if mode == "compare" and items: - print(f"[WebSearch] 📊 Comparing: {items}") - return _compare(items, aspect) - - results = _ddg_search(query) - raw = _format_ddg(query, results) - print(f"[WebSearch] ✅ DDG: {len(results)} result(s).") - # Let Ollama summarise the results for a cleaner spoken response - return _llm_summarize(query, raw) - - except Exception as e: - print(f"[WebSearch] ❌ Failed: {e}") - return f"Search failed, sir: {e}" +#web_search.py +import json +import sys +from pathlib import Path + +def _get_base_dir() -> Path: + if getattr(sys, "frozen", False): + return Path(sys.executable).parent + return Path(__file__).resolve().parent.parent + + +BASE_DIR = _get_base_dir() +API_CONFIG_PATH = BASE_DIR / "config" / "api_keys.json" + + +def _get_api_key() -> str: + with open(API_CONFIG_PATH, "r", encoding="utf-8") as f: + return json.load(f)["gemini_api_key"] + + +def _gemini_search(query: str) -> str: + from google import genai + + client = genai.Client(api_key=_get_api_key()) + response = client.models.generate_content( + model="gemini-2.5-flash", + contents=query, + config={"tools": [{"google_search": {}}]}, + ) + + text = "" + for part in response.candidates[0].content.parts: + if hasattr(part, "text") and part.text: + text += part.text + + text = text.strip() + if not text: + raise ValueError("Gemini returned an empty response.") + return text + + +def _ddg_search(query: str, max_results: int = 6) -> list[dict]: + try: + from ddgs import DDGS + except ImportError: + from duckduckgo_search import DDGS + + results = [] + with DDGS() as ddgs: + for r in ddgs.text(query, max_results=max_results): + results.append({ + "title": r.get("title", ""), + "snippet": r.get("body", ""), + "url": r.get("href", ""), + }) + return results + + +def _format_ddg(query: str, results: list[dict]) -> str: + if not results: + return f"No results found for: {query}" + + lines = [f"Search results for: {query}\n"] + for i, r in enumerate(results, 1): + if r.get("title"): lines.append(f"{i}. {r['title']}") + if r.get("snippet"): lines.append(f" {r['snippet']}") + if r.get("url"): lines.append(f" {r['url']}") + lines.append("") + return "\n".join(lines).strip() + +def _compare(items: list[str], aspect: str) -> str: + query = ( + f"Compare {', '.join(items)} in terms of {aspect}. " + "Give specific facts and data." + ) + try: + return _gemini_search(query) + except Exception as e: + print(f"[WebSearch] ⚠️ Gemini compare failed: {e} — falling back to DDG") + + # DDG fallback: fetch results per item and merge + all_results: dict[str, list] = {} + for item in items: + try: + all_results[item] = _ddg_search(f"{item} {aspect}", max_results=3) + except Exception: + all_results[item] = [] + + lines = [f"Comparison — {aspect.upper()}", "─" * 40] + for item in items: + lines.append(f"\n▸ {item}") + for r in all_results.get(item, [])[:2]: + if r.get("snippet"): + lines.append(f" • {r['snippet']}") + return "\n".join(lines) + +def web_search( + parameters: dict, + response=None, + player=None, + session_memory=None, +) -> str: + params = parameters or {} + query = params.get("query", "").strip() + mode = params.get("mode", "search").lower().strip() + items = params.get("items", []) + aspect = params.get("aspect", "general").strip() or "general" + + if not query and not items: + return "Please provide a search query, sir." + + if items and mode != "compare": + mode = "compare" + + if player: + player.write_log(f"[Search] {query or ', '.join(items)}") + + print(f"[WebSearch] 🔍 Query: {query!r} Mode: {mode}") + + try: + if mode == "compare" and items: + print(f"[WebSearch] 📊 Comparing: {items}") + result = _compare(items, aspect) + print("[WebSearch] ✅ Compare done.") + return result + + print("[WebSearch] 🌐 Trying Gemini...") + try: + result = _gemini_search(query) + print("[WebSearch] ✅ Gemini OK.") + return result + except Exception as e: + print(f"[WebSearch] ⚠️ Gemini failed ({e}) — trying DDG...") + results = _ddg_search(query) + result = _format_ddg(query, results) + print(f"[WebSearch] ✅ DDG: {len(results)} result(s).") + return result + + except Exception as e: + print(f"[WebSearch] ❌ All backends failed: {e}") + return f"Search failed, sir: {e}" \ No newline at end of file diff --git a/actions/youtube_video.py b/actions/youtube_video.py index 6cc8e70..aa51df9 100644 --- a/actions/youtube_video.py +++ b/actions/youtube_video.py @@ -9,6 +9,18 @@ from pathlib import Path from datetime import datetime from urllib.parse import quote_plus +try: + import pyautogui + _PYAUTOGUI = True +except ImportError: + _PYAUTOGUI = False + +try: + import numpy as np + _NUMPY = True +except ImportError: + _NUMPY = False + try: import requests _REQUESTS_OK = True @@ -45,6 +57,11 @@ HEADERS = { _YT_VIDEO_FILTER = "EgIQAQ%3D%3D" +def _get_api_key() -> str: + with open(API_CONFIG_PATH, "r", encoding="utf-8") as f: + return json.load(f)["gemini_api_key"] + + def _open_url(url: str) -> None: try: if is_mac(): @@ -150,20 +167,26 @@ def _get_transcript(video_id: str) -> str | None: def _summarize_with_gemini(transcript: str, video_url: str) -> str: - from core.llm_client import call_llm_text - system = ( - "You are JARVIS, an AI assistant. " - "Summarize YouTube video transcripts clearly and concisely. " - "Structure: 1-sentence overview, then 3-5 key points. " - "Be direct. Address the user as 'sir'. " - "Match the language of the transcript." - ) - max_chars = 12000 + from google import genai as _genai + from google.genai import types + + _client = _genai.Client(api_key=_get_api_key()) + max_chars = 80000 truncated = transcript[:max_chars] + ("..." if len(transcript) > max_chars else "") - return call_llm_text( - f"Please summarize this YouTube video transcript:\n\n{truncated}", - system=system, + response = _client.models.generate_content( + model="gemini-2.5-flash", + contents=f"Please summarize this YouTube video transcript:\n\n{truncated}", + config=types.GenerateContentConfig( + system_instruction=( + "You are JARVIS, an AI assistant. " + "Summarize YouTube video transcripts clearly and concisely. " + "Structure: 1-sentence overview, then 3-5 key points. " + "Be direct. Address the user as 'sir'. " + "Match the language of the transcript." + ) + ) ) + return response.text.strip() def _save_summary(content: str, video_url: str) -> str: diff --git a/config/__init__.py b/config/__init__.py index 13f1a5f..4ea0c24 100644 --- a/config/__init__.py +++ b/config/__init__.py @@ -1,10 +1,15 @@ # config/__init__.py -import json -import platform +import json, os, platform from pathlib import Path _CONFIG_PATH = Path(__file__).parent / "api_keys.json" +def _platform_os() -> str: + """Auto-detect OS when config file is absent.""" + return {"Windows": "windows", "Darwin": "mac", "Linux": "linux"}.get( + platform.system(), "linux" + ) + def get_config() -> dict: try: with open(_CONFIG_PATH, "r", encoding="utf-8") as f: @@ -13,13 +18,8 @@ def get_config() -> dict: return {} def get_os() -> str: - """Returns 'windows' | 'mac' | 'linux' — detected at runtime, not from config.""" - s = platform.system().lower() - if s == "darwin": - return "mac" - if s == "windows": - return "windows" - return "linux" + """Returns: 'windows' | 'mac' | 'linux'""" + return get_config().get("os_system", _platform_os()).lower() def is_windows() -> bool: return get_os() == "windows" def is_mac() -> bool: return get_os() == "mac" diff --git a/config/certs/jarvis.crt b/config/certs/jarvis.crt new file mode 100644 index 0000000..ce02b97 --- /dev/null +++ b/config/certs/jarvis.crt @@ -0,0 +1,19 @@ +-----BEGIN CERTIFICATE----- +MIIDAjCCAeqgAwIBAgIURVhBe4gYCY9qgmh+hvLhAsJW1YUwDQYJKoZIhvcNAQEL +BQAwKDEPMA0GA1UECgwGSkFSVklTMRUwEwYDVQQDDAxqYXJ2aXMubG9jYWwwHhcN +MjYwNjE2MTI1NjQwWhcNMzYwNjEzMTI1NjQwWjAoMQ8wDQYDVQQKDAZKQVJWSVMx +FTATBgNVBAMMDGphcnZpcy5sb2NhbDCCASIwDQYJKoZIhvcNAQEBBQADggEPADCC +AQoCggEBANg3svE1mhDUdxhnyR/XK/KnRNLTM2Tkw//UHQOTvRSOvFOOLLVbNypm ++fWBl4dZpfOGQTQxQZA3yLk+Gfnh3TxcvX8RwHcJFbJ72n0eq89aibILsU1X4gPU +m/A8Bz/417gTsQB6aRyRvR8dGAH7kjYTSvcmFOuftApOP7opRHbE221Y+TTZ9hdx +6aW/Fn4NAyQnugmqA55yHOys4yXhHRbsy9boYbUr4S4racwuEYgowDa/GAt7cj/Z +E2vWyV1AFQE4DOlhsX1Ves92vqNIQanakvrvuGCWx6TicapmyK7ftaGKdTEaL6Ix +8ilX8qci5qhqxr+sNK51Vg0CxW16+e8CAwEAAaMkMCIwIAYDVR0RBBkwF4IJbG9j +YWxob3N0hwR/AAABhwTAqAFrMA0GCSqGSIb3DQEBCwUAA4IBAQAd9+mMPvKyoghf +iBqyZN/XFBy1Va7pCMrVT2HDRO+NEIvUsgj1rl+LiODLqd7M0Hz0ItJzqdG9kuCg +obbbd3upT7HOxQi311UCg/ioeiZVP427Apt5Hc5qLSqIxeL1EK1kZIRbClwAxWhB +Za0ULqzJrcNx328s4loYJkDAFdDl3NTjtx5Gb5sOvE2Bsr5tFJ/u1SpBdO+M0hBB ++CQlhwLk5m8c6H841fW8Rex+2K+VBeRmGkze1ooKztpkC2nwV9o+QbUINsQ75p/x +JVF6JHKINERC/hQPub62CrEfhVE3UD/wT3bS1YQfdt/l6R0KNFctAQ1NdJJfufoq +cZdVMlqE +-----END CERTIFICATE----- diff --git a/config/certs/jarvis.key b/config/certs/jarvis.key new file mode 100644 index 0000000..b7b1138 --- /dev/null +++ b/config/certs/jarvis.key @@ -0,0 +1,27 @@ +-----BEGIN RSA PRIVATE KEY----- +MIIEowIBAAKCAQEA2Dey8TWaENR3GGfJH9cr8qdE0tMzZOTD/9QdA5O9FI68U44s +tVs3Kmb59YGXh1ml84ZBNDFBkDfIuT4Z+eHdPFy9fxHAdwkVsnvafR6rz1qJsgux +TVfiA9Sb8DwHP/jXuBOxAHppHJG9Hx0YAfuSNhNK9yYU65+0Ck4/uilEdsTbbVj5 +NNn2F3Hppb8Wfg0DJCe6CaoDnnIc7KzjJeEdFuzL1uhhtSvhLitpzC4RiCjANr8Y +C3tyP9kTa9bJXUAVATgM6WGxfVV6z3a+o0hBqdqS+u+4YJbHpOJxqmbIrt+1oYp1 +MRovojHyKVfypyLmqGrGv6w0rnVWDQLFbXr57wIDAQABAoIBACCIePdMTms73NOi +QiP1V1qCipuCRdhnh4pF2zu/jhK6y2FzIg0DtZA+wFuNbURjW9IBoGv/AVN24U/m +JThxIL3FYontHN2MU2RwuFZdOBLSF6fflIUDtTFIW1JRZfHdOnArHlXpFUnEMAWv +df5OYRib3IrFgx5VAF0Qmwwiy3dnLpgzDLLSGdWv8GSVqi90cafSJ/WFlPNVo6c4 +kTHov3rdfk82bBL7b8tLJmJMw4kYfJ9vft4cAiWqODTy4x57ffH5v8atRzqNjq5D +jYjfeeymiHZeOLnbEpdZTUG5PfBxBamFZfT8Ll1VTEDzh+s2+VNEROZx02Y+/Rts +DkKcH+ECgYEA8/SvI7Nkgtl8V9wxq3i/SZR4aTnrbupgtXQsO8GC/bmJ281An/BJ +eyNU7W2F0uRLE8vOdKVBKXFYpYVK1tFcriAfcsIqnhj0sXxRlylZpVtxuFNFzkyR +t3eKM1UM/fMLgFXDC+DjVGBKfDeUhkwRDegDH3sHMYE0mmxL/Zov4tECgYEA4uRv +qIt8IYx4g5K4ugrebYPaAENpeAt2fv6wQVqBB57i643a3cvaxZx6zOokG7NP4fN3 +DkOz9M5I0xDO+gN5XdvVf/KXUKoCuWOzvIhQaQ5nearxaHrHL9ajYmsXzuVKF/+B +H3iAf5OIqZL1vSBmpZOUvjmXpYi6lYGvwOBkwL8CgYAag8ahV7cd7atxFTKk02vf +AG8R6MSo+npmXmkMWZz2HLtaE3XprEEgTBfH8nZwFHYAgc8/qFJrwbED1Hfs3zZ2 +inFAlM/y/P1jaQKpLkpbCZRTl2Fn6bSQv0JbD+xokYj3etrHKLssiYau94+CBTR5 +UXU9b665d7SS8CBWQPLhcQKBgGyVbmp29p0HoNU27nj6oqA+raHaoSr2MWttb+SW +75E6FNcuOERVPwm7bXmLxGQNWHmGDsAg3oqV/nwkWymDaB3nCzIEYy4TbuJSOL5N +bk4YuqSbjpS1zYWzbtW28tntGZyKhr0M6lIjxQ+CaWBtlTJAgrrWzSBHYyQRwI7J +1O35AoGBAIlj3kff7/2vo8epr/zZHjQHh1YqqMfEKA/tlf5VdARpXDcXOJ3ffgpq +EgKmDoQqN1qaKLXR/5b0NDTP47h0DmPbB0u6cJd9GOUpsqdw1dDiK64ZcxaMHHg+ +Bt3IAC69opWKKvb6i3hch0uRjz30uPhhFVY1Cp4ECQzNeB4vCqE8 +-----END RSA PRIVATE KEY----- diff --git a/core/prompt.txt b/core/prompt.txt index ee3acfd..edb6f1a 100644 --- a/core/prompt.txt +++ b/core/prompt.txt @@ -1,51 +1,24 @@ 🧠 JARVIS CORE PROTOCOL -IDENTITY: You are Jarvis. Efficient, professional, direct assistant. No fluff. - -TOOL CALLING — ABSOLUTE RULE: -You have tools. For EVERY user request that involves an action, you MUST call the tool. -NEVER describe what you are doing in text. NEVER say "opening WhatsApp" or "-weather_report" — CALL THE TOOL. -NEVER simulate, pretend, or narrate an action. If a tool exists for it, USE IT. -Saying "Sure, let me open X" WITHOUT calling open_app is a critical failure. +IDENTITY: Efficient, professional, direct assistant. No fluff. EXECUTION RULES: -Action Flow: Call the tool immediately. For slow tools (search, vision), say ONE short sentence first. -Vision (screen_process): Say one curious sentence, then call the tool — total silence after. +Vision (screen_process): Say one curious sentence, then total silence (Vision module takes over). One-Call Policy: Never guess. Call tools exactly once. No retries. -Briefing: HARD LIMIT — 3 SHORT sentences maximum for ANY response. No exceptions. - • Confirmations / acks: 1 sentence only. - • Answers / results: 2-3 short sentences max. - • NEVER ask follow-up questions ("Would you like...?", "Is there anything else?"). - • NEVER pad with sympathy openers or filler phrases ("Of course!", "Great question!"). - • If you cannot fit the answer in 3 sentences, give the most important part only. +Briefing: 2-3 short sentence maximum. Report results briefly and always try to help the user. +Memory: Store critical user preferences/context automatically. Exit: Only call shutdown_jarvis if session termination is explicit. -Response time: Respond as fast as you can. -Always respond in the language that user is speaking. - -MEMORY RULE — MANDATORY, NO EXCEPTIONS: -Call save_memory IMMEDIATELY (in the same response, silently) whenever the user reveals ANY personal info. -Do NOT ask for permission. Do NOT announce that you are saving. Just call it. - -Trigger examples: -- "my name is Fatih" → save_memory(category="identity", key="name", value="Fatih") -- "I'm 22" / "I'm 22 years old" → save_memory(category="identity", key="age", value="22") -- "call me X" / "my name is X, not Y" → save_memory(category="identity", key="name", value="X") -- "I live in Istanbul" → save_memory(category="identity", key="city", value="Istanbul") -- "I'm a student / developer / designer" → save_memory(category="identity", key="job", value="...") -- "I prefer dark mode" / "I like X" → save_memory(category="preferences", key="...", value="...") -- "I'm working on X project" → save_memory(category="projects", key="...", value="...") -- Any name correction ("not Travis, Fatih") → save_memory identity/name immediately +Response time: Respond as fast as you can +Always call sir to user. +Act: Always act like Jarvis. If user says "Daddy's home" and other sentences like in the Iron Man movie, be like a real friendly assistant like Jarvis(like saying: "welcome home sir" and then continue to speak to help to user according to his text). +Always create 2-3 short sentence. Never less. TOOL ROUTING: computer_settings: ALL single OS actions (volume, brightness, wifi, close, shortcuts, power). -agent_task: ONLY for complex, multi-step planning (3+ steps) and call it if user it really specifies it. +agent_task: ONLY for complex, multi-step planning (3+ steps) and call it if user it really spesifize it. Language: Respond in user's language; extract parameters in English. -Do not call agent_task while you can accomplish it with a tool. +Do not call agent_task while you can accomplish it with a tool -OPEN vs MESSAGE rule (critical): -- "open WhatsApp", "launch WhatsApp", "pull up WhatsApp" → open_app (NOT send_message) -- "send a message to X saying Y" → send_message -- If user only says to open an app, NEVER call send_message, even if the app is a messaging platform. CRITICAL: Speak/Take action immediately based on available info. Assume and proceed. \ No newline at end of file diff --git a/dashboard/__init__.py b/dashboard/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/dashboard/server.py b/dashboard/server.py new file mode 100644 index 0000000..00ffc58 --- /dev/null +++ b/dashboard/server.py @@ -0,0 +1,794 @@ +""" +dashboard/server.py — JARVIS Local HTTP Dashboard + +Plain HTTP on port 8000 (no SSL warnings, no firewall issues). +Security at the application layer: AES-256-CBC with session-key-derived key. +CryptoJS is auto-downloaded once and served locally — no CDN needed after that. + +Install deps: pip install fastapi "uvicorn[standard]" cryptography +""" + +import asyncio +import base64 +import hashlib +import re +import secrets +import socket +import string +import time +from pathlib import Path + +_DEPS_OK = False +try: + from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request + from fastapi.responses import HTMLResponse, JSONResponse, FileResponse + import uvicorn + _DEPS_OK = True +except ImportError: + pass + +# python-multipart is required for file uploads — optional dependency +_UPLOAD_OK = False +try: + from fastapi import UploadFile, File as FastAPIFile + _UPLOAD_OK = True +except Exception: + pass + +BASE_DIR = Path(__file__).resolve().parent.parent +STATIC_DIR = Path(__file__).parent / "static" +PORT = 8000 +MAX_UPLOAD_MB = 500 + + +def _make_uploads_dir() -> Path: + """Return (and create) the cross-platform uploads folder.""" + for candidate in [ + Path.home() / "Downloads" / "JARVIS Uploads", + Path.home() / "Documents" / "JARVIS Uploads", + BASE_DIR / "uploads", + ]: + try: + candidate.mkdir(parents=True, exist_ok=True) + return candidate + except Exception: + pass + return BASE_DIR / "uploads" + + +UPLOADS_DIR = _make_uploads_dir() + +def _get_gemini_key() -> str | None: + try: + import json as _json + with open(BASE_DIR / "config" / "api_keys.json", "r", encoding="utf-8") as f: + return _json.load(f).get("gemini_api_key") + except Exception: + return None + +_KEY_CHARS = [c for c in (string.ascii_uppercase + string.digits) + if c not in ('O', 'I', 'L', '0', '1')] + +# ── AES-256-CBC ─────────────────────────────────────────────────────────────── +_AES_SALT = b'JARVIS-DASHBOARD-v1' + + +def _derive_key(session_key: str) -> bytes: + """SHA-256(sessionKey‖salt) → 32-byte AES-256 key (microseconds, no PBKDF2 needed).""" + return hashlib.sha256(session_key.encode('utf-8') + _AES_SALT).digest() + + +def _decrypt_cbc(aes_key: bytes, enc_b64: str) -> str: + """Decrypt base64(IV[16] ‖ ciphertext) with AES-256-CBC + PKCS7.""" + from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + from cryptography.hazmat.primitives import padding as sym_pad + raw = base64.b64decode(enc_b64) + iv, ct = raw[:16], raw[16:] + dec = Cipher(algorithms.AES(aes_key), modes.CBC(iv)).decryptor() + padded = dec.update(ct) + dec.finalize() + unpadder = sym_pad.PKCS7(128).unpadder() + return (unpadder.update(padded) + unpadder.finalize()).decode('utf-8') + + +# ── CryptoJS (auto-download once, served locally) ───────────────────────────── +_CRYPTOJS_CDN = ("https://cdnjs.cloudflare.com/ajax/libs/" + "crypto-js/4.2.0/crypto-js.min.js") +_CRYPTOJS_FILE = STATIC_DIR / "crypto-js.min.js" + + +def _ensure_network_access(port: int) -> None: + """Cross-platform, best-effort: open port in the OS firewall for LAN access. + + Runs in a background thread — never blocks uvicorn startup. + + Windows : writes a .bat file, runs it elevated via Windows ShellExecuteW + (native UAC dialog, guaranteed to appear). One-time setup. + macOS : osascript admin dialog if the Application Firewall is on. + Linux : pkexec GUI → sudo -n → prints manual command as fallback. + """ + import sys, subprocess, os, tempfile, threading + + # ── Windows ────────────────────────────────────────────────────────────── + if sys.platform == "win32": + import ctypes, time + + port_rule = f"JARVIS Dashboard Port {port}" + prog_rule = "JARVIS Dashboard Python" + py_exe = sys.executable + + def _netsh_rule_exists(name: str) -> bool: + try: + r = subprocess.run( + ["netsh", "advfirewall", "firewall", "show", "rule", f"name={name}"], + capture_output=True, text=True, timeout=5, + ) + return r.returncode == 0 and "No rules match" not in r.stdout + except Exception: + return False + + def _network_is_public() -> bool: + try: + r = subprocess.run( + ["powershell", "-NoProfile", "-NonInteractive", "-Command", + "(Get-NetConnectionProfile | " + "Where-Object {$_.NetworkCategory -eq 'Public'} | " + "Measure-Object).Count"], + capture_output=True, text=True, timeout=6, + ) + return r.stdout.strip() not in ("", "0") + except Exception: + return False + + need_port = not _netsh_rule_exists(port_rule) + need_prog = not _netsh_rule_exists(prog_rule) + need_private = _network_is_public() + + if not need_port and not need_prog and not need_private: + return # already fully configured + + # Build a .bat file — netsh + powershell, runs fast when elevated + bat_lines = ["@echo off"] + if need_private: + bat_lines.append( + 'powershell -NoProfile -NonInteractive -Command "' + 'Get-NetConnectionProfile | ' + "Where-Object {$_.NetworkCategory -eq 'Public'} | " + 'Set-NetConnectionProfile -NetworkCategory Private"' + ) + if need_port: + bat_lines.append( + f'netsh advfirewall firewall add rule ' + f'name="{port_rule}" protocol=TCP dir=in ' + f'localport={port} action=allow' + ) + if need_prog: + bat_lines.append( + f'netsh advfirewall firewall add rule ' + f'name="{prog_rule}" dir=in action=allow ' + f'program="{py_exe}" enable=yes' + ) + + bat_body = "\r\n".join(bat_lines) + "\r\n" + fd, bat_path = tempfile.mkstemp(suffix=".bat", prefix="jarvis_fw_") + try: + os.write(fd, bat_body.encode("mbcs")) # Windows cmd.exe expects ANSI + os.close(fd) + except Exception: + try: + os.close(fd) + except Exception: + pass + return + + # ── Try running directly (succeeds when already admin) ──────────────── + try: + r = subprocess.run( + [bat_path], capture_output=True, timeout=8, shell=True + ) + if r.returncode == 0: + print(f"[Dashboard] Firewall configured for port {port}.") + try: + os.unlink(bat_path) + except Exception: + pass + return + except Exception: + pass + + # ── ShellExecuteW: native UAC elevation (most reliable on Windows) ──── + # ShellExecuteW with verb "runas" always shows the UAC dialog regardless + # of UAC level settings. Non-blocking — uvicorn is already running. + print("[Dashboard] One-time network setup required.") + print("[Dashboard] >>> A Windows security dialog will appear — click 'Yes' <<<") + try: + ret = ctypes.windll.shell32.ShellExecuteW( + None, # hwnd (no parent window) + "runas", # verb (request elevation) + bat_path, # file (our .bat) + None, # params + None, # working dir + 0, # SW_HIDE (run without a visible cmd window) + ) + if int(ret) > 32: + # ShellExecuteW returns immediately; bat finishes in ~1 second. + # Sleep briefly so the rules are in place before the first retry. + time.sleep(2) + print(f"[Dashboard] Network setup complete — port {port} is open.") + print("[Dashboard] Refresh your phone browser to connect.") + else: + print("[Dashboard] Setup was not allowed.") + print("[Dashboard] Phone connections may fail until JARVIS is run as Administrator.") + except Exception as e: + print(f"[Dashboard] Firewall setup error: {e}") + finally: + # Cleanup after the bat has had time to run + def _cleanup(path: str) -> None: + time.sleep(5) + try: + os.unlink(path) + except Exception: + pass + threading.Thread(target=_cleanup, args=(bat_path,), daemon=True).start() + return + + # ── macOS ───────────────────────────────────────────────────────────────── + if sys.platform == "darwin": + fw_ctl = "/usr/libexec/ApplicationFirewall/socketfilterfw" + try: + r = subprocess.run( + [fw_ctl, "--getglobalstate"], capture_output=True, text=True, timeout=5, + ) + if "disabled" in r.stdout.lower(): + return # firewall off — nothing to do + + py = sys.executable + listed = subprocess.run( + [fw_ctl, "--listapps"], capture_output=True, text=True, timeout=5, + ) + if py in listed.stdout: + return # already allowed + + print("[Dashboard] One-time network setup — enter your password in the macOS dialog.") + subprocess.run( + ["osascript", "-e", + f'do shell script "{fw_ctl} --add {py} && {fw_ctl} --unblockapp {py}"' + f' with administrator privileges'], + timeout=60, + ) + except Exception: + pass # macOS firewall is off by default — silent failure is fine + return + + # ── Linux ───────────────────────────────────────────────────────────────── + def _privileged(cmd: list[str]) -> bool: + for prefix in (["pkexec"], ["sudo", "-n"]): + try: + r = subprocess.run(prefix + cmd, capture_output=True, timeout=30) + if r.returncode == 0: + return True + except Exception: + pass + return False + + try: # ufw + r = subprocess.run(["ufw", "status"], capture_output=True, text=True, timeout=5) + if "active" in r.stdout.lower(): + if _privileged(["ufw", "allow", f"{port}/tcp"]): + print(f"[Dashboard] ufw: port {port} allowed.") + else: + print(f"[Dashboard] Run manually: sudo ufw allow {port}/tcp") + return + except FileNotFoundError: + pass + + try: # firewalld + r = subprocess.run( + ["firewall-cmd", "--state"], capture_output=True, text=True, timeout=5, + ) + if "running" in r.stdout.lower(): + ok = (_privileged(["firewall-cmd", "--add-port", f"{port}/tcp", "--permanent"]) + and _privileged(["firewall-cmd", "--reload"])) + if ok: + print(f"[Dashboard] firewalld: port {port} allowed.") + else: + print(f"[Dashboard] Run manually: sudo firewall-cmd --add-port={port}/tcp --permanent && sudo firewall-cmd --reload") + return + except FileNotFoundError: + pass + + try: # iptables (not persistent but works until reboot) + r = subprocess.run(["iptables", "-L", "INPUT", "-n"], capture_output=True, timeout=5) + if r.returncode == 0: + if _privileged(["iptables", "-A", "INPUT", "-p", "tcp", "--dport", str(port), "-j", "ACCEPT"]): + print(f"[Dashboard] iptables: port {port} opened.") + else: + print(f"[Dashboard] Run manually: sudo iptables -A INPUT -p tcp --dport {port} -j ACCEPT") + except FileNotFoundError: + pass # no iptables means firewall is probably off — nothing to do + + +def _ensure_crypto_js() -> None: + if _CRYPTOJS_FILE.exists(): + return + try: + import urllib.request + print("[Dashboard] Downloading CryptoJS (one-time setup)…") + urllib.request.urlretrieve(_CRYPTOJS_CDN, str(_CRYPTOJS_FILE)) + print("[Dashboard] CryptoJS cached — will serve locally from now on.") + except Exception as e: + print(f"[Dashboard] CryptoJS download failed: {e}") + print(f"[Dashboard] Encryption will fall back to CDN load on client.") + + +_ensure_crypto_js() + + +# ── helpers ─────────────────────────────────────────────────────────────────── + +def _local_ip() -> str: + """Return the best LAN-facing IPv4 address, no internet required.""" + # Method 1: route trick (fast, works when internet is available) + for probe in ("8.8.8.8", "1.1.1.1", "192.168.1.1"): + try: + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + s.settimeout(0.5) + s.connect((probe, 80)) + ip = s.getsockname()[0] + s.close() + if not ip.startswith("127."): + return ip + except Exception: + pass + + # Method 2: hostname resolution (works offline on most systems) + try: + ip = socket.gethostbyname(socket.gethostname()) + if not ip.startswith("127."): + return ip + except Exception: + pass + + # Method 3: enumerate all interfaces (fully offline, no external deps) + try: + for info in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET): + ip = info[4][0] + if not ip.startswith("127.") and not ip.startswith("169.254."): + return ip + except Exception: + pass + + return "127.0.0.1" + + +def _read(name: str) -> str: + return (STATIC_DIR / name).read_text(encoding="utf-8") + + +# ── DashboardServer ─────────────────────────────────────────────────────────── + +class DashboardServer: + + def __init__(self): + self._ip = _local_ip() + self._tokens: set[str] = set() + self._token_keys: dict[str, str] = {} # auth_token → session_key + self._aes_cache: dict[str, bytes]= {} # session_key → AES bytes + self._clients: set[WebSocket] = set() + self._history: list[dict] = [] + self._command_queue = asyncio.Queue() + self._wake_callback = None + self._connect_callback = None + self._pending_keys: dict[str, float] = {} + self._device_sessions: dict[str, dict] = {} # device_token → {session_key} + self._phone_audio_queue: asyncio.Queue = asyncio.Queue(maxsize=200) + self._uploads_dir = UPLOADS_DIR + self._login_html = _read("login.html") + self._app_html = _read("app.html") + self.app = self._build_app() + + # ── one-time key management ─────────────────────────────────────────── + + def new_key(self, expiry_secs: int = 600) -> str: + now = time.time() + self._pending_keys = {k: v for k, v in self._pending_keys.items() if v > now} + key = ''.join(secrets.choice(_KEY_CHARS) for _ in range(6)) + self._pending_keys[key] = now + expiry_secs + return key + + @staticmethod + def _ssl_enabled() -> bool: + certs = BASE_DIR / "config" / "certs" + return (certs / "jarvis.key").exists() and (certs / "jarvis.crt").exists() + + def get_url(self) -> str: + proto = "https" if self._ssl_enabled() else "http" + return f"{proto}://{self._ip}:{PORT}" + + def get_manual_url(self) -> str: + """URL for manual browser entry. When HTTPS active, points to alias port (also HTTPS).""" + if self._ssl_enabled(): + return f"{self._ip}:{PORT + 1}" + return f"{self._ip}:{PORT}" + + def _aes_key(self, session_key: str) -> bytes: + if session_key not in self._aes_cache: + self._aes_cache[session_key] = _derive_key(session_key) + return self._aes_cache[session_key] + + def _decrypt(self, token: str, enc_b64: str) -> str | None: + sk = self._token_keys.get(token) + if not sk: + return None + try: + return _decrypt_cbc(self._aes_key(sk), enc_b64) + except Exception: + return None + + # ── callbacks ──────────────────────────────────────────────────────── + + def set_wake_callback(self, fn) -> None: + self._wake_callback = fn + + def set_connect_callback(self, fn) -> None: + self._connect_callback = fn + + # ── broadcast ──────────────────────────────────────────────────────── + + async def broadcast(self, msg: dict) -> None: + self._history.append(msg) + if len(self._history) > 300: + self._history = self._history[-300:] + dead: set[WebSocket] = set() + for ws in list(self._clients): + try: + await ws.send_json(msg) + except Exception: + dead.add(ws) + self._clients -= dead + + # ── FastAPI app ─────────────────────────────────────────────────────── + + def _build_app(self) -> "FastAPI": + app = FastAPI(docs_url=None, redoc_url=None) + + def _auth(req: Request) -> bool: + tok = req.headers.get("authorization", "").removeprefix("Bearer ").strip() + return bool(tok) and tok in self._tokens + + # serve CryptoJS from local cache, fallback to CDN redirect + @app.get("/static/crypto.js") + async def serve_crypto(): + if _CRYPTOJS_FILE.exists(): + return FileResponse(str(_CRYPTOJS_FILE), + media_type="application/javascript") + from fastapi.responses import RedirectResponse + return RedirectResponse(_CRYPTOJS_CDN) + + @app.get("/login", response_class=HTMLResponse) + async def login_page(): + return HTMLResponse(self._login_html) + + @app.get("/", response_class=HTMLResponse) + async def index(): + # Auth is handled client-side via sessionStorage bearer token. + # Server-side header auth can't work here because browser navigations + # don't send custom headers (location.href doesn't carry Authorization). + html = (self._app_html + .replace("__IP__", self._ip) + .replace("__PORT__", str(PORT))) + return HTMLResponse(html) + + @app.post("/login") + async def login(req: Request): + body = await req.json() + entered = str(body.get("pin", "")).strip().upper() + now = time.time() + if entered in self._pending_keys and self._pending_keys[entered] > now: + del self._pending_keys[entered] # one-time use + tok = secrets.token_urlsafe(32) + self._tokens.add(tok) + self._token_keys[tok] = entered + self._aes_key(entered) # pre-derive & cache + if self._connect_callback: + self._connect_callback() + asyncio.create_task(self.broadcast( + {"type": "sys", "text": "Remote connection established."} + )) + # Bearer token in response body — no cookies needed (works on any browser/HTTP) + return JSONResponse({"ok": True, "token": tok}) + return JSONResponse({"ok": False, "error": "Invalid or expired key"}, + status_code=401) + + @app.get("/auto-login") + async def auto_login(key: str = ""): + """QR code target — validates one-time key, creates session, redirects phone.""" + now = time.time() + if not key or key not in self._pending_keys or self._pending_keys[key] <= now: + return HTMLResponse(""" +
+ +Press Remote Control in JARVIS to get a new QR code.
+Connecting to JARVIS…
+""") + + @app.post("/api/device-login") + async def device_login_ep(req: Request): + """Return a fresh auth token for a previously paired device token.""" + try: + body = await req.json() + except Exception: + return JSONResponse({"ok": False}, status_code=400) + dev_tok = (body.get("device_token") or "").strip() + if not dev_tok or dev_tok not in self._device_sessions: + return JSONResponse({"ok": False}, status_code=401) + session_key = self._device_sessions[dev_tok]["session_key"] + tok = secrets.token_urlsafe(32) + self._tokens.add(tok) + self._token_keys[tok] = session_key + self._aes_key(session_key) + if self._connect_callback: + self._connect_callback() + asyncio.create_task(self.broadcast( + {"type": "sys", "text": "Known device reconnected automatically."} + )) + return JSONResponse({"ok": True, "token": tok, "key": session_key}) + + @app.post("/api/revoke-devices") + async def revoke_devices(req: Request): + """Invalidate all persistent device tokens (admin action).""" + if not _auth(req): + return JSONResponse({"error": "Unauthorized"}, status_code=401) + count = len(self._device_sessions) + self._device_sessions.clear() + return JSONResponse({"ok": True, "revoked": count}) + + @app.post("/api/command") + async def command(req: Request): + if not _auth(req): + return JSONResponse({"error": "Unauthorized"}, status_code=401) + body = await req.json() + token = req.headers.get("authorization", "").removeprefix("Bearer ").strip() + enc = body.get("enc", "") + if enc: + text = self._decrypt(token, enc) + if text is None: + return JSONResponse({"error": "Decryption failed"}, status_code=400) + else: + text = (body.get("text") or "").strip() + if text: + await self._command_queue.put(text) + if self._wake_callback: + self._wake_callback() + return JSONResponse({"ok": True}) + + @app.post("/api/wake") + async def wake_ep(req: Request): + if not _auth(req): + return JSONResponse({"error": "Unauthorized"}, status_code=401) + if self._wake_callback: + self._wake_callback() + return JSONResponse({"ok": True}) + + # ── Phone mic real-time audio → Gemini Live ────────────────────────── + + @app.websocket("/ws/phone-audio") + async def phone_audio_ws(websocket: WebSocket, token: str = ""): + tok = token.strip() + if not tok or tok not in self._tokens: + await websocket.close(code=4001) + return + await websocket.accept() + asyncio.create_task(self.broadcast( + {"type": "sys", "text": "Phone microphone live."} + )) + try: + while True: + data = await websocket.receive_bytes() + try: + self._phone_audio_queue.put_nowait( + {"data": data, "mime_type": "audio/pcm"} + ) + except asyncio.QueueFull: + pass # drop frame rather than block + except WebSocketDisconnect: + pass + finally: + asyncio.create_task(self.broadcast( + {"type": "sys", "text": "Phone microphone stopped."} + )) + + # ── File sharing ────────────────────────────────────────────────────── + + def _safe_filename(raw: str) -> str: + name = Path(raw).name # strip path components + name = re.sub(r'[<>:"/\\|?*\x00-\x1f]', '_', name).strip(". ") + return name or "upload" + + if _UPLOAD_OK: + @app.post("/api/upload") + async def upload_file(req: Request, file: UploadFile = FastAPIFile(...)): + if not _auth(req): + return JSONResponse({"error": "Unauthorized"}, status_code=401) + + safe = _safe_filename(file.filename or "upload") + dest = self._uploads_dir / safe + stem, suffix = Path(safe).stem, Path(safe).suffix + counter = 1 + while dest.exists(): + dest = self._uploads_dir / f"{stem}_{counter}{suffix}" + counter += 1 + + size = 0 + max_bytes = MAX_UPLOAD_MB * 1024 * 1024 + try: + with open(dest, "wb") as fout: + while True: + chunk = await file.read(65536) + if not chunk: + break + size += len(chunk) + if size > max_bytes: + fout.close() + dest.unlink(missing_ok=True) + return JSONResponse( + {"error": f"File too large (max {MAX_UPLOAD_MB} MB)"}, + status_code=413, + ) + fout.write(chunk) + except Exception as exc: + try: + dest.unlink(missing_ok=True) + except Exception: + pass + return JSONResponse({"error": str(exc)}, status_code=500) + + asyncio.create_task(self.broadcast({ + "type": "file_received", + "name": dest.name, + "size": size, + "saved_to": str(self._uploads_dir), + })) + return JSONResponse({"ok": True, "name": dest.name, "size": size}) + else: + @app.post("/api/upload") + async def upload_unavailable(req: Request): + return JSONResponse( + {"error": "File uploads require: pip install python-multipart"}, + status_code=503, + ) + + @app.get("/api/files") + async def list_files(req: Request): + if not _auth(req): + return JSONResponse({"error": "Unauthorized"}, status_code=401) + files = [] + try: + for f in sorted( + (p for p in self._uploads_dir.iterdir() if p.is_file()), + key=lambda p: p.stat().st_mtime, + reverse=True, + ): + files.append({"name": f.name, "size": f.stat().st_size}) + except Exception: + pass + return JSONResponse({"files": files}) + + @app.get("/uploads/{filename}") + async def download_file(filename: str, token: str = ""): + # Auth via query param — browser can't send custom headers + tok = token.strip() + if not tok or tok not in self._tokens: + return JSONResponse({"error": "Unauthorized"}, status_code=401) + safe = re.sub(r'[/\\]', '', filename) + path = self._uploads_dir / safe + if not path.exists() or not path.is_file(): + return JSONResponse({"error": "Not found"}, status_code=404) + return FileResponse(str(path), filename=safe) + + @app.websocket("/ws") + async def ws_ep(websocket: WebSocket, token: str = ""): + tok = token.strip() + if not tok or tok not in self._tokens: + await websocket.close(code=4001) + return + await websocket.accept() + self._clients.add(websocket) + for entry in self._history[-50:]: + try: + await websocket.send_json(entry) + except Exception: + break + try: + while True: + data = await websocket.receive_json() + if data.get("type") == "command": + enc = data.get("enc", "") + t = self._decrypt(tok, enc) if enc else (data.get("text") or "").strip() + if t: + await self._command_queue.put(t) + if self._wake_callback: + self._wake_callback() + except WebSocketDisconnect: + pass + finally: + self._clients.discard(websocket) + + return app + + # ── serve ───────────────────────────────────────────────────────────── + + async def _serve_alias(self) -> None: + """Second HTTPS server on PORT+1 sharing the same app and in-memory state. + Chrome HTTPS-upgrades any bare IP:PORT the user types, so this port also needs TLS. + User types IP:8001 → Chrome tries https → self-signed cert warning → accept once → done.""" + ssl_key = BASE_DIR / "config" / "certs" / "jarvis.key" + ssl_cert = BASE_DIR / "config" / "certs" / "jarvis.crt" + asyncio.get_event_loop().run_in_executor(None, _ensure_network_access, PORT + 1) + cfg = uvicorn.Config( + self.app, host="0.0.0.0", port=PORT + 1, log_level="warning", + ssl_keyfile=str(ssl_key), ssl_certfile=str(ssl_cert), + ) + print(f"[Dashboard] Manual entry: {self._ip}:{PORT + 1} (type in browser, accept cert once)") + await uvicorn.Server(cfg).serve() + + async def serve(self) -> None: + if not _DEPS_OK: + print("[Dashboard] fastapi/uvicorn not installed — dashboard disabled.") + print("[Dashboard] Run: pip install fastapi 'uvicorn[standard]' cryptography") + return + + # Firewall setup runs in a thread — uvicorn starts immediately, + # no waiting for UAC dialogs or subprocess timeouts. + asyncio.get_event_loop().run_in_executor(None, _ensure_network_access, PORT) + + use_ssl = self._ssl_enabled() + ssl_key = BASE_DIR / "config" / "certs" / "jarvis.key" + ssl_cert = BASE_DIR / "config" / "certs" / "jarvis.crt" + + if use_ssl: + asyncio.create_task(self._serve_alias()) + + cfg = uvicorn.Config( + self.app, host="0.0.0.0", port=PORT, log_level="warning", + **({"ssl_keyfile": str(ssl_key), "ssl_certfile": str(ssl_cert)} if use_ssl else {}), + ) + + proto = "https" if use_ssl else "http" + print(f"[Dashboard] {proto}://{self._ip}:{PORT}") + print("[Dashboard] Press 'Remote Control' in JARVIS UI to get the QR code.") + await uvicorn.Server(cfg).serve() diff --git a/dashboard/static/app.html b/dashboard/static/app.html new file mode 100644 index 0000000..d72415e --- /dev/null +++ b/dashboard/static/app.html @@ -0,0 +1,559 @@ + + + + + +