"""Issue #329 — wall junction limits, Python mirror of src/junction-limits.ts. The card refuses a WRITE that would ADD a junction violation; this module is the backend half of that contract for `config/set` and `plan/optimize` (#333) — the two commands a client can use to write arbitrary geometry. Import and backup restore are DELIBERATELY outside the gate: #329 §3 promises that a restore is never blocked, so a crafted import can still persist violations — they are then inherited, never legalised as new ones. It mirrors П1–П4 exactly, including the two legal incidences of П4 (a shared node and a T-joint into the middle of a foreign wall) and the П3 rule that measures the collinear same-thickness WALL RUN rather than one catalogue atom. П5 (the room keeps at least 25 cm² of interior) is deliberately NOT mirrored: it is a statement about the rendered wall bodies, and reproducing the mitre / inset pipeline in Python would be a second geometry implementation whose drift is a larger risk than the rule it guards. A document that violates only П5 is ugly, not corrupt — see docs/specs/329-junction-limits.md §5. """ from __future__ import annotations import logging import math from typing import Any from .wall_segment_model import ( WALL_SEGMENT_MODEL_VERSION, WallSegmentMigrationError, commit_wall_segment_model, ) _LOGGER = logging.getLogger(__name__) MIN_JUNCTION_ANGLE_DEG = 15.0 MAX_JUNCTION_VALENCE = 6 MIN_SEGMENT_LENGTH_CM = 20.0 MIN_NODE_DISTANCE_CM = 5.0 GRID_STEP_N = 1 / 240 _EPS = 1e-9 # #331 §2.1: below this two points are ONE node / a node is ON the wall — # floating debris of pre-canonicalisation arithmetic, not a near miss. _INCIDENT_EPS = 2e-7 _KEY_FACTOR = 1e7 class JunctionLimitError(ValueError): """A write introduced a wall-junction violation (#329).""" def __init__(self, space_id: str, rule: str, subject: str, actual: float, limit: float) -> None: self.space_id = space_id self.rule = rule self.subject = subject self.actual = actual self.limit = limit self.code = f"junction_limit_{rule}" super().__init__( f"space={space_id}; rule={rule}; subject={subject}; " f"actual={actual:.12g}; limit={limit:.12g}" ) def _finite_point(point: object) -> bool: return ( isinstance(point, (list, tuple)) and len(point) >= 2 and all(isinstance(value, (int, float)) and math.isfinite(value) for value in point[:2]) ) def _quantize_key_coord(value: float) -> float: """#331 §2.1: the canonicalisation formula — sign*floor(|v|*f+0.5)/f. Native round() is banker's rounding and parts ways with JS Math-style rounding on .5 ticks; coordinate canonicalisation already encodes this parity lesson, and node keys must follow it. -0 normalises to 0. """ rounded = math.copysign( math.floor(abs(value) * _KEY_FACTOR + 0.5), value, ) / _KEY_FACTOR return 0.0 if rounded == 0 else rounded def _key(point: list[float]) -> str: return f"{_quantize_key_coord(point[0])},{_quantize_key_coord(point[1])}" def _length(a: list[float], b: list[float]) -> float: return math.hypot(b[0] - a[0], b[1] - a[1]) def cm_to_units(cm: float, cell_cm: float, grid_pitch: float = GRID_STEP_N) -> float: return (cm / (cell_cm or 1)) * grid_pitch def limit_segments(space: dict[str, Any]) -> list[dict[str, Any]]: """Every current wall the limits judge: contour atoms and partitions.""" segments: list[dict[str, Any]] = [] for segment in space.get("wall_segments") or []: if _finite_point(segment.get("a")) and _finite_point(segment.get("b")): segments.append({ "id": str(segment.get("id") or ""), "a": segment["a"], "b": segment["b"], "cm": float(segment.get("cm") or 0), }) for partition in space.get("partitions") or []: if _finite_point(partition.get("a")) and _finite_point(partition.get("b")): segments.append({ "id": str(partition.get("id") or ""), "a": partition["a"], "b": partition["b"], "cm": float(partition.get("cm") or 0), }) return [segment for segment in segments if _length(segment["a"], segment["b"]) > _EPS] def check_nodes(segments: list[dict[str, Any]]) -> list[tuple[str, str, float, float]]: """П1 + П2: valence of a node and the narrowest wedge in it.""" rays: dict[str, list[float]] = {} for segment in segments: for start, end in ((segment["a"], segment["b"]), (segment["b"], segment["a"])): rays.setdefault(_key(start), []).append( math.atan2(end[1] - start[1], end[0] - start[0]) ) violations: list[tuple[str, str, float, float]] = [] for node, angles in rays.items(): if len(angles) > MAX_JUNCTION_VALENCE: violations.append( ("valence", node, float(len(angles)), float(MAX_JUNCTION_VALENCE)) ) if len(angles) < 2: continue ordered = sorted(angles) smallest = math.inf for index, value in enumerate(ordered): delta = ordered[(index + 1) % len(ordered)] - value if index == len(ordered) - 1: delta += math.pi * 2 degrees = (delta * 180) / math.pi # A ~0° pair is NOT a violation — and cannot be (#331 revision 4, # learned in the field): a shared wall of two adjacent rooms is # two co-located owner atoms ON ONE LINE, so every shared-wall # node carries a legitimate 0° pair by construction. An exact # duplicate stays a KNOWN LIMITATION of П1 at this level. if _EPS < degrees < smallest: smallest = degrees if smallest < MIN_JUNCTION_ANGLE_DEG - 1e-9: violations.append(("angle", node, smallest, MIN_JUNCTION_ANGLE_DEG)) return violations def _axis_degrees(segment: dict[str, Any]) -> float: degrees = math.degrees(math.atan2( segment["b"][1] - segment["a"][1], segment["b"][0] - segment["a"][0] )) return (degrees % 180 + 180) % 180 def _collinear(left: dict[str, Any], right: dict[str, Any], tolerance_deg: float = 1.0) -> bool: delta = abs(_axis_degrees(left) - _axis_degrees(right)) return min(delta, 180 - delta) <= tolerance_deg def _build_node_index(segments: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]: by_node: dict[str, list[dict[str, Any]]] = {} for item in segments: for point in (item["a"], item["b"]): by_node.setdefault(_key(point), []).append(item) return by_node def collinear_run_length_units( segment: dict[str, Any], segments: list[dict[str, Any]], by_node_index: dict[str, list[dict[str, Any]]] | None = None, ) -> float: """Length of the WALL a segment belongs to, not of the atom. #331 §2.3/§2.4: an iterative edge walk over the collinear component — no recursion (a 10 000-atom chain must answer, not overflow), no combinatorial DFS (every atom joins at most once, O(E)), no silently dropped fork. Collinearity is measured against the BASE segment's axis, so an arc of small per-atom turns cannot pose as one straight wall. #330 §4.3: the per-check caller builds the node index once. """ by_node = by_node_index if by_node_index is not None else _build_node_index(segments) visited = {id(segment)} total = _length(segment["a"], segment["b"]) frontier = [segment["a"], segment["b"]] base_cm = float(segment.get("cm") or 0) while frontier: node = frontier.pop() for candidate in by_node.get(_key(node), ()): if id(candidate) in visited: continue if not _collinear(candidate, segment): continue if float(candidate.get("cm") or 0) != base_cm: continue visited.add(id(candidate)) total += _length(candidate["a"], candidate["b"]) frontier.append(candidate["a"]) frontier.append(candidate["b"]) return total def check_segment_lengths( segments: list[dict[str, Any]], cell_cm: float, grid_pitch: float = GRID_STEP_N, ) -> list[tuple[str, str, float, float]]: """П3: a wall is at least 20 cm and never shorter than its own thickness.""" violations: list[tuple[str, str, float, float]] = [] by_node = _build_node_index(segments) for segment in segments: units = collinear_run_length_units(segment, segments, by_node) cm = (units / grid_pitch) * (cell_cm or 1) thickness = float(segment.get("cm") or 0) limit = max(MIN_SEGMENT_LENGTH_CM, thickness if thickness > 0 else 0.0) if cm < limit - 1e-9: violations.append(( "length", str(segment.get("id") or _key(segment["a"])), cm, limit, )) return violations def _distance_to_segment(point: list[float], a: list[float], b: list[float]) -> float: dx, dy = b[0] - a[0], b[1] - a[1] length_sq = dx * dx + dy * dy t = 0.0 if length_sq <= _EPS else max(0.0, min(1.0, ( (point[0] - a[0]) * dx + (point[1] - a[1]) * dy ) / length_sq)) return math.hypot(point[0] - (a[0] + dx * t), point[1] - (a[1] + dy * t)) def check_node_distances( segments: list[dict[str, Any]], cell_cm: float, grid_pitch: float = GRID_STEP_N, ) -> list[tuple[str, str, float, float]]: """П4: non-incident nodes and node-to-foreign-wall clearance. #330 §4.5: the all-pairs form cost 372 ms on 576 atoms. Nodes and padded segment bounding boxes are hashed into a grid with the threshold as cell size; each node is compared only against its 9-cell neighbourhood. Verdicts are identical — pinned by the equivalence unit and the parity suite. Pair order inside the subject follows the lexicographic key order, which replaces the i _EPS else 1.0 node_grid: dict[tuple[int, int], list[Any]] = {} for node_key, point in nodes.items(): node_grid.setdefault( (int(point[0] // size), int(point[1] // size)), [] ).append((node_key, point)) segment_grid: dict[tuple[int, int], list[Any]] = {} for segment in segments: x0 = min(segment["a"][0], segment["b"][0]) - min_units x1 = max(segment["a"][0], segment["b"][0]) + min_units y0 = min(segment["a"][1], segment["b"][1]) - min_units y1 = max(segment["a"][1], segment["b"][1]) + min_units for cx in range(int(x0 // size), int(x1 // size) + 1): for cy in range(int(y0 // size), int(y1 // size) + 1): segment_grid.setdefault((cx, cy), []).append(segment) violations: list[tuple[str, str, float, float]] = [] for node_key, point in nodes.items(): cx, cy = int(point[0] // size), int(point[1] // size) for dx in (-1, 0, 1): for dy in (-1, 0, 1): for other_key, other in node_grid.get((cx + dx, cy + dy), ()): if node_key >= other_key: continue distance = _length(point, other) # #331 §2.1: raw debris within the incidence quantum is ONE # node on two neighbouring keys — never a near miss. if distance <= _INCIDENT_EPS: continue if distance < min_units - 1e-9: violations.append(( "distance", f"{node_key} ↔ {other_key}", (distance / grid_pitch) * (cell_cm or 1), MIN_NODE_DISTANCE_CM, )) for segment in segment_grid.get((cx, cy), ()): # A node that belongs to the wall is a legal corner or T-joint. if _key(segment["a"]) == node_key or _key(segment["b"]) == node_key: continue distance = _distance_to_segment(point, segment["a"], segment["b"]) # Sitting exactly ON the wall is the other legal incidence. if distance <= _INCIDENT_EPS: continue if distance < min_units - 1e-9: violations.append(( "distance", f"{node_key} → {segment.get('id') or _key(segment['a'])}", (distance / grid_pitch) * (cell_cm or 1), MIN_NODE_DISTANCE_CM, )) return violations def space_violations(space: dict[str, Any]) -> list[tuple[str, str, float, float]]: """П1–П4 over one space, in the frontend's order.""" segments = limit_segments(space) cell_cm = float(space.get("cell_cm") or 1) return [ *check_nodes(segments), *check_segment_lengths(segments, cell_cm), *check_node_distances(segments, cell_cm), ] def _migrated_spaces(config: dict[str, Any] | None, *, side: str = "previous") -> dict[str, dict[str, Any]]: """Spaces of one document AFTER the wall-segment migration, by id. The limits read `wall_segments`, so a document that predates the catalogue reports NO walls at all — a legacy space would answer "no violations" regardless of its geometry. Comparing such a baseline against a candidate the client already migrated counts every inherited violation as new and refuses an unrelated edit (spec §3 forbids exactly that). Both sides are therefore judged after the SAME migration, mirroring the frontend barrier. A document that cannot be migrated is not a reason to refuse the write: the wall-model barrier owns that verdict and reports it with its own code. Here it simply means there is no baseline to inherit from. """ if not isinstance(config, dict): return {} # #330 §4.6: a document that already carries the current catalogue is # used as-is. The migration's purpose here is to never compare a # catalogue-less legacy baseline against a migrated candidate; a # current-version document has the catalogue, so the purpose is met with # zero work — while a no-op re-migration costs 0.8 s of _atomize on a # 576-atom plan. Equivalence "as-is == migrated" for current-version # documents is pinned by a parity case. if int(config.get("model_version") or 0) >= WALL_SEGMENT_MODEL_VERSION: migrated = config else: try: migrated, _ = commit_wall_segment_model(config) except (WallSegmentMigrationError, ValueError) as error: # The wall-model barrier owns this verdict and reports it with its # own code; here it merely means "no catalogue to read". _LOGGER.debug("junction limits: %s migration fell back: %s", side, error) migrated = config except Exception: # #331 §2.6 (AC6): a genuine migration bug on the CANDIDATE side # must surface as an honest WS error, not a silent "no # violations". The PREVIOUS side keeps the wide fallback — an # unprovable inheritance is no reason to block an unrelated write # (the exact P1 symptom this task fixes). if side == "candidate": raise _LOGGER.debug("junction limits: previous-side migration bug swallowed by design") migrated = config return { str(space.get("id")): space for space in (migrated or {}).get("spaces") or [] if isinstance(space, dict) } def space_violation_counts(spaces: dict[str, dict[str, Any]]) -> dict[str, dict[str, int]]: """Violation counts per space per rule — the shape the barrier consumes and the rev cache (#330 §4.2) stores. Documents are NOT retained.""" result: dict[str, dict[str, int]] = {} for space_id, space in spaces.items(): counts: dict[str, int] = {} for rule, _subject, _actual, _limit in space_violations(space): counts[rule] = counts.get(rule, 0) + 1 result[space_id] = counts return result def validate_junction_limits( config: dict[str, Any], previous: dict[str, Any] | None = None, *, baseline_counts: dict[str, dict[str, int]] | None = None, ) -> dict[str, dict[str, int]]: """Refuse a write that ADDS a junction violation; inherit the rest. Counted per RULE, not per subject: a structural write re-atomises and re-keys contour segments, so subject identity is not stable across the barrier and matching by it would report an inherited violation as new (the mistake that refused legitimate resizes on the frontend). Spec §3: a write may keep existing violations, it may never add one. A LEGACY document still goes through `commit_wall_segment_model` first, so a catalogue-less baseline is compared in the same terms as the candidate; a current-version document is used as-is (#330 §4.6). `baseline_counts` (#330 §4.2) replaces the previous document entirely: the caller caches the counts of the stored document by config_rev, so a repeated write does not even touch `previous`. Returns the candidate's counts — the value to cache after a successful write. """ if baseline_counts is not None: old_counts = baseline_counts else: old_counts = space_violation_counts(_migrated_spaces(previous)) new_spaces = _migrated_spaces(config, side="candidate") candidate_counts: dict[str, dict[str, int]] = {} for space_id, space in new_spaces.items(): # A brand-new space has nothing to inherit from — but neither is a # first write allowed to arrive already broken. before = old_counts.get(space_id) or {} after: dict[str, list[tuple[str, str, float, float]]] = {} for violation in space_violations(space): after.setdefault(violation[0], []).append(violation) candidate_counts[space_id] = { rule: len(items) for rule, items in after.items() } for rule, items in after.items(): if len(items) > before.get(rule, 0): _, subject, actual, limit = items[0] raise JunctionLimitError(space_id, rule, subject, actual, limit) return candidate_counts