107 lines
5.3 KiB
Python
107 lines
5.3 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
from ..catalog import get_catalog
|
|
from ..config import settings
|
|
from ..coverage import get_coverage_report
|
|
from ..modules.system.service import get_conda_envs, get_gpus
|
|
from ..modules.weights.service import load_manifest
|
|
|
|
|
|
def _run(command: list[str], cwd: Path | None = None, timeout: int = 60) -> dict:
|
|
result = subprocess.run(command, cwd=cwd, capture_output=True, text=True, timeout=timeout)
|
|
return {
|
|
"command": command,
|
|
"returncode": result.returncode,
|
|
"stdout": result.stdout[-4000:],
|
|
"stderr": result.stderr[-4000:],
|
|
"passed": result.returncode == 0,
|
|
}
|
|
|
|
|
|
def _fetch(url: str, timeout: int = 5) -> dict:
|
|
try:
|
|
with urllib.request.urlopen(url, timeout=timeout) as response:
|
|
body = response.read(200000).decode("utf-8", errors="replace")
|
|
return {"url": url, "status": response.status, "body": body, "passed": 200 <= response.status < 300}
|
|
except urllib.error.HTTPError as exc:
|
|
body = exc.read(200000).decode("utf-8", errors="replace")
|
|
return {"url": url, "status": exc.code, "body": body, "passed": False}
|
|
except Exception as exc:
|
|
return {"url": url, "error": str(exc), "passed": False}
|
|
|
|
|
|
def validate_project(run_build: bool = False) -> dict:
|
|
"""Validate current runtime readiness without launching heavy training."""
|
|
checks = []
|
|
catalog = get_catalog()
|
|
manifest = load_manifest()
|
|
coverage = get_coverage_report()
|
|
|
|
checks.append({"name": "catalog_has_yolo_heatmap", "passed": "yolo.heatmap" in catalog["task_types"]})
|
|
checks.append({"name": "catalog_has_mmseg_31_algs", "passed": len(catalog["mmseg_algorithms"]) >= 31})
|
|
checks.append({"name": "catalog_has_visual_tools", "passed": "visual.yolo11_heatmap_v2" in catalog["task_types"] and "visual.fps" in catalog["task_types"]})
|
|
checks.append({"name": "catalog_has_yolo_dataset_tools", "passed": "dataset.yolo_txt_sort" in catalog["task_types"] and "dataset.yolo_convert_png" in catalog["task_types"]})
|
|
checks.append({"name": "task_buildability", "passed": coverage["task_build_passed"], "detail": coverage["task_build_checks"]})
|
|
checks.append({"name": "script_coverage_user_facing", "passed": not coverage["unmapped_user_scripts"], "detail": coverage})
|
|
checks.append({"name": "weights_manifest_present", "passed": manifest.get("count", 0) >= 1})
|
|
checks.append({"name": "gpus_query", "passed": bool(get_gpus().get("available"))})
|
|
env_names = [item["name"] for item in get_conda_envs().get("envs", [])]
|
|
checks.append({"name": "seg_smp_env_exists", "passed": "seg_smp" in env_names})
|
|
|
|
smoke = _run(
|
|
[
|
|
"conda",
|
|
"run",
|
|
"-n",
|
|
"seg_smp",
|
|
"python",
|
|
"-c",
|
|
"import fastapi, uvicorn, torch, cv2, segmentation_models_pytorch, ultralytics, albumentations, mmengine, mmseg, mmcv; print(torch.__version__, torch.cuda.is_available())",
|
|
],
|
|
cwd=settings.project_root,
|
|
)
|
|
checks.append({"name": "seg_smp_backend_smoke", "passed": smoke["passed"], "detail": smoke})
|
|
|
|
no_weight = _run(["bash", "scripts/check_no_weight_git.sh"], cwd=settings.project_root)
|
|
checks.append({"name": "no_weight_in_git", "passed": no_weight["passed"], "detail": no_weight})
|
|
|
|
if os.getenv("SEG_VALIDATE_LIVE", "1") == "1":
|
|
backend_url = os.getenv("SEG_VALIDATE_BACKEND_URL", "http://127.0.0.1:8010")
|
|
frontend_url = os.getenv("SEG_VALIDATE_FRONTEND_URL", "http://127.0.0.1:5173")
|
|
health = _fetch(f"{backend_url}/api/health")
|
|
datasets = _fetch(f"{backend_url}/api/datasets")
|
|
live_coverage = _fetch(f"{backend_url}/api/coverage")
|
|
frontend = _fetch(frontend_url)
|
|
checks.append({"name": "live_backend_health", "passed": health["passed"] and '"ok":true' in health.get("body", "").replace(" ", ""), "detail": health})
|
|
checks.append({"name": "live_dataset_api", "passed": datasets["passed"] and datasets.get("body", "").lstrip().startswith("["), "detail": datasets})
|
|
checks.append({"name": "live_coverage_api", "passed": live_coverage["passed"] and '"task_build_passed":true' in live_coverage.get("body", "").replace(" ", ""), "detail": live_coverage})
|
|
checks.append({"name": "live_frontend_index", "passed": frontend["passed"] and "Seg Data Server" in frontend.get("body", ""), "detail": frontend})
|
|
|
|
if run_build:
|
|
tests = _run(["conda", "run", "-n", settings.backend_conda_env, "python", "-m", "pytest", "-q"], cwd=settings.project_root, timeout=120)
|
|
checks.append({"name": "backend_tests", "passed": tests["passed"], "detail": tests})
|
|
build = _run(["npm", "run", "build"], cwd=settings.project_root / "frontend", timeout=120)
|
|
checks.append({"name": "frontend_build", "passed": build["passed"], "detail": build})
|
|
|
|
passed = all(item["passed"] for item in checks)
|
|
return {
|
|
"agent": "validation_agent",
|
|
"passed": passed,
|
|
"checks": checks,
|
|
}
|
|
|
|
|
|
def write_validation_report(path: Path, run_build: bool = False) -> dict:
|
|
result = validate_project(run_build=run_build)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
return result
|