Add real workspace acceptance

backend/app/acceptance.py

@@ -12,6 +12,8 @@ from typing import Any
 
 from .config import settings
 
+IMAGE_SUFFIXES = {".png", ".jpg", ".jpeg", ".bmp", ".tif", ".tiff"}
+
 
 def _run_command(command: list[str], cwd: Path | None = None, timeout: int = 60) -> dict[str, Any]:
     try:
@@ -183,6 +185,23 @@ def _request_text(url: str, timeout: int = 10) -> dict[str, Any]:
         return {"passed": False, "error": str(exc)}
 
 
+def _content_type(path: Path) -> str:
+    suffix = path.suffix.lower()
+    if suffix in {".jpg", ".jpeg"}:
+        return "image/jpeg"
+    if suffix == ".png":
+        return "image/png"
+    if suffix in {".tif", ".tiff"}:
+        return "image/tiff"
+    if suffix == ".txt":
+        return "text/plain"
+    return "application/octet-stream"
+
+
+def _post_file(url: str, path: Path, timeout: int = 30) -> dict[str, Any]:
+    return _post_multipart(url, "files", path.name, path.read_bytes(), _content_type(path), timeout=timeout)
+
+
 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(
@@ -280,6 +299,61 @@ def _result_files(root: Path, suffixes: set[str]) -> list[Path]:
     return sorted(path for path in root.rglob("*") if path.is_file() and path.suffix.lower() in suffixes)
 
 
+def _files_by_stem(root: Path, suffixes: set[str], nonempty: bool = True) -> dict[str, Path]:
+    if not root.exists():
+        return {}
+    files: dict[str, Path] = {}
+    for path in sorted(root.iterdir()):
+        if not path.is_file() or path.suffix.lower() not in suffixes:
+            continue
+        if nonempty and path.stat().st_size <= 0:
+            continue
+        files.setdefault(path.stem, path)
+    return files
+
+
+def _find_stem_pair(left_root: Path, left_suffixes: set[str], right_root: Path, right_suffixes: set[str]) -> tuple[Path, Path] | None:
+    left = _files_by_stem(left_root, left_suffixes)
+    right = _files_by_stem(right_root, right_suffixes)
+    for stem in sorted(set(left) & set(right)):
+        return left[stem], right[stem]
+    return None
+
+
+def find_real_workspace_samples() -> dict[str, Any]:
+    """Find existing non-synthetic samples from the checked-out Seg workspace."""
+    source = settings.source_root
+    mask_pair = None
+    mask_candidates = []
+    for prefix in ("A", "B", "C"):
+        image_root = source / "DataSet_Own" / f"{prefix}_Ori"
+        mask_root = source / "DataSet_Own" / f"{prefix}_Label_Ori"
+        mask_candidates.append({"image_root": str(image_root), "mask_root": str(mask_root)})
+        pair = _find_stem_pair(image_root, IMAGE_SUFFIXES, mask_root, IMAGE_SUFFIXES)
+        if pair:
+            mask_pair = {"image": str(pair[0]), "mask": str(pair[1]), "dataset": prefix}
+            break
+
+    yolo_pair = None
+    yolo_candidates = []
+    yolo_dataset = source / "Seg_All_In_One_YoloModel" / "Yolo数据集构建" / "Data"
+    for split in ("train", "val"):
+        image_root = yolo_dataset / "images" / split
+        label_root = yolo_dataset / "labels" / split
+        yolo_candidates.append({"image_root": str(image_root), "label_root": str(label_root)})
+        pair = _find_stem_pair(image_root, IMAGE_SUFFIXES, label_root, {".txt"})
+        if pair:
+            yolo_pair = {"image": str(pair[0]), "label": str(pair[1]), "split": split}
+            break
+
+    return {
+        "passed": bool(mask_pair and yolo_pair),
+        "mask_pair": mask_pair,
+        "yolo_pair": yolo_pair,
+        "candidates": {"mask": mask_candidates, "yolo": yolo_candidates},
+    }
+
+
 def run_model_family_readiness() -> dict[str, Any]:
     """Exercise the model-family runtime stack without launching full training."""
     source = settings.source_root
@@ -366,6 +440,186 @@ def latest_deep_acceptance_report() -> dict[str, Any]:
     return json.loads(path.read_text(encoding="utf-8"))
 
 
+def latest_real_acceptance_report() -> dict[str, Any]:
+    path = settings.project_root / "var" / "acceptance" / "real_latest.json"
+    if not path.exists():
+        return {"available": False, "path": str(path)}
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def run_real_dataset_acceptance(base_url: str = "http://127.0.0.1:8010") -> dict[str, Any]:
+    """Run the upload/predict/heatmap path against existing non-synthetic Seg data."""
+    acceptance_root = settings.project_root / "var" / "acceptance"
+    run_id = uuid.uuid4().hex[:8]
+    fixture_root = acceptance_root / f"real_{run_id}"
+    fixture_root.mkdir(parents=True, exist_ok=True)
+
+    samples = find_real_workspace_samples()
+    checks: list[dict[str, Any]] = [
+        {"name": "real_workspace_samples_discovered", "passed": samples["passed"], "detail": samples}
+    ]
+    if not samples["passed"]:
+        report = {
+            "available": True,
+            "run_id": run_id,
+            "base_url": base_url,
+            "fixture_root": str(fixture_root),
+            "passed": False,
+            "checks": checks,
+            "created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+        }
+        (acceptance_root / "real_latest.json").write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
+        return report
+
+    dataset_name = f"real_acceptance_{run_id}"
+    created_dataset = _request_json("POST", f"{base_url}/api/datasets", {"name": dataset_name, "description": "real workspace acceptance"}, timeout=10)
+    checks.append({"name": "create_real_upload_dataset", "passed": created_dataset.get("passed", False), "detail": created_dataset})
+
+    mask_image = Path(samples["mask_pair"]["image"])
+    mask_file = Path(samples["mask_pair"]["mask"])
+    yolo_image = Path(samples["yolo_pair"]["image"])
+    yolo_label = Path(samples["yolo_pair"]["label"])
+
+    uploads = {
+        "real_mask_image_upload": _post_file(f"{base_url}/api/datasets/{dataset_name}/upload/images", mask_image, timeout=30),
+        "real_mask_upload": _post_file(f"{base_url}/api/datasets/{dataset_name}/upload/masks", mask_file, timeout=30),
+        "real_yolo_image_upload": _post_file(f"{base_url}/api/datasets/{dataset_name}/upload/images", yolo_image, timeout=30),
+        "real_yolo_label_upload": _post_file(f"{base_url}/api/datasets/{dataset_name}/upload/labels", yolo_label, timeout=30),
+    }
+    for name, detail in uploads.items():
+        checks.append({"name": name, "passed": detail.get("passed", False), "detail": detail})
+
+    validation = _request_json("GET", f"{base_url}/api/datasets/{dataset_name}/validate", timeout=20)
+    validation_json = validation.get("json") if validation.get("passed") else {}
+    checks.append(
+        {
+            "name": "real_dataset_validate_yolo_and_mask",
+            "passed": validation.get("passed", False)
+            and validation_json.get("ready", {}).get("yolo")
+            and validation_json.get("ready", {}).get("mask"),
+            "detail": validation,
+        }
+    )
+
+    yolo_yaml = _request_json("POST", f"{base_url}/api/datasets/{dataset_name}/yolo-yaml", {"class_names": ["object"]}, timeout=20)
+    checks.append({"name": "real_dataset_yolo_yaml", "passed": yolo_yaml.get("passed", False), "detail": yolo_yaml})
+
+    yolo_image_upload = uploads["real_yolo_image_upload"].get("json", {})
+    mask_image_upload = uploads["real_mask_image_upload"].get("json", {})
+    mask_upload = uploads["real_mask_upload"].get("json", {})
+    uploaded_yolo_image = yolo_image_upload.get("saved", [{}])[0].get("relative_path")
+    uploaded_mask_image = mask_image_upload.get("saved", [{}])[0].get("relative_path")
+    uploaded_mask = mask_upload.get("saved", [{}])[0].get("relative_path")
+
+    artifact_label = _request_text(f"{base_url}/api/artifacts/{uploads['real_yolo_label_upload'].get('json', {}).get('saved', [{}])[0].get('relative_path')}", timeout=10)
+    checks.append(
+        {
+            "name": "real_uploaded_label_artifact",
+            "passed": artifact_label.get("passed", False) and bool(artifact_label.get("body", "").strip()),
+            "detail": artifact_label,
+        }
+    )
+
+    yolo_weight = settings.source_root / "Seg_All_In_One_YoloModel" / "yolo11n-seg.pt"
+    predict_name = f"{dataset_name}_predict_real"
+    if uploaded_yolo_image:
+        predict = _create_job_and_wait(
+            base_url,
+            "yolo.predict_custom",
+            {
+                "weights": str(yolo_weight),
+                "source": uploaded_yolo_image,
+                "project": "var/custom_yolo_runs",
+                "name": predict_name,
+                "imgsz": 96,
+                "conf": 0.05,
+                "device": "cpu",
+                "exist_ok": True,
+            },
+            timeout=120,
+        )
+    else:
+        predict = {"passed": False, "error": "skipped because real_yolo_image_upload did not return a saved path"}
+    predict_root = settings.project_root / "var" / "custom_yolo_runs" / predict_name
+    predict_outputs = _result_files(predict_root, {".png", ".jpg", ".jpeg"})
+    checks.append(
+        {
+            "name": "real_workspace_yolo_predict_job_runner",
+            "passed": predict.get("passed", False) and bool(predict_outputs),
+            "detail": {**predict, "output_count": len(predict_outputs), "outputs": [_relative_to_project(path) for path in predict_outputs[:8]]},
+        }
+    )
+
+    heatmap_name = f"{dataset_name}_heatmap_real"
+    if uploaded_yolo_image:
+        heatmap = _create_job_and_wait(
+            base_url,
+            "yolo.heatmap_custom",
+            {
+                "weights": str(yolo_weight),
+                "source": uploaded_yolo_image,
+                "project": "var/custom_yolo_runs",
+                "name": heatmap_name,
+                "model_key": "YOLO11n-seg",
+                "pt_name": "best.pt",
+                "cam_method": "GradCAM",
+                "target_layers": "model.model.model[9]",
+                "limit": 1,
+            },
+            timeout=120,
+        )
+    else:
+        heatmap = {"passed": False, "error": "skipped because real_yolo_image_upload did not return a saved path"}
+    heatmap_root = settings.project_root / "var" / "custom_yolo_runs" / heatmap_name / "HeartMap_Visual"
+    heatmap_outputs = _result_files(heatmap_root, {".jpg", ".jpeg", ".png"})
+    checks.append(
+        {
+            "name": "real_workspace_yolo_heatmap_job_runner",
+            "passed": heatmap.get("passed", False) and len(heatmap_outputs) >= 2,
+            "detail": {**heatmap, "output_count": len(heatmap_outputs), "outputs": [_relative_to_project(path) for path in heatmap_outputs[:8]]},
+        }
+    )
+
+    stack_dir = fixture_root / "real_stack"
+    if uploaded_mask_image and uploaded_mask:
+        stack = _create_job_with_retry(
+            base_url,
+            "dataset.stack_single",
+            {
+                "image_path": str(settings.project_root / uploaded_mask_image),
+                "label_path": str(settings.project_root / uploaded_mask),
+                "result_dir": str(stack_dir),
+                "alpha": 0.35,
+            },
+            attempts=2,
+            timeout=90,
+        )
+    else:
+        stack = {"passed": False, "error": "skipped because real mask upload did not return saved paths"}
+    stack_outputs = _result_files(stack_dir, {".png", ".jpg", ".jpeg"})
+    checks.append(
+        {
+            "name": "real_workspace_stack_job_runner",
+            "passed": stack.get("passed", False) and bool(stack_outputs),
+            "detail": {**stack, "output_count": len(stack_outputs), "outputs": [_relative_to_project(path) for path in stack_outputs[:8]]},
+        }
+    )
+
+    report = {
+        "available": True,
+        "run_id": run_id,
+        "base_url": base_url,
+        "fixture_root": str(fixture_root),
+        "dataset_name": dataset_name,
+        "samples": samples,
+        "passed": all(item["passed"] for item in checks),
+        "checks": checks,
+        "created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+    }
+    (acceptance_root / "real_latest.json").write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
+    return report
+
+
 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"