"""Strict, read-only validation for console-owned point-cloud artifacts."""
|
|
from __future__ import annotations
|
|
import json
|
from datetime import UTC, datetime
|
from pathlib import Path
|
from typing import Any
|
|
|
def _json(path: Path) -> tuple[dict[str, Any] | None, str | None]:
|
try:
|
value = json.loads(path.read_text(encoding="utf-8"))
|
except FileNotFoundError:
|
return None, "metadata file is missing"
|
except OSError as exc:
|
return None, f"metadata cannot be read: {exc}"
|
except json.JSONDecodeError as exc:
|
return None, f"metadata JSON is invalid at line {exc.lineno}"
|
return (value, None) if isinstance(value, dict) else (None, "metadata JSON must be an object")
|
|
|
def _record(kind: str, artifact: Path, required: list[Path], errors: list[str]) -> dict[str, Any]:
|
missing = [path.name for path in required if not path.is_file()]
|
return {
|
"kind": kind,
|
"id": artifact.name,
|
"artifactRoot": artifact.as_posix(),
|
"status": "healthy" if not missing and not errors else "invalid",
|
"missing": missing,
|
"errors": errors,
|
}
|
|
|
def pointcloud_artifact_health(root: Path, limit: int = 80) -> dict[str, Any]:
|
"""Validate the artifact contract without attempting to repair or mutate it."""
|
output = root / "shared" / "outputs" / "05-3d-pointcloud"
|
records: list[dict[str, Any]] = []
|
contracts = (
|
("training_model", output / "training-runs", ("metrics.json", "model.pt", "predicted-semantic-preview.ply")),
|
("model_inference", output / "model-inference-runs", ("run_metadata.json", "predicted-semantic-preview.ply", "predicted-semantic-classified.las", "class-counts.csv", "prediction-summary.json")),
|
("automatic_annotation", output / "auto-annotation-runs", ("run_metadata.json", "predicted-semantic-preview.ply", "predicted-semantic-classified.las", "class-counts.csv", "prediction-summary.json", "automatic-annotation-candidates.json")),
|
("annotation_revision", output / "annotations", ("annotation.json",)),
|
)
|
for kind, directory, names in contracts:
|
if not directory.is_dir():
|
continue
|
for artifact in directory.iterdir():
|
if not artifact.is_dir():
|
continue
|
required = [artifact / name for name in names]
|
errors: list[str] = []
|
metadata_name = "annotation.json" if kind == "annotation_revision" else "metrics.json" if kind == "training_model" else "run_metadata.json"
|
metadata, issue = _json(artifact / metadata_name)
|
if issue:
|
errors.append(issue)
|
elif kind == "annotation_revision" and metadata.get("schema_version") != 1:
|
errors.append("annotation schema_version must be 1")
|
elif kind != "annotation_revision" and metadata.get("capability") != "05-3d-pointcloud":
|
errors.append("metadata capability must be 05-3d-pointcloud")
|
record = _record(kind, artifact, required, errors)
|
record["artifactRoot"] = artifact.relative_to(root).as_posix()
|
records.append(record)
|
records.sort(key=lambda item: (item["id"], item["kind"]), reverse=True)
|
records = records[:limit]
|
return {
|
"generatedAt": datetime.now(UTC).isoformat(),
|
"healthy": sum(item["status"] == "healthy" for item in records),
|
"invalid": sum(item["status"] != "healthy" for item in records),
|
"records": records,
|
}
|
