126 lines
4.6 KiB
Python
126 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import sys
|
|
from urllib.parse import urlparse
|
|
|
|
from common import ROOT, read_json
|
|
|
|
|
|
GPU_PROFILES = ROOT / "config" / "gpu_profiles.json"
|
|
KNOWN_SERVICES = {
|
|
"vllm-text",
|
|
"llama-gguf",
|
|
"translation-api",
|
|
"audio-api",
|
|
"video-api",
|
|
"image-api",
|
|
"model-chat-ui",
|
|
}
|
|
REQUIRED_PROFILES = {"default", "text", "audio", "video", "image", "gguf-1c"}
|
|
REQUIRED_FIELDS = {"id", "label", "command", "starts", "stops", "wait", "notes"}
|
|
|
|
|
|
def require(condition: bool, message: str, errors: list[str]) -> None:
|
|
if not condition:
|
|
errors.append(message)
|
|
|
|
|
|
def validate_service_list(profile_id: str, field: str, value: object, errors: list[str]) -> list[str]:
|
|
require(isinstance(value, list), f"{profile_id}.{field} must be a list", errors)
|
|
if not isinstance(value, list):
|
|
return []
|
|
|
|
services = []
|
|
for index, service in enumerate(value):
|
|
if not isinstance(service, str) or not service:
|
|
errors.append(f"{profile_id}.{field}[{index}] must be a non-empty string")
|
|
continue
|
|
services.append(service)
|
|
if service not in KNOWN_SERVICES:
|
|
errors.append(f"{profile_id}.{field}[{index}] has unknown service `{service}`")
|
|
if len(services) != len(set(services)):
|
|
errors.append(f"{profile_id}.{field} contains duplicate service ids")
|
|
return services
|
|
|
|
|
|
def validate_wait(profile_id: str, value: object, errors: list[str]) -> None:
|
|
require(isinstance(value, list), f"{profile_id}.wait must be a list", errors)
|
|
if not isinstance(value, list):
|
|
return
|
|
|
|
for index, item in enumerate(value):
|
|
if not isinstance(item, dict):
|
|
errors.append(f"{profile_id}.wait[{index}] must be an object")
|
|
continue
|
|
name = item.get("name")
|
|
url = item.get("url")
|
|
if not isinstance(name, str) or not name:
|
|
errors.append(f"{profile_id}.wait[{index}].name must be a non-empty string")
|
|
if not isinstance(url, str) or not url:
|
|
errors.append(f"{profile_id}.wait[{index}].url must be a non-empty string")
|
|
continue
|
|
parsed = urlparse(url)
|
|
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
|
errors.append(f"{profile_id}.wait[{index}].url is not an HTTP URL: {url}")
|
|
|
|
|
|
def validate_profile(profile_id: str, profile: object, errors: list[str]) -> None:
|
|
if not isinstance(profile, dict):
|
|
errors.append(f"{profile_id} must be an object")
|
|
return
|
|
|
|
missing = sorted(REQUIRED_FIELDS - set(profile))
|
|
if missing:
|
|
errors.append(f"{profile_id} missing field(s): {', '.join(missing)}")
|
|
|
|
require(profile.get("id") == profile_id, f"{profile_id}.id must equal `{profile_id}`", errors)
|
|
for field in ("label", "command", "notes"):
|
|
require(isinstance(profile.get(field), str) and bool(profile.get(field)), f"{profile_id}.{field} must be a non-empty string", errors)
|
|
|
|
command = str(profile.get("command") or "")
|
|
require(f"-Profile {profile_id}" in command, f"{profile_id}.command must include `-Profile {profile_id}`", errors)
|
|
|
|
starts = validate_service_list(profile_id, "starts", profile.get("starts"), errors)
|
|
stops = validate_service_list(profile_id, "stops", profile.get("stops"), errors)
|
|
overlap = sorted(set(starts) & set(stops))
|
|
if overlap:
|
|
errors.append(f"{profile_id} has service(s) in both starts and stops: {', '.join(overlap)}")
|
|
validate_wait(profile_id, profile.get("wait"), errors)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Validate GPU profile config.")
|
|
parser.add_argument("--print", action="store_true", help="Print valid profile ids.")
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
profiles = read_json(GPU_PROFILES)
|
|
except (OSError, ValueError) as exc:
|
|
print(f"GPU profile config error: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
errors: list[str] = []
|
|
missing_profiles = sorted(REQUIRED_PROFILES - set(profiles))
|
|
if missing_profiles:
|
|
errors.append(f"missing required profile(s): {', '.join(missing_profiles)}")
|
|
|
|
for profile_id, profile in sorted(profiles.items()):
|
|
validate_profile(str(profile_id), profile, errors)
|
|
|
|
if errors:
|
|
print("GPU profile validation failed:", file=sys.stderr)
|
|
for error in errors:
|
|
print(f"- {error}", file=sys.stderr)
|
|
return 1
|
|
|
|
print(f"Validated {len(profiles)} GPU profile(s).")
|
|
if args.print:
|
|
for profile_id in sorted(profiles):
|
|
print(f"- {profile_id}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|