"""Batch CDSL rebuild baseline and machine-readable reporting.""" from __future__ import annotations import argparse import json import os import signal import subprocess import sys import time from collections import Counter from pathlib import Path from typing import Any, Iterable from cdsl_importer.legacy_profile_adapter import lower_legacy_profiles from .semantic_validation import validate_semantic_cdsl def _write_json(path: Path, value: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8") class BatchBuildError(RuntimeError): def __init__(self, error: dict[str, Any], selector_resolution: list[dict[str, Any]] | None = None) -> None: super().__init__(str(error.get("message") or error.get("code") or "build failed")) self.error = error self.selector_resolution = selector_resolution or [] def _failure_category(report: dict[str, Any]) -> str: """Classify the terminal rebuild state without weakening any gate.""" if report.get("geometry_verified"): return "geometry_verified" if report.get("runtime_eligible") and not report.get("build_attempted"): return "runtime_eligible_not_built" comparison = report.get("numeric_comparison") or {} if comparison.get("classification") == "coordinate_frame_mismatch_candidate": return "coordinate_frame_mismatch_candidate" if report.get("built"): return "geometry_mismatch" blocker = report.get("first_blocker") or {} code = str(blocker.get("code") or "") if code in {"selector_not_found", "selector_ambiguous"}: return "selector_resolution" if code.startswith("unsupported_") or code in {"unknown_atomic", "unsupported_atomic"}: return "unsupported_capability" if code in { "unresolved_input", "missing_parameter", "missing_sketch", "missing_selector", "missing_host_face", "missing_hole_positions", "missing_extent_reference", "missing_reverse_extent_reference", "missing_revolve_axis", "missing_reference_axis_geometry", "missing_reference_orientation", "profile_no_closed_region", "profile_resolution_failed", "semantic_validation_failed", "no_solid_feature", "extent_target_not_reached", "missing_active_body", "invalid_hole_spec", "extent_target_not_in_direction", "non_uniform_extent_target", "invalid_extent_offset", }: return "input_incomplete" if code in {"build_timeout", "build_failed", "execution_failed"}: return "occ_execution_failure" return "preflight_blocked" def _runtime_atomic_ids() -> frozenset[str]: from .runtime import ALL_ATOMIC_IDS return ALL_ATOMIC_IDS def _truth_metrics(cdsl: dict[str, Any]) -> dict[str, Any] | None: truth = (cdsl.get("meta") or {}).get("document_truth") or {} mass = truth.get("mass_properties") or {} geometry = truth.get("geometry") or {} if not mass.get("available"): return None box_m = geometry.get("bounding_box") volume_m3 = mass.get("volume") area_m2 = mass.get("surface_area") solid_count = geometry.get("solid_body_count") if not isinstance(box_m, list) or len(box_m) != 6 or volume_m3 is None or area_m2 is None or solid_count is None: return None return { "bounding_box_mm": [float(value) * 1000.0 for value in box_m], "volume_mm3": float(volume_m3) * 1_000_000_000.0, "surface_area_mm2": float(area_m2) * 1_000_000.0, "solid_count": int(solid_count), } def _bounding_box_spans(box: list[float]) -> list[float]: return [box[index + 3] - box[index] for index in range(3)] def _verification_classification( *, actual_box: list[float], truth: dict[str, Any], box_delta: float, volume_relative_error: float, area_relative_error: float, solid_count_matches: bool, ) -> str: """Classify a strict truth comparison without weakening its threshold. Some exports carry internally consistent feature dimensions in a frame that differs from the source truth frame. Matching volume, area, solid count, and unordered bounding-box spans is useful evidence for diagnosing that condition, but is not enough to call a rebuild geometry-verified. """ metrics_match = volume_relative_error <= 0.001 and area_relative_error <= 0.001 and solid_count_matches spans_match = max( abs(actual - expected) for actual, expected in zip( sorted(_bounding_box_spans(actual_box)), sorted(_bounding_box_spans(truth["bounding_box_mm"])), ) ) <= 0.01 if box_delta <= 0.01 and metrics_match: return "verified" if metrics_match and spans_match: return "coordinate_frame_mismatch_candidate" return "geometry_mismatch" def _compare_truth(result: dict[str, Any], truth: dict[str, Any] | None) -> dict[str, Any] | None: if truth is None: return None actual_box = [*result["bbox_mm"]["min"], *result["bbox_mm"]["max"]] box_delta = max(abs(actual - expected) for actual, expected in zip(actual_box, truth["bounding_box_mm"])) volume = float(result["volume_mm3"]) volume_expected = float(truth["volume_mm3"]) volume_relative_error = abs(volume - volume_expected) / max(abs(volume_expected), 1e-9) from build123d import import_step rebuilt = import_step(str(result["out_step"])) area = float(rebuilt.area) area_expected = float(truth["surface_area_mm2"]) area_relative_error = abs(area - area_expected) / max(abs(area_expected), 1e-9) solid_count = len(rebuilt.solids()) solid_count_matches = solid_count == truth["solid_count"] classification = _verification_classification( actual_box=actual_box, truth=truth, box_delta=box_delta, volume_relative_error=volume_relative_error, area_relative_error=area_relative_error, solid_count_matches=solid_count_matches, ) return { "expected": truth, "actual": { "bounding_box_mm": actual_box, "volume_mm3": volume, "surface_area_mm2": area, "solid_count": solid_count, }, "bounding_box_max_delta_mm": box_delta, "volume_relative_error": volume_relative_error, "surface_area_relative_error": area_relative_error, "solid_count_matches": solid_count_matches, "classification": classification, "passed": classification == "verified", } def _run_isolated_build(cdsl_path: Path, out_step: Path, *, timeout_s: float) -> dict[str, Any]: """Keep an OCC boolean timeout local to one batch part.""" source = ( "import json\n" "from pathlib import Path\n" "from cdsl_engine.runtime import rebuild_cdsl, RuntimeExecutionError\n" "try:\n" f" result=rebuild_cdsl(json.loads(Path({str(cdsl_path)!r}).read_text(encoding='utf-8')), Path({str(out_step)!r}))\n" " result['engine']='cdsl_only'\n" " print(json.dumps(result))\n" "except RuntimeExecutionError as error:\n" " print(json.dumps({'_build_error': error.diagnostic.as_dict(), 'selector_resolution': error.selector_resolutions}))" ) process = subprocess.Popen( [sys.executable, "-c", source], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, start_new_session=True, ) try: stdout, stderr = process.communicate(timeout=timeout_s) except subprocess.TimeoutExpired as error: # Boolean operations can leave helper processes behind on some OCC # versions. Killing the dedicated process group prevents one part # from consuming a later shard's timeout budget. os.killpg(process.pid, signal.SIGKILL) process.communicate() raise RuntimeError(f"build_timeout: exceeded {timeout_s:g}s") from error if process.returncode: raise RuntimeError((stderr or stdout or "build subprocess failed").strip()) try: output = json.loads(stdout) except json.JSONDecodeError as error: raise RuntimeError(f"build subprocess returned invalid JSON: {stdout[-500:]}") from error if output.get("_build_error"): raise BatchBuildError(output["_build_error"], output.get("selector_resolution")) return output def analyze_document( path: Path, *, atomic_ids: frozenset[str] | None = None, out_step: Path | None = None, build_timeout_s: float | None = None, ) -> dict[str, Any]: started = time.monotonic() report: dict[str, Any] = { "part_id": path.name.removesuffix(".cdsl.json"), "cdsl_path": str(path), "semantic_valid": False, "runtime_eligible": False, "compiled": False, "built": False, "geometry_verified": False, "topology_observed": False, "feature_results": [], "unsupported_atomic_ids": [], "unsupported_profile_types": [], "unresolved_input": [], "selector_resolution": [], "numeric_comparison": None, "build_attempted": out_step is not None, "failure_category": "preflight_blocked", } try: cdsl = lower_legacy_profiles(json.loads(path.read_text(encoding="utf-8"))) report["part_id"] = str(cdsl.get("part_id") or report["part_id"]) semantic = validate_semantic_cdsl(cdsl) report["semantic_valid"] = True report["unresolved_input"] = semantic["unresolved"] from .runtime import analyze_cdsl # Profile resolution is runtime preflight, not a build-time surprise. # The optional atomic override keeps this function useful for focused # capability tests without changing the production runtime contract. if atomic_ids is None: analysis = analyze_cdsl(cdsl) else: from .capabilities import CapabilityAnalyzer, sketch_ids_required_by_contract from .sketch_solver import SHAPE_GENERATORS, resolve_required_sketches sketch_errors: dict[str, str] = {} resolved = resolve_required_sketches( cdsl, sketch_ids_required_by_contract(cdsl), errors=sketch_errors, ) analysis = CapabilityAnalyzer(atomic_ids=atomic_ids, profile_types=SHAPE_GENERATORS).analyze( resolved, sketch_errors=sketch_errors, ) report["runtime_eligible"] = analysis.runtime_eligible report["feature_results"] = [item.as_dict() for item in analysis.feature_results] blockers = [blocker for item in analysis.feature_results for blocker in item.blockers] blockers.extend(analysis.document_blockers) report["unsupported_atomic_ids"] = sorted({item.atomic_id for item in analysis.feature_results if item.resolved_status == "unsupported"}) report["unsupported_profile_types"] = sorted({blocker.detail.get("profile_type") for blocker in blockers if blocker.code == "unsupported_profile"}) if blockers: report["first_blocker"] = blockers[0].as_dict() if analysis.runtime_eligible: report["compiled"] = True if out_step is not None: from .rebuild import run_cdsl_only try: build_result = ( _run_isolated_build(path, out_step, timeout_s=build_timeout_s) if build_timeout_s is not None else run_cdsl_only(cdsl, out_step) ) report["built"] = True report["build_result"] = build_result report["selector_resolution"] = build_result.get("selector_resolution") or [] comparison = _compare_truth(build_result, _truth_metrics(cdsl)) report["numeric_comparison"] = comparison report["geometry_verified"] = bool(comparison and comparison["passed"]) from build123d import import_step rebuilt = import_step(str(out_step)) report["topology_observed"] = True report["topology"] = { "face_count": len(rebuilt.faces()), "edge_count": len(rebuilt.edges()), "vertex_count": len(rebuilt.vertices()), } except BatchBuildError as error: report["first_blocker"] = error.error report["selector_resolution"] = error.selector_resolution except Exception as error: code = "build_timeout" if str(error).startswith("build_timeout:") else "build_failed" report["first_blocker"] = {"code": code, "message": str(error)} except Exception as error: code = "profile_resolution_failed" if "analytic_contours" in str(error) or "profile" in str(error) else "semantic_validation_failed" report["first_blocker"] = {"code": code, "message": str(error)} report["failure_category"] = _failure_category(report) report["timings"] = {"analysis_s": round(time.monotonic() - started, 6)} return report def _write_summaries(out_dir: Path, reports: list[dict[str, Any]], *, input_directory: Path, total_count: int) -> dict[str, Any]: atomic_counter = Counter() blocker_counter = Counter() for report in reports: for feature in report.get("feature_results") or (): atomic_counter[feature["atomic_id"]] += 1 for blocker in feature["blockers"]: blocker_counter[blocker["code"]] += 1 first = report.get("first_blocker") or {} if not (report.get("feature_results") or ()) and first.get("code"): blocker_counter[first["code"]] += 1 manifest = { "schema": "cdsl.engine.batch-rebuild.v1", "input_directory": str(input_directory), "part_count": total_count, "completed_count": len(reports), "complete": len(reports) == total_count, "semantic_valid_count": sum(bool(report.get("semantic_valid")) for report in reports), "runtime_eligible_count": sum(bool(report.get("runtime_eligible")) for report in reports), "compiled_count": sum(bool(report.get("compiled")) for report in reports), "built_count": sum(bool(report.get("built")) for report in reports), "geometry_verified_count": sum(bool(report.get("geometry_verified")) for report in reports), "failure_category_counts": dict(sorted(Counter(report.get("failure_category") or "unknown" for report in reports).items())), "results": [{"part_id": report["part_id"], "report": f"parts/{report['part_id']}.report.json"} for report in reports], } _write_json(out_dir / "manifest.json", manifest) _write_json(out_dir / "summary-by-atomic.json", dict(sorted(atomic_counter.items()))) _write_json(out_dir / "summary-by-blocker.json", dict(sorted(blocker_counter.items()))) return manifest def batch_analyze( cdsl_dir: Path, out_dir: Path, *, atomic_ids: frozenset[str] | None = None, build: bool = False, overwrite: bool = False, max_parts: int | None = None, build_timeout_s: float | None = None, part_ids: Iterable[str] | None = None, ) -> dict[str, Any]: files = sorted(cdsl_dir.glob("*.cdsl.json")) if not files: raise ValueError(f"No *.cdsl.json files found in {cdsl_dir}") if part_ids is not None: # ``part_ids`` are exact CDSL filename stems. This is intentionally # independent from document ``part_id``: imported provenance may use # different case or normalisation than its collision-free artifact # filename. Phase pools return these same stems. requested = {str(part_id).strip() for part_id in part_ids if str(part_id).strip()} available = {path.name.removesuffix(".cdsl.json") for path in files} unknown = sorted(requested - available) if unknown: raise ValueError(f"Requested part ids do not exist in {cdsl_dir}: {', '.join(unknown)}") files = [path for path in files if path.name.removesuffix(".cdsl.json") in requested] if not files: raise ValueError("Part-id filter selected no CDSL documents") reports_by_part: dict[str, dict[str, Any]] = {} pending: list[Path] = [] for path in files: part_id = path.name.removesuffix(".cdsl.json") report_path = out_dir / "parts" / f"{part_id}.report.json" if report_path.exists() and not overwrite: existing = json.loads(report_path.read_text(encoding="utf-8")) if not build or existing.get("build_attempted"): # Reports are resumable artifacts. Derived classifications may # be added after an earlier checkpoint, so migrate them in # place without needlessly rebuilding the STEP artifact. category = _failure_category(existing) if existing.get("failure_category") != category: existing["failure_category"] = category _write_json(report_path, existing) reports_by_part[part_id] = existing continue pending.append(path) if max_parts is not None: pending = pending[:max(0, max_parts)] newly_analyzed = 0 for path in pending: part_id = path.name.removesuffix(".cdsl.json") report_path = out_dir / "parts" / f"{part_id}.report.json" report = analyze_document( path, atomic_ids=atomic_ids, out_step=(out_dir / "parts" / f"{part_id}.step") if build else None, build_timeout_s=build_timeout_s, ) _write_json(report_path, report) reports_by_part[part_id] = report newly_analyzed += 1 # Keep every successfully analyzed part recoverable during a long # build. A later invocation resumes from per-part report files; the # aggregate checkpoint is intentionally periodic to avoid O(n^2) # JSON writes during a thousand-part run. if newly_analyzed % 25 == 0: _write_summaries(out_dir, list(reports_by_part.values()), input_directory=cdsl_dir, total_count=len(files)) return _write_summaries(out_dir, list(reports_by_part.values()), input_directory=cdsl_dir, total_count=len(files)) def main() -> None: parser = argparse.ArgumentParser(description="Analyze semantic CDSL rebuild eligibility in batch") parser.add_argument("cdsl_dir", type=Path) parser.add_argument("--out", type=Path, required=True) parser.add_argument("--build", action="store_true", help="Rebuild every strict-runtime-eligible part via CDSL-only runtime") parser.add_argument("--overwrite", action="store_true", help="Ignore existing per-part reports and rerun the batch") parser.add_argument("--max-parts", type=int, default=None, help="Process at most this many pending parts; useful for resumable CI shards") parser.add_argument("--part-ids", default=None, help="Comma-separated exact part ids to include; omitted means every CDSL document") parser.add_argument("--phase", choices=("p3", "p4", "p6"), default=None, help="Run a documented static phase pool instead of every CDSL document") parser.add_argument("--build-timeout", type=float, default=10.0, help="Per-part build timeout in seconds when --build is used") args = parser.parse_args() if args.phase and args.part_ids: parser.error("--phase and --part-ids are mutually exclusive") selected_part_ids = args.part_ids.split(",") if args.part_ids else None if args.phase: from .phase_pools import select_static_phase_pool selected_part_ids = select_static_phase_pool(args.cdsl_dir, args.phase) manifest = batch_analyze( args.cdsl_dir, args.out, build=args.build, overwrite=args.overwrite, max_parts=args.max_parts, build_timeout_s=args.build_timeout if args.build else None, part_ids=selected_part_ids, ) print(json.dumps({key: value for key, value in manifest.items() if key != "results"}, ensure_ascii=False)) if __name__ == "__main__": main()