alexei.dolgolyov
f3d07fc47f
Lint & Test / test (push) Has been cancelled
- Dismissible donation/open-source banner after 3+ sessions (30-day snooze) - New About tab in Settings: version, repo link, license info - Centralize project URLs (REPO_URL, DONATE_URL) in __init__.py, served via /health - Center settings tab bar, reduce tab padding for 6-tab fit - External URL save button: icon button instead of full-width text button - Remove redundant settings footer close button - Footer "Source Code" link replaced with "About" opening settings - i18n keys for en/ru/zh
314 lines
10 KiB
Python
314 lines
10 KiB
Python
"""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__, REPO_URL, DONATE_URL
|
|
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 ( # noqa: E402
|
|
nvml_available as _nvml_available,
|
|
nvml as _nvml,
|
|
nvml_handle as _nvml_handle,
|
|
)
|
|
|
|
|
|
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(),
|
|
repo_url=REPO_URL,
|
|
donate_url=DONATE_URL,
|
|
)
|
|
|
|
|
|
@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()
|
|
from wled_controller.api.dependencies import get_asset_store
|
|
|
|
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,
|
|
get_asset_store,
|
|
]
|
|
for getter in store_getters:
|
|
try:
|
|
store = getter()
|
|
except RuntimeError as e:
|
|
logger.debug("Store not available during entity count: %s", e)
|
|
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("Failed to get displays: %s", e, exc_info=True)
|
|
raise HTTPException(status_code=500, detail="Internal server error")
|
|
|
|
|
|
@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("Failed to get processes: %s", e, exc_info=True)
|
|
raise HTTPException(status_code=500, detail="Internal server error")
|
|
|
|
|
|
@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] + "****" if len(key) >= 8 else "****"}
|
|
for label, key in config.auth.api_keys.items()
|
|
]
|
|
return {"keys": keys, "count": len(keys)}
|