172 lines
9.5 KiB
Python
172 lines
9.5 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 ..acceptance import run_deep_acceptance, run_live_acceptance
|
|
from ..capabilities import get_capability_matrix
|
|
from ..catalog import get_catalog
|
|
from ..config import settings
|
|
from ..coverage import get_coverage_report
|
|
from ..modules.results.service import scan_training_curves
|
|
from ..modules.system.service import get_conda_envs, get_gpus, get_runtime_readiness
|
|
from ..modules.weights.service import load_manifest
|
|
from ..progress import parse_progress
|
|
|
|
|
|
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, run_acceptance: bool | None = None, run_deep: bool | None = None) -> 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_yolo_custom_train", "passed": "yolo.train_custom" in catalog["task_types"]})
|
|
checks.append({"name": "catalog_has_yolo_custom_predict_heatmap", "passed": "yolo.predict_custom" in catalog["task_types"] and "yolo.heatmap_custom" 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})
|
|
curves = scan_training_curves()
|
|
checks.append({"name": "training_curves_detected", "passed": len(curves) >= 1, "detail": {"count": len(curves), "examples": [item["relative_path"] for item in curves[:5]]}})
|
|
parsed_progress = parse_progress("Epoch(train) [2][25/50] lr: 1e-3\n", status="running")
|
|
checks.append({"name": "job_progress_parser", "passed": parsed_progress["stage"] == "training" and parsed_progress["percent"] == 50, "detail": parsed_progress})
|
|
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": "task_env_exists", "passed": settings.task_conda_env in env_names, "detail": {"env": settings.task_conda_env}})
|
|
checks.append({"name": "mmseg_env_exists", "passed": settings.mmseg_conda_env in env_names, "detail": {"env": settings.mmseg_conda_env}})
|
|
runtime_readiness = get_runtime_readiness(force=True)
|
|
checks.append({"name": "runtime_env_readiness", "passed": runtime_readiness["passed"], "detail": runtime_readiness})
|
|
capability_matrix = get_capability_matrix()
|
|
checks.append({
|
|
"name": "capability_matrix_ready",
|
|
"passed": capability_matrix["passed"] and capability_matrix["summary"]["ready_domains"] == capability_matrix["summary"]["total_domains"],
|
|
"detail": capability_matrix,
|
|
})
|
|
|
|
smoke = _run(
|
|
[
|
|
"conda",
|
|
"run",
|
|
"-n",
|
|
settings.task_conda_env,
|
|
"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": "task_env_backend_smoke", "passed": smoke["passed"], "detail": smoke})
|
|
|
|
mmseg_smoke = _run(
|
|
[
|
|
"conda",
|
|
"run",
|
|
"-n",
|
|
settings.mmseg_conda_env,
|
|
"python",
|
|
"-c",
|
|
"import torch, cv2, mmcv, mmengine, mmseg; import mmcv._ext; print(torch.__version__, torch.version.cuda, cv2.__version__, mmcv.__version__, mmseg.__version__)",
|
|
],
|
|
cwd=settings.project_root,
|
|
)
|
|
checks.append({"name": "mmseg_env_full_mmcv_smoke", "passed": mmseg_smoke["passed"], "detail": mmseg_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_jobs = _fetch(f"{backend_url}/api/jobs")
|
|
live_readiness = _fetch(f"{backend_url}/api/system/readiness", timeout=20)
|
|
live_capabilities = _fetch(f"{backend_url}/api/capabilities", timeout=20)
|
|
live_coverage = _fetch(f"{backend_url}/api/coverage", timeout=20)
|
|
live_curves = _fetch(f"{backend_url}/api/results/curves", timeout=20)
|
|
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})
|
|
try:
|
|
live_job_items = json.loads(live_jobs.get("body", "[]")) if live_jobs.get("passed") else []
|
|
except Exception:
|
|
live_job_items = []
|
|
checks.append({
|
|
"name": "live_jobs_progress_api",
|
|
"passed": live_jobs["passed"] and isinstance(live_job_items, list) and (not live_job_items or "progress" in live_job_items[0]),
|
|
"detail": live_jobs,
|
|
})
|
|
checks.append({
|
|
"name": "live_runtime_readiness_api",
|
|
"passed": live_readiness["passed"] and '"passed":true' in live_readiness.get("body", "").replace(" ", ""),
|
|
"detail": live_readiness,
|
|
})
|
|
checks.append({
|
|
"name": "live_capability_matrix_api",
|
|
"passed": live_capabilities["passed"] and '"passed":true' in live_capabilities.get("body", "").replace(" ", ""),
|
|
"detail": live_capabilities,
|
|
})
|
|
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_training_curves_api", "passed": live_curves["passed"] and live_curves.get("body", "").lstrip().startswith("["), "detail": live_curves})
|
|
checks.append({"name": "live_frontend_index", "passed": frontend["passed"] and "Seg Data Server" in frontend.get("body", ""), "detail": frontend})
|
|
acceptance_enabled = run_acceptance if run_acceptance is not None else os.getenv("SEG_VALIDATE_ACCEPTANCE", "1") == "1"
|
|
deep_enabled = run_deep if run_deep is not None else os.getenv("SEG_VALIDATE_DEEP", "1") == "1"
|
|
if acceptance_enabled:
|
|
acceptance = run_live_acceptance(backend_url)
|
|
checks.append({"name": "live_acceptance_smoke", "passed": acceptance["passed"], "detail": acceptance})
|
|
if deep_enabled:
|
|
deep_acceptance = run_deep_acceptance()
|
|
checks.append({"name": "deep_training_acceptance", "passed": deep_acceptance["passed"], "detail": deep_acceptance})
|
|
|
|
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
|