run_pipeline.py

"""run_pipeline.py: unified pipeline runner for matchline.

Chains all pipeline stages:
  Stage 1: generate_building()           -> stage_01_building.json
  Stage 2: build_model()                 -> stage_02_model.json
  Stage 3: simplify_ring()               -> stage_03_simplified.json
  Stage 4: run_checks()                  -> stage_04_validation.json
  Stage 4b: _run_auto_triage()           -> stage_04b_auto_triage.json  [default]
  Stage 5: (fail-fast on validation)
  Stage 6: BEM export (gbXML + IFC4)     -> stage_06_bem/

Each stage writes intermediate JSON so operators can inspect any step.
Validation errors block export (exit code 1, not silent).

Modes (#998):
  full (default): stages 1-6, BEM construction and export included.
  progressive: stops after the takeoff (stage 1 symbols, areas, counts); no BEM
  construction, no review-queue gate, no export.
"""

from __future__ import annotations

import json
import sys
import time
from contextlib import contextmanager
from dataclasses import asdict
from pathlib import Path
from typing import Any

import numpy as np

from bem.bem_export import (
    BEMModel,
    BEMOpeningUnit,
    BEMSpace,
    _validate_out_path,
    validate_gbxml,
    validate_ifc4,
    write_gbxml,
    write_ifc4,
)
from bem.bem_helpers import _edge_facades
from construction_library import apply_construction_library
from datasets_adapter import (
    detections_from_yolo_json,
    extract_with_tags,
    load_aec_bench,
    parse_schedule_csv,
    rollup_takeoff,
)
from geometry.geometry_simplify import (  # noqa: F401  (footprint_from_regions reused below)
    footprint_from_regions,
    footprint_parts_from_regions,
    simplify_footprint,
    simplify_ring,
)
from link import build_model
from pipeline_exceptions import StageError
from review.convention_report import build_convention_report
from room_labels import attach_room_labels
from synth.multidiscipline import generate_building
from telemetry_emitter import TelemetryEmitter
from validate import export_gate, run_checks, validate_bem_conservation

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def write_json(p: Path, data: Any) -> None:
    p.parent.mkdir(parents=True, exist_ok=True)
    from atomic_io import write_text_atomic

    write_text_atomic(p, json.dumps(data, indent=2, default=str))


def _ensure_ccw(ring: list) -> list:
    """Ensure ring is counter-clockwise (for shoelace area > 0)."""
    s = 0.0
    n = len(ring)
    for i in range(n):
        x0, y0 = ring[i]
        x1, y1 = ring[(i + 1) % n]
        s += x0 * y1 - x1 * y0
    return list(reversed(ring)) if s < 0 else list(ring)


def _run_auto_triage(model) -> None:
    """Run auto-triage on every item currently in the review queue.

    This is called after validation checks so that triage can benefit from
    the confidence scores already computed.  Items are triaged in-place
    (mutating ``model.review_queue``) so that the enriched queue is
    reflected in the exported model JSON.

    Silently skips if the triage classifier is unavailable (all fields
    keep their safe defaults: ``needs_human=1.0``, ``urgency=1``).
    """
    for item in model.review_queue:
        model._triage_item(item)


# ---------------------------------------------------------------------------
# Adapter: BuildingModel -> BEMModel
# ---------------------------------------------------------------------------


def _north_up(pts) -> list:
    """Canonical plan frame (x right, y down) -> BEM frame (x east, y north)."""
    return [(float(p[0]), -float(p[1])) for p in pts]


def model_from_linked_model(
    model, simplified_ring: list, wall_height_m: float, simplify_tolerance: float
) -> BEMModel:
    """Build a BEMModel from the canonical BuildingModel.

    This adapter extracts the data needed for BEM export from the
    linked BuildingModel, converting to the BEMModel contract expected
    by write_gbxml and write_ifc4.
    """
    # --- spaces -----------------------------------------------------------
    from ifc.ifc_export import _use_fields
    from space_use_defaults import lighting_watts

    bem_spaces = []
    for i, sp in enumerate(model.spaces.values()):
        # canonical plan frame is y-down; gbXML/IFC want x=east, y=north
        poly = _ensure_ccw(_north_up(sp.polygon_m))
        # Compute area from polygon
        area = 0.0
        n = len(poly)
        for j in range(n):
            x0, y0 = poly[j]
            x1, y1 = poly[(j + 1) % n]
            area += x0 * y1 - x1 * y0
        area = abs(area) / 2.0
        name = f"{sp.name} {sp.number}".strip() or f"SPACE-{i + 1:02d}"
        bem_spaces.append(
            BEMSpace(
                sid=sp.id,
                name=name,
                number=sp.number,
                polygon_m=poly,
                area_m2=area,
                volume_m3=area * wall_height_m,
                lighting_w=lighting_watts(sp),  # fixtures, else LPD x area (#746)
                **_use_fields(sp),  # loads + schedules, as ifc_export does (#691)
                wall_u_value_w_m2k=getattr(sp, "wall_u_value_w_m2k", None),
                provenance=sp.core_provenance,
                history=list(sp.history),
            )
        )

    if not bem_spaces:
        raise ValueError("no spaces to export")

    # --- openings --------------------------------------------------------
    bem_openings = []
    skipped_openings = []
    for sp in model.spaces.values():
        for op in sp.openings:
            if op.width_m is None or op.height_m is None:
                skipped_openings.append(
                    {
                        "id": op.id,
                        "tag": op.tag,
                        "category": op.category,
                        "reason": "dimensions missing",
                    }
                )
                continue
            bem_openings.append(
                BEMOpeningUnit(
                    category=op.category,
                    tag=op.tag or "",
                    width_m=op.width_m,
                    height_m=op.height_m,
                    provenance=op.provenance,
                    history=list(op.history),
                    host_facade=op.host_facade or "",
                    space_sid=sp.id,
                    construction_id=op.construction_id or "",
                )
            )

    notes = []
    if skipped_openings:
        notes.append(
            f"{len(skipped_openings)} opening(s) dropped due to missing dimensions "
            f"(see skipped_openings)."
        )

    # --- envelope ring(s) -------------------------------------------------
    if hasattr(simplified_ring, "parts"):
        # #984: per-part result (#970): every wing and every courtyard hole
        # flows through to export. Wings wind CCW in the BEM frame; holes
        # wind CW so the writers' outward-normal convention points into the
        # courtyard void (outdoor air). ``ring_m`` stays the largest part's
        # ring for every existing single-ring consumer.
        rings_m = [
            {
                "ring": _ensure_ccw(_north_up(part.ring)),
                "holes": [list(reversed(_ensure_ccw(_north_up(h.ring)))) for h in hs],
            }
            for part, hs in zip(simplified_ring.parts, simplified_ring.holes)
        ]

        def _abs_area(ring):
            return abs(
                sum(
                    ring[i][0] * ring[(i + 1) % len(ring)][1]
                    - ring[(i + 1) % len(ring)][0] * ring[i][1]
                    for i in range(len(ring))
                )
            )

        ring_ccw = max(rings_m, key=lambda r: _abs_area(r["ring"]))["ring"]
        area_delta_pct = simplified_ring.area_delta_pct
    else:
        rings_m = []
        if hasattr(simplified_ring, "ring"):
            # plain SimplifyResult (single part, no holes): unwrap the ring
            simplified_ring = simplified_ring.ring
        ring_ccw = _ensure_ccw(_north_up(simplified_ring))

        # --- area delta from simplifier (approximate from ring change) ------
        # The simplifier reports relative area change; we pass it through
        # Original ring area vs simplified ring area
        orig_ring = footprint_from_regions([sp.polygon_m for sp in model.spaces.values()])
        orig_area = 0.0
        n = len(orig_ring)
        for i in range(n):
            x0, y0 = orig_ring[i]
            x1, y1 = orig_ring[(i + 1) % n]
            orig_area += x0 * y1 - x1 * y0
        orig_area = abs(orig_area) / 2.0

        simp_area = 0.0
        n = len(ring_ccw)
        for i in range(n):
            x0, y0 = ring_ccw[i]
            x1, y1 = ring_ccw[(i + 1) % n]
            simp_area += x0 * y1 - x1 * y0
        simp_area = abs(simp_area) / 2.0

        area_delta_pct = ((simp_area - orig_area) / orig_area * 100.0) if orig_area > 0 else 0.0

    from bem.bem_shading import shades_from_model

    shades, shade_notes = shades_from_model(
        model, lambda p: (p[0], -p[1]), [sp.polygon_m for sp in bem_spaces]
    )
    notes.extend(shade_notes)

    from constructions import (
        opening_constructions,
        roof_u_value,
        semi_exterior_floors,
        semi_exterior_walls,
        slab_u_value,
    )

    bem = BEMModel(
        building_name=model.name,
        spaces=bem_spaces,
        openings=bem_openings,
        # y flipped above (canonical y-down -> north-up), same as the IFC
        # and takeoff adapters; before this the ring was written mirrored N-S
        ring_facades=_edge_facades(ring_ccw, y_north=True),
        ring_m=ring_ccw,
        rings_m=rings_m,
        wall_height_m=wall_height_m,
        area_delta_pct=area_delta_pct,
        simplify_tolerance=simplify_tolerance,
        skipped_openings=skipped_openings,
        notes=notes,
        shades=shades,
        roof_u_value_w_m2k=roof_u_value(model),
        slab_u_value_w_m2k=slab_u_value(model),
        opening_constructions=opening_constructions(model),
        interior_wall_constructions=semi_exterior_walls(model),
        interior_floor_constructions=semi_exterior_floors(model),
    )

    return bem


# ---------------------------------------------------------------------------
# CLI parser
# ---------------------------------------------------------------------------


def parse_args():
    import argparse

    ap = argparse.ArgumentParser(
        prog="run_pipeline",
        description="Unified pipeline: generate + link + validate + BEM export.",
    )
    g = ap.add_mutually_exclusive_group(required=True)
    g.add_argument(
        "--seed", type=int, help="synthetic seed (mutually exclusive with --image and --aec-bench)"
    )
    g.add_argument(
        "--image",
        type=Path,
        help="real sheet image path (mutually exclusive with --seed and --aec-bench)",
    )
    g.add_argument(
        "--aec-bench",
        type=Path,
        help="AEC-Bench dataset root path (mutually exclusive with --seed and --image)",
    )
    g.add_argument(
        "--set",
        type=Path,
        dest="set_pdf",
        help="real PDF drawing set (#741): ingest, index, scale, walls/rooms, then the model",
    )
    g.add_argument(
        "--ifc",
        type=Path,
        dest="ifc_path",
        help="IFC model (#988): import, then the same simplify/validate/export stages",
    )
    ap.add_argument(
        "--acknowledge-review",
        action="store_true",
        help="mark open needs_review items acknowledged before validation "
        "(headless runs: the review queue is still written for humans)",
    )
    ap.add_argument(
        "--detections", type=Path, help="sahi_infer.py JSON predictions (required with --image)"
    )
    ap.add_argument(
        "--schedule-csv",
        type=Path,
        help="schedule CSV (required with --image, mutually exclusive with --schedule-table)",
    )
    ap.add_argument(
        "--weights", type=Path, default=Path("detector/best.pt"), help="YOLO weights path"
    )
    ap.add_argument(
        "--out-dir",
        type=Path,
        default=Path("bem_out"),
        help="output directory for stage JSONs (default: bem_out/)",
    )
    ap.add_argument(
        "--open-office-span",
        action="store_true",
        help="use open-office span layout instead of separate offices",
    )
    ap.add_argument(
        "--elevation-key",
        default="elev_grid",
        choices=["elev_grid", "elev_nogrid"],
        help="which elevation variant to link (default: elev_grid)",
    )
    ap.add_argument(
        "--detector-config",
        type=Path,
        default=None,
        help="JSON naming the door/window detection provider for --set (#743); "
        "default: none. See docs/detection_provider.md",
    )
    ap.add_argument(
        "--storey-height",
        type=float,
        default=None,
        help="storey height in metres for --set when the set states none (default 3.0, flagged)",
    )
    ap.add_argument(
        "--climate-zone",
        default=None,
        help="ASHRAE climate zone (e.g. 4A) for the construction library (#747); "
        "unset constructions stay unset without one",
    )
    ap.add_argument(
        "--building-category",
        default="Nonresidential",
        choices=["Nonresidential", "Residential", "Semiheated"],
        help="ASHRAE 90.1 Table 5.5 building category (default Nonresidential)",
    )
    ap.add_argument(
        "--simplify-tol",
        type=float,
        default=0.02,
        help="tolerance for geometry simplification (default: 0.02 = 2%%)",
    )
    ap.add_argument(
        "--no-roof-simplify",
        action="store_true",
        help="do not simplify the roof planes (#1275); on by default",
    )
    ap.add_argument(
        "--mode",
        default="full",
        choices=["full", "progressive"],
        help="pipeline mode (#998): 'full' runs everything through BEM export (default); "
        "'progressive' stops after the takeoff (symbols, areas, counts) without BEM "
        "construction, review-queue gate or export",
    )
    return ap.parse_args()


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------


def _merge_pdfs(pdfs: list[Path], out: Path) -> Path:
    """Concatenate PDFs in order into one multi-sheet set PDF.

    pdfium (pypdfium2, Apache-2.0/BSD-3) is already the set backend
    (``pdf_ingest``); no second PDF library is added for this (#988).
    """
    import pypdfium2 as pdfium

    dest = pdfium.PdfDocument.new()
    try:
        for pdf in pdfs:
            try:
                src = pdfium.PdfDocument(pdf)
            except Exception as e:  # corrupt or not a PDF after all
                raise StageError(
                    "Stage 1: drawing set",
                    1,
                    f"{pdf.name} could not be read as a PDF: {e}",
                    hint="Re-zip the set without the unreadable file, or run --set "
                    "on the PDFs directly.",
                ) from e
            try:
                dest.import_pages(src)
            finally:
                src.close()
        dest.save(out)
    finally:
        dest.close()
    return out


def _legend_tags_for(args, set_pdf: Path) -> dict | None:
    """The project's learned legend sidecar (#995): ``--legend-tags`` wins,
    else ``legend_tags.json`` beside the drawing set. Absent is normal (the
    run simply uses the built-in word rules); malformed is warned about and
    ignored rather than failing the whole run."""
    from legend_learning import SIDECAR_NAME, LegendTagsError, load_tags

    path = getattr(args, "legend_tags", None) or (set_pdf.parent / SIDECAR_NAME)
    try:
        tags = load_tags(path)
    except LegendTagsError as e:
        print(f"warning: legend sidecar ignored: {e}", file=sys.stderr)
        return None
    if tags:
        print(f"legend tags: {len(tags)} learned description(s) from {path}")
    return tags


def _drawing_set_pdf(args, out_dir: Path) -> Path:
    """Resolve ``--set`` to the PDF the set stages ingest (#988).

    A PDF passes straight through. A ``.zip`` is unpacked into
    ``out_dir/_sets/``: a single member PDF passes through unchanged, several
    are concatenated in filename order into one multi-sheet set so the rest of
    the set stages keep seeing one drawing set.
    """
    import zipfile

    pdf = Path(args.set_pdf)
    if pdf.suffix.lower() != ".zip":
        return pdf
    staging = out_dir / "_sets"
    staging.mkdir(parents=True, exist_ok=True)
    members: list[Path] = []
    staged: dict[str, Path] = {}
    try:
        with zipfile.ZipFile(pdf) as zf:
            for info in sorted(zf.infolist(), key=lambda i: i.filename):
                if info.is_dir() or Path(info.filename).suffix.lower() != ".pdf":
                    continue
                name = Path(info.filename).name
                if name in staged:  # two members share a basename: order would lie
                    raise StageError(
                        "Stage 1: drawing set",
                        1,
                        f"{pdf.name} holds two PDFs both named {name} (in "
                        f"{Path(info.filename).parent} and "
                        f"{Path(staged[name].stem).parent or '.'})",
                        hint="Rename one of them inside the zip so sheets stay in order.",
                    )
                dest = staging / name
                with zf.open(info) as src, open(dest, "wb") as dst:
                    dst.write(src.read())
                members.append(dest)
                staged[name] = dest
    except zipfile.BadZipFile as e:
        raise StageError("Stage 1: drawing set", 1, f"{pdf.name} is not a readable zip: {e}") from e
    if not members:
        raise StageError(
            "Stage 1: drawing set",
            1,
            f"{pdf.name} holds no PDF files",
            hint="Zip the drawing-set PDFs, or run --set on the PDF itself.",
        )
    if len(members) == 1:
        return members[0]
    return _merge_pdfs(members, staging / f"{pdf.stem}.pdf")


def _set_storey_height(args, config: dict | None):
    """Storey height for a drawing-set run: ``--storey-height`` first, then the run
    config's ``wall_height`` (a number for every level, or ``{level: metres}``) (#821).
    """
    flag = getattr(args, "storey_height", None)
    if flag is not None:
        return flag
    return (config or {}).get("wall_height")


def _progressive_stop(bldg_or_model: Any, bldg: dict | None) -> None:
    """Progressive mode (#998): report the takeoff, stop before BEM construction."""
    lines = ["progressive mode: stopping after the takeoff (stage 1)"]
    if isinstance(bldg, dict):
        if "takeoff" in bldg:
            t = bldg["takeoff"]
            lines.append(f"  takeoff: {sorted(t) if isinstance(t, dict) else type(t).__name__}")
        for k in ("spaces", "walls", "openings"):
            if k in bldg:
                lines.append(f"  {k}: {len(bldg[k])}")
    if bldg_or_model is not None and getattr(bldg_or_model, "spaces", None):
        lines.append(f"  spaces: {len(bldg_or_model.spaces)} (takeoff source model, no BEM build)")
    lines.append("  no BEM construction, no review-queue gate, no export")
    lines.append("  run with --mode full (the default) for the full pipeline")
    print("\n".join(lines))


def _stage_1_generate(args, out_dir: Path, config: dict | None = None):
    """Stage 1: Generate building or load real-sheet detections.

    Returns:
        tuple: (model, link_report, bldg) for synthetic seed path,
               or (model, None, None) for aec_bench path.
    """
    if getattr(args, "set_pdf", None):
        from detection_provider import (
            ProviderConfigError,
            ProviderLicenseError,
            provider_from_config,
        )
        from real_set import SetError, build_set_model

        set_pdf = _drawing_set_pdf(args, out_dir)  # unpacks a zip (#988)
        try:
            provider = provider_from_config(getattr(args, "detector_config", None))
            model, set_report = build_set_model(
                set_pdf,
                out_dir,
                storey_height_m=_set_storey_height(args, config),
                provider=provider,
                legend_tags=_legend_tags_for(args, set_pdf),
            )
        except (ProviderConfigError, ProviderLicenseError) as e:
            raise StageError(
                "Stage 1: drawing set",
                1,
                str(e),
                hint="--detector-config names the provider; see docs/detection_provider.md.",
            ) from e
        except SetError as e:
            raise StageError(
                "Stage 1: drawing set",
                1,
                str(e),
                hint="Each sheet's reason is in stage_00_set_report.json.",
            ) from e
        write_json(
            out_dir / "stage_01_building.json",
            {
                "source": str(args.set_pdf),
                "n_sheets": len(set_report.sheets),
                "levels": set_report.levels,
                "n_spaces": len(model.spaces),
                "failed_sheets": set_report.to_dict()["failed_sheets"],
            },
        )
        failed = set_report.to_dict()["failed_sheets"]
        if failed:
            print(
                f"WARNING: {len(failed)} sheet(s) could not be fully read: {', '.join(failed)} "
                f"(reasons in {out_dir / 'stage_00_set_report.json'})",
                file=sys.stderr,
            )
        return model, None, None
    if getattr(args, "ifc_path", None):  # IFC input path (#988)
        # Lazy import: ifc_import imports this module's StageError.
        from ifc.ifc_import import import_ifc

        model = import_ifc(str(args.ifc_path))
        write_json(
            out_dir / "stage_01_building.json",
            {
                "source": str(args.ifc_path),
                "schema": getattr(model, "source_schema", None) or "IFC4",
                "n_levels": len(model.levels),
                "n_spaces": len(model.spaces),
            },
        )
        return model, None, None
    if getattr(args, "image", None):  # cli Namespace carries no --image (#753)
        dets = detections_from_yolo_json(getattr(args, "detections", None), args.image.stem)
        schedule_csv = getattr(args, "schedule_csv", None)
        schedule = parse_schedule_csv(schedule_csv) if schedule_csv else {}
        # Stage 1b (#668): read schedule tags beside each symbol; no-op without OCR
        from PIL import Image

        sheet = np.asarray(Image.open(args.image).convert("L"), dtype=np.float64)
        dets = extract_with_tags(dets, sheet, config=config, schedule_tags=set(schedule))
        takeoff = rollup_takeoff(dets, schedule, drawing_type="floor_plan")
        bldg = {
            "building_id": args.image.stem,
            "seed": None,
            "detections": [d.__dict__ for d in dets],
            "schedule": {k: v.__dict__ for k, v in schedule.items()},
            "takeoff": takeoff,
        }
        write_json(
            out_dir / "stage_01_building.json",
            {
                "source": str(args.image),
                "n_detections": len(dets),
                "n_scheduled": len(schedule),
            },
        )
        if getattr(args, "mode", "full") == "progressive":
            # Progressive mode (#998): the takeoff is the deliverable; a single
            # sheet image cannot build a model, so stop here instead of raising.
            write_json(
                out_dir / "stage_01_takeoff.json",
                {
                    "source": str(args.image),
                    "detections": bldg["detections"],
                    "schedule": bldg["schedule"],
                    "takeoff": bldg["takeoff"],
                },
            )
            print("  takeoff: stage_01_takeoff.json")
            return bldg, None, bldg
        # A single raster sheet gives symbols and a takeoff (written above) but no
        # walls or rooms: raster wall finding is not built, so there is nothing to
        # put the symbols in. Fail loudly with the route that does build a model.
        raise StageError(
            "Stage 2: build_model",
            2,
            f"a single sheet image gives symbols and a takeoff ({len(dets)} detections, "
            f"{len(schedule)} scheduled tags, in stage_01_building.json) but no walls or "
            "rooms, so no building model can be built from it",
            hint="Run the PDF drawing set instead: matchline run --set drawings.pdf --out out/ "
            "(walls and rooms come from vector floor plans; see docs/real_set.md).",
        )
    elif args.aec_bench:
        samples, takeoff_result = load_aec_bench(args.aec_bench)
        write_json(
            out_dir / "stage_01_building.json",
            {
                "source": str(args.aec_bench),
                "n_samples": len(samples),
                "n_regions": len(takeoff_result.regions),
            },
        )
        model = _build_minimal_model_from_regions(takeoff_result, bldg_id="aec-bench")
        return model, None, None
    else:
        bldg = generate_building(args.seed, open_office_span=args.open_office_span)
        write_json(out_dir / "stage_01_building.json", bldg)
        return None, None, bldg


def _stage_2_build_model(bldg, args, out_dir: Path, model=None):
    """Stage 2: Build linked model from building data.

    Returns:
        tuple: (model, link_report)
    """
    if model is not None:
        # real set / AEC-Bench: stage 1 already built the model
        write_json(
            out_dir / "stage_02_model.json", {"model": _model_to_dict(model), "link_report": None}
        )
        return model, None
    if args.aec_bench:
        link_report = None
        return None, link_report
    else:
        model, link_report = build_model(
            bldg, elevation_key=args.elevation_key, building_name=bldg["building_id"]
        )
        write_json(
            out_dir / "stage_02_model.json",
            {
                "model": json.loads(model.to_json())
                if hasattr(model, "to_json")
                else _model_to_dict(model),
                "link_report": asdict(link_report) if link_report else None,
            },
        )
        return model, link_report


def _roof_simplify_enabled(args, config: dict | None) -> bool:
    """Is the roof-plane simplification on for this run? (default: yes)

    Opt-out, following the ``--simplify-tol`` / ``simplify_tolerance``
    conventions: the CLI flag ``--no-roof-simplify`` and the YAML config key
    ``roof_simplify: false`` each turn it off; anything else leaves it on.
    """
    enabled = True
    if config:
        value = config.get("roof_simplify", True)
        if isinstance(value, str):
            enabled = value.strip().lower() not in ("false", "no", "0", "off")
        else:
            enabled = bool(value)
    if getattr(args, "no_roof_simplify", False):
        enabled = False
    return enabled


def _stage_3_roof_simplify(model, enabled: bool = True) -> dict:
    """Simplify the model's roof planes and return the audit record (#1275).

    On by default; the ``roof_solar_aperture`` check compares the simplified
    roof against ``model.source_roof_planes``, which this call populates.
    Skips are recorded, never raised: a model with no roof planes or no
    usable site latitude keeps its roof as it is and the reason is written
    into the stage artifact.
    """
    from geometry.roof_simplify import apply_roof_simplification, roof_simplify_report

    planes = list(getattr(model, "roof_planes", None) or [])
    if not enabled:
        return {
            "method": "roof_simplify",
            "applied": False,
            "reason": "disabled (--no-roof-simplify / roof_simplify: false)",
            "n_planes_before": len(planes),
            "n_planes_after": len(planes),
        }
    if not planes:
        return {
            "method": "roof_simplify",
            "applied": False,
            "reason": "no roof planes to simplify",
            "n_planes_before": 0,
            "n_planes_after": 0,
        }
    lat = getattr(model, "site_latitude_deg", None)
    try:
        res = apply_roof_simplification(model)
    except ValueError as e:
        # The wrapper raises ValueError only when no latitude is available
        # (neither an explicit one nor model.site_latitude_deg); anything
        # else is a real failure and must not be swallowed.
        if lat is not None or "latitude" not in str(e).lower():
            raise
        return {
            "method": "roof_simplify",
            "applied": False,
            "reason": f"no site latitude; roof simplification skipped ({e})",
            "n_planes_before": len(planes),
            "n_planes_after": len(planes),
        }
    report = roof_simplify_report(res)
    # NOTE: the report's own "skipped" key is the merge-skip list from
    # roof_simplify_report; the stage-level outcome rides on "applied".
    report["applied"] = True
    report["reason"] = None
    return report


def _stage_3_simplify(
    model,
    simplify_tol: float,
    wall_height: float | None,
    out_dir: Path,
    roof_simplify: bool = True,
):
    """Stage 3: Simplify building geometry.

    Simplifies the footprint (walls) and, by default, the roof planes
    (#1275). Returns:
        tuple: (simplify_result, wall_height)
    """
    if wall_height is None:
        wall_height = model.levels[0].wall_height_m
    parts = footprint_parts_from_regions([sp.polygon_m for sp in model.spaces.values()])
    sres = simplify_footprint(parts, tol=simplify_tol, wall_height=wall_height)
    payload = {
        "original_count": sres.original_count,
        "simplified_count": sres.simplified_count,
        "simplified_ring": sres.ring,
        "area_delta_pct": sres.area_delta_pct,
        "tolerance_pct": sres.tol * 100.0,
    }
    if hasattr(sres, "parts"):
        # #970: per-part simplification (courtyards + wings); record every
        # ring's own delta alongside the combined shared-budget delta.
        payload["method"] = sres.method
        payload["per_part"] = [
            {
                "original_count": p.original_count,
                "simplified_count": p.simplified_count,
                "simplified_ring": p.ring,
                "area_delta_pct": p.area_delta_pct,
                "holes": [
                    {
                        "original_count": h.original_count,
                        "simplified_count": h.simplified_count,
                        "simplified_ring": h.ring,
                        "area_delta_pct": h.area_delta_pct,
                    }
                    for h in hs
                ],
            }
            for p, hs in zip(sres.parts, sres.holes)
        ]
    # #1275: the roof planes are simplified here too (on by default), so the
    # roof_solar_aperture check has a "before" roof to compare against.
    payload["roof_simplify"] = _stage_3_roof_simplify(model, enabled=roof_simplify)
    write_json(out_dir / "stage_03_simplified.json", payload)
    return sres, wall_height


def _acknowledge_review_queue(model) -> int:
    """Mark every open needs_review item acknowledged (headless runs).

    Mirrors the export gate's own criteria (validate/export.py
    review_queue_acknowledged) so a headless run can clear the same set a
    human reviewer would acknowledge in the review UI.
    """
    n = 0
    for item in model.review_queue:
        if (
            item.needs_review
            and not item.acknowledged
            and item.status not in ("confirmed", "rejected")
        ):
            item.acknowledged = True
            n += 1
    return n


def _stage_4_validation(model, sres, min_review_confidence, out_dir: Path):
    """Stage 4: Run validation checks on the model.

    Returns:
        ValidationReport
    """
    report = run_checks(model, sres=sres, min_review_confidence=min_review_confidence)
    write_json(out_dir / "stage_04_validation.json", report.to_dict())
    return report


def _stage_4b_auto_triage(model, out_dir: Path):
    """Stage 4b: Run auto-triage on review queue items."""
    import building_model

    building_model.ENABLE_AUTO_TRIAGE = True
    _run_auto_triage(model)
    write_json(
        out_dir / "stage_04b_auto_triage.json",
        {"auto_triage": True, "items": [asdict(i) for i in model.review_queue]},
    )


def _stage_5_fail_fast(report, min_review_confidence=None):
    """Stage 5: Fail fast if validation errors exist."""
    if not export_gate(report, getattr(report, "model", None), min_review_confidence):
        print(f"VALIDATION FAILED: {len(report.errors)} error(s)", file=sys.stderr)
        for r in report.errors:
            print(f"  - {r.check_id}: {r.message}", file=sys.stderr)
        sys.exit(1)


def _stage_6_conservation_gate(bem_model: BEMModel) -> None:
    """Block export if conservation law checks fail."""
    results = validate_bem_conservation(bem_model)
    violations = [r for r in results if r.severity == "error"]
    if violations:
        msgs = "; ".join(f"{r.name}: {r.message}" for r in violations)
        raise StageError(
            stage_name="validate_bem_conservation",
            stage_index=6,
            msg=f"Conservation law violation: {msgs}",
            hint="Check area_delta_pct and simplify_tolerance thresholds",
        )


def _stage_6_bem_export(model, sres, wall_height, simplify_tol, out_dir: Path):
    """Stage 6: Export BEM model to gbXML and IFC4 formats."""
    bem_model = model_from_linked_model(
        model=model,
        simplified_ring=sres,
        wall_height_m=wall_height,
        simplify_tolerance=simplify_tol * 100.0,
    )
    if len(model.levels) > 1:
        # several storeys (#639): levels, floors, ceilings and roofs
        from bem.bem_levels import add_levels

        add_levels(model, bem_model)
    # scheduled zone airflow onto the gbXML thermal zones (#746)
    from zone_airflow import apply_zone_airflow

    apply_zone_airflow(model, bem_model)
    # scheduled air handlers as gbXML air loops (#746, option 3)
    from air_loops import apply_air_loops

    apply_air_loops(model, bem_model)
    _stage_6_conservation_gate(bem_model)
    bem_dir = out_dir / "stage_06_bem"
    bem_dir.mkdir(parents=True, exist_ok=True)

    gbxml_path = bem_dir / f"{model.name or 'building'}.xml"
    write_gbxml(bem_model, gbxml_path)
    validate_gbxml(gbxml_path)

    ifc_path = bem_dir / f"{model.name or 'building'}.ifc"
    write_ifc4(bem_model, ifc_path)
    validate_ifc4(ifc_path)

    return gbxml_path, ifc_path


def _print_pipeline_complete(out_dir: Path, report):
    """Print completion summary of all pipeline stages."""
    print(f"Pipeline complete: {out_dir}")
    print("  stage_01_building.json")
    print("  stage_02_model.json")
    print("  stage_03_simplified.json")
    print(f"  stage_04_validation.json (ok={report.ok})")
    bem_dir = out_dir / "stage_06_bem"
    if bem_dir.exists():
        for f in bem_dir.iterdir():
            print(f"  stage_06_bem/{f.name}")


def _run_inputs(args, config: dict | None) -> list:
    """What this run read, for the trust report: input files (hashed there) and
    the settings that change the output."""
    out = []
    for role, attr in (
        ("drawing set", "set_pdf"),
        ("sheet image", "image"),
        ("detections", "detections"),
        ("schedule csv", "schedule_csv"),
        ("aec-bench", "aec_bench"),
        ("ifc model", "ifc_path"),
        ("detector config", "detector_config"),
    ):
        v = getattr(args, attr, None)
        if v:
            out.append({"role": role, "path": str(v)})
    seed = getattr(args, "seed", None)
    if seed is not None:
        out.append({"role": "synthetic seed", "value": seed})
    for role, attr in (
        ("climate zone", "climate_zone"),
        ("building category", "building_category"),
    ):
        v = (config or {}).get(attr, getattr(args, attr, None))
        if v:
            out.append({"role": role, "value": v})
    if config:
        out.append({"role": "config", "value": json.dumps(config, sort_keys=True, default=str)})
    return out


@contextmanager
def _stage_telemetry(emitter: TelemetryEmitter, name: str):
    """Record one pipeline stage's outcome and duration (#1037)."""
    t0 = time.monotonic()
    try:
        yield
    except Exception:
        emitter.record_stage(name, "error", time.monotonic() - t0)
        raise
    emitter.record_stage(name, "ok", time.monotonic() - t0)


def _make_emitter(args, config: dict | None) -> TelemetryEmitter:
    """The consent-gated emitter: --telemetry or config telemetry_consent (#1037)."""
    consent = bool(getattr(args, "telemetry", False)) or bool(
        (config or {}).get("telemetry_consent", False)
    )
    local_only = bool((config or {}).get("telemetry_local_only", False))
    return TelemetryEmitter(consent=consent, local_only=local_only)


def _capture_stage_failure(out_dir: Path, err: StageError) -> None:
    """Capture a failed stage as local friction feedback (#1038); never fails the run."""
    from review.feedback_capture import FeedbackCaptureError, FeedbackLog

    cause = type(err.__cause__).__name__ if err.__cause__ else "stage_error"
    # stage identity is the stage number: stage names like "Stage 1:
    # generate_building" are refused by the exclusion list ("building")
    try:
        FeedbackLog(out_dir / "feedback.jsonl").log_failed_parse(f"stage_{err.stage_index}", cause)
    except (FeedbackCaptureError, OSError):
        pass


def _emit_telemetry(emitter: TelemetryEmitter, out_dir: Path) -> None:
    """Write the payload when emission is on; telemetry never fails a run (#1037)."""
    from telemetry_emitter import EmitterBuildError

    try:
        payload = emitter.build_payload()
    except EmitterBuildError as e:
        print(f"warning: telemetry payload skipped: {e}", file=sys.stderr)
        return
    if payload is not None:
        write_json(out_dir / "telemetry.json", payload)


def main(args, config: dict | None = None) -> None:
    out_dir = _validate_out_path(args.out_dir)
    out_dir.mkdir(parents=True, exist_ok=True)
    emitter = _make_emitter(args, config)
    try:
        # Config overrides from YAML
        simplify_tol = args.simplify_tol
        wall_height = None
        min_review_confidence = None
        roof_simplify = _roof_simplify_enabled(args, config)
        if config:
            simplify_tol = config.get("simplify_tolerance", simplify_tol)
            wall_height = config.get("wall_height", None)
            if isinstance(wall_height, dict):
                # per-level heights went to the set model (#821); simplify uses its first level
                wall_height = None
            min_review_confidence = config.get("review_confidence", None)

        # --- Stage 1: generate building or load real-sheet detections ----------
        try:
            with _stage_telemetry(emitter, "1.generate"):
                bldg_or_model, link_report, bldg = _stage_1_generate(args, out_dir, config)
        except StageError:
            raise
        except Exception as e:
            raise StageError(
                "Stage 1: generate_building",
                1,
                str(e),
                hint="Check input data format. For --aec-bench ensure dataset path is valid. "
                "For --seed ensure the seed is an integer. "
                "For --ifc ensure the file is a readable IFC.",
            ) from e

        # --- Progressive mode (#998): the takeoff is the deliverable ---------
        if getattr(args, "mode", "full") == "progressive":
            _progressive_stop(bldg_or_model, bldg)
            _emit_telemetry(emitter, out_dir)
            return

        # --- Stage 2: build model --------------------------------------------
        try:
            with _stage_telemetry(emitter, "2.build_model"):
                model, link_report = _stage_2_build_model(bldg, args, out_dir, bldg_or_model)
        except StageError:
            raise
        except Exception as e:
            raise StageError(
                "Stage 2: build_model",
                2,
                str(e),
                hint="Check that Stage 1 output is valid. Ensure elevation_key is correct "
                "and building detections contain required categories.",
            ) from e

        # --- Stage 3: simplify geometry --------------------------------------
        try:
            with _stage_telemetry(emitter, "3.simplify"):
                sres, wall_height = _stage_3_simplify(
                    model, simplify_tol, wall_height, out_dir, roof_simplify=roof_simplify
                )
        except StageError:
            raise
        except Exception as e:
            raise StageError(
                "Stage 3: simplify_ring",
                3,
                str(e),
                hint="Check simplify_tolerance setting. "
                "Try increasing --simplify-tol (default 0.02). "
                "Ensure wall_height is valid (> 0).",
            ) from e

        # --- Stage 3b: construction library (#747) ---------------------------
        with _stage_telemetry(emitter, "3b.constructions"):
            climate_zone = getattr(args, "climate_zone", None)
            category = getattr(args, "building_category", None) or "Nonresidential"
            if config:
                climate_zone = config.get("climate_zone", climate_zone)
                category = config.get("building_category", category)
            constructions = apply_construction_library(model, climate_zone or "", category)
            write_json(out_dir / "stage_03b_constructions.json", constructions.to_dict())

        if getattr(args, "acknowledge_review", False):
            n_ack = _acknowledge_review_queue(model)
            print(f"  acknowledged {n_ack} review queue item(s) (--acknowledge-review)")

        # --- Stage 4: validation checks --------------------------------------
        try:
            with _stage_telemetry(emitter, "4.validation"):
                report = _stage_4_validation(model, sres, min_review_confidence, out_dir)
        except StageError:
            raise
        except Exception as e:
            raise StageError(
                "Stage 4: run_checks",
                4,
                str(e),
                hint="Check model structure and validation rules. Ensure all required "
                "fields are populated in the BuildingModel.",
            ) from e

        from telemetry_emitter import EmitterBuildError

        for _r in report.errors:
            try:
                emitter.record_check_failure(_r.check_id)
            except EmitterBuildError:
                pass  # a non-conforming check id is a skipped metric, never a failed run

        # --- Stage 4b: auto-triage (default, no opt-in) -----------------------
        with _stage_telemetry(emitter, "4b.auto_triage"):
            _stage_4b_auto_triage(model, out_dir)

        # --- HTML review report, written before the fail-fast gate (#749) ---
        from review.review_report import write_review

        with _stage_telemetry(emitter, "review.report"):
            review_dir = write_review(out_dir, model)
        print(f"  review: {review_dir / 'review.html'}")

        # --- Stage 5: fail-fast on validation errors ------------------------
        _t5 = time.monotonic()
        try:
            _stage_5_fail_fast(report, min_review_confidence)
        except SystemExit:
            emitter.record_stage("5.fail_fast", "error", time.monotonic() - _t5)
            _emit_telemetry(emitter, out_dir)
            raise
        emitter.record_stage("5.fail_fast", "ok", time.monotonic() - _t5)

        # --- Stage 6: BEM export ---------------------------------------------
        try:
            _t6 = time.monotonic()
            gbxml_path, ifc_path = _stage_6_bem_export(
                model, sres, wall_height, simplify_tol, out_dir
            )
            conv = build_convention_report(model, report, sres, constructions)
            write_json(out_dir / "stage_06_bem" / "convention_report.json", conv)
            # export package with the one-page trust report (#750)
            from export_package import build_package

            pkg = build_package(
                out_dir,
                model,
                report,
                conv,
                [gbxml_path, ifc_path],
                _run_inputs(args, config),
                review_dir=review_dir,
                require_gate=True,
            )
            print(f"  package: {pkg}")
            print(f"  package zip: {out_dir / 'package.zip'}")
            _print_pipeline_complete(out_dir, report)
            emitter.record_stage("6.export", "ok", time.monotonic() - _t6)
        except StageError:
            emitter.record_stage("6.export", "error", time.monotonic() - _t6)
            raise
        except Exception as e:
            emitter.record_stage("6.export", "error", time.monotonic() - _t6)
            raise StageError(
                "Stage 6: BEM_export",
                6,
                str(e),
                hint="Check BEM export dependencies and output directory permissions. "
                "Ensure write_gbxml and write_ifc4 can write to the output directory.",
            ) from e
        _emit_telemetry(emitter, out_dir)
    except StageError as e:
        # the stage name, the reason and the recovery hint, so a failed run says why
        print(f"ERROR {e}", file=sys.stderr)
        if e.hint:
            print(f"  hint: {e.hint}", file=sys.stderr)
        _capture_stage_failure(out_dir, e)
        _emit_telemetry(emitter, out_dir)
        sys.exit(1)


def _build_minimal_model_from_regions(takeoff_result, bldg_id: str):
    """Build a minimal BuildingModel from AEC-Bench regions.

    Creates a canonical model with one Level, one Space, minimal openings,
    and review-queue entries to satisfy validation checks (errors blocked,
    warnings acceptable).  This is not architecturally accurate — it is the
    minimum viable model needed to run validation and produce gbXML.
    """
    from shapely import MultiPolygon as ShapelyMultiPolygon
    from shapely import Polygon as ShapelyPolygon
    from shapely.ops import unary_union

    from building_model import (
        BuildingModel,
        EnvelopeWall,
        Level,
        Provenance,
        ReviewItem,
        Space,
        SpaceHVAC,
        SpaceLighting,
        SpaceOpening,
    )

    regions = takeoff_result.regions
    by_cat: dict[str, list] = {}
    for r in regions:
        by_cat.setdefault(r.category, []).append(r)

    # AEC-Bench annotations are in pixel coordinates.  Use a rough scale
    # (1 px = 0.001 m ≈ 200 DPI architectural drawing) to produce plausible
    # metre-level geometry for BEM export.
    scale = 0.001  # m per px

    def px_to_m(pt):
        return [float(pt[0]) * scale, float(pt[1]) * scale]

    # --- Polygon: union of all floor_area regions in metres, fallback bounding box ---
    floor_regions = by_cat.get("floor_area", [])
    if floor_regions:
        polys_shapely = []
        for fr in floor_regions:
            if len(fr.polygon_px) >= 3:
                pts_m = [px_to_m(p) for p in fr.polygon_px]
                polys_shapely.append(ShapelyPolygon(pts_m))
        if polys_shapely:
            merged = unary_union(polys_shapely)
            if isinstance(merged, ShapelyMultiPolygon):
                merged = max(merged.geoms, key=lambda g: g.area)
            ring_m = list(merged.exterior.coords)
            if ring_m and ring_m[0] == ring_m[-1]:
                ring_m = ring_m[:-1]
            poly = ring_m
        else:
            poly = [[0.0, 0.0], [10.0, 0.0], [10.0, 10.0], [0.0, 10.0]]
    else:
        # No floor areas: use bounding box of all wall regions in metres
        wall_regions = by_cat.get("wall", [])
        if wall_regions:
            all_pts = []
            for wr in wall_regions:
                all_pts.extend(wr.polygon_px)
            if all_pts:
                xs = [p[0] for p in all_pts]
                ys = [p[1] for p in all_pts]
                min_x, max_x = min(xs), max(xs)
                min_y, max_y = min(ys), max(ys)
                poly = [
                    px_to_m((min_x, min_y)),
                    px_to_m((max_x, min_y)),
                    px_to_m((max_x, max_y)),
                    px_to_m((min_x, max_y)),
                ]
            else:
                poly = [[0.0, 0.0], [10.0, 0.0], [10.0, 10.0], [0.0, 10.0]]
        else:
            poly = [[0.0, 0.0], [10.0, 0.0], [10.0, 10.0], [0.0, 10.0]]

    # --- Compute area from polygon (shoelace, in m²) ---
    area = 0.0
    n = len(poly)
    for i in range(n):
        x0, y0 = poly[i]
        x1, y1 = poly[(i + 1) % n]
        area += x0 * y1 - x1 * y0
    area = abs(area) / 2.0
    wall_height = 3.0  # metres
    volume = area * wall_height

    level = Level(id="L1", name="Level 1", wall_height_m=wall_height)
    labeled = attach_room_labels(takeoff_result, None)
    if labeled.spaces:
        first_labeled = labeled.spaces[0]
        lconf = first_labeled.label_confidence
        lsrc = first_labeled.label_source or "room_label_parse"
        note = f"room_label: name={first_labeled.name}, number={first_labeled.number}"
        prov = Provenance(
            sheet_id=bldg_id,
            revision=0,
            method=lsrc,
            confidence=lconf,
            note=note,
        )
    else:
        prov = Provenance(sheet_id=bldg_id, revision=0, method="aec-bench-minimal", confidence=1.0)

    # --- Space with minimal openings ---
    openings: list[SpaceOpening] = []
    # Wall regions are walls, not openings: doors come only from door
    # symbols (below), so a sheet's door count is never inflated by its walls.
    for i, dr in enumerate(by_cat.get("door", [])):
        tag = f"D{i:02d}"
        # Plan position of the door symbol (centroid of its detected
        # polygon), so space_merge can tell which rooms a door connects.
        # A degenerate polygon leaves it None rather than guessing.
        center = None
        if len(dr.polygon_px) >= 3:
            c = ShapelyPolygon([px_to_m(pt) for pt in dr.polygon_px]).centroid
            if not c.is_empty:
                center = [round(c.x, 6), round(c.y, 6)]
        openings.append(
            SpaceOpening(
                id=tag,
                tag=tag,
                category="door",
                width_m=0.9,
                height_m=2.1,
                area_m2=0.9 * 2.1,
                provenance=prov,
                needs_review=False,
                plan_center_m=center,
            )
        )

    for i, wr in enumerate(by_cat.get("window", [])):
        tag = f"WD{i:02d}"
        openings.append(
            SpaceOpening(
                id=tag,
                tag=tag,
                category="window",
                width_m=1.2,
                height_m=1.5,
                area_m2=1.2 * 1.5,
                provenance=prov,
                needs_review=False,
            )
        )

    space = Space(
        id="L1-1",
        level_id="L1",
        name="SPACE-1",
        number="1",
        polygon_m=poly,
        area_m2=area,
        volume_m3=volume,
        openings=openings,
        lighting=SpaceLighting(total_w=area * 5.0),  # 5 W/m2 default
        hvac=SpaceHVAC(),
        core_provenance=prov,
        label_confidence=1.0,
    )

    # --- Envelope wall: single facade from bounding box ---
    n = len(poly)
    perimeter = sum(
        ((poly[i][0] - poly[(i + 1) % n][0]) ** 2 + (poly[i][1] - poly[(i + 1) % n][1]) ** 2) ** 0.5
        for i in range(n)
    )

    envelope = [
        EnvelopeWall(
            id="ENV-1",
            facade="south",
            from_m=[float(v) for v in poly[0]],
            to_m=[float(v) for v in poly[1 if n > 1 else 0]],
            height_m=wall_height,
            area_m2=perimeter * wall_height,
            provenance=prov,
        )
    ]

    # --- Review queue: suppress fixture/opening errors ---
    # These are known limitations (not reviewed); needs_review=False with
    # confidence < 1.0 per the conservation-law invariant.
    review_queue = [
        ReviewItem(
            id="rq-fixture",
            kind="fixture_schedule",
            description="AEC-Bench minimal model: fixture schedule unavailable",
            confidence=0.85,
            provenance=prov,
            needs_review=False,
        ),
        ReviewItem(
            id="rq-opening",
            kind="window_room_link",
            description="AEC-Bench minimal model: opening dimensions estimated",
            confidence=0.85,
            provenance=prov,
            needs_review=False,
        ),
    ]

    model = BuildingModel(
        name=bldg_id,
        levels=[level],
        spaces={"L1-1": space},
        zones={},
        envelope=envelope,
        bim_elements=[],
        schedules={},  # empty: review queue suppresses errors
        review_queue=review_queue,
    )

    # Route low-confidence labels to review queue
    if labeled.spaces:
        for ls in labeled.spaces:
            if ls.label_confidence < 0.80:
                model.flag_for_review(
                    kind="space_no_geometry",
                    description=f"Low-confidence room label: name={ls.name}, number={ls.number}",
                    confidence=ls.label_confidence,
                    provenance=prov,
                )
    for ul in labeled.unmatched_labels:
        if ul.confidence < 0.80:
            model.flag_for_review(
                kind="window_room_link",
                description=f"Low-confidence unmatched label: text={ul.text}",
                confidence=ul.confidence,
                provenance=prov,
            )

    return model


def _model_to_dict(model) -> dict:
    """Convert BuildingModel to dict for JSON serialization."""
    return {
        "name": model.name,
        "model_version": model.model_version,
        "levels": [asdict(l) for l in model.levels],
        "spaces": {k: asdict(v) for k, v in model.spaces.items()},
        "zones": {k: asdict(v) for k, v in model.zones.items()},
        "envelope": [asdict(e) for e in model.envelope],
        "schedules": model.schedules,
        "review_queue": [asdict(r) for r in model.review_queue],
    }


if __name__ == "__main__":
    main(parse_args())