Spaces:
Sleeping
Sleeping
Aaron
Sync backend with GitHub main (0aa3b71): round-25 AI robustness + authoritative SVG shapes
1ef2a66 Download processors/shape_assembler.py from lylron/wonq-backend: direct link, hf CLI and curl.
- Browser
- Download file 12.7 kB
-
https://huggingface.co/spaces/lylron/wonq-backend/resolve/main/processors/shape_assembler.py
- Command line
-
hf download hf://spaces/lylron/wonq-backend/processors/shape_assembler.py
-
curl -L -o shape_assembler.py https://huggingface.co/spaces/lylron/wonq-backend/resolve/main/processors/shape_assembler.py
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} | |