166 lines
8.8 KiB
Python
166 lines
8.8 KiB
Python
"""Fixed offline acceptance harness for the 2D CAD adapter.
|
||
|
||
This validates the production worker and deterministic outputs. It does not claim
|
||
AutoCAD, ZWCAD, or GstarCAD interoperability; those remain manual target-machine gates.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import importlib.util
|
||
import json
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
from uuid import uuid4
|
||
|
||
WORKER_PATH = Path(__file__).with_name("worker.py")
|
||
SPEC = importlib.util.spec_from_file_location("zcbot_cad2d_worker", WORKER_PATH)
|
||
assert SPEC and SPEC.loader
|
||
worker = importlib.util.module_from_spec(SPEC)
|
||
SPEC.loader.exec_module(worker)
|
||
|
||
|
||
def _base(title: str, sheet: str = "A3", drawing_type: str = "floor_plan") -> dict:
|
||
walls = [
|
||
{"id": "south", "start": [0, 0], "end": [12000, 0], "thickness": 240, "kind": "exterior"},
|
||
{"id": "east", "start": [12000, 0], "end": [12000, 9000], "thickness": 240, "kind": "exterior"},
|
||
{"id": "north", "start": [12000, 9000], "end": [0, 9000], "thickness": 240, "kind": "exterior"},
|
||
{"id": "west", "start": [0, 9000], "end": [0, 0], "thickness": 240, "kind": "exterior"},
|
||
{"id": "partition", "start": [5000, 0], "end": [5000, 9000], "thickness": 120, "kind": "interior"},
|
||
]
|
||
return {
|
||
"schema_version": 1, "inputs": [], "outputs": [],
|
||
"operation": {"drawing": {
|
||
"type": drawing_type, "recipe_version": 1, "title": title, "unit": "mm",
|
||
"sheet": {"size": sheet, "orientation": "landscape", "scale": 50},
|
||
"walls": walls,
|
||
"openings": [
|
||
{"id": "entry", "wall_id": "south", "offset": 900, "width": 1000, "type": "single_door", "hinge": "start", "swing": "in"},
|
||
{"id": "window1", "wall_id": "north", "offset": 1200, "width": 1800, "type": "window"},
|
||
],
|
||
"areas": [{"id": "room", "boundary": [[0,0],[12000,0],[12000,9000],[0,9000],[0,0]], "name": "房间", "label_position": [6000,4500], "show_area": True}],
|
||
"objects": [{"id": "desk1", "type": "desk", "position": [2500,2500], "rotation": 0, "label": "试验台"}],
|
||
"dimensions": [{"id": "overall1", "type": "overall", "points": [[0,0],[5000,0],[12000,0]], "offset": -700}],
|
||
"notes": [{"id": "note1", "position": [8000,8000], "text": "尺寸单位：mm"}],
|
||
}},
|
||
}
|
||
|
||
|
||
def _cases() -> dict[str, tuple[dict, str | None]]:
|
||
small = _base("中文小三居", "A3")
|
||
small["operation"]["drawing"]["walls"] += [
|
||
{"id":"left_split","start":[0,4500],"end":[5000,4500],"thickness":120,"kind":"interior"},
|
||
{"id":"right_split","start":[5000,5200],"end":[12000,5200],"thickness":120,"kind":"interior"},
|
||
]
|
||
small["operation"]["drawing"]["areas"] = [
|
||
{"id":"bed1","boundary":[[0,0],[5000,0],[5000,4500],[0,4500],[0,0]],"name":"卧室一","label_position":[2500,3200],"show_area":True},
|
||
{"id":"bed2","boundary":[[0,4500],[5000,4500],[5000,9000],[0,9000],[0,4500]],"name":"卧室二","label_position":[2500,6800],"show_area":True},
|
||
{"id":"bed3","boundary":[[5000,5200],[12000,5200],[12000,9000],[5000,9000],[5000,5200]],"name":"卧室三","label_position":[8500,7200],"show_area":True},
|
||
{"id":"living","boundary":[[5000,0],[12000,0],[12000,5200],[5000,5200],[5000,0]],"name":"客餐厅","label_position":[8500,3000],"show_area":True},
|
||
]
|
||
two_room = _base("两室一厅", "A4")
|
||
two_room["operation"]["drawing"]["sheet"]["scale"] = 100
|
||
lab = _base("实验室设备布置", "A2", "equipment_layout")
|
||
lab["operation"]["drawing"]["objects"] += [
|
||
{"id": "hood", "type": "fume_hood", "position": [8000,2500], "label": "通风柜"},
|
||
{"id": "exit", "type": "safety_exit", "position": [10500,7000], "label": "安全出口"},
|
||
]
|
||
workshop = _base("车间布置", "A2", "equipment_layout")
|
||
workshop["operation"]["drawing"]["objects"] = [
|
||
{"id": "machine1", "type": "machine", "position": [3000,3000], "size": [2200,1600], "label": "成型机"},
|
||
{"id": "tank1", "type": "tank", "position": [9000,5000], "size": [1800,1800], "label": "储罐"},
|
||
]
|
||
dense = _base("密集门窗尺寸", "A2")
|
||
dense["operation"]["drawing"]["dimensions"] = [{"id":"chain1","type":"chain","points":[[x,0] for x in range(0,12001,1000)],"offset":-900}]
|
||
outside = _base("门窗越界")
|
||
outside["operation"]["drawing"]["openings"][0]["offset"] = 11900
|
||
overlap = _base("洞口重叠")
|
||
overlap["operation"]["drawing"]["openings"].append({"id":"entry2","wall_id":"south","offset":1500,"width":1200,"type":"opening"})
|
||
unclosed = _base("区域不闭合")
|
||
unclosed["operation"]["drawing"]["areas"][0]["boundary"][-1] = [1, 0]
|
||
collision = _base("对象碰撞警告")
|
||
collision["operation"]["drawing"]["objects"][0]["position"] = [1000, 500]
|
||
return {
|
||
"small_three_bedroom": (small, None), "two_bedroom_living": (two_room, None),
|
||
"laboratory_layout": (lab, None), "workshop_layout": (workshop, None),
|
||
"dense_openings_dimensions": (dense, None),
|
||
"opening_outside_wall": (outside, "CAD2D_OPENING_OUTSIDE_WALL"),
|
||
"overlapping_openings": (overlap, "CAD2D_OPENINGS_OVERLAP"),
|
||
"unclosed_area": (unclosed, "CAD2D_AREA_NOT_CLOSED"),
|
||
"object_collision_warning": (collision, None),
|
||
}
|
||
|
||
|
||
def _digest(path: Path) -> str:
|
||
return hashlib.sha256(path.read_bytes()).hexdigest()
|
||
|
||
|
||
def _execute(root: Path, name: str, request: dict, expected_error: str | None, repeat: int) -> dict:
|
||
digests: list[dict[str, str]] = []
|
||
result: dict = {"name": name, "passed": False, "expected_error": expected_error}
|
||
for index in range(repeat):
|
||
job = root / f"验证 输出 {name}-{index + 1}"
|
||
(job / "request").mkdir(parents=True)
|
||
record = {"job_id": str(uuid4()), "lease_id": str(uuid4()), "request_digest": "a" * 64, "request": request}
|
||
(job / "request" / "request.json").write_text(json.dumps(record, ensure_ascii=False), encoding="utf-8")
|
||
try:
|
||
artifacts = worker.run(job)
|
||
if expected_error:
|
||
result["detail"] = "expected failure but worker succeeded"
|
||
return result
|
||
current = {item["filename"]: _digest(job / "output" / (".meta/provenance.json" if item["filename"] == "provenance.json" else item["filename"])) for item in artifacts if item["filename"] not in {"provenance.json", "drawing.pdf"}}
|
||
digests.append(current)
|
||
report = json.loads((job / "output" / "validation-report.json").read_text(encoding="utf-8"))
|
||
if not report["passed"] or report["checks"]["version"] != "R2013":
|
||
result["detail"] = "mechanical validation failed"
|
||
return result
|
||
except worker.DrawingError as exc:
|
||
if exc.code != expected_error:
|
||
result["detail"] = f"unexpected error: {exc.code}"
|
||
return result
|
||
result.update({"passed": True, "observed_error": exc.code})
|
||
return result
|
||
result["passed"] = all(item == digests[0] for item in digests[1:])
|
||
result["deterministic_repeats"] = len(digests)
|
||
result["warnings"] = report["warnings"]
|
||
return result
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument("--work-root", type=Path, required=True)
|
||
parser.add_argument("--repeat", type=int, default=3, choices=range(1, 6))
|
||
parser.add_argument("--case", action="append", choices=sorted(_cases()))
|
||
args = parser.parse_args()
|
||
if args.work_root.exists():
|
||
print("[ERR] --work-root must not already exist", file=sys.stderr)
|
||
return 2
|
||
args.work_root.mkdir(parents=True)
|
||
selected = args.case or list(_cases())
|
||
results = [_execute(args.work_root, name, *_cases()[name], args.repeat) for name in selected]
|
||
# Cancellation/cleanup gate: terminate a child before it can touch a job directory.
|
||
child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"])
|
||
child.terminate()
|
||
try:
|
||
child.wait(timeout=5)
|
||
cancellation = {"passed": True, "returncode": child.returncode}
|
||
except subprocess.TimeoutExpired:
|
||
child.kill(); child.wait(); cancellation = {"passed": False, "returncode": child.returncode}
|
||
report = {
|
||
"passed": all(item["passed"] for item in results) and cancellation["passed"],
|
||
"adapter_version": worker.ADAPTER_VERSION, "cases": results,
|
||
"cancellation_and_cleanup": cancellation,
|
||
"path_coverage": "work root supports spaces and Chinese characters",
|
||
"manual_compatibility_pending": ["AutoCAD", "ZWCAD", "GstarCAD"],
|
||
}
|
||
(args.work_root / "acceptance-report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
||
print("[OK] 2D CAD acceptance passed." if report["passed"] else "[ERR] 2D CAD acceptance failed.")
|
||
return 0 if report["passed"] else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|