autocad-mcp-pro/scripts/field_test_takeoff.py

"""Run both takeoffs on a real P&ID and layout, and write the report next to them.

LOCAL ONLY. This is the field test of track H (spec §15): the user points it at
a client's P&ID and layout, compares what comes out with the hand-made
workbooks, and keeps the result on their own disk. Nothing it reads or writes
may enter this repository (spec §14), so:

* the report is written next to the P&ID (or into ``--out``) and **the script
  refuses any output folder inside this repository**, before reading a byte;
* the console gets file names, row counts and the scale verdict - **never a
  quantity**, unless ``--show`` is given on the user's own terminal;
* it is never run in CI and never by a test on a real drawing; the test next
  to it (tests/test_field_test_takeoff.py) runs it on the synthetic plant pair.

What it writes: ``pipe_takeoff.xlsx`` / ``cable_takeoff.xlsx`` (CSV beside
them always, XLSX when the ``office`` extra is installed) through the same
``write_workbook`` the tools use, and ``field_test_report.json`` with the scale
check and both results in full.

    uv run --frozen python scripts/field_test_takeoff.py --pid D:/job/pid.dxf --layout D:/job/layout.dxf
    uv run --frozen python scripts/field_test_takeoff.py --pid ... --layout ... --lang en --show
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

#: Exit code for a refused run (output inside the repository, a missing file).
REFUSED = 2


def _inside_repository(folder: Path) -> bool:
    resolved = folder.resolve()
    return resolved == ROOT or ROOT in resolved.parents


def _parse(argv) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Pipe and cable takeoffs on a real P&ID + layout (local only)."
    )
    parser.add_argument("--pid", required=True, help="the P&ID, a .dxf")
    parser.add_argument("--layout", required=True, help="the layout, a .dxf")
    parser.add_argument("--out", default=None, help="report folder (default: the P&ID's folder)")
    parser.add_argument("--lang", default="tr", choices=("tr", "en", "ru"))
    parser.add_argument("--allowance", type=float, default=0.20)
    parser.add_argument(
        "--show", action="store_true", help="also print the totals (quantities) to the console"
    )
    return parser.parse_args(argv)


def main(argv=None) -> int:
    args = _parse(argv)
    pid_path, layout_path = Path(args.pid), Path(args.layout)
    out = Path(args.out) if args.out else pid_path.resolve().parent
    if _inside_repository(out):
        print(
            f"refused: {out} is inside the repository ({ROOT}); client drawings and "
            "anything derived from them never enter it. Pass --out outside it."
        )
        return REFUSED
    for path in (pid_path, layout_path):
        if not path.is_file():
            print(f"refused: {path} does not exist")
            return REFUSED

    from engineering.understand.network import build_network
    from engineering.understand.report import write_workbook
    from engineering.understand.scale import scale_check
    from engineering.understand.snapshot import read_snapshot
    from engineering.understand.takeoff import cable_rows, pipe_rows

    out.mkdir(parents=True, exist_ok=True)
    pid = read_snapshot(str(pid_path))
    layout = read_snapshot(str(layout_path))
    scale = scale_check(pid, layout)
    pipe = pipe_rows(build_network(pid), pid, layout, scale=scale, allowance=args.allowance)
    cable = cable_rows(pid, layout, allowance=args.allowance)
    written = [
        write_workbook(pipe, kind="pipe", lang=args.lang, path=str(out / "pipe_takeoff.xlsx")),
        write_workbook(cable, kind="cable", lang=args.lang, path=str(out / "cable_takeoff.xlsx")),
    ]
    report = out / "field_test_report.json"
    report.write_text(
        json.dumps(
            {"scale": scale, "pipe": pipe, "cable": cable, "written": written},
            indent=2,
            ensure_ascii=False,
            default=str,
        ),
        encoding="utf-8",
    )
    print(
        json.dumps(
            {
                "scale_verdict": scale["verdict"],
                "pipe_rows": len(pipe["rows"]),
                "cable_rows": len(cable["rows"]),
                "report": str(report),
                "workbooks": [
                    path
                    for item in written
                    for path in ([item["xlsx"]] if item["xlsx"] else []) + list(item["csv"])
                ],
            },
            indent=2,
        )
    )
    if args.show:
        print(
            json.dumps(
                {"pipe": pipe["rows"], "cable": cable["rows"]},
                indent=2,
                ensure_ascii=False,
                default=str,
            )
        )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())