Hybrid GPU Manager: battery impact estimation and GPU state detection v1

cortex/cli.py

@@ -1,6 +1,7 @@
 import argparse
 import logging
 import os
+import subprocess
 import sys
 import time
 from datetime import datetime
@@ -19,6 +20,18 @@
     format_package_list,
 )
 from cortex.env_manager import EnvironmentManager, get_env_manager
+from cortex.gpu_manager import (
+    apply_gpu_mode_switch,
+    detect_gpu_switch_backend,
+    get_app_gpu_preference,
+    get_per_app_gpu_env,
+    list_app_gpu_preferences,
+    plan_gpu_mode_switch,
+    remove_app_gpu_preference,
+    run_command_with_env,
+    set_app_gpu_preference,
+)
+from cortex.hardware_detection import detect_gpu_mode, estimate_gpu_battery_impact
 from cortex.installation_history import InstallationHistory, InstallationStatus, InstallationType
 from cortex.llm.interpreter import CommandInterpreter
 from cortex.network_config import NetworkConfig
@@ -1013,6 +1026,246 @@ def status(self):
         doctor = SystemDoctor()
         return doctor.run_checks()
 
+    def gpu_battery(self, args: argparse.Namespace | None = None) -> int:
+        """
+        Estimate battery impact based on current GPU usage and mode.
+
+        Prints:
+        - Heuristic estimates (unless --measured-only)
+        - Measured battery discharge (if available)
+        - Measured NVIDIA GPU power draw (when supported by nvidia-smi)
+        """
+        try:
+            data = estimate_gpu_battery_impact()
+        except (OSError, PermissionError, ValueError, RuntimeError) as e:
+            cx_print(f"Failed to probe battery/GPU info: {e}", "error")
+            return 2
+
+        measured_only = bool(getattr(args, "measured_only", False))
+
+        mode = data.get("mode", "Unknown")
+        estimates = data.get("estimates") or {}
+
+        measured = data.get("measured") or {}
+        battery = measured.get("battery") or {}
+
+        # NVIDIA power draw (watts) - emitted by estimate_gpu_battery_impact() as "nvidia_power_w"
+        nvidia_watts = measured.get("nvidia_power_w")
+
+        has_battery_data = bool(battery)
+        has_nvidia_data = nvidia_watts is not None
+
+        cx_print(f"GPU Mode: {mode}", "info")
+        print()
+
+        if not measured_only:
+            cx_print("Estimated power draw:", "info")
+            print(f"- Integrated GPU only: {estimates.get('integrated', {}).get('power', 'N/A')}")
+            print(f"- Hybrid (idle dGPU): {estimates.get('hybrid_idle', {}).get('power', 'N/A')}")
+            print(f"- NVIDIA active: {estimates.get('nvidia_active', {}).get('power', 'N/A')}")
+            print()
+
+            cx_print("Estimated battery impact:", "info")
+            print(f"- Hybrid idle: {estimates.get('hybrid_idle', {}).get('impact', 'N/A')}")
+            print(f"- NVIDIA active: {estimates.get('nvidia_active', {}).get('impact', 'N/A')}")
+            print()
+
+        if has_battery_data or has_nvidia_data:
+            cx_print("Measured (if available):", "info")
+
+            if has_battery_data:
+                status = battery.get("status")
+                percent = battery.get("percent")
+                power_watts = battery.get("power_watts")
+                hours_remaining = battery.get("hours_remaining")
+
+                if status:
+                    print(f"- Battery status: {status}")
+                if percent is not None:
+                    print(f"- Battery: {percent}%")
+                if power_watts is not None:
+                    try:
+                        print(f"- Battery draw: ~{float(power_watts):.2f} W")
+                    except (TypeError, ValueError):
+                        print(f"- Battery draw: ~{power_watts} W")
+                if hours_remaining is not None:
+                    try:
+                        print(f"- Est. time remaining: ~{float(hours_remaining):.2f} h")
+                    except (TypeError, ValueError):
+                        print(f"- Est. time remaining: ~{hours_remaining} h")
+
+            if has_nvidia_data:
+                try:
+                    print(f"- NVIDIA power draw: ~{float(nvidia_watts):.2f} W")
+                except (TypeError, ValueError):
+                    print(f"- NVIDIA power draw: ~{nvidia_watts} W")
+
+            print()
+        else:
+            if measured_only:
+                cx_print("No real measurements available on this system.", "warning")
+                cx_print("Tip: NVIDIA GPU power requires nvidia-smi.", "info")
+                cx_print(
+                    "Tip: Battery draw requires BAT* metrics (may be unavailable on WSL).", "info"
+                )
+                return 2
+
+        if not measured_only:
+            cx_print(
+                "Note: Estimates are heuristic and vary by hardware and workload.",
+                "warning",
+            )
+        return 0
+
+    def gpu(self, args: argparse.Namespace) -> int:
+        """Handle GPU management commands (status, set, run, app).
+
+        Args:
+            args: Parsed command-line arguments containing gpu_command subcommand.
+
+        Returns:
+            int: 0 on success, 1 on error, 2 on missing backend or invalid usage.
+        """
+
+        if args.gpu_command == "status":
+            backend = detect_gpu_switch_backend()
+            mode = detect_gpu_mode()
+            apps = list_app_gpu_preferences()
+            cx_print(f"GPU mode: {mode}")
+            cx_print(f"Switch backend: {backend.value}")
+            cx_print(f"Per-app assignments: {len(apps)}")
+            cx_print(
+                "Tip: `cortex gpu set integrated|hybrid|nvidia --dry-run` to preview, or add `--execute` to apply."
+            )
+            return 0
+
+        if args.gpu_command == "set":
+            # Defensive guard (argparse should prevent this, but keep runtime safety)
+            if getattr(args, "dry_run", False) and getattr(args, "execute", False):
+                cx_print("Error: --dry-run and --execute cannot be used together.", "error")
+                return 2
+
+            # Plan the switch; invalid modes raise ValueError
+            try:
+                plan = plan_gpu_mode_switch(args.mode)
+            except ValueError as e:
+                cx_print(f"Invalid target GPU mode: {e}", "error")
+                return 1
+
+            if plan is None:
+                cx_print("No supported GPU switch backend found.", "error")
+                cx_print(
+                    "Tip: install/configure prime-select or system76-power, then retry.",
+                    "info",
+                )
+                return 2
+
+            # Always show the plan
+            cx_print(f"Backend: {plan.backend.value}")
+            cx_print(f"Target mode: {plan.target_mode}")
+            cx_print("Commands:")
+            for c in plan.commands:
+                cx_print("  " + " ".join(c))
+            cx_print(f"Restart required: {plan.requires_restart}")
+            if getattr(plan, "notes", None):
+                cx_print(f"Notes: {plan.notes}")
+
+            # Honor --dry-run explicitly: show plan only, do not prompt, do not execute
+            if getattr(args, "dry_run", False):
+                return 0
+
+            # Execute only when --execute is set (otherwise plan-only)
+            if getattr(args, "execute", False):
+                if not getattr(args, "yes", False):
+                    console.print("\n⚠ This will run GPU switch commands with sudo.")
+                    console.print("Proceed? [y/N]: ", end="")
+                    try:
+                        resp = input().strip().lower()
+                    except (EOFError, KeyboardInterrupt):
+                        resp = ""
+                    if resp not in ("y", "yes"):
+                        cx_print("Operation cancelled", "info")
+                        return 0
+
+                return apply_gpu_mode_switch(plan, execute=True)
+
+            # Default: plan-only (no prompt, no execution)
+            return 0
+
+        if args.gpu_command == "run":
+            cmd = list(args.cmd or [])
+            if cmd and cmd[0] == "--":
+                cmd = cmd[1:]
+            if not cmd:
+                cx_print("Missing command.", "error")
+                return 2
+
+            # Resolve env for the requested mode/app
+            if getattr(args, "nvidia", False):
+                env = get_per_app_gpu_env(use_nvidia=True)
+            elif getattr(args, "integrated", False):
+                env = get_per_app_gpu_env(use_nvidia=False)
+            elif getattr(args, "app", None):
+                env = get_per_app_gpu_env(app=args.app)
+            else:
+                cx_print("Specify one of: --app, --nvidia, or --integrated", "error")
+                return 2
+
+            # If we're launching an integrated run and env is empty, we must also
+            # explicitly UNSET PRIME vars that may be set in the parent shell.
+            integrated_case = bool(getattr(args, "integrated", False))
+            app_case = bool(getattr(args, "app", None))
+
+            if (integrated_case or app_case) and not env:
+                cmd = [
+                    "env",
+                    "-u",
+                    "__NV_PRIME_RENDER_OFFLOAD",
+                    "-u",
+                    "__GLX_VENDOR_LIBRARY_NAME",
+                    "-u",
+                    "__VK_LAYER_NV_optimus",
+                    "--",
+                ] + cmd
+
+            return run_command_with_env(cmd, extra_env=env)
+
+        if args.gpu_command == "app":
+            if args.app_action == "set":
+                try:
+                    set_app_gpu_preference(args.app, args.mode)
+                except ValueError as e:
+                    cx_print(f"Invalid per-app GPU assignment: {e}", "error")
+                    return 1
+
+                cx_print(f"Saved: {args.app} -> {args.mode}")
+                return 0
+
+            if args.app_action == "get":
+                pref = get_app_gpu_preference(args.app)
+                if pref is None:
+                    cx_print(f"{args.app}: GPU preference not set", "warning")
+                    return 2
+                cx_print(f"{args.app}: {pref}")
+                return 0
+
+            if args.app_action == "list":
+                apps = list_app_gpu_preferences()
+                for k, v in apps.items():
+                    console.print(f"{k} -> {v}")
+                return 0
+            if args.app_action == "remove":
+                removed = remove_app_gpu_preference(args.app)
+                if removed:
+                    cx_print(f"GPU preference removed for app '{args.app}'", "success")
+                    return 0
+
+                cx_print(f"No GPU preference found for app '{args.app}'", "warning")
+                return 1
+
+        cx_print("Unknown gpu command")
+        return 2
+
     def wizard(self):
         """Interactive setup wizard for API key configuration"""
         show_banner()
@@ -2030,6 +2283,8 @@ def show_rich_help():
     table.add_row("demo", "See Cortex in action")
     table.add_row("wizard", "Configure API key")
     table.add_row("status", "System status")
+    table.add_row("gpu", "Hybrid GPU manager tools")
+    table.add_row("gpu-battery", "Estimate battery impact of current GPU usage")
     table.add_row("install <pkg>", "Install software")
     table.add_row("import <file>", "Import deps from package files")
     table.add_row("history", "View history")
@@ -2130,6 +2385,64 @@ def main():
     # Status command (includes comprehensive health checks)
     subparsers.add_parser("status", help="Show comprehensive system status and health checks")
 
+    # GPU battery estimation
+    gpu_battery_parser = subparsers.add_parser(
+        "gpu-battery",
+        help="Estimate battery impact of current GPU usage",
+    )
+
+    gpu_battery_parser.add_argument(
+        "--measured-only",
+        action="store_true",
+        help="Show only real measurements (if available)",
+    )
+
+    gpu_parser = subparsers.add_parser("gpu", help="Hybrid GPU manager tools")
+    gpu_sub = gpu_parser.add_subparsers(dest="gpu_command", required=True)
+
+    gpu_sub.add_parser("status", help="Show GPU status")
+
+    gpu_set = gpu_sub.add_parser("set", help="Switch GPU mode")
+    gpu_set.add_argument("mode", choices=["integrated", "hybrid", "nvidia"])
+
+    # Make --dry-run and --execute mutually exclusive
+    gpu_set_flags = gpu_set.add_mutually_exclusive_group()
+    gpu_set_flags.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Show the switch plan only (no sudo, no changes applied)",
+    )
+    gpu_set_flags.add_argument(
+        "--execute",
+        action="store_true",
+        help="Execute GPU switch commands (sudo required)",
+    )
+
+    gpu_set.add_argument("-y", "--yes", action="store_true", help="Skip confirmation prompt")
+
+    gpu_app = gpu_sub.add_parser("app", help="Per-app GPU assignment")
+    gpu_app_sub = gpu_app.add_subparsers(dest="app_action", required=True)
+
+    gpu_app_sub.add_parser("list")
+    app_set = gpu_app_sub.add_parser("set")
+    app_set.add_argument("app")
+    app_set.add_argument("mode", choices=["nvidia", "integrated"])
+
+    app_get = gpu_app_sub.add_parser("get")
+    app_get.add_argument("app")
+
+    app_rm = gpu_app_sub.add_parser("remove")
+    app_rm.add_argument("app")
+
+    gpu_run = gpu_sub.add_parser("run")
+
+    gpu_run_mode = gpu_run.add_mutually_exclusive_group(required=True)
+    gpu_run_mode.add_argument("--nvidia", action="store_true", help="Force NVIDIA GPU")
+    gpu_run_mode.add_argument("--integrated", action="store_true", help="Force integrated GPU")
+    gpu_run_mode.add_argument("--app", help="Use saved per-app GPU preference")
+
+    gpu_run.add_argument("cmd", nargs=argparse.REMAINDER)
+
     # Ask command
     ask_parser = subparsers.add_parser("ask", help="Ask a question about your system")
     ask_parser.add_argument("question", type=str, help="Natural language question")
@@ -2531,6 +2844,11 @@ def main():
             return 1
         elif args.command == "env":
             return cli.env(args)
+        elif args.command == "gpu-battery":
+            return cli.gpu_battery(args)
+        elif args.command == "gpu":
+            return cli.gpu(args)
+
         else:
             parser.print_help()
             return 1