cdsl-cad/cadfs_to_cdsl/reports.py

from __future__ import annotations

import csv, json, re
from collections import Counter, defaultdict
from datetime import datetime
from pathlib import Path
from typing import Any


def write_json(path: Path, value: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(json.dumps(value, ensure_ascii=True, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    temporary.replace(path)


def read_json(path: Path) -> Any: return json.loads(path.read_text(encoding="utf-8"))


def write_manifest(path: Path, records: list[dict[str, Any]]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("".join(json.dumps(item, ensure_ascii=True, sort_keys=True) + "\n" for item in records), encoding="utf-8")


def generate_reports(output: Path, records: list[dict[str, Any]]) -> dict[str, Any]:
    statuses = Counter(str(item.get("status") or "unknown") for item in records)
    operations: Counter[str] = Counter(); reasons: Counter[str] = Counter(); gaps: dict[str, list[str]] = defaultdict(list)
    rows = []
    for item in records:
        sample_dir = output / "samples" / str(item["sample_id"])
        diagnostics_path = sample_dir / "diagnostics.json"
        if diagnostics_path.exists():
            for diagnostic in read_json(diagnostics_path):
                reasons[str(diagnostic.get("code") or "unknown")] += 1
                gap = diagnostic.get("capability") or diagnostic.get("operation")
                if gap: gaps[str(gap)].append(str(item["sample_id"]))
        history_path = sample_dir / "history.json"
        if history_path.exists():
            for step in read_json(history_path): operations[str(step.get("operation") or "unknown")] += 1
        comparison_path = sample_dir / "comparison.json"
        if comparison_path.exists():
            comparison = read_json(comparison_path); metrics = comparison["raw"]["metrics"]
            rows.append({"sample_id": item["sample_id"], "decision": comparison["decision"], "bbox_max_delta_mm": metrics["bbox_max_delta_mm"], "volume_relative_error": metrics["volume_relative_error"], "surface_area_relative_error": metrics["surface_area_relative_error"]})
    summary = {"schema": "cadfs_to_cdsl.summary.v1", "total_models": len(records), "statuses": dict(statuses), "operation_counts": dict(operations), "failure_reasons": dict(reasons), "capability_gap_counts": {key: len(set(value)) for key, value in gaps.items()}}
    write_json(output / "summary.json", summary)
    gap_payload = {key: {"sample_count": len(set(ids)), "sample_ids": sorted(set(ids))} for key, ids in sorted(gaps.items())}; write_json(output / "capability_gaps.json", gap_payload)
    lines = ["# Unsupported CADFS capabilities", ""]
    for name, value in gap_payload.items(): lines.extend([f"## {name}", "", f"Affected models: {value['sample_count']}", "", "Sample IDs: " + ", ".join(value["sample_ids"]), ""])
    (output / "unsupported_capabilities.md").write_text("\n".join(lines), encoding="utf-8")
    with (output / "comparison_summary.csv").open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=["sample_id", "decision", "bbox_max_delta_mm", "volume_relative_error", "surface_area_relative_error"]); writer.writeheader(); writer.writerows(rows)
    return summary


def _pct(count: int, total: int) -> str:
    return "0.00%" if total <= 0 else f"{count / total * 100:.2f}%"


def _table(headers: list[str], rows: list[list[Any]]) -> list[str]:
    lines = ["| " + " | ".join(headers) + " |", "| " + " | ".join(["---"] * len(headers)) + " |"]
    for row in rows:
        lines.append("| " + " | ".join(str(value).replace("\n", " ") for value in row) + " |")
    return lines


def _read_optional_json(path: Path) -> Any | None:
    return read_json(path) if path.exists() else None


def _normalize_error(message: str) -> str:
    message = re.sub(r"/Users/[^ ]+", "<path>", message)
    message = re.sub(r"0x[0-9a-fA-F]+", "0x...", message)
    message = re.sub(r"\d+\.\d{4,}", "<float>", message)
    return message[:180] if len(message) > 180 else message


def generate_markdown_report(
    output: Path,
    records: list[dict[str, Any]],
    *,
    input_root: Path | None = None,
    command: str | None = None,
    report_name: str = "full_run_report.md",
) -> Path:
    summary = generate_reports(output, records)
    total = len(records)
    sample_root = output / "samples"
    status_records = []
    modality_missing: Counter[str] = Counter()
    alignment_fallbacks = 0
    dataset_index = _read_optional_json(output / "dataset_index.json") or {}
    indexed_records = {str(item.get("sample_id")): item for item in dataset_index.get("records") or []}
    for record in records:
        indexed = indexed_records.get(str(record["sample_id"]), {})
        for diagnostic in indexed.get("diagnostics") or record.get("diagnostics") or []:
            if diagnostic.get("code") == "missing_modality":
                modality_missing[str(diagnostic.get("modality") or "unknown")] += 1
            if diagnostic.get("code") == "alignment_fallback":
                alignment_fallbacks += 1
        status = _read_optional_json(sample_root / str(record["sample_id"]) / "status.json") or record
        status_records.append(status)

    statuses = Counter(str(item.get("status") or "unknown") for item in status_records)
    conversion_statuses = Counter(str(item.get("conversion_status") or "missing") for item in status_records)
    file_counts = {
        "candidate.cdsl.json": sum(1 for _ in sample_root.glob("*/candidate.cdsl.json")),
        "bound.cdsl.json": sum(1 for _ in sample_root.glob("*/bound.cdsl.json")),
        "rebuild.step": sum(1 for _ in sample_root.glob("*/rebuild.step")),
        "comparison.json": sum(1 for _ in sample_root.glob("*/comparison.json")),
        "status.json": sum(1 for _ in sample_root.glob("*/status.json")),
    }

    comparison_decisions: Counter[str] = Counter()
    strict_pass = 0
    rp_pass = 0
    comparison_failures: Counter[str] = Counter()
    comparison_errors: Counter[str] = Counter()
    comparison_error_examples: dict[str, list[str]] = defaultdict(list)
    metric_rows = []
    for status in status_records:
        error = status.get("comparison_error") or {}
        if error:
            key = f"{error.get('type') or 'Error'}: {_normalize_error(str(error.get('message') or ''))}"
            comparison_errors[key] += 1
            if len(comparison_error_examples[key]) < 5:
                comparison_error_examples[key].append(str(status.get("sample_id") or "unknown"))
    for path in sample_root.glob("*/comparison.json"):
        comparison = _read_optional_json(path)
        if not comparison:
            continue
        sample_id = path.parent.name
        comparison_decisions[str(comparison.get("decision") or "unknown")] += 1
        strict_pass += 1 if ((comparison.get("strict") or {}).get("passed")) else 0
        rp_pass += 1 if ((comparison.get("rp") or {}).get("passed")) else 0
        for reason in ((comparison.get("raw") or {}).get("failure_reasons") or []):
            comparison_failures[str(reason)] += 1
        metrics = ((comparison.get("raw") or {}).get("metrics") or {})
        metric_rows.append((
            sample_id,
            comparison.get("decision"),
            metrics.get("bbox_max_delta_mm"),
            metrics.get("volume_relative_error"),
            metrics.get("surface_area_relative_error"),
        ))

    rebuild_statuses: Counter[str] = Counter()
    rebuild_errors: Counter[str] = Counter()
    rebuild_error_examples: dict[str, list[str]] = defaultdict(list)
    for path in sample_root.glob("*/rebuild.json"):
        rebuild = _read_optional_json(path)
        if not rebuild:
            continue
        rebuild_statuses[str(rebuild.get("status") or "unknown")] += 1
        error = rebuild.get("error") or {}
        if error:
            key = f"{error.get('type') or 'Error'}: {_normalize_error(str(error.get('message') or ''))}"
            rebuild_errors[key] += 1
            if len(rebuild_error_examples[key]) < 5:
                rebuild_error_examples[key].append(path.parent.name)

    diagnostic_counts: Counter[str] = Counter()
    diagnostic_examples: dict[str, list[str]] = defaultdict(list)
    capability_examples: dict[str, list[str]] = defaultdict(list)
    capability_source: dict[str, str] = {}
    for record in records:
        sample_id = str(record["sample_id"])
        diagnostics = _read_optional_json(sample_root / sample_id / "diagnostics.json") or []
        for diagnostic in diagnostics:
            code = str(diagnostic.get("code") or "unknown")
            diagnostic_counts[code] += 1
            if len(diagnostic_examples[code]) < 8:
                diagnostic_examples[code].append(sample_id)
            capability = diagnostic.get("capability") or diagnostic.get("operation")
            if capability:
                capability = str(capability)
                if len(capability_examples[capability]) < 10:
                    capability_examples[capability].append(sample_id)
                capability_source.setdefault(capability, str(diagnostic.get("operation") or capability))

    gap_payload = _read_optional_json(output / "capability_gaps.json") or {}
    top_gaps = sorted(
        ((name, int(value.get("sample_count") or 0), ", ".join((value.get("sample_ids") or [])[:8])) for name, value in gap_payload.items()),
        key=lambda item: (-item[1], item[0]),
    )

    unsupported_ops = {
        "shell", "sweep", "draft", "thicken", "split", "booleanBodies", "circularPattern",
        "moveFace", "replaceFace", "deleteFace", "import", "derive",
    }
    exact_mappings = {
        "extrude": "extrude_add_blind / extrude_add_two_sided / extrude_cut_blind",
        "loft": "loft_add (simple closed sketch profiles only)",
        "revolve": "revolve_add / revolve_cut",
        "fillet": "fillet",
        "chamfer": "chamfer",
        "hole": "hole_wizard",
        "mirror": "pattern_mirror",
        "cPlane": "reference_plane (OFFSET only)",
    }
    try:
        from engine.cdsl_engine.runtime import EXECUTORS
        engine_atomic_ids = sorted(EXECUTORS)
    except Exception:
        engine_atomic_ids = []

    operation_rows = sorted(
        ((name, count) for name, count in (summary.get("operation_counts") or {}).items()),
        key=lambda item: (-item[1], item[0]),
    )
    status_rows = [[name, count, _pct(count, total)] for name, count in sorted(statuses.items(), key=lambda item: (-item[1], item[0]))]
    conversion_rows = [[name, count, _pct(count, total)] for name, count in sorted(conversion_statuses.items(), key=lambda item: (-item[1], item[0]))]

    lines: list[str] = [
        "# CADFS full conversion report",
        "",
        f"- Generated at: {datetime.now().isoformat(timespec='seconds')}",
        f"- Input: `{input_root}`" if input_root else "- Input: not recorded",
        f"- Output: `{output}`",
        f"- Command: `{command}`" if command else "- Command: not recorded",
        f"- Total samples: {total}",
        f"- Manifest rows: {sum(1 for _ in (output / 'manifest.jsonl').open(encoding='utf-8')) if (output / 'manifest.jsonl').exists() else 'missing'}",
        "",
        "## Acceptance summary",
        "",
        f"- RP accepted samples: {rp_pass} ({_pct(rp_pass, total)})",
        f"- Strict accepted samples: {strict_pass} ({_pct(strict_pass, total)})",
        f"- Rebuilt STEP files present: {file_counts['rebuild.step']}",
        f"- Comparison reports present: {file_counts['comparison.json']}",
        f"- Candidate CDSL files present: {file_counts['candidate.cdsl.json']}",
        f"- Bound CDSL files present: {file_counts['bound.cdsl.json']}",
        "",
        "## Final statuses",
        "",
        *_table(["Status", "Count", "Share"], status_rows),
        "",
        "## Conversion statuses",
        "",
        *_table(["Conversion status", "Count", "Share"], conversion_rows),
        "",
        "## Modality and alignment",
        "",
    ]
    if modality_missing:
        lines.extend(_table(["Missing modality", "Count"], sorted(modality_missing.items())))
    else:
        lines.append("- No missing local modalities were recorded in the manifest.")
    lines.extend(["", f"- JSONL content alignment fallbacks: {alignment_fallbacks}", ""])

    lines.extend([
        "## Comparison results",
        "",
        *_table(["Decision", "Count"], sorted(comparison_decisions.items(), key=lambda item: (-item[1], item[0]))),
        "",
        "Top strict/RP comparison failure checks:",
        "",
    ])
    if comparison_failures:
        lines.extend(_table(["Failure check", "Count"], sorted(comparison_failures.items(), key=lambda item: (-item[1], item[0]))[:12]))
    else:
        lines.append("- No comparison failure checks were recorded.")
    lines.extend(["", "Comparison worker errors:", ""])
    if comparison_errors:
        rows = [[name, count, ", ".join(comparison_error_examples[name])] for name, count in sorted(comparison_errors.items(), key=lambda item: (-item[1], item[0]))[:12]]
        lines.extend(_table(["Error", "Count", "Examples"], rows))
    else:
        lines.append("- No comparison worker errors were recorded.")

    lines.extend([
        "",
        "## Rebuild outcomes",
        "",
        *_table(["Rebuild status", "Count"], sorted(rebuild_statuses.items(), key=lambda item: (-item[1], item[0]))),
        "",
        "Top rebuild/runtime errors:",
        "",
    ])
    if rebuild_errors:
        rows = [[name, count, ", ".join(rebuild_error_examples[name])] for name, count in sorted(rebuild_errors.items(), key=lambda item: (-item[1], item[0]))[:12]]
        lines.extend(_table(["Error", "Count", "Examples"], rows))
    else:
        lines.append("- No rebuild errors were recorded.")

    lines.extend([
        "",
        "## Diagnostics",
        "",
        *_table(
            ["Diagnostic code", "Count", "Example samples"],
            [[name, count, ", ".join(diagnostic_examples[name])] for name, count in sorted(diagnostic_counts.items(), key=lambda item: (-item[1], item[0]))],
        ),
        "",
        "## Capability gaps",
        "",
        *_table(["Capability", "Affected samples", "Example samples"], top_gaps[:25]),
        "",
        "## FeatureScript operation counts",
        "",
        *_table(["Operation", "Occurrences"], operation_rows),
        "",
        "## Exact mapping policy",
        "",
        "- The converter keeps FeatureScript operation identity in `history.json` and diagnostics.",
        "- Unsupported operations are not rewritten as substitute atomics.",
        "- Parameters are statically evaluated from FeatureScript only; STEP geometry is not used to infer or tune CDSL parameters.",
        "",
        *_table(["FeatureScript operation", "CDSL atomic policy"], sorted(exact_mappings.items())),
        "",
        "Known unsupported FeatureScript operations recorded as capability gaps:",
        "",
        ", ".join(sorted(unsupported_ops)),
        "",
        "Engine executor atomic IDs:",
        "",
        ", ".join(engine_atomic_ids) if engine_atomic_ids else "Unable to import engine executor registry while generating this report.",
        "",
        "## Regression check",
        "",
    ])
    regression = _read_optional_json(sample_root / "00000173" / "comparison.json")
    if regression:
        metrics = ((regression.get("raw") or {}).get("metrics") or {})
        lines.extend([
            "- Sample `00000173` decision: `" + str(regression.get("decision")) + "`",
            "- RP passed: `" + str((regression.get("rp") or {}).get("passed")) + "`, strict passed: `" + str((regression.get("strict") or {}).get("passed")) + "`",
            f"- BBox max delta: `{metrics.get('bbox_max_delta_mm')}` mm",
            f"- Volume relative error: `{metrics.get('volume_relative_error')}`",
            f"- Surface area relative error: `{metrics.get('surface_area_relative_error')}`",
            "- This is consistent with the known CADFS RP radius quantization case: FeatureScript uses 9.53 mm while the source STEP is about 9.525 mm.",
        ])
    else:
        lines.append("- Sample `00000173` has no comparison report.")

    lines.extend([
        "",
        "## Output locations",
        "",
        f"- Per-sample artifacts: `{sample_root}/<sample_id>/`",
        f"- Manifest: `{output / 'manifest.jsonl'}`",
        f"- Summary JSON: `{output / 'summary.json'}`",
        f"- Capability gaps JSON: `{output / 'capability_gaps.json'}`",
        f"- Unsupported capabilities Markdown: `{output / 'unsupported_capabilities.md'}`",
        f"- Comparison CSV: `{output / 'comparison_summary.csv'}`",
        "",
        "## Notes",
        "",
        "- The source CADFS directory was treated as read-only by the pipeline.",
        "- `workers=1` was used for OCC stability and reproducibility.",
        "- Accepted samples require a generated CDSL candidate, rebuilt STEP, and comparison report.",
        "- Deferred or rejected samples retain evidence in `diagnostics.json`, `history.json`, `rebuild.json`, or `comparison.json`.",
        "",
    ])
    if metric_rows:
        worst_bbox = sorted(metric_rows, key=lambda item: (item[2] is None, item[2] or 0), reverse=True)[:5]
        worst_volume = sorted(metric_rows, key=lambda item: (item[3] is None, item[3] or 0), reverse=True)[:5]
        lines.extend(["## Largest observed comparison deltas", "", "BBox delta:", ""])
        lines.extend(_table(["Sample", "Decision", "BBox max delta mm", "Volume rel err", "Area rel err"], worst_bbox))
        lines.extend(["", "Volume relative error:", ""])
        lines.extend(_table(["Sample", "Decision", "BBox max delta mm", "Volume rel err", "Area rel err"], worst_volume))
        lines.append("")

    report_path = output / report_name
    report_path.write_text("\n".join(lines), encoding="utf-8")
    return report_path