Merge pull request #1265 from ispyisail/feature/qet-mcp-layout-check

Add qet_layout_check: score a drawing and fix wires that jog
This commit is contained in:
ispyisail
2026-10-03 15:52:16 +13:00
committed by GitHub
3 changed files with 983 additions and 6 deletions
+34 -1
View File
@@ -40,6 +40,7 @@ here read the model.
| `qet_project_new` | **start from nothing** — an empty project with a title and folios |
| `qet_element_search` | **find a symbol** in a collection by name (any language), type or terminal count |
| `qet_check` | **design-rule checks** — duplicate labels, unlabelled masters, unnumbered conductors, empty folios |
| `qet_layout_check` | **does the drawing read well?** — a 0–100 score; wires that jog because two symbols are a few pixels out of line, symbols off the grid, wires through symbols, overlaps, crossings; and the moves that fix them, ready for `qet_edit` |
| `qet_query` | **ask the project database** — read-only SQL over the views and tables |
| `qet_about` | **start here** — where QElectroTech keeps things, what is switched on, the stored scripts, the calls a script can make (from `qet-assistant.json`) |
| `qet_script_api` | **what a script can call** — every `qet.*` call of this build, and the header that makes a script a button |
@@ -188,7 +189,7 @@ clicked, and stop working until it is turned on:
| | |
|---|---|
| need `QET_ENABLE_SCRIPTING=1` | `qet_query`, `qet_continuity`, `qet_check`, `qet_project_new`, `qet_edit`, `qet_script_api`, `qet_script_test`, `qet_script_install`, `qet_script_remove` |
| need `QET_ENABLE_SCRIPTING=1` | `qet_query`, `qet_continuity`, `qet_check`, `qet_layout_check`, `qet_project_new`, `qet_edit`, `qet_script_api`, `qet_script_test`, `qet_script_install`, `qet_script_remove` |
| unaffected | everything else — they read the `.qet` directly, or, in `qet_export`'s case, use a plain CLI flag |
The variable goes in the environment this server is started in, which for an
@@ -377,6 +378,38 @@ open but never shows the token.
Four elements moved by one uniform delta; nothing was relabelled. That is
the answer a screenshot gave wrongly.
**Tidy a drawing: straight wires, symbols in line**
A wire is straight only when its two terminals are exactly in line. A symbol
is placed by its origin and its terminals sit at an offset from it, so
symbols placed "under each other" by eye are often a few pixels apart and
the wire jogs. After drawing:
```json
{"name": "qet_layout_check", "arguments": {"project": "drawn.qet"}}
```
```json
{"ok": true, "style": "iec", "score": 40,
"summary": {"wires": 4, "straight_wires": 0, "avoidable_bends": 4, "off_grid": 0, ...},
"findings": [{"rule": "avoidable_bend", "folio": 1, "offset": 3.0, ...}],
"fixes": [{"op": "move_element", "folio": 0, "element": "{...}", "dx": -3.0, "dy": 0.0}, ...]}
```
Pass `fixes` as they are, all in one call, to `qet_edit`, then check again;
the same drawing then scores 100. The moves are planned together: a wire
that is already straight pins its two symbols, a move never puts a symbol
on another one or across another wire, and symbols lined up with each other
go onto the grid together. A jog no move can fix (two symbols whose
terminals are not spaced alike) is reported with `"conflict": true`.
`style` is `iec` (current paths as columns, wires mostly vertical), `nfpa`
(ladder rungs as rows, wires mostly horizontal) or `auto`, which goes by the
drawing. The check is read-only. On a QElectroTech build without
`conductorPath()`, a wire whose two terminals both carry other wires cannot
be read; the answer names those in `unread_wires` and leaves them out of the
score.
**Draw something, and check it landed**
```json
+656 -4
View File
@@ -1026,7 +1026,8 @@ def _run_qet(binary: str, args: list[str], timeout: int = 180,
"QET_ENABLE_SCRIPTING=1 to the environment this server is "
"started in -- in an MCP client that is the \"env\" block of "
"its entry in the client configuration. Only qet_query, "
"qet_continuity, qet_check, qet_project_new, qet_edit, "
"qet_continuity, qet_check, qet_layout_check, qet_project_new, "
"qet_edit, "
"qet_script_api, qet_script_test, qet_script_install, "
"qet_script_remove and qet_recording_check need it; every "
"other tool either reads "
@@ -2455,6 +2456,614 @@ def tool_check(binary: str, project: str, checks: list | None = None,
return answer
# --------------------------------------------------------------------------
# Layout check: does the drawing read well?
# --------------------------------------------------------------------------
#
# A wire is straight only when its two terminals share an x or a y exactly.
# An assistant places a symbol by its origin, and its terminals sit at an
# offset from that origin it cannot see, so "under K1" lands a few pixels
# off and QElectroTech draws a jog. This check finds those, with the move
# that removes each one.
#
# The drawn path of a wire is not in the file: a wire on QElectroTech's
# default path is saved with no <segment> at all. So the geometry comes from
# QElectroTech, through the read calls of its scripting API, and is scored
# here. Nothing is saved.
LAYOUT_STYLES = ["auto", "iec", "nfpa"]
LAYOUT_GRID = 10.0 # Diagram::xGrid / yGrid
LAYOUT_RULES = {
"avoidable_bend": {
"severity": "warning",
"note": "The two terminals face each other along one axis but are a few "
"pixels out of line, so the wire jogs. Moving one symbol makes it "
"straight; \"fix\" is that symbol's move from \"fixes\". "
"\"conflict\": no move is offered, because both symbols are "
"already lined up by other wires along this axis or moving either "
"would put it on another symbol or across another wire.",
},
"extra_bends": {
"severity": "info",
"note": "The wire bends more often than its two terminals need. Often a "
"segment moved by hand; route_conductor redraws it.",
},
"wire_through_symbol": {
"severity": "warning",
"note": "A wire runs through a symbol it is not connected to. A symbol "
"drawn around one of the wire's own ends is a frame and is left "
"out, as the router does, and so is a symbol with no terminals. "
"Cable tags and shields are drawn across wires on purpose: ignore "
"the finding for those.",
},
"overlapping_symbols": {
"severity": "warning",
"note": "Two symbols overlap by more than one grid step (a symbol's box "
"is its declared size, so side-by-side symbols can share a few "
"pixels of it). Frames, drawn around another symbol, and symbols "
"with no terminals (tags, shields, label holders) are left out.",
},
"off_grid": {
"severity": "warning",
"note": "The symbol's origin is off the 10 px grid QElectroTech snaps "
"symbols to, so its terminals are off the grid the other symbols "
"are on. An axis a straight wire lines it up on is left alone. "
"\"fix\" moves it onto the grid.",
},
"crossing": {
"severity": "info",
"note": "Two wires cross. Counted so two drafts can be compared; some "
"crossings cannot be avoided, so they do not lower the score.",
},
}
_LAYOUT_SEG = re.compile(r"^\s*\d+:\s*\(([^,]+),([^)]+)\)-\(([^,]+),([^)]+)\)")
# One launch, read-only. Per folio: every symbol's geometry and terminals,
# every wire's ends and drawn path. conductorPath() reads any wire by its
# uuid; a build without it reads a wire through one of its ends, which
# conductorSegments() refuses on a terminal carrying a second wire.
_LAYOUT_JS = r"""
var only = @FOLIO@;
var byUuid = typeof qet.conductorPath === 'function';
for (var f = 0; f < qet.folioCount(); f++) {
if (only >= 0 && f !== only) continue;
var els = qet.elementUuids(f), E = [];
for (var i = 0; i < els.length; i++) {
E.push({uuid: els[i], name: qet.elementName(f, els[i]),
label: qet.elementLabel(f, els[i]), g: qet.elementGeometry(f, els[i]),
terminals: qet.elementTerminals(f, els[i]).length});
}
var cu = qet.conductorUuids(f), lines = qet.conductors(f), C = [];
for (var j = 0; j < cu.length; j++) {
var ends = qet.conductorEnds(f, cu[j]), path = null, segs = null;
if (byUuid) {
path = qet.conductorPath(f, cu[j]);
} else if (ends.length === 2) {
for (var k = 0; k < 2 && segs === null; k++) {
if (ends[k] === '?') continue;
var n = 0;
for (var l = 0; l < lines.length; l++) {
var p = lines[l].split(' : ')[0].split(' -- ');
if (p[0] === ends[k] || p[1] === ends[k]) n++;
}
if (n !== 1) continue;
var m = ends[k].split(' terminal ');
segs = qet.conductorSegments(f, m[0], parseInt(m[1], 10));
}
}
C.push({uuid: cu[j], ends: ends, path: path, segs: segs});
}
qet.log(@MARKER@ + JSON.stringify({kind: 'layout', folio: f, elements: E,
conductors: C}));
}
"""
def _layout_points(wire: dict) -> list | None:
"""The wire's drawn path as points, from either read call; None if it
could not be read."""
if wire.get("path"):
try:
return [(float(p["x"]), float(p["y"])) for p in wire["path"]]
except (KeyError, TypeError, ValueError):
return None
segs = wire.get("segs")
if not segs:
return None
pts = []
for line in segs:
mt = _LAYOUT_SEG.match(line)
if not mt:
return None
x1, y1, x2, y2 = (float(v) for v in mt.groups())
if not pts:
pts.append((x1, y1))
pts.append((x2, y2))
return pts if len(pts) >= 2 else None
def _simplify(pts: list) -> list:
"""Drop zero-length steps and merge straight runs, so what is left has a
corner at every inner point."""
out = []
for p in pts:
if out and abs(p[0] - out[-1][0]) < 1e-6 and abs(p[1] - out[-1][1]) < 1e-6:
continue
if len(out) >= 2:
a, b = out[-2], out[-1]
if ((abs(a[0] - b[0]) < 1e-6 and abs(b[0] - p[0]) < 1e-6)
or (abs(a[1] - b[1]) < 1e-6 and abs(b[1] - p[1]) < 1e-6)):
out[-1] = p
continue
out.append(p)
return out
def _facing(dock: tuple, nxt: tuple) -> tuple | None:
"""Which way a terminal sends its wire: the unit step from the dock point
to the next distinct point of the path, or None if that is not along an
axis."""
dx, dy = nxt[0] - dock[0], nxt[1] - dock[1]
if abs(dx) < 1e-6 and abs(dy) > 1e-6:
return (0, 1 if dy > 0 else -1)
if abs(dy) < 1e-6 and abs(dx) > 1e-6:
return (1 if dx > 0 else -1, 0)
return None
def _box(g: dict) -> tuple | None:
try:
return (float(g["left"]), float(g["top"]), float(g["right"]), float(g["bottom"]))
except (KeyError, TypeError, ValueError):
return None
def _contains(outer: tuple, inner: tuple) -> bool:
return (outer[0] <= inner[0] and outer[1] <= inner[1]
and outer[2] >= inner[2] and outer[3] >= inner[3] and outer != inner)
def _overlap(a: tuple, b: tuple, margin: float = 1.0) -> bool:
return (min(a[2], b[2]) - max(a[0], b[0]) > margin
and min(a[3], b[3]) - max(a[1], b[1]) > margin)
def _segment_through(p: tuple, q: tuple, box: tuple, margin: float = 1.0) -> bool:
"""Does the axis-aligned segment p-q run through the inside of box?"""
l, t, r, b = box[0] + margin, box[1] + margin, box[2] - margin, box[3] - margin
if l >= r or t >= b:
return False
if abs(p[1] - q[1]) < 1e-6: # horizontal
lo, hi = sorted((p[0], q[0]))
return t < p[1] < b and min(hi, r) - max(lo, l) > 1e-6
if abs(p[0] - q[0]) < 1e-6: # vertical
lo, hi = sorted((p[1], q[1]))
return l < p[0] < r and min(hi, b) - max(lo, t) > 1e-6
return False
def _crosses(a: tuple, b: tuple, c: tuple, d: tuple) -> bool:
"""Do a horizontal and a vertical segment cross inside both?"""
if abs(a[1] - b[1]) < 1e-6 and abs(c[0] - d[0]) < 1e-6:
h, v = (a, b), (c, d)
elif abs(a[0] - b[0]) < 1e-6 and abs(c[1] - d[1]) < 1e-6:
h, v = (c, d), (a, b)
else:
return False
x, y = v[0][0], h[0][1]
hx = sorted((h[0][0], h[1][0]))
vy = sorted((v[0][1], v[1][1]))
return hx[0] + 1e-6 < x < hx[1] - 1e-6 and vy[0] + 1e-6 < y < vy[1] - 1e-6
def _end_element(end: str) -> str:
return end.split(" terminal ")[0] if " terminal " in end else ""
def _grid_offset(v: float) -> float:
"""How far v must move to reach the nearest grid line."""
return round(v / LAYOUT_GRID) * LAYOUT_GRID - v
def _layout_folio(data: dict, max_shift: float) -> dict:
"""Score one folio's dump and plan the moves that fix it. Pure: no
QElectroTech, so every rule is testable with made-up geometry.
Fixes are planned together, one move per symbol, because they interact:
two jogs can ask one symbol to move two ways, and snapping a symbol to
the grid can bend a straight wire. So straight wires are taken first and
pin their two symbols on their axis; jogs then move a symbol not yet
pinned, preferring a move that lands it on the grid; grid snaps come
last and only on an axis nothing pinned.
Applying all of "fixes" at once is therefore consistent; applying each
finding's fix on its own, one after the other, is not.
"""
folio = int(data.get("folio", 0))
symbols = {}
for el in data.get("elements") or []:
box = _box(el.get("g") or {})
if box is None:
continue
g = el["g"]
symbols[el["uuid"]] = {"uuid": el["uuid"], "name": el.get("name", ""),
"label": el.get("label", ""),
# No terminals: a label holder or a drawing
# aid, put on top of other symbols on purpose.
"annotation": int(el.get("terminals", 1) or 0) == 0,
"x": float(g.get("x", 0)), "y": float(g.get("y", 0)),
"xy": (float(g.get("x", 0)), float(g.get("y", 0))),
"box": box, "docks": []}
wire_count = {}
wires, unread = [], []
for w in data.get("conductors") or []:
ends = [_end_element(e) for e in (w.get("ends") or [])]
for e in ends:
if e:
wire_count[e] = wire_count.get(e, 0) + 1
pts = _layout_points(w)
if pts is None:
unread.append(w.get("uuid", ""))
continue
ends = (ends + ["", ""])[:2]
# The path runs from the first end's terminal to the second's.
for end, dock in ((ends[0], pts[0]), (ends[1], pts[-1])):
if end in symbols:
symbols[end]["docks"].append(dock)
wires.append({"uuid": w.get("uuid", ""), "ends": ends, "pts": _simplify(pts)})
findings = []
dirty_wires, dirty_symbols = set(), set()
length = {"vertical": 0.0, "horizontal": 0.0}
move = {} # symbol uuid -> [dx, dy], the one planned move
locked = set() # (symbol uuid, axis) a jog fix already decided
def add(rule, **fields):
findings.append({"rule": rule, "severity": LAYOUT_RULES[rule]["severity"],
"folio": folio + 1, **fields})
def on_grid(v):
return abs(_grid_offset(v)) < 1e-6
solid = [x for x in symbols.values() if not x["annotation"]]
# Symbols lined up on an axis by straight wires (as planned) form a
# group; the grid snap moves a group together so it stays in line.
parent = {}
def find(k):
parent.setdefault(k, k)
while parent[k] != k:
parent[k] = parent[parent[k]]
k = parent[k]
return k
def union(a, b):
parent[find(a)] = find(b)
def shifted(box, d):
return (box[0] + d[0], box[1] + d[1], box[2] + d[0], box[3] + d[1])
def blocked(uuid, total):
"""Would moving this symbol by total (its whole planned move) put it
on another symbol, or across a wire it is not on, where it was not
before? A fix must not trade a jog for a collision."""
me = symbols[uuid]
if me["annotation"]:
return False
old, new = me["box"], shifted(me["box"], total)
for o in solid:
if o["uuid"] == uuid:
continue
ob = shifted(o["box"], move.get(o["uuid"], [0, 0]))
if _contains(ob, old) or _contains(old, ob) or _contains(ob, new) or _contains(new, ob):
continue
if _overlap(new, ob, margin=LAYOUT_GRID) and not _overlap(old, o["box"], margin=LAYOUT_GRID):
return True
for w in wires:
if uuid in w["ends"]:
continue
for p, q in zip(w["pts"], w["pts"][1:]):
if _segment_through(p, q, new) and not _segment_through(p, q, old):
return True
return False
for w in wires:
pts = w["pts"]
for p, q in zip(pts, pts[1:]):
length["vertical" if abs(p[0] - q[0]) < 1e-6 else "horizontal"] += (
abs(p[0] - q[0]) + abs(p[1] - q[1]))
w["bends"] = max(0, len(pts) - 2)
# Every wire whose terminals face each other along one axis: straight
# ones first, so they pin their symbols on that axis (a fix must not
# trade one straight wire for another), then jogs, smallest first.
in_line = []
for w in wires:
pts, bends = w["pts"], w["bends"]
if len(pts) < 2:
continue
a, b = pts[0], pts[-1]
fa, fb = _facing(a, pts[1]), _facing(b, pts[-2])
if not (fa and fb):
continue
if fa[0] == 0 and fb[0] == 0: # both vertical
axis, i = "x", 0
facing = fa[1] == (1 if b[1] > a[1] else -1) and fb[1] == -fa[1]
elif fa[1] == 0 and fb[1] == 0: # both horizontal
axis, i = "y", 1
facing = fa[0] == (1 if b[0] > a[0] else -1) and fb[0] == -fa[0]
else: # an L at best
if bends > 1:
add("extra_bends", conductor=w["uuid"], bends=bends, needed=1,
note=LAYOUT_RULES["extra_bends"]["note"],
fix={"op": "route_conductor", "folio": folio, "conductor": w["uuid"]})
dirty_wires.add(w["uuid"])
continue
offset = b[i] - a[i]
straight = facing and abs(offset) < 1e-6
jog = facing and 1e-6 <= abs(offset) <= max_shift and bends > 0
if straight or jog:
in_line.append((0 if straight else 1, abs(offset), w, axis, i, a, b, jog))
need = 0 if straight else 2
if not jog and bends > need:
add("extra_bends", conductor=w["uuid"], bends=bends, needed=need,
note=LAYOUT_RULES["extra_bends"]["note"],
fix={"op": "route_conductor", "folio": folio, "conductor": w["uuid"]})
dirty_wires.add(w["uuid"])
for _, _, w, axis, i, a, b, jog in sorted(in_line, key=lambda t: t[:2]):
ea, eb = w["ends"]
# As it will be once the moves planned so far are applied.
left = (b[i] + move.get(eb, [0, 0])[i]) - (a[i] + move.get(ea, [0, 0])[i])
mover = None
if abs(left) > 1e-6:
# Prefer the end that lands on the grid, then the one with fewer
# other wires (on a tie the second, so the first anchors a chain).
def rank(e):
delta = -left if e == eb else left
origin = symbols[e]["xy"][i] + move.get(e, [0, 0])[i] + delta
return (not on_grid(origin), wire_count.get(e, 0), e != eb)
free = [e for e in (ea, eb) if e in symbols and (e, axis) not in locked]
for cand in sorted(free, key=rank):
total = list(move.get(cand, [0.0, 0.0]))
total[i] += -left if cand == eb else left
if blocked(cand, total):
continue
mover = cand
move[cand] = total
break
locked.update((e, axis) for e in (ea, eb) if e in symbols)
if (abs(left) < 1e-6 or mover is not None) and ea in symbols and eb in symbols:
union((ea, axis), (eb, axis))
if jog:
w["mover"] = mover
w["conflict"] = abs(left) > 1e-6 and mover is None
add("avoidable_bend", conductor=w["uuid"],
offset=round(abs(b[i] - a[i]), 3), bends=w["bends"],
note=LAYOUT_RULES["avoidable_bend"]["note"])
dirty_wires.add(w["uuid"])
# Off the grid: the symbol's origin, which is what QElectroTech's own
# grid snaps. A symbol lined up with others by straight wires moves only
# with its whole group, and only when they are all off by the same
# amount -- straight wires matter more than the grid, and snapping one
# member alone would bend them, so the next run would undo it.
groups = {}
for (uuid, axis) in locked:
groups.setdefault(find((uuid, axis)), set()).add(uuid)
def offset(u, i):
return _grid_offset(symbols[u]["xy"][i] + move.get(u, [0, 0])[i])
step = {u: [0.0, 0.0] for u in symbols}
for i, axis in ((0, "x"), (1, "y")):
for u in symbols:
if (u, axis) not in locked:
step[u][i] = offset(u, i)
for root, members in groups.items():
if root[1] != axis:
continue
offs = {round(offset(u, i), 6) for u in members}
if len(offs) == 1:
d = offs.pop()
for u in members:
step[u][i] = d
def total(u):
m = move.get(u, [0.0, 0.0])
return [m[0] + step[u][0], m[1] + step[u][1]]
# A blocked member holds its whole group back on that axis.
for u in [u for u in symbols if any(abs(v) > 1e-6 for v in step[u])]:
if not blocked(u, total(u)):
continue
symbols[u]["blocked"] = True
for i, axis in ((0, "x"), (1, "y")):
if (u, axis) in locked and abs(step[u][i]) > 1e-6:
for v in groups.get(find((u, axis)), {u}):
step[v][i] = 0.0
snapped = []
for u, s in symbols.items():
if abs(step[u][0]) > 1e-6 or abs(step[u][1]) > 1e-6:
if blocked(u, total(u)):
s["blocked"] = True
else:
s["blocked"] = False
move[u] = total(u)
snapped.append(s)
elif s.get("blocked"):
snapped.append(s)
def move_op(uuid):
if uuid not in move:
return None
dx, dy = move[uuid]
return {"op": "move_element", "folio": folio, "element": uuid,
"dx": round(dx, 3) + 0.0, "dy": round(dy, 3) + 0.0}
by_wire = {w["uuid"]: w for w in wires}
for f in findings:
if f["rule"] == "avoidable_bend":
w = by_wire[f["conductor"]]
f["fix"] = move_op(w.get("mover"))
if w.get("conflict"):
f["conflict"] = True
for s in snapped:
extra = {"conflict": True} if s.get("blocked") else {}
add("off_grid", element=s["uuid"], label=s["label"], name=s["name"],
x=s["x"], y=s["y"], note=LAYOUT_RULES["off_grid"]["note"],
fix=None if s.get("blocked") else move_op(s["uuid"]), **extra)
dirty_symbols.add(s["uuid"])
# Wires through symbols. A symbol drawn around either end's own symbol
# is a frame (conductorrouter.cpp), not an obstacle.
for w in wires:
own = [symbols[e]["box"] for e in w["ends"] if e in symbols]
for uuid, s in symbols.items():
if (s["annotation"] or uuid in w["ends"]
or any(_contains(s["box"], o) for o in own)):
continue
pts = w["pts"]
if any(_segment_through(p, q, s["box"]) for p, q in zip(pts, pts[1:])):
add("wire_through_symbol", conductor=w["uuid"], element=uuid,
label=s["label"], name=s["name"],
note=LAYOUT_RULES["wire_through_symbol"]["note"],
fix={"op": "route_conductor", "folio": folio, "conductor": w["uuid"]})
dirty_wires.add(w["uuid"])
# A symbol's box is its declared size, rounded up to the grid, so two
# symbols drawn side by side can share up to one grid step of it.
for i, s in enumerate(solid):
for t in solid[i + 1:]:
if _contains(s["box"], t["box"]) or _contains(t["box"], s["box"]):
continue
if _overlap(s["box"], t["box"], margin=LAYOUT_GRID):
add("overlapping_symbols", elements=[s["uuid"], t["uuid"]],
labels=[s["label"], t["label"]], names=[s["name"], t["name"]],
note=LAYOUT_RULES["overlapping_symbols"]["note"])
dirty_symbols.update((s["uuid"], t["uuid"]))
crossings = 0
for i, w in enumerate(wires):
sw = list(zip(w["pts"], w["pts"][1:]))
for v in wires[i + 1:]:
n = sum(1 for a, b in sw for c, d in zip(v["pts"], v["pts"][1:])
if _crosses(a, b, c, d))
if n:
crossings += n
add("crossing", conductors=[w["uuid"], v["uuid"]], count=n,
note=LAYOUT_RULES["crossing"]["note"])
return {"folio": folio, "symbols": len(symbols), "wires": len(wires),
"unread": unread, "findings": findings, "dirty_wires": dirty_wires,
"dirty_symbols": dirty_symbols, "length": length, "crossings": crossings,
"straight": sum(1 for w in wires if w.get("bends") == 0),
"fixes": [move_op(u) for u in move]}
def _layout_answer(folios: list, style: str, limit: int) -> dict:
"""Combine per-folio results into the tool's answer."""
symbols = sum(f["symbols"] for f in folios)
wires = sum(f["wires"] for f in folios)
vertical = sum(f["length"]["vertical"] for f in folios)
horizontal = sum(f["length"]["horizontal"] for f in folios)
total = vertical + horizontal
if style == "auto":
style = "nfpa" if horizontal > vertical else "iec"
findings = [x for f in folios for x in f["findings"]]
order = {"error": 0, "warning": 1, "info": 2}
findings.sort(key=lambda x: (order[x["severity"]], x["folio"], x["rule"]))
count = {r: sum(1 for x in findings if x["rule"] == r) for r in LAYOUT_RULES}
clean_wires = wires - sum(len(f["dirty_wires"]) for f in folios)
clean_symbols = symbols - sum(len(f["dirty_symbols"]) for f in folios)
score = 100.0 * (0.6 * (clean_wires / wires if wires else 1.0)
+ 0.4 * (clean_symbols / symbols if symbols else 1.0))
unread = [u for f in folios for u in f["unread"]]
answer = {
"ok": True,
"style": style,
"score": round(score),
"summary": {
"folios": len(folios), "symbols": symbols, "wires": wires,
"unread_wires": len(unread),
"straight_wires": sum(f["straight"] for f in folios),
"avoidable_bends": count["avoidable_bend"],
"extra_bends": count["extra_bends"],
"wires_through_symbols": count["wire_through_symbol"],
"overlaps": count["overlapping_symbols"],
"off_grid": count["off_grid"],
"crossings": sum(f["crossings"] for f in folios),
"flow": {"vertical": round(vertical / total, 3) if total else 0.0,
"horizontal": round(horizontal / total, 3) if total else 0.0},
},
"findings": findings[:limit],
# One move per symbol, all findings' moves combined: apply them
# together in one qet_edit call.
"fixes": [op for f in folios for op in f["fixes"]],
}
if len(findings) > limit:
answer["truncated"] = len(findings) - limit
if unread:
answer["unread_wires"] = unread[:limit]
answer["note"] = (f"{len(unread)} wire(s) could not be read: both of their "
"terminals carry other wires too, and this QElectroTech build "
"has no conductorPath() to read them by uuid. They are left "
"out of the score.")
return answer
def tool_layout_check(binary: str, project: str, folio: int | None = None,
style: str = "auto", max_shift: float = 40, limit: int = 50,
elements_dir: str | None = None, timeout: int = 180) -> dict:
"""Score how well a project's drawing reads: straight wires, symbols in
line and on the grid, nothing overlapping. Read-only."""
proj = Path(project).expanduser()
if not proj.is_file():
raise ValueError(f"no such project: {proj}")
if style not in LAYOUT_STYLES:
raise ValueError(f"unknown style {style!r}; expected one of {', '.join(LAYOUT_STYLES)}")
if (isinstance(max_shift, bool) or not isinstance(max_shift, (int, float))
or max_shift < 0):
raise ValueError("max_shift must be a number >= 0")
if isinstance(limit, bool) or not isinstance(limit, int) or limit < 0:
raise ValueError("limit must be an integer >= 0")
if folio is not None and (isinstance(folio, bool) or not isinstance(folio, int)
or folio < 1):
raise ValueError("folio is counted from 1, as qet_elements numbers them")
script = (_LAYOUT_JS.replace("@FOLIO@", str(folio - 1 if folio else -1))
.replace("@MARKER@", json.dumps(_MARKER)))
result = _run_qet(binary, [str(proj)], timeout=timeout,
elements_dir=elements_dir, script=script, tail=20_000_000)
folios = []
for line in (result.get("stdout", "") + "\n" + result.get("stderr", "")).splitlines():
idx = line.find(_MARKER)
if idx < 0:
continue
try:
rec = json.loads(line[idx + len(_MARKER):])
except json.JSONDecodeError:
continue
if rec.get("kind") == "layout":
folios.append(_layout_folio(rec, float(max_shift)))
if not folios:
answer = {"ok": False,
"hint": result.get("hint") or (
"no layout came back: the folio does not exist, or this build's "
"scripting API predates the read calls this check needs")}
if result.get("exit_code") is not None:
answer["exit_code"] = result["exit_code"]
return answer
answer = _layout_answer(folios, style, limit)
if result.get("hint"):
answer["ok"] = False
answer["hint"] = result["hint"]
return answer
def tool_project_new(binary: str, output: str, title: str = "Untitled",
folios=1, author: str = "", overwrite: bool = False,
elements_dir: str | None = None, timeout: int = 180) -> dict:
@@ -2798,7 +3407,8 @@ SERVER_INSTRUCTIONS = (
"general script, qet_recording_check it until it matches, then "
"qet_script_install it.\n"
"Verify edits by reading the result (qet_diff, qet_elements), not by "
"assuming them.")
"assuming them. After drawing, run qet_layout_check and apply its fixes: "
"a wire is straight only when its two terminals are exactly in line.")
def tool_about() -> dict:
@@ -3987,6 +4597,47 @@ TOOLS = [
a.get("sample", 10), a.get("elements_dir"),
a.get("timeout", 180)),
},
{
"name": "qet_layout_check",
"description": "Score how well a drawing reads, 0-100, and list what spoils it: "
"wires that jog because two symbols are a few pixels out of line "
"(avoidable_bend), wires with more bends than needed, wires "
"running through a symbol, overlapping symbols, symbols off the "
"10 px grid, and crossings. \"fixes\" is the list of "
"move_element operations that removes the jogs and off-grid "
"symbols, one move per symbol, planned together: pass the whole "
"list to one qet_edit call (folio counted from 0, as qet_edit "
"counts; findings report \"folio\" counted from 1). Run it "
"after drawing, apply \"fixes\", run it again. "
"One QElectroTech launch; read-only, nothing is saved.",
"inputSchema": {
"type": "object",
"properties": {
"binary": {"type": "string", "description": "the qelectrotech executable; leave it out to use the one this server is configured with. Any other is refused unless its configuration allows it"},
"project": {"type": "string"},
"folio": {"type": "integer", "description": "only this folio, counted from 1; omit for all"},
"style": {"type": "string", "enum": LAYOUT_STYLES, "default": "auto",
"description": "iec: current paths are columns, wires mostly "
"vertical. nfpa: ladder rungs are rows, wires "
"mostly horizontal. auto: from the drawing"},
"max_shift": {"type": "number", "default": 40,
"description": "the largest move, in pixels, an "
"avoidable_bend fix may suggest; a bigger "
"jog is taken as intended"},
"limit": {"type": "integer", "default": 50,
"description": "how many findings to return; the summary "
"counts them all"},
"elements_dir": {"type": "string"},
"timeout": {"type": "integer", "default": 180},
},
"required": ["project"],
},
"handler": lambda a: tool_layout_check(a["binary"], a["project"], a.get("folio"),
a.get("style", "auto"),
a.get("max_shift", 40), a.get("limit", 50),
a.get("elements_dir"),
a.get("timeout", 180)),
},
{
"name": "qet_element_build",
"description": "Write a .elmt element definition: named in one or more "
@@ -4356,6 +5007,7 @@ _DATA_PATHS = {
"qet_query": {"read": ("project",)},
"qet_continuity": {"read": ("project",)},
"qet_check": {"read": ("project",)},
"qet_layout_check": {"read": ("project",)},
"qet_project_new": {"write": ("output",)},
"qet_element_build": {"write": ("output",)},
# The scripts folder is chosen by scripts_dir(), never by the client,
@@ -4368,8 +5020,8 @@ _DATA_PATHS = {
# Tools that launch QElectroTech, and so take "binary" and "elements_dir".
_LAUNCHES_QET = {"qet_export", "qet_edit", "qet_query", "qet_continuity",
"qet_check", "qet_project_new", "qet_script_api", "qet_script_test",
"qet_recording_check"}
"qet_check", "qet_layout_check", "qet_project_new", "qet_script_api",
"qet_script_test", "qet_recording_check"}
# Tools that launch QElectroTech only when given this argument.
_LAUNCHES_QET_WITH = {"qet_script_install": "test_project"}
+293 -1
View File
@@ -184,7 +184,7 @@ class ToolRegistry(unittest.TestCase):
"qet_live_run_stored", "qet_live_command", "qet_live_show_folio",
"qet_live_undo_last", "qet_live_screenshot", "qet_about",
"qet_recording_list", "qet_recording_read", "qet_recording_check",
"qet_recording_remove"})
"qet_recording_remove", "qet_layout_check"})
class EditValidation(unittest.TestCase):
@@ -1303,6 +1303,231 @@ class CheckAndContinuityAnswers(unittest.TestCase):
self.assertEqual(m.tool_continuity("qet", str(self.qet), folio=0)["finding_count"], 0)
def _sym(uuid, x, y, w=20, h=40, label="", terminals=2, name="S"):
"""A symbol for the layout rules: origin x/y, box centred on it."""
return {"uuid": uuid, "name": name, "label": label, "terminals": terminals,
"g": {"x": x, "y": y, "rotation": 0, "left": x - w / 2, "top": y - h / 2,
"right": x + w / 2, "bottom": y + h / 2}}
def _wire(uuid, a, b, points):
return {"uuid": uuid, "ends": [f"{a} terminal 1", f"{b} terminal 0"],
"path": [{"x": x, "y": y} for x, y in points], "segs": None}
def _vjog(uuid, a, b, xa, xb, y0=120, y1=160):
"""A wire leaving a's bottom terminal down and entering b's top one."""
mid = (y0 + y1) / 2
return _wire(uuid, a, b, [(xa, y0), (xa, y0 + 10), (xa, mid), (xb, mid),
(xb, y1 - 10), (xb, y1)])
class LayoutRules(unittest.TestCase):
"""qet_layout_check's scoring and fix planning, on made-up geometry: no
QElectroTech needed, so every rule is pinned exactly."""
def folio(self, elements, conductors, max_shift=40, folio=0):
return m._layout_folio({"folio": folio, "elements": elements,
"conductors": conductors}, max_shift)
def rules(self, r):
return sorted(f["rule"] for f in r["findings"])
def test_simplify_drops_zero_steps_and_merges_runs(self):
self.assertEqual(m._simplify([(0, 0), (0, 10), (0, 10), (0, 30), (5, 30), (9, 30)]),
[(0, 0), (0, 30), (9, 30)])
def test_points_from_segments_and_from_path(self):
segs = ["0: (560,150)-(560,160) vertical static",
"1: (560,160)-(560,290.5) vertical movable"]
self.assertEqual(m._layout_points({"segs": segs}),
[(560.0, 150.0), (560.0, 160.0), (560.0, 290.5)])
self.assertEqual(m._layout_points({"path": [{"x": 1, "y": 2}, {"x": 1, "y": 9}]}),
[(1.0, 2.0), (1.0, 9.0)])
self.assertIsNone(m._layout_points({"segs": ["garbage"]}))
self.assertIsNone(m._layout_points({"segs": None, "path": None}))
def test_vertical_jog_moves_the_end_that_lands_on_the_grid(self):
r = self.folio([_sym("A", 100, 100), _sym("B", 103, 180)],
[_vjog("W", "A", "B", 100, 103)])
[f] = r["findings"]
self.assertEqual(f["rule"], "avoidable_bend")
self.assertEqual((f["offset"], f["bends"], f["folio"]), (3.0, 2, 1))
self.assertEqual(f["fix"], {"op": "move_element", "folio": 0, "element": "B",
"dx": -3.0, "dy": 0.0})
self.assertEqual(r["fixes"], [f["fix"]])
def test_horizontal_jog_nfpa(self):
w = _wire("W", "A", "B", [(110, 100), (120, 100), (130, 100), (130, 96),
(140, 96), (150, 96)])
r = self.folio([_sym("A", 100, 100, 20, 20), _sym("B", 160, 96, 20, 20)], [w])
[f] = r["findings"]
self.assertEqual(f["fix"], {"op": "move_element", "folio": 0, "element": "B",
"dx": 0.0, "dy": 4.0})
def test_terminals_not_facing_are_not_a_jog(self):
# both terminals send their wire downwards: a U, never straight
w = _wire("W", "A", "B", [(100, 120), (100, 140), (103, 140), (103, 120)])
r = self.folio([_sym("A", 100, 100), _sym("B", 103, 100, 2, 2)], [w])
self.assertNotIn("avoidable_bend", self.rules(r))
def test_jog_beyond_max_shift_is_left_alone(self):
r = self.folio([_sym("A", 100, 100), _sym("B", 160, 180)],
[_vjog("W", "A", "B", 100, 160)], max_shift=40)
self.assertEqual(r["findings"], [])
self.assertEqual(r["fixes"], [])
def test_a_straight_wire_pins_its_symbols(self):
# A-B straight; B-C jogs: C moves, not B
ab = _wire("AB", "A", "B", [(100, 120), (100, 160)])
bc = _vjog("BC", "B", "C", 100, 104, 200, 240)
r = self.folio([_sym("A", 100, 100), _sym("B", 100, 180), _sym("C", 104, 260)], [ab, bc])
self.assertEqual(r["fixes"], [{"op": "move_element", "folio": 0, "element": "C",
"dx": -4.0, "dy": 0.0}])
def test_conflict_when_both_ends_are_pinned(self):
# A and B each held in line by a straight wire; the A-B jog cannot move
wires = [_wire("AX", "X", "A", [(100, 40), (100, 80)]),
_wire("BY", "B", "Y", [(104, 200), (104, 240)]),
_vjog("AB", "A", "B", 100, 104)]
r = self.folio([_sym("X", 100, 20), _sym("A", 100, 100), _sym("B", 104, 180),
_sym("Y", 104, 260)], wires)
[f] = [f for f in r["findings"] if f["rule"] == "avoidable_bend"]
self.assertIsNone(f["fix"])
self.assertTrue(f["conflict"])
def test_a_move_onto_another_symbol_is_not_offered(self):
# B can only line up by moving onto D, so A moves instead
r = self.folio([_sym("A", 100, 100), _sym("B", 120, 180), _sym("D", 100, 180)],
[_vjog("W", "A", "B", 100, 120)])
[f] = [f for f in r["findings"] if f["rule"] == "avoidable_bend"]
self.assertEqual(f["fix"]["element"], "A")
self.assertEqual(f["fix"]["dx"], 20.0)
def test_off_grid_symbol_without_wires(self):
r = self.folio([_sym("A", 103, 97)], [])
[f] = r["findings"]
self.assertEqual(f["rule"], "off_grid")
self.assertEqual(f["fix"], {"op": "move_element", "folio": 0, "element": "A",
"dx": -3.0, "dy": 3.0})
def test_a_lined_up_group_snaps_together(self):
# three symbols in line at x=103, off the grid together: all move,
# and the wires between them stay straight
wires = [_wire("AB", "A", "B", [(103, 120), (103, 160)]),
_wire("BC", "B", "C", [(103, 200), (103, 240)])]
r = self.folio([_sym("A", 103, 100), _sym("B", 103, 180), _sym("C", 103, 260)], wires)
self.assertEqual(sorted((f["element"], f["dx"]) for f in r["fixes"]),
[("A", -3.0), ("B", -3.0), ("C", -3.0)])
def test_wire_through_symbol_but_not_frame_or_annotation(self):
w = _wire("W", "A", "B", [(100, 120), (100, 300)])
elements = [_sym("A", 100, 100), _sym("B", 100, 320),
_sym("K", 100, 200, name="Coil"), # in the way
_sym("F", 100, 200, 300, 600, name="Cabinet"), # frame round A
_sym("T", 100, 250, terminals=0, name="Tag")] # annotation
r = self.folio(elements, [w])
hits = [f for f in r["findings"] if f["rule"] == "wire_through_symbol"]
self.assertEqual([f["element"] for f in hits], ["K"])
self.assertEqual(hits[0]["fix"], {"op": "route_conductor", "folio": 0,
"conductor": "W"})
def test_overlap_needs_more_than_one_grid_step(self):
touching = self.folio([_sym("A", 100, 100), _sym("B", 110, 100)], []) # 10 px
self.assertNotIn("overlapping_symbols", self.rules(touching))
r = self.folio([_sym("A", 100, 100), _sym("B", 105, 100)], []) # 15 px
self.assertIn("overlapping_symbols", self.rules(r))
def test_crossing_counted_not_at_shared_ends(self):
h = _wire("H", "A", "B", [(0, 50), (200, 50)])
v = _wire("V", "C", "D", [(100, 0), (100, 200)])
t = _wire("T", "A", "E", [(0, 50), (0, 200)]) # meets H at its end
r = self.folio([], [h, v, t])
self.assertEqual(r["crossings"], 1)
def test_extra_bends_for_an_l(self):
w = _wire("W", "A", "B", [(100, 120), (100, 140), (120, 140), (120, 160),
(150, 160)])
r = self.folio([], [w])
[f] = r["findings"]
self.assertEqual((f["rule"], f["bends"], f["needed"]), ("extra_bends", 3, 1))
def test_answer_score_style_and_limit(self):
clean = self.folio([_sym("A", 100, 100), _sym("B", 100, 180)],
[_wire("W", "A", "B", [(100, 120), (100, 160)])])
a = m._layout_answer([clean], "auto", 50)
self.assertEqual((a["score"], a["style"], a["summary"]["straight_wires"]), (100, "iec", 1))
self.assertEqual(a["summary"]["flow"], {"vertical": 1.0, "horizontal": 0.0})
jog = self.folio([_sym("A", 100, 100), _sym("B", 103, 180), _sym("C", 300, 301)],
[_vjog("W", "A", "B", 100, 103)])
a = m._layout_answer([jog], "nfpa", 1)
# wires 0/1 clean, symbols 2/3 clean (C off the grid)
self.assertEqual(a["score"], round(100 * (0.6 * 0 + 0.4 * 2 / 3)))
self.assertEqual(a["style"], "nfpa")
self.assertEqual(a["truncated"], 1)
self.assertEqual(len(a["fixes"]), 2)
def test_unread_wires_are_named_not_scored(self):
r = self.folio([], [{"uuid": "U", "ends": ["{a} terminal 0", "{b} terminal 0"],
"path": None, "segs": None}])
a = m._layout_answer([r], "auto", 50)
self.assertEqual(a["summary"]["unread_wires"], 1)
self.assertEqual(a["unread_wires"], ["U"])
self.assertIn("conductorPath", a["note"])
self.assertEqual(a["score"], 100)
class LayoutCheckTool(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.qet = Path(self.tmp.name) / "p.qet"
self.qet.write_text("<project><diagram/></project>", encoding="utf-8")
def tearDown(self):
self.tmp.cleanup()
def test_bad_arguments(self):
for kw in ({"style": "ansi"}, {"folio": 0}, {"folio": True},
{"max_shift": -1}, {"max_shift": "4"}, {"limit": -1}):
with self.subTest(kw=kw), self.assertRaises(ValueError):
m.tool_layout_check("qet", str(self.qet), **kw)
with self.assertRaises(ValueError):
m.tool_layout_check("qet", str(self.qet) + ".missing")
def test_script_reads_the_chosen_folio_only(self):
seen = {}
def run(binary, args, **kw):
seen.update(kw)
return {"stdout": "", "stderr": ""}
with mock.patch.object(m, "_run_qet", run):
r = m.tool_layout_check("qet", str(self.qet), folio=3)
self.assertIn("var only = 2;", seen["script"])
self.assertNotIn("save", seen["script"])
self.assertFalse(r["ok"])
self.assertIn("no layout came back", r["hint"])
def test_answer_from_log_lines(self):
rec = {"kind": "layout", "folio": 0,
"elements": [_sym("A", 100, 100), _sym("B", 103, 180)],
"conductors": [_vjog("W", "A", "B", 100, 103)]}
out = "noise\n" + m._MARKER + "{bad json\n" + m._MARKER + json.dumps(rec)
with mock.patch.object(m, "_run_qet", lambda *a, **k: {"stdout": out, "stderr": ""}):
r = m.tool_layout_check("qet", str(self.qet))
self.assertTrue(r["ok"])
self.assertEqual(r["summary"]["avoidable_bends"], 1)
self.assertEqual(r["fixes"][0]["element"], "B")
def test_a_launch_hint_is_passed_on(self):
rec = {"kind": "layout", "folio": 0, "elements": [], "conductors": []}
out = m._MARKER + json.dumps(rec)
with mock.patch.object(m, "_run_qet",
lambda *a, **k: {"stdout": out, "stderr": "", "hint": "boom"}):
r = m.tool_layout_check("qet", str(self.qet))
self.assertFalse(r["ok"])
self.assertEqual(r["hint"], "boom")
class ElementSearch(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
@@ -5268,6 +5493,73 @@ class Integration(unittest.TestCase):
self.assertFalse(r["ok"])
@needs_elements
class LayoutIntegration(unittest.TestCase):
"""A drawing made the way an assistant makes one -- symbols placed by
eye a few pixels out of line -- comes out straight and on the grid
after one round of qet_layout_check's fixes, in both styles."""
def setUp(self):
self.sb = Sandbox()
def tearDown(self):
self.sb.close()
def draw(self, style):
if style == "iec": # current paths: columns, wires vertical
pos = [(100, 100), (103, 200), (96, 300), (301, 100), (298, 200), (300, 300)]
else: # rungs: rows, wires horizontal
pos = [(100, 100), (200, 104), (300, 97), (100, 301), (200, 298), (300, 300)]
ops = []
for k, (x, y) in enumerate(pos):
ops.append({"op": "add_element", "folio": 0, "path": [TERMINAL, SLAVE, COIL][k % 3],
"x": x, "y": y, "id": f"e{k}"})
if style == "nfpa":
ops.append({"op": "rotate_element", "folio": 0, "element": f"$e{k}",
"angle": 270})
for c in (0, 3):
# borne_2's bottom terminal is index 2 (index 1 is its side one)
ops.append({"op": "add_conductor", "folio": 0, "from": f"$e{c}", "from_terminal": 2,
"to": f"$e{c + 1}", "to_terminal": 0})
ops.append({"op": "add_conductor", "folio": 0, "from": f"$e{c + 1}",
"from_terminal": 1, "to": f"$e{c + 2}", "to_terminal": 0})
r = self.sb.edit(self.sb.new(), ops, out=f"{style}.qet")
self.assertTrue(r["ok"], r.get("hint"))
return r["output"]
def check(self, path, **kw):
r = m.tool_layout_check(BINARY, path, elements_dir=ELEMENTS, **kw)
self.assertTrue(r["ok"], r.get("hint"))
return r
def round_trip(self, style):
drawn = self.draw(style)
before_bytes = Path(drawn).read_bytes()
before = self.check(drawn)
self.assertEqual(Path(drawn).read_bytes(), before_bytes, "the check must not save")
self.assertEqual(before["style"], style)
self.assertEqual(before["summary"]["avoidable_bends"], 4)
self.assertEqual(before["summary"]["straight_wires"], 0)
fixed = self.sb.edit(drawn, before["fixes"], out=f"{style}-fixed.qet")
self.assertTrue(fixed["ok"], fixed.get("hint"))
after = self.check(fixed["output"])
self.assertEqual(after["score"], 100, after["findings"])
self.assertEqual(after["summary"]["straight_wires"], 4)
self.assertEqual(after["fixes"], [])
def test_iec_columns(self):
self.round_trip("iec")
def test_nfpa_rungs(self):
self.round_trip("nfpa")
def test_one_folio_only(self):
drawn = self.draw("iec")
self.assertEqual(self.check(drawn, folio=1)["summary"]["folios"], 1)
r = m.tool_layout_check(BINARY, drawn, folio=5, elements_dir=ELEMENTS)
self.assertFalse(r["ok"])
@needs_binary
class PlcIntegration(unittest.TestCase):
"""PLC IO table and PLC-slave linking, against the fixtures in