537 lines
30 KiB
Python
537 lines
30 KiB
Python
"""Deterministic runtime-only 2D CAD adapter with a closed declarative recipe."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import os
|
|
import sys
|
|
from collections import Counter
|
|
from copy import deepcopy
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
|
|
ADAPTER_VERSION = "1.0.0"
|
|
PILLOW_VERSION = "12.0.0"
|
|
MAX_REQUEST_BYTES = 524_288
|
|
MAX_ENTITIES = 1024
|
|
LAYERS = (
|
|
"A-WALL-EXT", "A-WALL-INT", "A-DOOR", "A-WINDOW", "A-AREA",
|
|
"A-FURN", "A-EQUIP", "A-SAFE", "A-DIMS", "A-TEXT", "A-HATCH", "A-FRAME",
|
|
)
|
|
SHEET_MM = {"A4": (210, 297), "A3": (297, 420), "A2": (420, 594), "A1": (594, 841), "A0": (841, 1189)}
|
|
OBJECT_DEFAULTS = {
|
|
"desk": (1400, 700), "chair": (500, 500), "cabinet": (900, 450),
|
|
"lab_bench": (1800, 750), "fume_hood": (1500, 850), "machine": (1800, 1200),
|
|
"tank": (1200, 1200), "safety_exit": (900, 300), "extinguisher": (300, 300),
|
|
}
|
|
OUTPUT_MEDIA = {
|
|
"drawing.dxf": ("drawing_dxf", "image/vnd.dxf"),
|
|
"drawing.svg": ("drawing_svg", "image/svg+xml"),
|
|
"drawing.png": ("drawing_png", "image/png"),
|
|
"drawing.pdf": ("drawing_pdf", "application/pdf"),
|
|
"drawing-recipe.json": ("drawing_recipe", "application/json"),
|
|
"drawing-manifest.json": ("drawing_manifest", "application/json"),
|
|
"validation-report.json": ("validation_report", "application/json"),
|
|
"provenance.json": ("provenance", "application/json"),
|
|
}
|
|
|
|
|
|
class DrawingError(ValueError):
|
|
def __init__(self, code: str, detail: str = "") -> None:
|
|
super().__init__(code if not detail else f"{code}: {detail}")
|
|
self.code = code
|
|
|
|
|
|
def _atomic_bytes(path: Path, data: bytes) -> None:
|
|
temporary = path.with_name(path.name + ".tmp-" + os.urandom(8).hex())
|
|
try:
|
|
with temporary.open("wb") as handle:
|
|
handle.write(data)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.replace(temporary, path)
|
|
finally:
|
|
temporary.unlink(missing_ok=True)
|
|
|
|
|
|
def _atomic_json(path: Path, value: Any) -> None:
|
|
_atomic_bytes(path, (json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True, allow_nan=False) + "\n").encode())
|
|
|
|
|
|
def _sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def _artifact(path: Path) -> dict[str, Any]:
|
|
artifact_id, media_type = OUTPUT_MEDIA[path.name]
|
|
return {"artifact_id": artifact_id, "filename": path.name, "media_type": media_type,
|
|
"size_bytes": path.stat().st_size, "sha256": _sha256(path)}
|
|
|
|
|
|
def _terminal(record: dict[str, Any], status: str, artifacts: list[dict[str, Any]], code: str = "", detail: str = "") -> dict[str, Any]:
|
|
return {"job_id": record.get("job_id", ""), "lease_id": record.get("lease_id", ""),
|
|
"request_digest": record.get("request_digest", ""), "status": status,
|
|
"artifacts": artifacts, "error": {} if not code else {"code": code, "detail": detail[:500]},
|
|
"finished_at": datetime.now(timezone.utc).isoformat()}
|
|
|
|
|
|
def _point(value: Iterable[float]) -> tuple[float, float]:
|
|
x, y = value
|
|
result = (float(x), float(y))
|
|
if not all(math.isfinite(item) and abs(item) <= 1_000_000 for item in result):
|
|
raise DrawingError("CAD2D_COORDINATE_INVALID")
|
|
return result
|
|
|
|
|
|
def _distance(a: tuple[float, float], b: tuple[float, float]) -> float:
|
|
return math.hypot(b[0] - a[0], b[1] - a[1])
|
|
|
|
|
|
def _cross(a: tuple[float, float], b: tuple[float, float], c: tuple[float, float]) -> float:
|
|
return (b[0] - a[0]) * (c[1] - a[1]) - (b[1] - a[1]) * (c[0] - a[0])
|
|
|
|
|
|
def _segments_intersect(a: tuple[float, float], b: tuple[float, float], c: tuple[float, float], d: tuple[float, float]) -> bool:
|
|
def orientation(p: tuple[float, float], q: tuple[float, float], r: tuple[float, float]) -> int:
|
|
value = _cross(p, q, r)
|
|
return 1 if value > 1e-8 else -1 if value < -1e-8 else 0
|
|
def on_segment(p: tuple[float, float], q: tuple[float, float], r: tuple[float, float]) -> bool:
|
|
return min(p[0], r[0]) - 1e-8 <= q[0] <= max(p[0], r[0]) + 1e-8 and min(p[1], r[1]) - 1e-8 <= q[1] <= max(p[1], r[1]) + 1e-8
|
|
first, second = orientation(a, b, c), orientation(a, b, d)
|
|
third, fourth = orientation(c, d, a), orientation(c, d, b)
|
|
if first != second and third != fourth:
|
|
return True
|
|
return ((first == 0 and on_segment(a, c, b)) or (second == 0 and on_segment(a, d, b))
|
|
or (third == 0 and on_segment(c, a, d)) or (fourth == 0 and on_segment(c, b, d)))
|
|
|
|
|
|
def _polygon_area(points: list[tuple[float, float]]) -> float:
|
|
return abs(sum(a[0] * b[1] - b[0] * a[1] for a, b in zip(points, points[1:]))) / 2
|
|
|
|
|
|
def _point_in_polygon(point: tuple[float, float], polygon: list[tuple[float, float]]) -> bool:
|
|
inside = False
|
|
x, y = point
|
|
for a, b in zip(polygon, polygon[1:]):
|
|
if (a[1] > y) != (b[1] > y):
|
|
intersection = (b[0] - a[0]) * (y - a[1]) / (b[1] - a[1]) + a[0]
|
|
if x < intersection:
|
|
inside = not inside
|
|
return inside
|
|
|
|
|
|
def _rotated_rectangle(position: tuple[float, float], size: tuple[float, float], rotation: float) -> list[tuple[float, float]]:
|
|
x, y = position
|
|
width, height = size
|
|
angle = math.radians(rotation)
|
|
cos_a, sin_a = math.cos(angle), math.sin(angle)
|
|
points = []
|
|
for dx, dy in ((-width / 2, -height / 2), (width / 2, -height / 2), (width / 2, height / 2), (-width / 2, height / 2)):
|
|
points.append((x + dx * cos_a - dy * sin_a, y + dx * sin_a + dy * cos_a))
|
|
return points + [points[0]]
|
|
|
|
|
|
def _bbox(points: Iterable[tuple[float, float]]) -> list[float]:
|
|
values = list(points)
|
|
return [min(p[0] for p in values), min(p[1] for p in values), max(p[0] for p in values), max(p[1] for p in values)]
|
|
|
|
|
|
def _bbox_overlap(a: list[float], b: list[float]) -> bool:
|
|
return a[0] < b[2] and a[2] > b[0] and a[1] < b[3] and a[3] > b[1]
|
|
|
|
|
|
def _validate_and_normalize(drawing: dict[str, Any]) -> tuple[dict[str, Any], list[dict[str, str]], dict[str, Any]]:
|
|
value = deepcopy(drawing)
|
|
arrays = [value[name] for name in ("walls", "openings", "areas", "objects", "dimensions", "notes")]
|
|
if sum(len(items) for items in arrays) > MAX_ENTITIES:
|
|
raise DrawingError("CAD2D_ENTITY_LIMIT_EXCEEDED")
|
|
identifiers = [item["id"] for items in arrays for item in items]
|
|
if len(identifiers) != len(set(identifiers)):
|
|
raise DrawingError("CAD2D_DUPLICATE_ID")
|
|
walls: dict[str, dict[str, Any]] = {}
|
|
wall_boxes: list[list[float]] = []
|
|
for wall in value["walls"]:
|
|
start, end = _point(wall["start"]), _point(wall["end"])
|
|
length = _distance(start, end)
|
|
if length <= 1e-6:
|
|
raise DrawingError("CAD2D_ZERO_LENGTH_WALL", wall["id"])
|
|
if not 50 <= float(wall["thickness"]) <= 2000:
|
|
raise DrawingError("CAD2D_WALL_THICKNESS_INVALID", wall["id"])
|
|
wall["start"], wall["end"] = list(start), list(end)
|
|
wall["length"] = round(length, 6)
|
|
walls[wall["id"]] = wall
|
|
half = float(wall["thickness"]) / 2
|
|
wall_boxes.append([min(start[0], end[0]) - half, min(start[1], end[1]) - half,
|
|
max(start[0], end[0]) + half, max(start[1], end[1]) + half])
|
|
openings_by_wall: dict[str, list[tuple[float, float, str]]] = {}
|
|
for opening in value["openings"]:
|
|
wall = walls.get(opening["wall_id"])
|
|
if wall is None:
|
|
raise DrawingError("CAD2D_WALL_REFERENCE_MISSING", opening["id"])
|
|
start = float(opening["offset"])
|
|
end = start + float(opening["width"])
|
|
if start < 0 or end > wall["length"] + 1e-6:
|
|
raise DrawingError("CAD2D_OPENING_OUTSIDE_WALL", opening["id"])
|
|
intervals = openings_by_wall.setdefault(opening["wall_id"], [])
|
|
if any(start < old_end - 1e-6 and end > old_start + 1e-6 for old_start, old_end, _ in intervals):
|
|
raise DrawingError("CAD2D_OPENINGS_OVERLAP", opening["id"])
|
|
intervals.append((start, end, opening["id"]))
|
|
area_values: dict[str, float] = {}
|
|
all_points: list[tuple[float, float]] = []
|
|
warnings: list[dict[str, str]] = []
|
|
for area in value["areas"]:
|
|
points = [_point(point) for point in area["boundary"]]
|
|
if _distance(points[0], points[-1]) > 1e-6:
|
|
raise DrawingError("CAD2D_AREA_NOT_CLOSED", area["id"])
|
|
for index, (a, b) in enumerate(zip(points, points[1:])):
|
|
for other, (c, d) in enumerate(zip(points, points[1:])):
|
|
if abs(index - other) <= 1 or {index, other} == {0, len(points) - 2}:
|
|
continue
|
|
if _segments_intersect(a, b, c, d):
|
|
raise DrawingError("CAD2D_AREA_SELF_INTERSECTION", area["id"])
|
|
size = _polygon_area(points)
|
|
if size <= 1e-6:
|
|
raise DrawingError("CAD2D_AREA_ZERO", area["id"])
|
|
if size < 500_000 or size > 1_000_000_000:
|
|
warnings.append({"code": "CAD2D_AREA_SIZE_UNUSUAL", "entity_id": area["id"], "detail": "区域面积异常，请复核 Recipe。"})
|
|
area_values[area["id"]] = round(size, 3)
|
|
area["boundary"] = [list(point) for point in points]
|
|
all_points.extend(points)
|
|
label_position = _point(area["label_position"])
|
|
all_points.append(label_position)
|
|
if not _point_in_polygon(label_position, points):
|
|
warnings.append({"code": "CAD2D_LABEL_OUTSIDE_AREA", "entity_id": area["id"], "detail": "区域标签位于边界之外。"})
|
|
object_boxes: dict[str, list[float]] = {}
|
|
for item in value["objects"]:
|
|
position = _point(item["position"])
|
|
kind = item["type"]
|
|
if kind == "custom_circle":
|
|
radius = float(item.get("radius", 0))
|
|
if radius <= 0 or "size" in item:
|
|
raise DrawingError("CAD2D_OBJECT_SIZE_INVALID", item["id"])
|
|
box = [position[0] - radius, position[1] - radius, position[0] + radius, position[1] + radius]
|
|
else:
|
|
if kind == "custom_rectangle" and "size" not in item:
|
|
raise DrawingError("CAD2D_OBJECT_SIZE_INVALID", item["id"])
|
|
size = tuple(float(v) for v in item.get("size", OBJECT_DEFAULTS.get(kind, (0, 0))))
|
|
if len(size) != 2 or min(size) <= 0 or "radius" in item:
|
|
raise DrawingError("CAD2D_OBJECT_SIZE_INVALID", item["id"])
|
|
item["size"] = list(size)
|
|
box = _bbox(_rotated_rectangle(position, size, float(item.get("rotation", 0))))
|
|
item["position"] = list(position)
|
|
object_boxes[item["id"]] = box
|
|
all_points.extend(((box[0], box[1]), (box[2], box[3])))
|
|
if any(_bbox_overlap(box, wall_box) for wall_box in wall_boxes):
|
|
warnings.append({"code": "CAD2D_OBJECT_OVERLAPS_WALL", "entity_id": item["id"], "detail": "对象与墙体包围范围相交。"})
|
|
for opening in value["openings"]:
|
|
if "door" not in opening["type"]:
|
|
continue
|
|
wall = walls[opening["wall_id"]]
|
|
a, b = _point(wall["start"]), _point(wall["end"])
|
|
ratio = float(opening["offset"]) / wall["length"]
|
|
hinge = (a[0] + (b[0] - a[0]) * ratio, a[1] + (b[1] - a[1]) * ratio)
|
|
radius = float(opening["width"])
|
|
swing_box = [hinge[0] - radius, hinge[1] - radius, hinge[0] + radius, hinge[1] + radius]
|
|
for object_id, box in object_boxes.items():
|
|
if _bbox_overlap(swing_box, box):
|
|
warnings.append({"code": "CAD2D_DOOR_SWING_OBJECT_COLLISION", "entity_id": opening["id"], "detail": f"门扇可能与对象 {object_id} 碰撞。"})
|
|
for wall in walls.values():
|
|
all_points.extend((_point(wall["start"]), _point(wall["end"])))
|
|
for note in value["notes"]:
|
|
all_points.append(_point(note["position"]))
|
|
if not all_points:
|
|
raise DrawingError("CAD2D_EMPTY_DRAWING")
|
|
bounds = _bbox(all_points)
|
|
sheet = value["sheet"]
|
|
paper = SHEET_MM[sheet["size"]]
|
|
if sheet["orientation"] == "landscape":
|
|
paper = (paper[1], paper[0])
|
|
usable = ((paper[0] - 20) * sheet["scale"], (paper[1] - 20) * sheet["scale"])
|
|
if bounds[2] - bounds[0] > usable[0] + 1e-6 or bounds[3] - bounds[1] > usable[1] + 1e-6:
|
|
raise DrawingError("CAD2D_DRAWING_OUTSIDE_SHEET")
|
|
if len(identifiers) > 700:
|
|
warnings.append({"code": "CAD2D_INFORMATION_DENSE", "entity_id": "drawing", "detail": "图纸信息密度较高，建议人工检查可读性。"})
|
|
for dimension in value["dimensions"]:
|
|
source_points = dimension["points"] if "points" in dimension else [dimension["start"], dimension["end"]]
|
|
points = [_point(p) for p in source_points]
|
|
offset = float(dimension["offset"])
|
|
all_points.extend(points)
|
|
if dimension["type"] == "vertical":
|
|
all_points.extend((point[0] + offset, point[1]) for point in points)
|
|
elif dimension["type"] in {"horizontal", "chain", "overall"}:
|
|
all_points.extend((point[0], point[1] + offset) for point in points)
|
|
else:
|
|
all_points.extend((point[0] - abs(offset), point[1] - abs(offset)) for point in points)
|
|
all_points.extend((point[0] + abs(offset), point[1] + abs(offset)) for point in points)
|
|
if len(points) > 12:
|
|
warnings.append({"code": "CAD2D_DIMENSION_TEXT_MAY_OVERLAP", "entity_id": dimension["id"], "detail": "连续尺寸较密，文字可能重叠。"})
|
|
return value, warnings, {"bbox": [round(v, 3) for v in bounds], "areas_mm2": area_values}
|
|
|
|
|
|
def _add(entities: list[dict[str, Any]], kind: str, layer: str, **values: Any) -> None:
|
|
entities.append({"kind": kind, "layer": layer, **values})
|
|
|
|
|
|
def _entities(drawing: dict[str, Any], metrics: dict[str, Any]) -> list[dict[str, Any]]:
|
|
entities: list[dict[str, Any]] = []
|
|
for wall in drawing["walls"]:
|
|
layer = "A-WALL-EXT" if wall["kind"] == "exterior" else "A-WALL-INT"
|
|
a, b = _point(wall["start"]), _point(wall["end"])
|
|
length, half = wall["length"], float(wall["thickness"]) / 2
|
|
nx, ny = -(b[1] - a[1]) / length * half, (b[0] - a[0]) / length * half
|
|
_add(entities, "polyline", layer, points=[(a[0]+nx,a[1]+ny),(b[0]+nx,b[1]+ny),(b[0]-nx,b[1]-ny),(a[0]-nx,a[1]-ny)], closed=True)
|
|
for opening in drawing["openings"]:
|
|
wall = next(item for item in drawing["walls"] if item["id"] == opening["wall_id"])
|
|
a, b = _point(wall["start"]), _point(wall["end"])
|
|
ux, uy = (b[0]-a[0])/wall["length"], (b[1]-a[1])/wall["length"]
|
|
start = (a[0]+ux*opening["offset"], a[1]+uy*opening["offset"])
|
|
end = (start[0]+ux*opening["width"], start[1]+uy*opening["width"])
|
|
layer = "A-DOOR" if "door" in opening["type"] else "A-WINDOW"
|
|
_add(entities, "line", layer, start=start, end=end)
|
|
if "door" in opening["type"]:
|
|
_add(entities, "arc", layer, center=start, radius=opening["width"], start_angle=0, end_angle=90)
|
|
for area in drawing["areas"]:
|
|
_add(entities, "polyline", "A-AREA", points=[_point(p) for p in area["boundary"][:-1]], closed=True)
|
|
label = area["name"] + (f" {metrics['areas_mm2'][area['id']]/1_000_000:.2f} m²" if area["show_area"] else "")
|
|
_add(entities, "text", "A-TEXT", point=_point(area["label_position"]), text=label, height=250)
|
|
for item in drawing["objects"]:
|
|
layer = "A-SAFE" if item["type"] in {"safety_exit", "extinguisher"} else "A-FURN" if item["type"] in {"desk", "chair", "cabinet"} else "A-EQUIP"
|
|
position = _point(item["position"])
|
|
if item["type"] == "custom_circle":
|
|
_add(entities, "circle", layer, center=position, radius=item["radius"])
|
|
else:
|
|
_add(entities, "polyline", layer, points=_rotated_rectangle(position, tuple(item["size"]), float(item.get("rotation", 0)))[:-1], closed=True)
|
|
if item.get("label"):
|
|
_add(entities, "text", "A-TEXT", point=position, text=item["label"], height=220)
|
|
for dimension in drawing["dimensions"]:
|
|
source_points = dimension["points"] if "points" in dimension else [dimension["start"], dimension["end"]]
|
|
points = [_point(p) for p in source_points]
|
|
pairs = list(zip(points, points[1:]))
|
|
if dimension["type"] == "overall":
|
|
pairs = [(points[0], points[-1])]
|
|
for a, b in pairs:
|
|
offset = float(dimension["offset"])
|
|
if dimension["type"] == "vertical":
|
|
aa, bb = (a[0]+offset,a[1]), (b[0]+offset,b[1]); distance = abs(b[1]-a[1])
|
|
elif dimension["type"] == "horizontal":
|
|
aa, bb = (a[0],a[1]+offset), (b[0],b[1]+offset); distance = abs(b[0]-a[0])
|
|
else:
|
|
length = _distance(a, b); nx, ny = (-(b[1]-a[1])/length, (b[0]-a[0])/length)
|
|
aa, bb = (a[0]+nx*offset,a[1]+ny*offset), (b[0]+nx*offset,b[1]+ny*offset); distance = length
|
|
_add(entities, "line", "A-DIMS", start=aa, end=bb)
|
|
_add(entities, "text", "A-DIMS", point=((aa[0]+bb[0])/2,(aa[1]+bb[1])/2), text=f"{distance:.0f}", height=180)
|
|
for note in drawing["notes"]:
|
|
_add(entities, "text", "A-TEXT", point=_point(note["position"]), text=note["text"], height=220, rotation=float(note.get("rotation", 0)))
|
|
bounds = metrics["bbox"]
|
|
margin = max(500.0, max(bounds[2]-bounds[0], bounds[3]-bounds[1]) * 0.03)
|
|
_add(entities, "polyline", "A-FRAME", points=[(bounds[0]-margin,bounds[1]-margin),(bounds[2]+margin,bounds[1]-margin),(bounds[2]+margin,bounds[3]+margin),(bounds[0]-margin,bounds[3]+margin)], closed=True)
|
|
return entities
|
|
|
|
|
|
def _dxf_pair(code: int, value: Any) -> str:
|
|
if isinstance(value, float):
|
|
value = format(value, ".9f").rstrip("0").rstrip(".") or "0"
|
|
return f"{code}\r\n{value}\r\n"
|
|
|
|
|
|
def _write_dxf(path: Path, entities: list[dict[str, Any]]) -> None:
|
|
parts = [_dxf_pair(0,"SECTION"),_dxf_pair(2,"HEADER"),_dxf_pair(9,"$ACADVER"),_dxf_pair(1,"AC1027"),_dxf_pair(9,"$INSUNITS"),_dxf_pair(70,4),_dxf_pair(0,"ENDSEC"),
|
|
_dxf_pair(0,"SECTION"),_dxf_pair(2,"TABLES"),_dxf_pair(0,"TABLE"),_dxf_pair(2,"LAYER"),_dxf_pair(70,len(LAYERS))]
|
|
colors = [1, 8, 2, 4, 3, 6, 5, 1, 7, 7, 9, 7]
|
|
for layer, color in zip(LAYERS, colors):
|
|
parts += [_dxf_pair(0,"LAYER"),_dxf_pair(2,layer),_dxf_pair(70,0),_dxf_pair(62,color),_dxf_pair(6,"CONTINUOUS")]
|
|
parts += [_dxf_pair(0,"ENDTAB"),_dxf_pair(0,"ENDSEC"),_dxf_pair(0,"SECTION"),_dxf_pair(2,"ENTITIES")]
|
|
for entity in entities:
|
|
kind, layer = entity["kind"], entity["layer"]
|
|
if kind == "line":
|
|
a,b=entity["start"],entity["end"]; parts += [_dxf_pair(0,"LINE"),_dxf_pair(8,layer),_dxf_pair(10,a[0]),_dxf_pair(20,a[1]),_dxf_pair(11,b[0]),_dxf_pair(21,b[1])]
|
|
elif kind == "polyline":
|
|
parts += [_dxf_pair(0,"LWPOLYLINE"),_dxf_pair(8,layer),_dxf_pair(90,len(entity["points"])),_dxf_pair(70,1 if entity.get("closed") else 0)]
|
|
for point in entity["points"]: parts += [_dxf_pair(10,point[0]),_dxf_pair(20,point[1])]
|
|
elif kind == "circle":
|
|
c=entity["center"]; parts += [_dxf_pair(0,"CIRCLE"),_dxf_pair(8,layer),_dxf_pair(10,c[0]),_dxf_pair(20,c[1]),_dxf_pair(40,entity["radius"])]
|
|
elif kind == "arc":
|
|
c=entity["center"]; parts += [_dxf_pair(0,"ARC"),_dxf_pair(8,layer),_dxf_pair(10,c[0]),_dxf_pair(20,c[1]),_dxf_pair(40,entity["radius"]),_dxf_pair(50,entity["start_angle"]),_dxf_pair(51,entity["end_angle"])]
|
|
else:
|
|
p=entity["point"]; parts += [_dxf_pair(0,"TEXT"),_dxf_pair(8,layer),_dxf_pair(10,p[0]),_dxf_pair(20,p[1]),_dxf_pair(40,entity["height"]),_dxf_pair(1,entity["text"]),_dxf_pair(50,entity.get("rotation",0))]
|
|
parts += [_dxf_pair(0,"ENDSEC"),_dxf_pair(0,"EOF")]
|
|
_atomic_bytes(path, "".join(parts).encode("utf-8"))
|
|
|
|
|
|
def _reopen_dxf(path: Path) -> dict[str, Any]:
|
|
lines = path.read_text(encoding="utf-8").splitlines()
|
|
if len(lines) % 2 or "AC1027" not in lines or lines[-1] != "EOF":
|
|
raise DrawingError("CAD2D_DXF_REOPEN_FAILED")
|
|
pairs = [(lines[i].strip(), lines[i+1]) for i in range(0, len(lines), 2)]
|
|
layers = {value for index,(code,value) in enumerate(pairs) if code == "0" and value == "LAYER" for next_code,value in pairs[index+1:index+4] if next_code == "2"}
|
|
entity_names = {"LINE","LWPOLYLINE","CIRCLE","ARC","TEXT"}
|
|
kinds = Counter(value for code,value in pairs if code == "0" and value in entity_names)
|
|
if layers != set(LAYERS) or not kinds:
|
|
raise DrawingError("CAD2D_DXF_STRUCTURE_INVALID")
|
|
points: list[tuple[float, float]] = []
|
|
for index, (code, name) in enumerate(pairs):
|
|
if code != "0" or name not in entity_names:
|
|
continue
|
|
block: list[tuple[str, str]] = []
|
|
for pair in pairs[index + 1:]:
|
|
if pair[0] == "0":
|
|
break
|
|
block.append(pair)
|
|
def values(group: str) -> list[float]:
|
|
return [float(value) for item_code, value in block if item_code == group]
|
|
xs, ys = values("10"), values("20")
|
|
if name == "LINE":
|
|
xs += values("11"); ys += values("21")
|
|
if name in {"CIRCLE", "ARC"} and xs and ys:
|
|
radius = values("40")[0]
|
|
points.extend(((xs[0] - radius, ys[0] - radius), (xs[0] + radius, ys[0] + radius)))
|
|
else:
|
|
points.extend(zip(xs, ys))
|
|
if not points:
|
|
raise DrawingError("CAD2D_DXF_BOUNDS_INVALID")
|
|
return {"version": "R2013", "unit": "mm", "layers": sorted(layers), "entity_types": dict(sorted(kinds.items())), "entity_count": sum(kinds.values()), "bbox": [round(value, 3) for value in _bbox(points)]}
|
|
|
|
|
|
def _entity_bbox(entities: list[dict[str, Any]]) -> list[float]:
|
|
points: list[tuple[float, float]] = []
|
|
for entity in entities:
|
|
if entity["kind"] == "line":
|
|
points.extend((entity["start"], entity["end"]))
|
|
elif entity["kind"] == "polyline":
|
|
points.extend(entity["points"])
|
|
elif entity["kind"] in {"circle", "arc"}:
|
|
x, y = entity["center"]; radius = float(entity["radius"])
|
|
points.extend(((x - radius, y - radius), (x + radius, y + radius)))
|
|
else:
|
|
points.append(entity["point"])
|
|
return [round(value, 3) for value in _bbox(points)]
|
|
|
|
|
|
def _xml(value: str) -> str:
|
|
return value.replace("&","&").replace("<","<").replace(">",">").replace('"',""")
|
|
|
|
|
|
def _svg(path: Path, entities: list[dict[str, Any]], bounds: list[float]) -> None:
|
|
width,height=max(1,bounds[2]-bounds[0]),max(1,bounds[3]-bounds[1]); stroke=max(width,height)/1200
|
|
parts=[f'<svg xmlns="http://www.w3.org/2000/svg" viewBox="{bounds[0]} {-bounds[3]} {width} {height}">', '<rect width="100%" height="100%" fill="white"/>', '<g fill="none" stroke="#111" stroke-width="%s">'%stroke]
|
|
for e in entities:
|
|
if e["kind"]=="line": parts.append(f'<line x1="{e["start"][0]}" y1="{-e["start"][1]}" x2="{e["end"][0]}" y2="{-e["end"][1]}"/>')
|
|
elif e["kind"]=="polyline": parts.append('<polygon points="%s"/>'%' '.join(f'{p[0]},{-p[1]}' for p in e["points"]))
|
|
elif e["kind"]=="circle": parts.append(f'<circle cx="{e["center"][0]}" cy="{-e["center"][1]}" r="{e["radius"]}"/>')
|
|
elif e["kind"]=="arc": parts.append(f'<circle cx="{e["center"][0]}" cy="{-e["center"][1]}" r="{e["radius"]}" stroke-dasharray="{e["radius"]*1.57} {e["radius"]*4.71}"/>')
|
|
parts.append('</g><g font-family="Microsoft YaHei,SimHei,sans-serif" fill="#111" text-anchor="middle">')
|
|
for e in entities:
|
|
if e["kind"]=="text": parts.append(f'<text x="{e["point"][0]}" y="{-e["point"][1]}" font-size="{e["height"]}">{_xml(str(e["text"]))}</text>')
|
|
parts.append('</g></svg>')
|
|
_atomic_bytes(path, ("\n".join(parts)+"\n").encode())
|
|
|
|
|
|
def _font_path() -> Path:
|
|
root = Path(os.environ.get("WINDIR", r"C:\Windows")) / "Fonts"
|
|
for name in ("msyh.ttc", "simhei.ttf"):
|
|
candidate = root / name
|
|
if candidate.is_file(): return candidate
|
|
raise DrawingError("CAD2D_CHINESE_FONT_UNAVAILABLE")
|
|
|
|
|
|
def _raster(path: Path, entities: list[dict[str, Any]], bounds: list[float]) -> None:
|
|
from PIL import Image, ImageDraw, ImageFont
|
|
width,height=1600,1100; margin=60; sx=(width-2*margin)/max(1,bounds[2]-bounds[0]); sy=(height-2*margin)/max(1,bounds[3]-bounds[1]); scale=min(sx,sy)
|
|
def pixel(point: tuple[float,float]) -> tuple[int,int]: return (round(margin+(point[0]-bounds[0])*scale),round(height-margin-(point[1]-bounds[1])*scale))
|
|
image=Image.new("RGB",(width,height),"white"); draw=ImageDraw.Draw(image); font=ImageFont.truetype(str(_font_path()),18)
|
|
for e in entities:
|
|
if e["kind"]=="line": draw.line([pixel(e["start"]),pixel(e["end"])],fill="#111111",width=2)
|
|
elif e["kind"]=="polyline":
|
|
points=[pixel(p) for p in e["points"]]; draw.line(points+[points[0]],fill="#111111",width=3 if "WALL" in e["layer"] else 1)
|
|
elif e["kind"]=="circle":
|
|
c=pixel(e["center"]); r=max(1,round(e["radius"]*scale)); draw.ellipse((c[0]-r,c[1]-r,c[0]+r,c[1]+r),outline="#111111",width=2)
|
|
elif e["kind"]=="arc":
|
|
c=pixel(e["center"]); r=max(1,round(e["radius"]*scale)); draw.arc((c[0]-r,c[1]-r,c[0]+r,c[1]+r),180,270,fill="#333333",width=2)
|
|
else:
|
|
p=pixel(e["point"]); draw.text(p,str(e["text"]),font=font,fill="#111111",anchor="mm")
|
|
image.save(path, format="PNG", optimize=False, compress_level=9)
|
|
|
|
|
|
def _pdf_from_png(png: Path, pdf: Path) -> None:
|
|
import time
|
|
from PIL import Image
|
|
with Image.open(png) as image:
|
|
fixed_time = time.gmtime(0)
|
|
image.convert("RGB").save(
|
|
pdf, format="PDF", resolution=150.0, title="zcbot 2D CAD drawing",
|
|
creationDate=fixed_time, modDate=fixed_time,
|
|
)
|
|
|
|
|
|
def _record(job_dir: Path) -> dict[str, Any]:
|
|
raw = (job_dir / "request" / "request.json").read_bytes()
|
|
if len(raw) > MAX_REQUEST_BYTES + 131_072: raise DrawingError("CAD2D_REQUEST_TOO_LARGE")
|
|
value = json.loads(raw)
|
|
if not isinstance(value, dict) or not isinstance(value.get("request"), dict): raise DrawingError("CAD2D_JOB_REQUEST_INVALID")
|
|
return value
|
|
|
|
|
|
def run(job_dir: Path) -> list[dict[str, Any]]:
|
|
job_dir=job_dir.resolve(strict=True); record=_record(job_dir); request=record["request"]
|
|
if request.get("inputs"): raise DrawingError("CAD2D_INPUTS_MUST_BE_EMPTY")
|
|
if request.get("outputs"): raise DrawingError("CAD2D_OUTPUTS_MUST_BE_EMPTY")
|
|
drawing,warnings,metrics=_validate_and_normalize(request["operation"]["drawing"])
|
|
output=(job_dir/"output").resolve(); output.mkdir(exist_ok=True)
|
|
if not output.is_relative_to(job_dir): raise DrawingError("CAD2D_OUTPUT_PATH_ESCAPES_JOB")
|
|
metadata=output/".meta"; metadata.mkdir(exist_ok=True)
|
|
entities=_entities(drawing,metrics)
|
|
dxf=output/"drawing.dxf"; svg=output/"drawing.svg"; png=output/"drawing.png"; pdf=output/"drawing.pdf"
|
|
_write_dxf(dxf,entities); reopened=_reopen_dxf(dxf)
|
|
if reopened["entity_count"] != len(entities): raise DrawingError("CAD2D_ENTITY_COUNT_MISMATCH")
|
|
if reopened["bbox"] != _entity_bbox(entities): raise DrawingError("CAD2D_DXF_BOUNDS_MISMATCH")
|
|
bounds=metrics["bbox"]; padding=max(500.0,max(bounds[2]-bounds[0],bounds[3]-bounds[1])*.05); render_bounds=[bounds[0]-padding,bounds[1]-padding,bounds[2]+padding,bounds[3]+padding]
|
|
_svg(svg,entities,render_bounds); _raster(png,entities,render_bounds); _pdf_from_png(png,pdf)
|
|
recipe_path=output/"drawing-recipe.json"; report_path=output/"validation-report.json"
|
|
_atomic_json(recipe_path,{"schema_version":1,"drawing":drawing})
|
|
observed_layer_counts=Counter(e["layer"] for e in entities)
|
|
layer_counts={layer: observed_layer_counts.get(layer, 0) for layer in LAYERS}
|
|
report={"passed":True,"errors":[],"warnings":warnings,"checks":{"dxf_reopened":True,"version":"R2013","unit":"mm","layers":list(LAYERS),"entity_count":len(entities),"bbox":bounds,"dxf_bbox":reopened["bbox"],"areas_mm2":metrics["areas_mm2"]}}
|
|
_atomic_json(report_path,report)
|
|
summaries=[]
|
|
for path in (dxf,svg,png,pdf,recipe_path,report_path): summaries.append({"filename":path.name,"size_bytes":path.stat().st_size,"sha256":_sha256(path)})
|
|
manifest={"adapter_version":ADAPTER_VERSION,"recipe_version":1,"drawing_type":drawing["type"],"unit":"mm","sheet":drawing["sheet"],"bbox":bounds,"areas_mm2":metrics["areas_mm2"],"layers":list(LAYERS),"layer_entity_counts":layer_counts,"entity_count":len(entities),"warnings":warnings,"artifacts":summaries}
|
|
manifest_path=output/"drawing-manifest.json"; _atomic_json(manifest_path,manifest)
|
|
provenance_path=metadata/"provenance.json"; _atomic_json(provenance_path,{"adapter":"cad.drawing.author@v1","adapter_version":ADAPTER_VERSION,"request_digest":record.get("request_digest"),"recipe_sha256":_sha256(recipe_path),"manifest_sha256":_sha256(manifest_path),"renderer":{"font":_font_path().name,"png":"Pillow","svg":"builtin","pdf":"Pillow","dxf":"builtin-r2013"}})
|
|
paths=[dxf,svg,png,pdf,recipe_path,manifest_path,report_path,provenance_path]
|
|
if any(not p.is_file() or p.stat().st_size==0 for p in paths): raise DrawingError("CAD2D_REQUIRED_OUTPUT_MISSING")
|
|
return [_artifact(path) for path in paths]
|
|
|
|
|
|
def probe() -> int:
|
|
payload={"adapter_version":ADAPTER_VERSION,"software":"二维 CAD 工程制图","software_version":None,"health":"unavailable","detail":""}
|
|
try:
|
|
from PIL import __version__ as pillow_version
|
|
if pillow_version != PILLOW_VERSION:
|
|
raise DrawingError("CAD2D_PILLOW_VERSION_UNSUPPORTED", f"expected {PILLOW_VERSION}, found {pillow_version}")
|
|
font=_font_path(); payload.update({"software_version":f"Pillow {pillow_version}","health":"ready","detail":f"Runtime-only renderer ready; Chinese font: {font.name}; DXF R2013 reopen validator ready"})
|
|
except (ImportError,OSError,DrawingError) as exc: payload["detail"]=str(exc)[:500]
|
|
print(json.dumps(payload,ensure_ascii=False)); return 0
|
|
|
|
|
|
def main() -> int:
|
|
if sys.argv[1:]==["--probe"]: return probe()
|
|
if len(sys.argv)!=2: print("[ERR] Usage: worker.py <job-directory>",file=sys.stderr); return 2
|
|
job_dir=Path(sys.argv[1]); record:dict[str,Any]={}
|
|
try:
|
|
record=_record(job_dir); artifacts=run(job_dir); _atomic_json(job_dir/"artifacts.json",artifacts); _atomic_json(job_dir/"terminal.json",_terminal(record,"succeeded",artifacts)); print("[OK] 2D CAD drawing authored and verified."); return 0
|
|
except Exception as exc:
|
|
code=exc.code if isinstance(exc,DrawingError) else "CAD2D_ADAPTER_FAILED"
|
|
try: _atomic_json(job_dir/"terminal.json",_terminal(record,"failed",[],code,str(exc)))
|
|
except OSError: pass
|
|
print(f"[ERR] {code}: {exc}",file=sys.stderr); return 1
|
|
|
|
|
|
if __name__ == "__main__": raise SystemExit(main())
|