wled-screen-controller-mixed/server/src/wled_controller/api/routes/system.py

"""System routes: health, version, displays, performance, tags, api-keys.

Backup/restore and settings routes are in backup.py and system_settings.py.
"""

import asyncio
import platform
import subprocess
import sys
from datetime import datetime, timezone
from typing import Optional

import psutil
from fastapi import APIRouter, Depends, HTTPException, Query

from wled_controller import __version__
from wled_controller.api.auth import AuthRequired, is_auth_enabled
from wled_controller.api.dependencies import (
    get_audio_source_store,
    get_audio_template_store,
    get_automation_store,
    get_color_strip_store,
    get_device_store,
    get_output_target_store,
    get_pattern_template_store,
    get_picture_source_store,
    get_pp_template_store,
    get_processor_manager,
    get_scene_preset_store,
    get_sync_clock_store,
    get_template_store,
    get_value_source_store,
)
from wled_controller.api.schemas.system import (
    DisplayInfo,
    DisplayListResponse,
    GpuInfo,
    HealthResponse,
    PerformanceResponse,
    ProcessListResponse,
    VersionResponse,
)
from wled_controller.config import get_config, is_demo_mode
from wled_controller.core.capture.screen_capture import get_available_displays
from wled_controller.utils import get_logger
from wled_controller.storage.base_store import EntityNotFoundError

# Re-export load_external_url so existing callers still work
from wled_controller.api.routes.system_settings import load_external_url  # noqa: F401

logger = get_logger(__name__)

# Prime psutil CPU counter (first call always returns 0.0)
psutil.cpu_percent(interval=None)

# GPU monitoring (initialized once in utils.gpu, shared with metrics_history)
from wled_controller.utils.gpu import nvml_available as _nvml_available, nvml as _nvml, nvml_handle as _nvml_handle  # noqa: E402


def _get_cpu_name() -> str | None:
    """Get a human-friendly CPU model name (cached at module level)."""
    try:
        if platform.system() == "Windows":
            import winreg

            key = winreg.OpenKey(
                winreg.HKEY_LOCAL_MACHINE,
                r"HARDWARE\DESCRIPTION\System\CentralProcessor\0",
            )
            name, _ = winreg.QueryValueEx(key, "ProcessorNameString")
            winreg.CloseKey(key)
            return name.strip()
        elif platform.system() == "Linux":
            with open("/proc/cpuinfo") as f:
                for line in f:
                    if "model name" in line:
                        return line.split(":")[1].strip()
        elif platform.system() == "Darwin":
            return (
                subprocess.check_output(
                    ["sysctl", "-n", "machdep.cpu.brand_string"]
                )
                .decode()
                .strip()
            )
    except Exception as e:
        logger.warning("CPU name detection failed: %s", e)
    return platform.processor() or None


_cpu_name: str | None = _get_cpu_name()

router = APIRouter()


@router.get("/health", response_model=HealthResponse, tags=["Health"])
async def health_check():
    """Check service health status.

    Returns basic health information including status, version, and timestamp.
    """
    logger.debug("Health check requested")

    return HealthResponse(
        status="healthy",
        timestamp=datetime.now(timezone.utc),
        version=__version__,
        demo_mode=get_config().demo,
        auth_required=is_auth_enabled(),
    )


@router.get("/api/v1/version", response_model=VersionResponse, tags=["Info"])
async def get_version():
    """Get version information.

    Returns application version, Python version, and API version.
    """
    logger.debug("Version info requested")

    return VersionResponse(
        version=__version__,
        python_version=f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}",
        api_version="v1",
        demo_mode=get_config().demo,
    )


@router.get("/api/v1/tags", tags=["Tags"])
async def list_all_tags(_: AuthRequired):
    """Get all tags used across all entities."""
    all_tags: set[str] = set()
    store_getters = [
        get_device_store, get_output_target_store, get_color_strip_store,
        get_picture_source_store, get_audio_source_store, get_value_source_store,
        get_sync_clock_store, get_automation_store, get_scene_preset_store,
        get_template_store, get_audio_template_store, get_pp_template_store,
        get_pattern_template_store,
    ]
    for getter in store_getters:
        try:
            store = getter()
        except RuntimeError:
            continue
        # BaseJsonStore subclasses provide get_all(); DeviceStore provides get_all_devices()
        fn = getattr(store, "get_all", None) or getattr(store, "get_all_devices", None)
        items = fn() if fn else None
        if items:
            for item in items:
                all_tags.update(item.tags)
    return {"tags": sorted(all_tags)}


@router.get("/api/v1/config/displays", response_model=DisplayListResponse, tags=["Config"])
async def get_displays(
    _: AuthRequired,
    engine_type: Optional[str] = Query(None, description="Engine type to get displays for"),
):
    """Get list of available displays.

    Returns information about all available monitors/displays that can be captured.
    When ``engine_type`` is provided, returns displays specific to that engine
    (e.g. ``scrcpy`` returns connected Android devices instead of desktop monitors).
    """
    logger.info(f"Listing available displays (engine_type={engine_type})")

    try:
        from wled_controller.core.capture_engines import EngineRegistry

        if engine_type:
            engine_cls = EngineRegistry.get_engine(engine_type)
            display_dataclasses = await asyncio.to_thread(engine_cls.get_available_displays)
        elif is_demo_mode():
            # In demo mode, use the best available engine (demo engine at priority 1000)
            # instead of the mss-based real display detection
            best = EngineRegistry.get_best_available_engine()
            if best:
                engine_cls = EngineRegistry.get_engine(best)
                display_dataclasses = await asyncio.to_thread(engine_cls.get_available_displays)
            else:
                display_dataclasses = await asyncio.to_thread(get_available_displays)
        else:
            display_dataclasses = await asyncio.to_thread(get_available_displays)

        # Convert dataclass DisplayInfo to Pydantic DisplayInfo
        displays = [
            DisplayInfo(
                index=d.index,
                name=d.name,
                width=d.width,
                height=d.height,
                x=d.x,
                y=d.y,
                is_primary=d.is_primary,
                refresh_rate=d.refresh_rate,
            )
            for d in display_dataclasses
        ]

        logger.info(f"Found {len(displays)} displays")

        return DisplayListResponse(
            displays=displays,
            count=len(displays),
        )

    except EntityNotFoundError as e:
        raise HTTPException(status_code=404, detail=str(e))


    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except Exception as e:
        logger.error(f"Failed to get displays: {e}")
        raise HTTPException(
            status_code=500,
            detail=f"Failed to retrieve display information: {str(e)}"
        )


@router.get("/api/v1/system/processes", response_model=ProcessListResponse, tags=["Config"])
async def get_running_processes(_: AuthRequired):
    """Get list of currently running process names.

    Returns a sorted list of unique process names for use in automation conditions.
    """
    from wled_controller.core.automations.platform_detector import PlatformDetector

    try:
        detector = PlatformDetector()
        processes = await detector.get_running_processes()
        sorted_procs = sorted(processes)
        return ProcessListResponse(processes=sorted_procs, count=len(sorted_procs))
    except Exception as e:
        logger.error(f"Failed to get processes: {e}")
        raise HTTPException(
            status_code=500,
            detail=f"Failed to retrieve process list: {str(e)}"
        )


@router.get(
    "/api/v1/system/performance",
    response_model=PerformanceResponse,
    tags=["Config"],
)
def get_system_performance(_: AuthRequired):
    """Get current system performance metrics (CPU, RAM, GPU).

    Uses sync ``def`` so FastAPI runs it in a thread pool — the psutil
    and NVML calls are blocking and would stall the event loop if run
    in an ``async def`` handler.
    """
    mem = psutil.virtual_memory()

    gpu = None
    if _nvml_available:
        try:
            util = _nvml.nvmlDeviceGetUtilizationRates(_nvml_handle)
            mem_info = _nvml.nvmlDeviceGetMemoryInfo(_nvml_handle)
            temp = _nvml.nvmlDeviceGetTemperature(
                _nvml_handle, _nvml.NVML_TEMPERATURE_GPU
            )
            gpu = GpuInfo(
                name=_nvml.nvmlDeviceGetName(_nvml_handle),
                utilization=float(util.gpu),
                memory_used_mb=round(mem_info.used / 1024 / 1024, 1),
                memory_total_mb=round(mem_info.total / 1024 / 1024, 1),
                temperature_c=float(temp),
            )
        except Exception as e:
            logger.debug("NVML query failed: %s", e)

    return PerformanceResponse(
        cpu_name=_cpu_name,
        cpu_percent=psutil.cpu_percent(interval=None),
        ram_used_mb=round(mem.used / 1024 / 1024, 1),
        ram_total_mb=round(mem.total / 1024 / 1024, 1),
        ram_percent=mem.percent,
        gpu=gpu,
        timestamp=datetime.now(timezone.utc),
    )


@router.get("/api/v1/system/metrics-history", tags=["Config"])
async def get_metrics_history(
    _: AuthRequired,
    manager=Depends(get_processor_manager),
):
    """Return the last ~2 minutes of system and per-target metrics.

    Used by the dashboard to seed charts on page load so history
    survives browser refreshes.
    """
    return manager.metrics_history.get_history()


@router.get("/api/v1/system/api-keys", tags=["System"])
def list_api_keys(_: AuthRequired):
    """List API key labels (read-only; keys are defined in the YAML config file)."""
    config = get_config()
    keys = [
        {"label": label, "masked": key[:4] + "****" + key[-4:] if len(key) >= 8 else "****"}
        for label, key in config.auth.api_keys.items()
    ]
    return {"keys": keys, "count": len(keys)}