533 lines
22 KiB
Python
533 lines
22 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import uuid
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .config import settings
|
|
|
|
|
|
def _run_command(command: list[str], cwd: Path | None = None, timeout: int = 60) -> dict[str, Any]:
|
|
try:
|
|
result = subprocess.run(
|
|
command,
|
|
cwd=str(cwd or settings.project_root),
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout,
|
|
)
|
|
return {
|
|
"passed": result.returncode == 0,
|
|
"returncode": result.returncode,
|
|
"stdout": result.stdout[-4000:],
|
|
"stderr": result.stderr[-4000:],
|
|
}
|
|
except subprocess.TimeoutExpired as exc:
|
|
return {
|
|
"passed": False,
|
|
"returncode": None,
|
|
"stdout": (exc.stdout or "")[-4000:] if isinstance(exc.stdout, str) else "",
|
|
"stderr": (exc.stderr or "")[-4000:] if isinstance(exc.stderr, str) else "",
|
|
"error": f"command timed out after {timeout}s",
|
|
}
|
|
except Exception as exc:
|
|
return {"passed": False, "returncode": None, "stdout": "", "stderr": "", "error": str(exc)}
|
|
|
|
|
|
def _run_snippet(code: str, cwd: Path | None = None, timeout: int = 60) -> dict[str, Any]:
|
|
return _run_command([sys.executable, "-c", code], cwd=cwd, timeout=timeout)
|
|
|
|
|
|
def _run_conda_snippet(env_name: str, code: str, cwd: Path | None = None, timeout: int = 60) -> dict[str, Any]:
|
|
detail = _run_command(["conda", "run", "-n", env_name, "python", "-c", code], cwd=cwd, timeout=timeout)
|
|
detail["env"] = env_name
|
|
return detail
|
|
|
|
|
|
MMSEG_FULL_BUILD_SNIPPET = (
|
|
"from mmseg.utils import register_all_modules; "
|
|
"register_all_modules(init_default_scope=True); "
|
|
"from mmengine.config import Config; "
|
|
"from mmseg.registry import MODELS; "
|
|
"import mmcv._ext; "
|
|
"cfg=Config.fromfile({config_path!r}); "
|
|
"model=MODELS.build(cfg.model); "
|
|
"print(type(model).__name__)"
|
|
)
|
|
|
|
|
|
SEGMODEL_TRAIN_STEP_SNIPPET = (
|
|
"import torch, segmentation_models_pytorch as smp; "
|
|
"torch.manual_seed(7); "
|
|
"model=smp.Unet(encoder_name='resnet18', encoder_weights=None, classes=2).train(); "
|
|
"inputs=torch.randn(2,3,64,64); "
|
|
"targets=torch.randint(0,2,(2,64,64)); "
|
|
"optimizer=torch.optim.SGD(model.parameters(), lr=1e-3); "
|
|
"outputs=model(inputs); "
|
|
"loss=torch.nn.functional.cross_entropy(outputs, targets); "
|
|
"loss.backward(); optimizer.step(); "
|
|
"print('loss', round(float(loss.detach()), 6), 'shape', tuple(outputs.shape))"
|
|
)
|
|
|
|
|
|
def _yolo_tiny_train_snippet(root: Path, weight: Path) -> str:
|
|
custom_train = settings.project_root / "backend" / "app" / "modules" / "yolo" / "custom_train.py"
|
|
yolo_dir = settings.source_root / "Seg_All_In_One_YoloModel"
|
|
return (
|
|
"import shutil, subprocess, sys, cv2, numpy as np; "
|
|
"from pathlib import Path; "
|
|
f"root=Path({str(root)!r}); weight={str(weight)!r}; "
|
|
f"custom_train=Path({str(custom_train)!r}); yolo_dir=Path({str(yolo_dir)!r}); "
|
|
"shutil.rmtree(root, ignore_errors=True); "
|
|
"[ (root / item).mkdir(parents=True, exist_ok=True) for item in ['images/train','images/val','labels/train','labels/val','runs'] ]; "
|
|
"image=np.zeros((64,64,3), dtype=np.uint8); "
|
|
"cv2.rectangle(image, (16,16), (48,48), (255,255,255), -1); "
|
|
"label='0 0.25 0.25 0.75 0.25 0.75 0.75 0.25 0.75\\n'; "
|
|
"\nfor split in ['train','val']:\n"
|
|
" cv2.imwrite(str(root / 'images' / split / 'sample.jpg'), image)\n"
|
|
" (root / 'labels' / split / 'sample.txt').write_text(label, encoding='utf-8')\n"
|
|
"(root / 'data.yaml').write_text('path: '+str(root)+'\\ntrain: images/train\\nval: images/val\\nnc: 1\\nnames:\\n 0: object\\n', encoding='utf-8'); "
|
|
"subprocess.run([sys.executable, str(custom_train), '--data', str(root/'data.yaml'), '--model', str(weight), '--epochs', '1', '--imgsz', '64', '--batch', '1', '--workers', '0', '--device', 'cpu', '--project', str(root/'runs'), '--name', 'tiny', '--exist-ok'], cwd=str(yolo_dir), check=True); "
|
|
"results=root/'runs'/'tiny'/'results.csv'; best=root/'runs'/'tiny'/'weights'/'best.pt'; "
|
|
"assert results.exists() and results.stat().st_size > 0; "
|
|
"assert best.exists() and best.stat().st_size > 0; "
|
|
"print('results', results, results.stat().st_size, 'best', best.stat().st_size)"
|
|
)
|
|
|
|
|
|
def _yolo_heatmap_snippet(root: Path) -> str:
|
|
script_path = settings.source_root / "Seg_All_In_One_YoloModel" / "yolo_predict_visualize_nn.py"
|
|
return (
|
|
"from pathlib import Path; "
|
|
"import importlib.util, shutil, sys, types; "
|
|
"fake=types.ModuleType('yolo_config'); "
|
|
"fake.MODEL_CONFIGS={'YOLO11n-seg': {}}; "
|
|
"fake.TEST_IMAGE_DIR=''; fake.PREDICT_BEST_MODEL_DIR=Path('.'); fake.show_config_summary=lambda: None; "
|
|
"sys.modules['yolo_config']=fake; "
|
|
f"script=Path({str(script_path)!r}); "
|
|
"spec=importlib.util.spec_from_file_location('yolo_heatmap_mod', script); "
|
|
"mod=importlib.util.module_from_spec(spec); spec.loader.exec_module(mod); "
|
|
f"root=Path({str(root)!r}); "
|
|
"base=root/'runs'/'tiny'; "
|
|
"heatmap_root=base/'HeartMap_Visual'; "
|
|
"shutil.rmtree(heatmap_root, ignore_errors=True); "
|
|
"mod.visualize_nn_comprehensive(str(base/'weights'/'best.pt'), str(root/'images'/'val'/'sample.jpg'), base, 'best.pt', 'GradCAM', 'model.model.model[9]', 'YOLO11n-seg'); "
|
|
"outputs=sorted(heatmap_root.rglob('*.jpg')); "
|
|
"assert len(outputs) >= 2; "
|
|
"print('heatmaps', len(outputs), [str(item.relative_to(base)) for item in outputs[:4]])"
|
|
)
|
|
|
|
|
|
def _mmseg_train_step_snippet(config_path: Path) -> str:
|
|
return (
|
|
"import torch; "
|
|
"from mmengine.config import Config; "
|
|
"from mmengine.structures import PixelData; "
|
|
"from mmseg.registry import MODELS; "
|
|
"from mmseg.structures import SegDataSample; "
|
|
"from mmseg.utils import register_all_modules; "
|
|
"register_all_modules(init_default_scope=True); "
|
|
f"cfg=Config.fromfile({str(config_path)!r}); "
|
|
"cfg.model.backbone.init_cfg=None; cfg.model.pretrained=None; "
|
|
"model=MODELS.build(cfg.model).train(); "
|
|
"sample=SegDataSample(); "
|
|
"sample.gt_sem_seg=PixelData(data=torch.randint(0,19,(1,64,64), dtype=torch.long)); "
|
|
"losses=model(torch.randn(1,3,64,64), [sample], mode='loss'); "
|
|
"loss=sum(value if torch.is_tensor(value) else sum(value) for value in losses.values()); "
|
|
"optimizer=torch.optim.SGD(model.parameters(), lr=1e-4); "
|
|
"loss.backward(); optimizer.step(); "
|
|
"print('loss', round(float(loss.detach()), 6), sorted(losses.keys()))"
|
|
)
|
|
|
|
|
|
def _request_json(method: str, url: str, payload: dict[str, Any] | None = None, timeout: int = 10) -> dict[str, Any]:
|
|
data = None
|
|
headers = {"Accept": "application/json"}
|
|
if payload is not None:
|
|
data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
|
headers["Content-Type"] = "application/json"
|
|
request = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=timeout) as response:
|
|
body = response.read().decode("utf-8", errors="replace")
|
|
return {"passed": 200 <= response.status < 300, "status": response.status, "body": body, "json": json.loads(body) if body else None}
|
|
except urllib.error.HTTPError as exc:
|
|
body = exc.read().decode("utf-8", errors="replace")
|
|
return {"passed": False, "status": exc.code, "body": body}
|
|
except Exception as exc:
|
|
return {"passed": False, "error": str(exc)}
|
|
|
|
|
|
def _request_text(url: str, timeout: int = 10) -> dict[str, Any]:
|
|
try:
|
|
with urllib.request.urlopen(url, timeout=timeout) as response:
|
|
body = response.read().decode("utf-8", errors="replace")
|
|
return {"passed": 200 <= response.status < 300, "status": response.status, "body": body}
|
|
except Exception as exc:
|
|
return {"passed": False, "error": str(exc)}
|
|
|
|
|
|
def _post_multipart(url: str, field: str, filename: str, content: bytes, content_type: str = "text/plain", timeout: int = 10) -> dict[str, Any]:
|
|
boundary = f"----SegAcceptance{uuid.uuid4().hex}"
|
|
body = b"".join(
|
|
[
|
|
f"--{boundary}\r\n".encode(),
|
|
f'Content-Disposition: form-data; name="{field}"; filename="{filename}"\r\n'.encode(),
|
|
f"Content-Type: {content_type}\r\n\r\n".encode(),
|
|
content,
|
|
f"\r\n--{boundary}--\r\n".encode(),
|
|
]
|
|
)
|
|
request = urllib.request.Request(
|
|
url,
|
|
data=body,
|
|
method="POST",
|
|
headers={"Content-Type": f"multipart/form-data; boundary={boundary}", "Accept": "application/json"},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=timeout) as response:
|
|
text = response.read().decode("utf-8", errors="replace")
|
|
return {"passed": 200 <= response.status < 300, "status": response.status, "body": text, "json": json.loads(text) if text else None}
|
|
except urllib.error.HTTPError as exc:
|
|
text = exc.read().decode("utf-8", errors="replace")
|
|
return {"passed": False, "status": exc.code, "body": text}
|
|
except Exception as exc:
|
|
return {"passed": False, "error": str(exc)}
|
|
|
|
|
|
def _poll_job(base_url: str, job_id: str, timeout: int = 90) -> dict[str, Any]:
|
|
deadline = time.time() + timeout
|
|
last = None
|
|
while time.time() < deadline:
|
|
result = _request_json("GET", f"{base_url}/api/jobs/{job_id}", timeout=10)
|
|
last = result
|
|
job = result.get("json") if result.get("passed") else None
|
|
if job and job.get("status") in {"success", "failed", "cancelled"}:
|
|
return {"passed": job.get("status") == "success", "job": job}
|
|
time.sleep(1)
|
|
return {"passed": False, "error": "job timed out", "last": last}
|
|
|
|
|
|
def _create_job_and_wait(base_url: str, task_type: str, params: dict[str, Any], timeout: int = 90) -> dict[str, Any]:
|
|
created = _request_json("POST", f"{base_url}/api/jobs", {"type": task_type, "params": params}, timeout=10)
|
|
if not created.get("passed") or not created.get("json"):
|
|
return {"passed": False, "created": created}
|
|
job_id = created["json"]["id"]
|
|
polled = _poll_job(base_url, job_id, timeout=timeout)
|
|
events = _request_text(f"{base_url}/api/jobs/{job_id}/events", timeout=10)
|
|
return {"passed": polled["passed"], "created": created["json"], "polled": polled, "events": events}
|
|
|
|
|
|
def _create_job_with_retry(base_url: str, task_type: str, params: dict[str, Any], attempts: int = 2, timeout: int = 90) -> dict[str, Any]:
|
|
results = []
|
|
for _ in range(attempts):
|
|
result = _create_job_and_wait(base_url, task_type, params, timeout=timeout)
|
|
results.append(result)
|
|
if result.get("passed"):
|
|
return {"passed": True, "attempts": results}
|
|
time.sleep(1)
|
|
return {"passed": False, "attempts": results}
|
|
|
|
|
|
def _write_acceptance_images(root: Path) -> tuple[Path, Path, Path]:
|
|
import cv2
|
|
import numpy as np
|
|
|
|
image_dir = root / "images"
|
|
label_dir = root / "labels"
|
|
result_dir = root / "stacked"
|
|
image_dir.mkdir(parents=True, exist_ok=True)
|
|
label_dir.mkdir(parents=True, exist_ok=True)
|
|
result_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
image = np.zeros((16, 16, 3), dtype=np.uint8)
|
|
image[:, :, 1] = 180
|
|
label = np.zeros((16, 16, 3), dtype=np.uint8)
|
|
label[4:12, 4:12, 2] = 255
|
|
image_path = image_dir / "sample.png"
|
|
label_path = label_dir / "sample.png"
|
|
cv2.imwrite(str(image_path), image)
|
|
cv2.imwrite(str(label_path), label)
|
|
return image_path, label_path, result_dir
|
|
|
|
|
|
def run_model_family_readiness() -> dict[str, Any]:
|
|
"""Exercise the model-family runtime stack without launching full training."""
|
|
source = settings.source_root
|
|
yolo_weight = source / "Seg_All_In_One_YoloModel" / "yolo11n-seg.pt"
|
|
mmseg_config = source / "Seg_All_In_One_MMSeg" / "configs" / "fcn" / "fcn_r18-d8_4xb2-80k_cityscapes-512x1024.py"
|
|
mmseg_pretrained = source / "Seg_All_In_One_MMSeg" / "My_Local_Model" / "mmcls" / "resnet18.pth"
|
|
|
|
checks = [
|
|
{
|
|
"name": "segmodel_smp_forward",
|
|
"required": True,
|
|
"detail": _run_snippet(
|
|
"import torch, segmentation_models_pytorch as smp; "
|
|
"m=smp.Unet(encoder_name='resnet18', encoder_weights=None, classes=2).eval(); "
|
|
"torch.set_grad_enabled(False); y=m(torch.randn(1,3,64,64)); "
|
|
"print(tuple(y.shape))"
|
|
),
|
|
},
|
|
{
|
|
"name": "yolo_seg_predict_cpu",
|
|
"required": True,
|
|
"detail": _run_snippet(
|
|
"from ultralytics import YOLO; import numpy as np; "
|
|
f"model=YOLO({str(yolo_weight)!r}); "
|
|
"r=model.predict(np.zeros((64,64,3), dtype=np.uint8), imgsz=64, verbose=False, save=False, device='cpu'); "
|
|
"print(len(r), r[0].orig_shape)"
|
|
),
|
|
},
|
|
{
|
|
"name": "mmseg_config_parse",
|
|
"required": True,
|
|
"detail": _run_snippet(
|
|
"from mmengine.config import Config; "
|
|
f"cfg=Config.fromfile({str(mmseg_config)!r}); "
|
|
"print(cfg.model.type, cfg.train_dataloader.batch_size)"
|
|
),
|
|
},
|
|
{
|
|
"name": "mmseg_local_pretrained_weight",
|
|
"required": True,
|
|
"detail": {"passed": mmseg_pretrained.exists(), "path": str(mmseg_pretrained), "size": mmseg_pretrained.stat().st_size if mmseg_pretrained.exists() else 0},
|
|
},
|
|
{
|
|
"name": "mmseg_full_env_imports",
|
|
"required": True,
|
|
"detail": _run_conda_snippet(
|
|
settings.mmseg_conda_env,
|
|
"import torch, cv2, mmcv, mmengine, mmseg; "
|
|
"import mmcv._ext; "
|
|
"print(torch.__version__, torch.version.cuda, cv2.__version__, mmcv.__version__, mmseg.__version__)",
|
|
timeout=90,
|
|
),
|
|
},
|
|
{
|
|
"name": "mmseg_full_model_build",
|
|
"required": True,
|
|
"detail": _run_conda_snippet(
|
|
settings.mmseg_conda_env,
|
|
MMSEG_FULL_BUILD_SNIPPET.format(config_path=str(mmseg_config)),
|
|
timeout=90,
|
|
),
|
|
},
|
|
]
|
|
for check in checks:
|
|
check["passed"] = bool(check["detail"].get("passed"))
|
|
return {
|
|
"passed": all(item["passed"] for item in checks if item["required"]),
|
|
"warnings": [item for item in checks if not item["required"] and not item["passed"]],
|
|
"checks": checks,
|
|
}
|
|
|
|
|
|
def latest_acceptance_report() -> dict[str, Any]:
|
|
path = settings.project_root / "var" / "acceptance" / "latest.json"
|
|
if not path.exists():
|
|
return {"available": False, "path": str(path)}
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def latest_deep_acceptance_report() -> dict[str, Any]:
|
|
path = settings.project_root / "var" / "acceptance" / "deep_latest.json"
|
|
if not path.exists():
|
|
return {"available": False, "path": str(path)}
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def run_deep_acceptance() -> dict[str, Any]:
|
|
"""Run minimal training loops for each model family without full datasets."""
|
|
acceptance_root = settings.project_root / "var" / "acceptance"
|
|
run_id = uuid.uuid4().hex[:8]
|
|
fixture_root = acceptance_root / f"deep_{run_id}"
|
|
fixture_root.mkdir(parents=True, exist_ok=True)
|
|
|
|
yolo_weight = settings.source_root / "Seg_All_In_One_YoloModel" / "yolo11n-seg.pt"
|
|
mmseg_config = settings.source_root / "Seg_All_In_One_MMSeg" / "configs" / "fcn" / "fcn_r18-d8_4xb2-80k_cityscapes-512x1024.py"
|
|
|
|
yolo_root = fixture_root / "yolo_tiny"
|
|
checks = [
|
|
{
|
|
"name": "segmodel_tiny_train_step",
|
|
"passed": False,
|
|
"detail": _run_snippet(SEGMODEL_TRAIN_STEP_SNIPPET, timeout=90),
|
|
},
|
|
]
|
|
yolo_train = {
|
|
"name": "yolo_tiny_segment_train_epoch",
|
|
"passed": False,
|
|
"detail": _run_snippet(_yolo_tiny_train_snippet(yolo_root, yolo_weight), timeout=180),
|
|
}
|
|
checks.append(yolo_train)
|
|
if yolo_train["detail"].get("passed"):
|
|
checks.append(
|
|
{
|
|
"name": "yolo_tiny_heatmap_generation",
|
|
"passed": False,
|
|
"detail": _run_snippet(_yolo_heatmap_snippet(yolo_root), timeout=90),
|
|
}
|
|
)
|
|
else:
|
|
checks.append(
|
|
{
|
|
"name": "yolo_tiny_heatmap_generation",
|
|
"passed": False,
|
|
"detail": {"passed": False, "error": "skipped because yolo_tiny_segment_train_epoch failed"},
|
|
}
|
|
)
|
|
checks.append(
|
|
{
|
|
"name": "mmseg_tiny_train_step",
|
|
"passed": False,
|
|
"detail": _run_conda_snippet(settings.mmseg_conda_env, _mmseg_train_step_snippet(mmseg_config), timeout=120),
|
|
}
|
|
)
|
|
for check in checks:
|
|
check["passed"] = bool(check["detail"].get("passed"))
|
|
|
|
report = {
|
|
"available": True,
|
|
"run_id": run_id,
|
|
"fixture_root": str(fixture_root),
|
|
"passed": all(item["passed"] for item in checks),
|
|
"checks": checks,
|
|
"created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
|
}
|
|
latest = acceptance_root / "deep_latest.json"
|
|
latest.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
return report
|
|
|
|
|
|
def run_live_acceptance(base_url: str = "http://127.0.0.1:8010") -> dict[str, Any]:
|
|
"""Run a lightweight end-to-end smoke against the live API and job runner."""
|
|
acceptance_root = settings.project_root / "var" / "acceptance"
|
|
run_id = uuid.uuid4().hex[:8]
|
|
fixture_root = acceptance_root / f"run_{run_id}"
|
|
fixture_root.mkdir(parents=True, exist_ok=True)
|
|
|
|
checks: list[dict[str, Any]] = []
|
|
|
|
dataset_name = f"acceptance_{run_id}"
|
|
created_dataset = _request_json("POST", f"{base_url}/api/datasets", {"name": dataset_name, "description": "acceptance smoke"}, timeout=10)
|
|
checks.append({"name": "create_dataset_api", "passed": created_dataset.get("passed", False), "detail": created_dataset})
|
|
|
|
import cv2
|
|
import numpy as np
|
|
|
|
image = np.zeros((16, 16, 3), dtype=np.uint8)
|
|
image[:, :, 1] = 160
|
|
mask = np.zeros((16, 16), dtype=np.uint8)
|
|
mask[4:12, 4:12] = 255
|
|
_, image_encoded = cv2.imencode(".png", image)
|
|
_, mask_encoded = cv2.imencode(".png", mask)
|
|
|
|
upload_image = _post_multipart(
|
|
f"{base_url}/api/datasets/{dataset_name}/upload/images",
|
|
"files",
|
|
"sample.png",
|
|
image_encoded.tobytes(),
|
|
"image/png",
|
|
timeout=10,
|
|
)
|
|
checks.append({"name": "upload_image_api", "passed": upload_image.get("passed", False), "detail": upload_image})
|
|
|
|
upload = _post_multipart(
|
|
f"{base_url}/api/datasets/{dataset_name}/upload/labels",
|
|
"files",
|
|
"sample.txt",
|
|
b"0 0.5 0.5 0.25 0.25\n",
|
|
"text/plain",
|
|
timeout=10,
|
|
)
|
|
checks.append({"name": "upload_label_api", "passed": upload.get("passed", False), "detail": upload})
|
|
|
|
upload_mask = _post_multipart(
|
|
f"{base_url}/api/datasets/{dataset_name}/upload/masks",
|
|
"files",
|
|
"sample.png",
|
|
mask_encoded.tobytes(),
|
|
"image/png",
|
|
timeout=10,
|
|
)
|
|
checks.append({"name": "upload_mask_api", "passed": upload_mask.get("passed", False), "detail": upload_mask})
|
|
|
|
artifact_ok = False
|
|
artifact_detail: dict[str, Any] = {"skipped": True}
|
|
try:
|
|
relative_path = upload["json"]["saved"][0]["relative_path"]
|
|
artifact_detail = _request_text(f"{base_url}/api/artifacts/{relative_path}", timeout=10)
|
|
artifact_ok = artifact_detail.get("passed", False) and "0 0.5" in artifact_detail.get("body", "")
|
|
except Exception as exc:
|
|
artifact_detail = {"error": str(exc)}
|
|
checks.append({"name": "artifact_api_for_uploaded_label", "passed": artifact_ok, "detail": artifact_detail})
|
|
|
|
dataset_validation = _request_json("GET", f"{base_url}/api/datasets/{dataset_name}/validate", timeout=10)
|
|
validation_json = dataset_validation.get("json") if dataset_validation.get("passed") else {}
|
|
checks.append(
|
|
{
|
|
"name": "dataset_validate_api",
|
|
"passed": dataset_validation.get("passed", False) and validation_json.get("ready", {}).get("yolo") and validation_json.get("ready", {}).get("mask"),
|
|
"detail": dataset_validation,
|
|
}
|
|
)
|
|
|
|
yolo_yaml = _request_json("POST", f"{base_url}/api/datasets/{dataset_name}/yolo-yaml", {"class_names": ["object"]}, timeout=10)
|
|
yolo_yaml_json = yolo_yaml.get("json") if yolo_yaml.get("passed") else {}
|
|
checks.append(
|
|
{
|
|
"name": "dataset_yolo_yaml_api",
|
|
"passed": yolo_yaml.get("passed", False) and "dataset.yaml" in str(yolo_yaml_json.get("relative_path", "")),
|
|
"detail": yolo_yaml,
|
|
}
|
|
)
|
|
|
|
mock = _create_job_and_wait(base_url, "mock.echo", {"message": f"acceptance {run_id}"}, timeout=45)
|
|
mock_log = mock.get("polled", {}).get("job", {}).get("log_tail", "")
|
|
checks.append({"name": "mock_job_runner", "passed": mock.get("passed", False) and f"acceptance {run_id}" in mock_log, "detail": mock})
|
|
|
|
image_path, label_path, result_dir = _write_acceptance_images(fixture_root)
|
|
stack = _create_job_with_retry(
|
|
base_url,
|
|
"dataset.stack_single",
|
|
{
|
|
"image_path": str(image_path),
|
|
"label_path": str(label_path),
|
|
"result_dir": str(result_dir),
|
|
"alpha": 0.35,
|
|
},
|
|
attempts=2,
|
|
timeout=90,
|
|
)
|
|
output_path = result_dir / "sample.png"
|
|
checks.append(
|
|
{
|
|
"name": "legacy_stack_job_runner",
|
|
"passed": stack.get("passed", False) and output_path.exists() and output_path.stat().st_size > 0,
|
|
"detail": {**stack, "output_path": str(output_path), "output_exists": output_path.exists()},
|
|
}
|
|
)
|
|
|
|
readiness = run_model_family_readiness()
|
|
checks.append({"name": "model_family_readiness", "passed": readiness["passed"], "detail": readiness})
|
|
|
|
report = {
|
|
"available": True,
|
|
"run_id": run_id,
|
|
"base_url": base_url,
|
|
"passed": all(item["passed"] for item in checks),
|
|
"checks": checks,
|
|
"model_family_readiness": readiness,
|
|
"created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
|
}
|
|
latest = acceptance_root / "latest.json"
|
|
latest.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
return report
|