wonq-backend / processors /shape_assembler.py
Aaron
Sync backend with GitHub main (0aa3b71): round-25 AI robustness + authoritative SVG shapes
1ef2a66
Raw History Blame Contribute Delete
12.7 kB
"""
Shape Assembler — from segment soup to drawing shapes.
=========================================================
Takes the line extractor's segments (endpoints already corner- and
T-snapped, rounds 20-21) and assembles the topology they form:
- endpoints within _NODE_TOL_PX cluster into shared graph nodes,
- dead-end spurs (degree-1 chains: dimension lines, leader lines,
unattached strokes) are iteratively stripped and reported as OPEN
polyline chains,
- the remaining graph — where every node has degree >= 2 — is traversed
as a planar embedding (half-edge walk, next-most-clockwise rule) so
its FACES emerge as CLOSED polygons: the rooms and outlines of a
floor plan, not just the walls that bound them. The outer face (the
one enclosing everything, identified as the largest absolute area) is
dropped; interior faces below _MIN_FACE_AREA_PX2 (slivers from
double-drawn walls) are dropped too.
Pure deterministic geometry — same ethos as rounds 14/19/20: nothing here
can hallucinate a shape that the extracted segments don't actually form.
Used by svg_generator (closed <path> elements with subtle fills, open
chains as polylines) and by the line extractor's visualisation (closed
shapes tinted so "did it form the room?" is visible at a glance).
"""
from __future__ import annotations
import math
from typing import Any, Dict, List, Tuple
_NODE_TOL_PX = 8.0
_MIN_FACE_AREA_PX2 = 400.0 # ~20x20px — smaller "rooms" are wall slivers
# Round 25: assembled stroke faces above this isoperimetric ratio
# (perimeter² / 4πA; 1.0 = circle) are jagged Hough-fragment cycles, not
# drawing shapes, and render as the "odd shapes" the user reported. Real
# architectural faces are compact-to-elongated: even a 35:1 wall sliver
# only measures ~12, while fragment tangles measure 20-100+.
_SPIKY_RATIO_MAX = 12.0
def split_solid_polygons(
segments: List[Dict[str, Any]],
) -> Tuple[List[List[Tuple[float, float]]], List[Dict[str, Any]]]:
"""Separate solid-region outline segments (tagged with solid_id /
poly_idx by the line extractor, round 25) from stroke segments, and
reconstruct the outlines' closed polygons in vertex order.
The polygons come back as authoritative shapes: they were closed by
construction (contours of filled regions) and must NOT be re-derived
through graph assembly — mixing them into the segment soup was
measured producing mangled hybrid faces whose boundaries jump between
a building's outline and unrelated stroke fragments ("odd shapes").
Segment endpoints may have shifted a few px from corner/T snapping;
each polygon vertex is taken from its edge's start point, which keeps
the ring consistent."""
by_solid: Dict[int, List[Dict[str, Any]]] = {}
strokes: List[Dict[str, Any]] = []
for s in segments:
sid = s.get("solid_id")
if sid is None:
strokes.append(s)
else:
by_solid.setdefault(sid, []).append(s)
polys: List[List[Tuple[float, float]]] = []
for sid in sorted(by_solid):
edges = sorted(by_solid[sid], key=lambda s: s["poly_idx"])
poly = [(float(e["x1"]), float(e["y1"])) for e in edges]
if len(poly) >= 3:
polys.append(poly)
return polys, strokes
def isoperimetric_ratio(poly: List[Tuple[float, float]]) -> float:
"""perimeter² / (4π·area); 1.0 for a circle, grows with elongation
and raggedness. Degenerate (zero-area) polygons return inf."""
n = len(poly)
if n < 3:
return float("inf")
perim = sum(math.dist(poly[i], poly[(i + 1) % n]) for i in range(n))
area = abs(sum(
poly[i][0] * poly[(i + 1) % n][1] - poly[(i + 1) % n][0] * poly[i][1]
for i in range(n)
)) / 2.0
if area <= 0:
return float("inf")
return (perim * perim) / (4.0 * math.pi * area)
def filter_spiky_faces(
faces: List[List[Tuple[float, float]]],
) -> List[List[Tuple[float, float]]]:
"""Drop assembled faces whose isoperimetric ratio marks them as
fragment tangles rather than drawing shapes (see _SPIKY_RATIO_MAX)."""
return [f for f in faces if isoperimetric_ratio(f) <= _SPIKY_RATIO_MAX]
def _cluster_nodes(
segments: List[Dict[str, Any]],
) -> Tuple[List[Tuple[float, float]], List[Tuple[int, int]]]:
"""Union-find endpoints into nodes; returns (node_positions, edges) with
edges as (node_a, node_b) per input segment (self-loops dropped)."""
pts = []
for s in segments:
pts.append((float(s["x1"]), float(s["y1"])))
pts.append((float(s["x2"]), float(s["y2"])))
n = len(pts)
parent = list(range(n))
def find(a: int) -> int:
while parent[a] != a:
parent[a] = parent[parent[a]]
a = parent[a]
return a
def union(a: int, b: int) -> None:
ra, rb = find(a), find(b)
if ra != rb:
parent[ra] = rb
for i in range(n):
for j in range(i + 1, n):
if (abs(pts[i][0] - pts[j][0]) <= _NODE_TOL_PX
and abs(pts[i][1] - pts[j][1]) <= _NODE_TOL_PX
and math.dist(pts[i], pts[j]) <= _NODE_TOL_PX):
union(i, j)
root_to_node: Dict[int, int] = {}
members: List[List[int]] = []
point_node = [0] * n
for i in range(n):
r = find(i)
if r not in root_to_node:
root_to_node[r] = len(members)
members.append([])
node_id = root_to_node[r]
members[node_id].append(i)
point_node[i] = node_id
positions = [
(sum(pts[i][0] for i in m) / len(m), sum(pts[i][1] for i in m) / len(m))
for m in members
]
edges = []
for k in range(len(segments)):
a, b = point_node[2 * k], point_node[2 * k + 1]
if a != b:
edges.append((a, b))
return positions, edges
def _split_edges_at_nodes(
positions: List[Tuple[float, float]], edges: List[Tuple[int, int]],
) -> List[Tuple[int, int]]:
"""Noding: an endpoint T-snapped ONTO another segment's interior
(line_extractor round 21) creates a node that lies ON an edge without
splitting it — graph-wise the junction doesn't exist, the T-wall's
node has degree 1, and it gets stripped as a spur (observed directly:
a floor plan whose wireframe visually showed closed rooms assembled to
0 closed shapes). Split every edge at the nodes on its interior."""
result: List[Tuple[int, int]] = []
for a, b in edges:
ax, ay = positions[a]
bx, by = positions[b]
vx, vy = bx - ax, by - ay
seg_len2 = vx * vx + vy * vy
if seg_len2 < 1e-9:
continue
cuts = []
for n, (nx, ny) in enumerate(positions):
if n == a or n == b:
continue
t = ((nx - ax) * vx + (ny - ay) * vy) / seg_len2
if t <= 0.0 or t >= 1.0:
continue
qx, qy = ax + t * vx, ay + t * vy
if math.dist((nx, ny), (qx, qy)) <= _NODE_TOL_PX:
cuts.append((t, n))
cuts.sort()
prev = a
for _, n in cuts:
if n != prev:
result.append((prev, n))
prev = n
if prev != b:
result.append((prev, b))
return result
def _strip_spurs(
positions: List[Tuple[float, float]], edges: List[Tuple[int, int]],
) -> Tuple[List[List[int]], List[Tuple[int, int]]]:
"""Iteratively remove degree-1 nodes, collecting the removed chains as
open node-paths; returns (open_chains, remaining_edges)."""
adj: Dict[int, List[int]] = {}
edge_set = set()
for idx, (a, b) in enumerate(edges):
adj.setdefault(a, []).append(idx)
adj.setdefault(b, []).append(idx)
edge_set.add(idx)
removed_at: Dict[int, List[int]] = {}
changed = True
while changed:
changed = False
for node, incident in list(adj.items()):
live = [e for e in incident if e in edge_set]
if len(live) == 1:
e = live[0]
edge_set.discard(e)
removed_at.setdefault(node, []).append(e)
changed = True
# Reconstruct open chains from removed edges by walking them
removed_edges = [edges[i] for i in range(len(edges)) if i not in edge_set]
chain_adj: Dict[int, List[int]] = {}
for a, b in removed_edges:
chain_adj.setdefault(a, []).append(b)
chain_adj.setdefault(b, []).append(a)
visited_pairs = set()
chains: List[List[int]] = []
for start, nbrs in chain_adj.items():
if len(nbrs) != 1:
continue # chain interiors and junction attach-points start elsewhere
nxt = nbrs[0]
if (start, nxt) in visited_pairs:
continue
chain = [start]
prev, cur = start, nxt
visited_pairs.add((start, nxt)); visited_pairs.add((nxt, start))
chain.append(cur)
while True:
options = [w for w in chain_adj.get(cur, [])
if w != prev and (cur, w) not in visited_pairs]
if len(options) != 1:
break
prev, cur = cur, options[0]
visited_pairs.add((prev, cur)); visited_pairs.add((cur, prev))
chain.append(cur)
chains.append(chain)
remaining = [edges[i] for i in sorted(edge_set)]
return chains, remaining
def _trace_faces(
positions: List[Tuple[float, float]], edges: List[Tuple[int, int]],
) -> List[List[int]]:
"""Planar half-edge face traversal: at each node, incident half-edges
are angularly sorted; from half-edge (u -> v) the walk continues with
the half-edge leaving v immediately clockwise of (v -> u). Every face
of the embedding is traced exactly once."""
incident: Dict[int, List[int]] = {}
for a, b in edges:
incident.setdefault(a, []).append(b)
incident.setdefault(b, []).append(a)
def angle(u: int, v: int) -> float:
return math.atan2(positions[v][1] - positions[u][1],
positions[v][0] - positions[u][0])
order: Dict[int, List[int]] = {
u: sorted(vs, key=lambda v: angle(u, v)) for u, vs in incident.items()
}
visited = set()
faces: List[List[int]] = []
for a, b in edges:
for u, v in ((a, b), (b, a)):
if (u, v) in visited:
continue
face = []
cu, cv = u, v
while (cu, cv) not in visited:
visited.add((cu, cv))
face.append(cu)
nbrs = order[cv]
i = nbrs.index(cu)
w = nbrs[(i - 1) % len(nbrs)] # next clockwise after the reverse edge
cu, cv = cv, w
if len(face) >= 3:
faces.append(face)
return faces
def _polygon_area(positions: List[Tuple[float, float]], face: List[int]) -> float:
s = 0.0
for i in range(len(face)):
x1, y1 = positions[face[i]]
x2, y2 = positions[face[(i + 1) % len(face)]]
s += x1 * y2 - x2 * y1
return s / 2.0
def assemble_shapes(segments: List[Dict[str, Any]]) -> Dict[str, List[List[Tuple[float, float]]]]:
"""
Parameters
----------
segments : the extractor's straight segments (curves excluded by the
caller — their many short polyline pieces belong to arcs, not wall
topology).
Returns
-------
{"closed": [poly, ...], "open": [poly, ...]} — each poly a list of
(x, y) points. Closed polys are the interior faces (rooms/outlines);
open polys are spur chains (dimension/leader lines etc.). Segments
whose endpoints merged into a single node (tiny) are dropped.
"""
if not segments:
return {"closed": [], "open": []}
positions, edges = _cluster_nodes(segments)
edges = _split_edges_at_nodes(positions, edges)
open_chains, cycle_edges = _strip_spurs(positions, edges)
closed: List[List[Tuple[float, float]]] = []
if cycle_edges:
faces = _trace_faces(positions, cycle_edges)
if faces:
areas = [_polygon_area(positions, f) for f in faces]
# outer face = largest absolute area; drop it, keep real rooms
outer = max(range(len(faces)), key=lambda i: abs(areas[i]))
for i, f in enumerate(faces):
if i == outer or abs(areas[i]) < _MIN_FACE_AREA_PX2:
continue
closed.append([positions[n] for n in f])
open_polys = [[positions[n] for n in chain] for chain in open_chains
if len(chain) >= 2]
return {"closed": closed, "open": open_polys}