diff --git a/backends/com_backend.py b/backends/com_backend.py index 8fd3ebd..b0aa28d 100644 --- a/backends/com_backend.py +++ b/backends/com_backend.py @@ -616,6 +616,12 @@ def _entity_info(entity) -> EntityInfo: props["x_scale"] = entity.XScaleFactor props["y_scale"] = entity.YScaleFactor props["rotation_deg"] = rad2deg(entity.Rotation) + # ezdxf parity: a block reference whose Normal points -Z is drawn + # on the mirror image (x -> -x) about its insertion point. + try: + props["mirrored"] = float(entity.Normal[2]) < 0 + except Exception: + props["mirrored"] = False geometry_bbox = _com_geometry_bbox(entity) if geometry_bbox is not None: props["geometry_bbox"] = geometry_bbox diff --git a/backends/ezdxf_backend.py b/backends/ezdxf_backend.py index 3254222..465d1d8 100644 --- a/backends/ezdxf_backend.py +++ b/backends/ezdxf_backend.py @@ -845,6 +845,11 @@ def _entity_info_dxf(ent) -> EntityInfo: props["x_scale"] = ent.dxf.xscale if ent.dxf.hasattr("xscale") else 1.0 props["y_scale"] = ent.dxf.yscale if ent.dxf.hasattr("yscale") else 1.0 props["rotation_deg"] = ent.dxf.get("rotation", 0.0) + # MIRROR3D and foreign DXFs store a mirror as extrusion -Z (the + # stored x unnegated); `insertion` above is already WCS, but the + # block's content is drawn on the mirror image about it, so a + # reader that places catalogue ports needs to know. + props["mirrored"] = float(extrusion[2]) < 0 # The drawn symbol alone. ``bounding_box`` below takes the ATTRIBs # with it, so a TAG above a valve pushes the box past the body and # a line ending on the body's real edge reads as *inside* it. diff --git a/engineering/pid/drawlines.py b/engineering/pid/drawlines.py index 5586565..6761a8b 100644 --- a/engineering/pid/drawlines.py +++ b/engineering/pid/drawlines.py @@ -65,8 +65,10 @@ async def resolve_endpoint(backend: AutoCADBackend, ref: dict) -> dict: ins = info.properties["insertion"] x_scale = float(info.properties.get("x_scale", 1.0)) # The full INSERT transform: a mirrored symbol (``entity_mirror`` writes - # ``y_scale = -1``) has its ports on the mirrored geometry, not on the - # unmirrored one; a stretched symbol is refused by ``transform_port``. + # ``y_scale = -1``; MIRROR3D and foreign DXFs store extrusion -Z, which + # the engines report as ``mirrored``) has its ports on the mirrored + # geometry, not on the unmirrored one; a stretched symbol is refused + # by ``transform_port``. try: world = transform_port( Port(port_name, float(lx), float(ly), ldir, kind, float(radius)), @@ -75,6 +77,7 @@ async def resolve_endpoint(backend: AutoCADBackend, ref: dict) -> dict: float(info.properties.get("rotation_deg", 0.0)), x_scale, float(info.properties.get("y_scale", x_scale)), + bool(info.properties.get("mirrored", False)), ) except ValueError as exc: raise ValueError(f"{handle}: {exc}") from exc diff --git a/engineering/pid/graph.py b/engineering/pid/graph.py index a1f76a8..1e52a28 100644 --- a/engineering/pid/graph.py +++ b/engineering/pid/graph.py @@ -66,6 +66,8 @@ class _Node: y: float rotation_deg: float scale: float + y_scale: float + mirrored: bool layer: str space: str bbox: tuple[float, float, float, float] | None @@ -75,6 +77,7 @@ class _Node: ports: dict = field(default_factory=dict) tag: str | None = None tag_source: str | None = None + notes: list[str] = field(default_factory=list) def to_dict(self) -> dict: parsed = None @@ -93,12 +96,15 @@ class _Node: "y": self.y, "rotation_deg": self.rotation_deg, "scale": self.scale, + "y_scale": self.y_scale, + "mirrored": self.mirrored, "layer": self.layer, "space": self.space, "ports": self.ports, "description": self.attributes.get("DESC") or None, "source": self.source, "confidence": self.confidence, + "notes": list(self.notes), } @@ -164,24 +170,30 @@ class _Edge: class _Grid: - """Uniform grid for nearest-within-tolerance queries.""" + """Uniform grid for nearest-within-tolerance queries, one per space. + + Every layout is its own sheet: a line end in SHEET-2 may only attach to + what is drawn in SHEET-2. Keying the cells by space is what keeps a + ``scope="all"`` read from joining two sheets that happen to share + coordinates — every layout starts at the origin. + """ def __init__(self, cell: float): self.cell = cell - self.cells: dict[tuple[int, int], list] = {} + self.cells: dict[tuple[str, int, int], list] = {} - def _key(self, x: float, y: float) -> tuple[int, int]: - return (int(math.floor(x / self.cell)), int(math.floor(y / self.cell))) + def _key(self, space: str, x: float, y: float) -> tuple[str, int, int]: + return (space, int(math.floor(x / self.cell)), int(math.floor(y / self.cell))) - def add(self, x: float, y: float, item) -> None: - self.cells.setdefault(self._key(x, y), []).append((x, y, item)) + def add(self, space: str, x: float, y: float, item) -> None: + self.cells.setdefault(self._key(space, x, y), []).append((x, y, item)) - def near(self, x: float, y: float, radius: float): - kx, ky = self._key(x, y) + def near(self, space: str, x: float, y: float, radius: float): + _, kx, ky = self._key(space, x, y) reach = int(math.ceil(radius / self.cell)) for dx in range(-reach, reach + 1): for dy in range(-reach, reach + 1): - for px, py, item in self.cells.get((kx + dx, ky + dy), []): + for px, py, item in self.cells.get((space, kx + dx, ky + dy), []): d = math.hypot(px - x, py - y) if d <= radius: yield d, item @@ -368,6 +380,7 @@ async def build_graph( rotation = float(info.properties.get("rotation_deg", 0.0)) scale = float(info.properties.get("x_scale", 1.0)) y_scale = float(info.properties.get("y_scale", scale)) + mirrored = bool(info.properties.get("mirrored", False)) base = dict( id=info.handle, block_name=name, @@ -375,6 +388,8 @@ async def build_graph( y=float(ins[1]), rotation_deg=rotation, scale=scale, + y_scale=y_scale, + mirrored=mirrored, layer=info.layer, space=space, bbox=_bbox(info), @@ -391,18 +406,24 @@ async def build_graph( confidence=CONFIDENCE[source], **base, ) + stretched = abs(abs(scale) - abs(y_scale)) > 1e-9 for pname, (lx, ly, ldir, kind, radius) in payload.get("ports", {}).items(): - # Measured from the INSERT as it is: a mirrored symbol carries - # y_scale = -1 and its ports sit on the mirrored geometry. + # Measured from the INSERT as it is (spec §9.4): a mirror is + # either y_scale = -1 (entity_mirror) or extrusion -Z + # (MIRROR3D, foreign DXFs), and the ports sit on the mirrored + # geometry either way. A stretched INSERT (someone scaled one + # axis by hand) is measured per axis, which is exact for a + # point port; only a bubble has no circle to be exact on, and + # that is said in the node's confidence — never with + # `inferred`, which spec §9.1 reserves for foreign-block ports. local = Port(pname, float(lx), float(ly), ldir, kind, float(radius)) - try: - port = transform_port(local, node.x, node.y, rotation, scale, y_scale) - except ValueError: - # A stretched INSERT (someone scaled one axis by hand): the - # reader never refuses a drawing, so it measures along the - # X factor and says so instead of raising. - port = transform_port(local, node.x, node.y, rotation, scale) - port["inferred"] = True + port = transform_port( + local, node.x, node.y, rotation, scale, y_scale, mirrored, strict=False + ) + if stretched and local.radius > 0: + node.confidence = min(node.confidence, CONFIDENCE["heuristic"]) + if "non-uniform scale" not in node.notes: + node.notes.append("non-uniform scale") node.ports[pname] = {**port, "edges": []} nodes[node.id] = node continue @@ -479,7 +500,7 @@ async def build_graph( # a bubble's line starts on its circle, not at its centre radial_ports.append((node.id, pname, port["x"], port["y"], port["radius"])) else: - grid.add(port["x"], port["y"], ("port", node.id, pname)) + grid.add(node.space, port["x"], port["y"], ("port", node.id, pname)) # 4 — resolve endpoints junctions: list[dict] = [] @@ -494,11 +515,13 @@ async def build_graph( has side effects.""" x, y = _end(edge, end) for node_id, pname, cx, cy, r in radial_ports: + if nodes[node_id].space != edge.space: + continue if abs(math.hypot(x - cx, y - cy) - r) <= tolerance: nodes[node_id].ports[pname]["edges"].append(edge.id) return {"node": node_id, "port": pname} best = None - for d, item in grid.near(x, y, tolerance): + for d, item in grid.near(edge.space, x, y, tolerance): if item[0] == "port" and (best is None or d < best[0]): best = (d, item) if best: @@ -512,6 +535,7 @@ async def build_graph( n for n in list(nodes.values()) + list(candidates.values()) if n.source in ("heuristic", "inferred") + and n.space == edge.space and n.bbox and _on_bbox_boundary((x, y), n.bbox, tolerance) ), @@ -532,34 +556,40 @@ async def build_graph( "inferred": True, "edges": [edge.id], } - grid.add(x, y, ("port", node.id, pname)) + grid.add(node.space, x, y, ("port", node.id, pname)) return {"node": node.id, "port": pname} return None - def _junction_at(x, y, edge_id, other_id) -> dict: + def _junction_at(space, x, y, edge_id, other_id) -> dict: for j in junctions: - if math.hypot(j["x"] - x, j["y"] - y) <= tolerance: + if j["space"] == space and math.hypot(j["x"] - x, j["y"] - y) <= tolerance: for e in (edge_id, other_id): if e not in j["edges"]: j["edges"].append(e) return {"junction": j["id"]} - j = {"id": f"J{len(junctions) + 1}", "x": x, "y": y, "edges": [edge_id, other_id]} + j = { + "id": f"J{len(junctions) + 1}", + "x": x, + "y": y, + "edges": [edge_id, other_id], + "space": space, + } junctions.append(j) return {"junction": j["id"]} def _junction_or_dangling(edge: _Edge, end: str) -> dict | None: x, y = _end(edge, end) - for _d, item in grid.near(x, y, tolerance): + for _d, item in grid.near(edge.space, x, y, tolerance): if item[0] == "vertex" and item[1] != edge.id: - return _junction_at(x, y, edge.id, item[1]) + return _junction_at(edge.space, x, y, edge.id, item[1]) for other in edges.values(): - if other.id == edge.id: + if other.id == edge.id or other.space != edge.space: continue for a, b, bulge in other.segments(): if _point_segment_distance((x, y), a, b, bulge) <= tolerance: - return _junction_at(x, y, edge.id, other.id) + return _junction_at(edge.space, x, y, edge.id, other.id) nearest = None - for d, item in grid.near(x, y, 10.0 * tolerance): + for d, item in grid.near(edge.space, x, y, 10.0 * tolerance): if item[0] == "port" and (nearest is None or d < nearest[0]): nearest = (d, item) dangling.append( @@ -594,7 +624,7 @@ async def build_graph( # 4b — junctions and dangling ends over the edges that are in the graph for edge in edges.values(): for v in edge.vertices: - grid.add(v[0], v[1], ("vertex", edge.id)) + grid.add(edge.space, v[0], v[1], ("vertex", edge.id)) for edge in edges.values(): if edge.from_ref is None: edge.from_ref = _junction_or_dangling(edge, "from") @@ -604,13 +634,29 @@ async def build_graph( # 5 — tags and line numbers from attributes or nearby text texts = [(info, space) for info, space in entities if info.type in TEXT_TYPES] - def _text_near(x, y, radius, pattern=None): + def _text_near(space, x, y, radius, pattern=None, above=None, exclude=None): + """Nearest text (by insertion) within ``radius`` of (x, y) in ``space``. + + ``pattern`` keeps only matching content; ``exclude`` drops it. With + ``above`` (a box top), only a text whose insertion is at or above + ``above - height / 2`` counts — an equipment tag is lettered over the + symbol (spec §9.2 step 6), and a full circle around the top-centre + let a note or line label under the body win over the tag above it. + """ best = None - for info, _space in texts: + for info, text_space in texts: + if text_space != space: + continue ins = info.properties.get("insertion") content = str(info.properties.get("text", "")).strip() if not ins or not content or (pattern and not pattern.match(content)): continue + if exclude and exclude.match(content): + continue + if above is not None: + height = info.properties.get("height", info.properties.get("char_height", 0.0)) + if float(ins[1]) < above - 0.5 * float(height or 0.0): + continue d = math.hypot(float(ins[0]) - x, float(ins[1]) - y) if d <= radius and (best is None or d < best[0]): best = (d, content) @@ -628,18 +674,22 @@ async def build_graph( if node.tag is None: radius = max((p.get("radius", 0.0) for p in node.ports.values()), default=0.0) if node.kind == "instrument" and radius: - found = _text_near(node.x, node.y, radius) + found = _text_near(node.space, node.x, node.y, radius) else: cx = ((node.bbox[0] + node.bbox[2]) / 2.0) if node.bbox else node.x top = node.bbox[3] if node.bbox else node.y - found = _text_near(cx, top, label_search) + found = _text_near( + node.space, cx, top, label_search, above=top, exclude=LINE_NUMBER_RE + ) if found: node.tag, node.tag_source = found, "text" for edge in edges.values(): if edge.line_number: continue a, b = max(zip(edge.vertices, edge.vertices[1:], strict=False), key=lambda s: math.dist(*s)) - found = _text_near((a[0] + b[0]) / 2.0, (a[1] + b[1]) / 2.0, label_search, LINE_NUMBER_RE) + found = _text_near( + edge.space, (a[0] + b[0]) / 2.0, (a[1] + b[1]) / 2.0, label_search, LINE_NUMBER_RE + ) if found: edge.line_number, edge.number_source = found, "label" diff --git a/engineering/pid/symbols.py b/engineering/pid/symbols.py index da1cb2a..ed32d92 100644 --- a/engineering/pid/symbols.py +++ b/engineering/pid/symbols.py @@ -183,19 +183,32 @@ def transform_port( rotation_deg: float, scale: float, y_scale: float | None = None, + mirrored: bool = False, + strict: bool = True, ) -> dict: """Symbol-local port -> WCS, for an INSERT at (x, y) rotated by ``rotation_deg``. ``scale`` is the X factor; ``y_scale`` defaults to it (a uniform INSERT). A negative factor is a mirror — ``entity_mirror`` on an INSERT writes ``y_scale = -1`` — and is applied in full, to the point, the direction and - the radius, so the port lands where the mirrored geometry actually is. The - magnitudes must match: a stretched symbol has no circle for a radial port to - sit on, so it is refused instead of resolving to a point off the symbol. + the radius, so the port lands where the mirrored geometry actually is. + + ``mirrored`` is the other spelling of a mirror: MIRROR3D and foreign DXFs + store it as extrusion -Z, whose OCS-to-WCS map is x -> -x, y -> y about the + (WCS) insertion point. It applies *after* rotation and scale, which is the + order the INSERT's own matrix composes them in — the same order ezdxf's + ``Insert.matrix44`` reports. + + The scale magnitudes must match unless ``strict`` is False: a stretched + symbol has no circle for a radial port to sit on, so the drawing tools + refuse it instead of resolving to a point off the symbol. The reader, + which never refuses a drawing, passes ``strict=False``: a point port then + scales per axis (exact), and a radial port keeps the smaller factor's + circle — the caller says so in the node's confidence. """ sx = float(scale) sy = sx if y_scale is None else float(y_scale) - if abs(abs(sx) - abs(sy)) > 1e-9: + if strict and abs(abs(sx) - abs(sy)) > 1e-9: raise ValueError( f"non-uniform INSERT scale (x_scale={sx:g}, y_scale={sy:g}): P&ID symbols " "are placed with one scale factor (a mirror may flip its sign); re-insert " @@ -204,27 +217,34 @@ def transform_port( a = math.radians(rotation_deg) cos_a, sin_a = math.cos(a), math.sin(a) lx, ly = port.x * sx, port.y * sy - wx = x + lx * cos_a - ly * sin_a - wy = y + lx * sin_a + ly * cos_a + dx = lx * cos_a - ly * sin_a + dy = lx * sin_a + ly * cos_a + if mirrored: + dx = -dx + wx, wy = x + dx, y + dy if port.direction_deg is None: direction = None else: # Reflect first (a flipped X negates the angle about 90°, a flipped Y # negates it about 0°), then rotate — exact, so the uniform case keeps - # the same numbers it always had. + # the same numbers it always had. The frame mirror comes last, as it + # does for the point. d = float(port.direction_deg) if sx < 0: d = 180.0 - d if sy < 0: d = -d - direction = (d + rotation_deg) % 360.0 + d = d + rotation_deg + if mirrored: + d = 180.0 - d + direction = d % 360.0 return { "name": port.name, "x": wx, "y": wy, "direction_deg": direction, "kind": port.kind, - "radius": port.radius * abs(sx), + "radius": port.radius * min(abs(sx), abs(sy)), "inferred": False, } diff --git a/server.py b/server.py index 41f8c8f..86e3059 100644 --- a/server.py +++ b/server.py @@ -6566,9 +6566,11 @@ async def pid_graph( Nodes: symbol INSERTs classified by catalogue name (confidence 1.0), ACADMCP_PID payload (0.95), tag attributes / block-name keywords (0.6) or a line touching an unknown INSERT (0.3); every node reports its `source`. - Ports come from the INSERT's real rotation and scale, measured. Edges: - lines and polylines; each end resolves to a port, a junction on another - line, or `dangling` with the nearest port as a hint. `stats.confidence_min` + Ports come from the INSERT's real rotation, scale and mirror, measured + (a stretched bubble drops the node to 0.6 with a note). Edges: lines and + polylines; each end resolves to a port, a junction on another line, or + `dangling` with the nearest port as a hint — always within its own + space, so `scope="all"` never joins two sheets. `stats.confidence_min` is the number to read before trusting a foreign drawing. """ from engineering.pid.graph import build_graph diff --git a/tests/test_pid_graph.py b/tests/test_pid_graph.py index 9f84833..160160d 100644 --- a/tests/test_pid_graph.py +++ b/tests/test_pid_graph.py @@ -606,3 +606,228 @@ def test_an_approximate_geometry_bbox_is_tightened_by_the_attribute_inclusive_bo (95.0, 100.0), _bbox(info(geometry_bbox=loose, bounding_box=outer)), 0.5 ) assert not _on_bbox_boundary((95.0, 100.0), _bbox(info(geometry_bbox=loose)), 0.5) + + +# ── per-space resolution, mirrored inserts, tags above the box, per-axis scale ── + + +async def test_scope_all_resolves_each_line_in_its_own_space(backend): + """The grids, box candidates and texts were pooled across layouts, so a + line in SHEET-2 attached to the Model gate that shares its coordinates, + and a line under a Model-only pump 'connected' across sheets.""" + hv1 = (await place_symbol(backend, "gate", 0, 0, tag="HV-1"))["handle"] + await backend.layout_create("SHEET-2") + await backend.layout_set_current("SHEET-2") + hv2 = (await place_symbol(backend, "gate", 0, 0, tag="HV-2"))["handle"] + to_valve = (await backend.entity_create_line(-40, 0, -4, 0, layer="PROCESS-PIPING-MAIN")).handle + await backend.layout_set_current("Model") + pump = (await place_symbol(backend, "centrifugal_pump", 100, 100, tag="P-1"))["handle"] + await backend.layout_set_current("SHEET-2") + under_pump = ( + await backend.entity_create_line(100, 150, 100, 106, layer="PROCESS-PIPING-MAIN") + ).handle + await backend.layout_set_current("Model") + graph = await build_graph(backend, scope="all") + nodes = {n["id"]: n for n in graph["nodes"]} + edges = {e["id"]: e for e in graph["edges"]} + assert nodes[hv1]["space"] == "Model" and nodes[hv2]["space"] == "SHEET-2" + assert edges[to_valve]["space"] == "SHEET-2" + assert edges[to_valve]["to"] == {"node": hv2, "port": "in"} + assert nodes[hv2]["ports"]["in"]["edges"] == [to_valve] + assert nodes[hv1]["ports"]["in"]["edges"] == [] + assert edges[under_pump]["to"] is None + assert nodes[pump]["ports"]["discharge"]["edges"] == [] + dangling = {(d["edge"], d["end"]): d for d in graph["dangling"]} + assert set(dangling) == {(to_valve, "from"), (under_pump, "from"), (under_pump, "to")} + assert dangling[(under_pump, "to")]["nearest"] is None # the pump is on another sheet + + +async def test_scope_all_keeps_junctions_and_tags_within_a_space(backend): + """A Model line's vertex must not make a junction for a SHEET-2 line + ending on the same coordinates, and a Model tag must not label a + SHEET-2 symbol.""" + await backend.entity_create_line(0, 0, 50, 0, layer="PROCESS-PIPING-MAIN") + await backend.entity_create_text("P-9", 100, 113, 2.5) + await backend.layout_create("SHEET-2") + await backend.layout_set_current("SHEET-2") + branch = (await backend.entity_create_line(25, 0, 25, -30, layer="PROCESS-PIPING-MAIN")).handle + blk = backend._doc.blocks.new(name="CENTRIFUGAL_PUMP") + blk.add_circle((0, 0), 5) + pump = (await backend.block_insert("CENTRIFUGAL_PUMP", 100, 100)).handle + await backend.layout_set_current("Model") + graph = await build_graph(backend, scope="all") + edge = next(e for e in graph["edges"] if e["id"] == branch) + assert edge["from"] is None and graph["junctions"] == [] + node = next(n for n in graph["nodes"] if n["id"] == pump) + assert node["tag"] is None and node["space"] == "SHEET-2" + + +async def test_a_mirrored_insert_has_its_ports_on_the_mirror_image(backend): + """MIRROR3D and foreign DXFs store a mirror as extrusion -Z with the + stored insertion x unnegated. The reader took the catalogue ports on + the unmirrored image at confidence 1.0.""" + from ezdxf.math import Vec3 + + pump = (await place_symbol(backend, "centrifugal_pump", 50, 20, rotation=30.0, tag="P-1"))[ + "handle" + ] + ent = backend._doc.entitydb.get(pump) + ent.dxf.extrusion = (0, 0, -1) + ent.dxf.insert = (-50, 20, 0) # the same WCS insertion point + info = (await backend.entity_get(pump)).properties + assert info["insertion"] == pytest.approx([50.0, 20.0]) + assert info["mirrored"] is True + m = ent.matrix44() + local = {"discharge": (0.0, 6.0), "suction": (-4.0, 0.0)} + truth = {name: m.transform(Vec3(lx, ly, 0)) for name, (lx, ly) in local.items()} + graph = await build_graph(backend) + node = next(n for n in graph["nodes"] if n["id"] == pump) + assert node["mirrored"] is True and node["confidence"] == 1.0 + ports = node["ports"] + assert (ports["discharge"]["x"], ports["discharge"]["y"]) == pytest.approx( + (53.0, 25.196), abs=1e-3 + ) + assert (ports["suction"]["x"], ports["suction"]["y"]) == pytest.approx((53.464, 18.0), abs=1e-3) + for name, p in truth.items(): + assert (ports[name]["x"], ports[name]["y"]) == pytest.approx((p.x, p.y), abs=1e-9) + assert ports[name]["inferred"] is False + # 90° discharge turned to 120° reads as 60° in the mirror image + assert ports["discharge"]["direction_deg"] == pytest.approx(60.0) + assert ports["suction"]["direction_deg"] == pytest.approx((180.0 - 210.0) % 360.0) + line = ( + await backend.entity_create_line(53.0, 40.0, 53.0, 25.196, layer="PROCESS-PIPING-MAIN") + ).handle + graph = await build_graph(backend) + edge = next(e for e in graph["edges"] if e["id"] == line) + assert edge["to"] == {"node": pump, "port": "discharge"} + + +async def test_an_unmirrored_insert_reports_mirrored_false(backend): + pump = (await place_symbol(backend, "centrifugal_pump", 50, 20, tag="P-1"))["handle"] + assert (await backend.entity_get(pump)).properties["mirrored"] is False + graph = await build_graph(backend) + assert graph["nodes"][0]["mirrored"] is False and graph["nodes"][0]["notes"] == [] + + +async def test_equipment_tag_is_the_text_above_the_box_not_the_line_label_below(backend): + """Spec §9.2 step 6: the tag is the nearest text *above* the box. A full + circle around the top-centre let a line number lettered under the pump + win over the tag above it.""" + blk = backend._doc.blocks.new(name="CENTRIFUGAL_PUMP") + blk.add_circle((0, 0), 5) + pump = (await backend.block_insert("CENTRIFUGAL_PUMP", 100, 100)).handle + await backend.entity_create_text("P-101", 100, 113, 2.5) + await backend.entity_create_text('4"-P-100-CS1', 100, 99, 2.5) + graph = await build_graph(backend) + node = next(n for n in graph["nodes"] if n["id"] == pump) + assert node["tag"] == "P-101" and node["tag_source"] == "text" + + +async def test_valve_tag_is_the_text_above_the_box_not_the_note_below(backend): + blk = backend._doc.blocks.new(name="GATE_VALVE") + blk.add_lwpolyline([(-4, -2), (4, -2), (4, 2), (-4, 2)], close=True) + valve = (await backend.block_insert("GATE_VALVE", 50, 50)).handle + await backend.entity_create_text("HV-9", 50, 58, 2.5) + await backend.entity_create_text("NC", 50, 47, 2.5) + graph = await build_graph(backend) + node = next(n for n in graph["nodes"] if n["id"] == valve) + assert node["tag"] == "HV-9" and node["tag_source"] == "text" + + +async def test_a_line_number_above_the_box_is_never_an_equipment_tag(backend): + blk = backend._doc.blocks.new(name="GATE_VALVE") + blk.add_lwpolyline([(-4, -2), (4, -2), (4, 2), (-4, 2)], close=True) + valve = (await backend.block_insert("GATE_VALVE", 50, 50)).handle + await backend.entity_create_text('4"-P-100-CS1', 50, 55, 2.5) + await backend.entity_create_text("HV-9", 50, 60, 2.5) + graph = await build_graph(backend) + node = next(n for n in graph["nodes"] if n["id"] == valve) + assert node["tag"] == "HV-9" + + +async def test_a_text_flush_on_the_box_top_still_counts_as_above_it(backend): + """Insertion y >= top - 0.5 * height: a label whose baseline sits a hair + under the box top (lettered flush on the body) is still above it, while + one clearly inside the body is not.""" + blk = backend._doc.blocks.new(name="GATE_VALVE") + blk.add_lwpolyline([(-4, -2), (4, -2), (4, 2), (-4, 2)], close=True) + valve = (await backend.block_insert("GATE_VALVE", 50, 50)).handle + # top is 52, so the threshold is 50.75: HV-9 (1.56 from the top-centre) + # counts as above it, NC (1.3 away, nearer) does not + await backend.entity_create_text("HV-9", 51.0, 50.8, 2.5) + await backend.entity_create_text("NC", 50.0, 50.7, 2.5) + graph = await build_graph(backend) + node = next(n for n in graph["nodes"] if n["id"] == valve) + assert node["tag"] == "HV-9" + + +async def test_a_non_uniformly_scaled_symbol_is_measured_per_axis_not_inferred(backend): + """The reader re-ran the transform with the X factor for both axes and + marked the port `inferred` — a flag spec §9.1 reserves for foreign-block + ports. Per-axis scale is exact for a point port.""" + pump = (await place_symbol(backend, "centrifugal_pump", 0, 0, tag="P-1"))["handle"] + backend._doc.entitydb.get(pump).dxf.yscale = 2.0 + line = (await backend.entity_create_line(0, 40, 0, 12, layer="PROCESS-PIPING-MAIN")).handle + graph = await build_graph(backend) + node = next(n for n in graph["nodes"] if n["id"] == pump) + discharge = node["ports"]["discharge"] + assert (discharge["x"], discharge["y"]) == pytest.approx((0.0, 12.0)) + assert discharge["inferred"] is False and discharge["direction_deg"] == pytest.approx(90.0) + assert node["confidence"] == 1.0 and node["notes"] == [] + assert node["scale"] == pytest.approx(1.0) and node["y_scale"] == pytest.approx(2.0) + suction = node["ports"]["suction"] + assert (suction["x"], suction["y"]) == pytest.approx((-4.0, 0.0)) + edge = next(e for e in graph["edges"] if e["id"] == line) + assert edge["to"] == {"node": pump, "port": "discharge"} + assert discharge["edges"] == [line] + + +async def test_a_stretched_bubble_keeps_the_smaller_radius_and_says_so(backend): + """A stretched instrument bubble is an ellipse; no circle is exact, so the + radial port keeps min(sx, sy)·r and the node's confidence drops to 0.6 + with a note — never `inferred`, which is for foreign-block ports.""" + fic = ( + await place_symbol(backend, "instrument", 0, 0, tag="FIC-1", type="dcs", location="primary") + )["handle"] + graph = await build_graph(backend) + radius = graph["nodes"][0]["ports"]["signal"]["radius"] + assert radius > 0 + backend._doc.entitydb.get(fic).dxf.xscale = 2.0 + graph = await build_graph(backend) + node = graph["nodes"][0] + port = node["ports"]["signal"] + assert port["radius"] == pytest.approx(radius) and port["inferred"] is False + assert node["confidence"] == 0.6 and node["notes"] == ["non-uniform scale"] + assert node["source"] == "catalog" + assert graph["stats"]["confidence_min"] == 0.6 + + +def test_com_block_reference_reports_mirrored_from_its_normal(): + from types import SimpleNamespace + + from backends.com_backend import _entity_info + + def refuse(_name): + raise RuntimeError("no block table in this fake") + + def blockref(**extra): + return SimpleNamespace( + ObjectName="AcDbBlockReference", + Name="PID_X", + InsertionPoint=(50.0, 20.0, 0.0), + XScaleFactor=1.0, + YScaleFactor=1.0, + Rotation=0.0, + Handle="B1", + Layer="0", + Color=256, + Linetype="ByLayer", + Visible=True, + Document=SimpleNamespace(Blocks=SimpleNamespace(Item=refuse)), + GetBoundingBox=lambda: ((45.0, 15.0, 0.0), (55.0, 25.0, 0.0)), + **extra, + ) + + assert _entity_info(blockref(Normal=(0.0, 0.0, -1.0))).properties["mirrored"] is True + assert _entity_info(blockref(Normal=(0.0, 0.0, 1.0))).properties["mirrored"] is False + assert _entity_info(blockref()).properties["mirrored"] is False diff --git a/tests/test_pid_line_draw.py b/tests/test_pid_line_draw.py index 4e82522..7401307 100644 --- a/tests/test_pid_line_draw.py +++ b/tests/test_pid_line_draw.py @@ -171,6 +171,33 @@ async def test_mirrored_symbol_ports_follow_the_mirrored_geometry(backend): assert (ep["x"], ep["y"], ep["direction_deg"]) == (100.0, 194.0, 270.0) +async def test_an_extrusion_mirrored_symbol_resolves_its_port_on_the_mirror_image(backend): + """MIRROR3D (and foreign DXFs) store a mirror as extrusion -Z rather than + a negative scale; the port is where the mirrored geometry is.""" + from ezdxf.math import Vec3 + + pump = await place_symbol(backend, "centrifugal_pump", 50, 20, rotation=30.0, tag="P-1") + ent = backend._doc.entitydb.get(pump["handle"]) + ent.dxf.extrusion = (0, 0, -1) + ent.dxf.insert = (-50, 20, 0) # the same WCS insertion point + truth = ent.matrix44().transform(Vec3(0, 6, 0)) + ep = await resolve_endpoint(backend, {"handle": pump["handle"], "port": "discharge"}) + assert (ep["x"], ep["y"]) == pytest.approx((truth.x, truth.y)) + assert (ep["x"], ep["y"]) == pytest.approx((53.0, 25.196), abs=1e-3) + assert ep["direction_deg"] == pytest.approx(60.0) + # unrotated: the suction that faced -x at (46, 20) now faces +x at (54, 20) + pump = await place_symbol(backend, "centrifugal_pump", 150, 20, tag="P-2") + ent = backend._doc.entitydb.get(pump["handle"]) + ent.dxf.extrusion = (0, 0, -1) + ent.dxf.insert = (-150, 20, 0) + ep = await resolve_endpoint(backend, {"handle": pump["handle"], "port": "suction"}) + assert (ep["x"], ep["y"], ep["direction_deg"]) == pytest.approx((154.0, 20.0, 0.0)) + result = await draw_line( + backend, {"handle": pump["handle"], "port": "suction"}, {"x": 200, "y": 20} + ) + assert result["vertices"][0] == pytest.approx([154.0, 20.0]) + + async def test_stretched_symbol_is_refused_not_resolved_off_the_geometry(backend): pump = await place_symbol(backend, "centrifugal_pump", 100, 100, tag="P-1") payload = await read_payload(backend, pump["handle"])