from __future__ import annotations import base64 import binascii import hashlib import json import math import mimetypes import os import re import tempfile from collections import deque from copy import deepcopy from dataclasses import dataclass, field, replace from pathlib import Path from typing import Any try: from unfolding import MAX_LINES, UnfoldingDescription, UnfoldingLineSummary, clockwise_sort, order_points_clockwise except ImportError: # pragma: no cover - supports package imports in tests/tools. from .unfolding import MAX_LINES, UnfoldingDescription, UnfoldingLineSummary, clockwise_sort, order_points_clockwise MIN_NEIGHBOR_DISTANCE_MM = 34.0 DEFAULT_GRID_STEP_MM = MIN_NEIGHBOR_DISTANCE_MM MAX_NEIGHBOR_TOLERANCE_MM = 2.0 METER_LIKE_MAX_SPAN = 10.0 _DOCUMENT_KNOWN_FIELDS = { "background", "circles", "coordinate_unit", "grid_step_mm", "leds", "lines", "unit", "unfolding", } _LED_POINT_KNOWN_FIELDS = { "block_id", "block_index", "col", "column", "id", "line_id", "line_index", "manual", "row", "source_id", "x", "y", } class NominalGridDistance(float): """Grid spacing carrying a tolerance expressed in the same coordinate unit.""" def __new__(cls, value: float, tolerance: float): # noqa: ANN204 instance = super().__new__(cls, value) instance.tolerance = tolerance return instance def load_json_document(path: str | Path) -> dict[str, Any]: path = Path(path) if path.exists(): try: if path.stat().st_size == 0: path.write_text('{"circles": []}', encoding="utf-8") else: content = path.read_bytes() if not content.strip(): path.write_text('{"circles": []}', encoding="utf-8") except Exception: pass text, was_converted = _read_text_as_utf8(path) if was_converted: path.write_text(text, encoding="utf-8") data = _loads_lenient_json(text) if isinstance(data, list): return {"circles": data} if isinstance(data, dict): return data raise ValueError("JSON must contain an object or an array of LED points.") def _read_text_as_utf8(path: Path) -> tuple[str, bool]: raw = path.read_bytes() try: text = raw.decode("utf-8-sig") if _looks_like_json_text(text): return text, False except UnicodeDecodeError: pass for encoding in ("utf-16", "utf-16-le", "utf-16-be", "cp1251", "cp1252"): try: text = raw.decode(encoding) except UnicodeDecodeError: continue if _looks_like_json_text(text): return text.lstrip("\ufeff"), True return raw.decode("utf-8", errors="replace"), True def _looks_like_json_text(text: str) -> bool: if "\x00" in text[:64]: return False stripped = text.lstrip("\ufeff \t\r\n") return stripped.startswith("{") or stripped.startswith("[") def _loads_lenient_json(text: str) -> Any: try: return json.loads(text) except json.JSONDecodeError as original_error: fixed_text = _sanitize_raw_json_text(text) if fixed_text == text: raise original_error try: return json.loads(fixed_text) except json.JSONDecodeError: raise original_error def _sanitize_raw_json_text(text: str) -> str: result: list[str] = [] in_string = False escaped = False length = len(text) index = 0 while index < length: char = text[index] if in_string: result.append(char) if escaped: escaped = False elif char == "\\": escaped = True elif char == '"': in_string = False index += 1 continue if char == '"': in_string = True result.append(char) elif char == "," and _next_non_space(text, index + 1) in {"}", "]"}: pass elif char == "." and result and result[-1].isdigit() and not _next_char_is_digit(text, index + 1): result.append(".0") else: result.append(char) index += 1 return "".join(result) def _next_non_space(text: str, start: int) -> str | None: for index in range(start, len(text)): if not text[index].isspace(): return text[index] return None def _next_char_is_digit(text: str, start: int) -> bool: return start < len(text) and text[start].isdigit() @dataclass class LedPoint: uid: int source_id: int x: float y: float row: int col: int line_id: int | None = None block_id: int | None = None line_index: int | None = None block_index: int | None = None coordinate_unit: str = "auto" manual: bool = False extra_payload: dict[str, Any] = field(default_factory=dict) @dataclass class LedBlock: id: int name: str point_ids: list[int] = field(default_factory=list) @dataclass class LedLine: id: int name: str color: str blocks: list[LedBlock] = field(default_factory=list) @dataclass class BackgroundImage: path: str = "" offset_x: float = 0.0 offset_y: float = 0.0 scale_x_percent: float = 100.0 scale_y_percent: float = 100.0 opacity_percent: float = 100.0 visible: bool = True embedded_image_base64: str = "" embedded_image_mime_type: str = "" embedded_image_name: str = "" embedded_image_sha256: str = "" @classmethod def from_payload(cls, payload) -> "BackgroundImage": # noqa: ANN001 if not isinstance(payload, dict): return cls() embedded = payload.get("embedded_image") embedded_data = "" embedded_mime_type = "" embedded_name = "" embedded_sha256 = "" if isinstance(embedded, dict): candidate_data = str(embedded.get("data") or "") try: decoded = base64.b64decode(candidate_data, validate=True) if candidate_data else b"" except (binascii.Error, ValueError, TypeError): decoded = b"" candidate_sha256 = str(embedded.get("sha256") or "").strip().lower() actual_sha256 = hashlib.sha256(decoded).hexdigest() if decoded else "" if decoded and (not candidate_sha256 or candidate_sha256 == actual_sha256): embedded_data = candidate_data embedded_mime_type = str(embedded.get("mime_type") or "application/octet-stream") embedded_name = Path(str(embedded.get("name") or "background")).name embedded_sha256 = actual_sha256 scale = _optional_float(payload.get("scale_percent")) scale_x = _optional_float(payload.get("scale_x_percent")) scale_y = _optional_float(payload.get("scale_y_percent")) if scale_x is None: scale_x = 100.0 if scale is None else scale if scale_y is None: scale_y = 100.0 if scale is None else scale offset_x = _optional_float(payload.get("offset_x")) offset_y = _optional_float(payload.get("offset_y")) opacity = _optional_float(payload.get("opacity_percent")) return cls( path=str(payload.get("path") or ""), offset_x=0.0 if offset_x is None else offset_x, offset_y=0.0 if offset_y is None else offset_y, scale_x_percent=max(1.0, float(scale_x)), scale_y_percent=max(1.0, float(scale_y)), opacity_percent=min(100.0, max(0.0, 100.0 if opacity is None else float(opacity))), visible=bool(payload.get("visible", True)), embedded_image_base64=embedded_data, embedded_image_mime_type=embedded_mime_type, embedded_image_name=embedded_name, embedded_image_sha256=embedded_sha256, ) def to_payload(self) -> dict[str, Any]: payload: dict[str, Any] = { "path": self.path, "offset_x": self.offset_x, "offset_y": self.offset_y, "scale_x_percent": self.scale_x_percent, "scale_y_percent": self.scale_y_percent, "opacity_percent": self.opacity_percent, "visible": self.visible, } if self.has_embedded_image(): payload["embedded_image"] = { "encoding": "base64", "mime_type": self.embedded_image_mime_type or "application/octet-stream", "name": self.embedded_image_name or "background", "sha256": self.embedded_image_sha256 or hashlib.sha256(self.embedded_image_bytes()).hexdigest(), "data": self.embedded_image_base64, } return payload def is_enabled(self) -> bool: return bool(self.path) or self.has_embedded_image() def has_embedded_image(self) -> bool: return bool(self.embedded_image_base64) def embedded_image_bytes(self) -> bytes: if not self.embedded_image_base64: return b"" try: return base64.b64decode(self.embedded_image_base64, validate=True) except (binascii.Error, ValueError, TypeError): return b"" def with_embedded_image(self, base_dir: Path | None = None) -> "BackgroundImage": if self.has_embedded_image() or not self.path: return self image_path = Path(self.path) if not image_path.is_absolute() and base_dir is not None: image_path = base_dir / image_path image_bytes = image_path.read_bytes() if not image_bytes: raise ValueError("Файл подложки пуст и не может быть встроен в схему.") mime_type = mimetypes.guess_type(image_path.name)[0] or "application/octet-stream" return replace( self, path="", embedded_image_base64=base64.b64encode(image_bytes).decode("ascii"), embedded_image_mime_type=mime_type, embedded_image_name=image_path.name, embedded_image_sha256=hashlib.sha256(image_bytes).hexdigest(), ) class MappingDocument: def __init__(self) -> None: self.path: Path | None = None self.collection_name = "circles" self.points: list[LedPoint] = [] self.lines: list[LedLine] = [] self.background = BackgroundImage() self.unfolding = UnfoldingDescription() self.coordinate_unit = "mm" self.grid_step_mm = DEFAULT_GRID_STEP_MM self.extra_payload: dict[str, Any] = {} self._next_line_id = 1 self._next_block_id = 1 def clear(self) -> None: self.path = None self.collection_name = "circles" self.points.clear() self.lines.clear() self.background = BackgroundImage() self.unfolding = UnfoldingDescription() self.coordinate_unit = "mm" self.grid_step_mm = DEFAULT_GRID_STEP_MM self.extra_payload.clear() self._next_line_id = 1 self._next_block_id = 1 def load(self, path: str | Path) -> None: self.clear() self.path = Path(path) data = load_json_document(self.path) self.extra_payload = { key: value for key, value in data.items() if key not in _DOCUMENT_KNOWN_FIELDS } self.coordinate_unit = str(data.get("coordinate_unit", data.get("unit", "auto"))).strip().lower() self.grid_step_mm = _positive_finite_float(data.get("grid_step_mm"), DEFAULT_GRID_STEP_MM) points_data = data.get("leds") self.collection_name = "leds" if not isinstance(points_data, list): points_data = data.get("circles") self.collection_name = "circles" if not isinstance(points_data, list): raise ValueError("JSON must contain a 'circles' or 'leds' array.") for index, item in enumerate(points_data): self.points.append( LedPoint( uid=index, source_id=int(item.get("source_id", item.get("id", index))), x=float(item["x"]), y=float(item["y"]), row=_optional_int(item.get("row")) if item.get("row") is not None else index * 2, col=( _optional_int(item.get("col")) if item.get("col") is not None else _optional_int(item.get("column")) if item.get("column") is not None else 0 ), line_id=_optional_int(item.get("line_id")), block_id=_optional_int(item.get("block_id")), line_index=_optional_int(item.get("line_index")), block_index=_optional_int(item.get("block_index")), coordinate_unit=self.coordinate_unit, manual=bool(item.get("manual", False)), extra_payload={ key: value for key, value in item.items() if key not in _LED_POINT_KNOWN_FIELDS }, ) ) self._canonicalize_coordinate_unit() self._load_lines(data) self.background = BackgroundImage.from_payload(data.get("background")) self.unfolding = UnfoldingDescription.from_payload(data.get("unfolding")) self.unfolding.max_lines = MAX_LINES if not self.lines and any(point.line_id is not None for point in self.points): self._rebuild_lines_from_points() self._deduplicate_block_memberships() self._validate_line_limit() self.reindex_all() def _load_lines(self, data: dict[str, Any]) -> None: lines_data = data.get("lines", []) if not isinstance(lines_data, list): return source_to_uid = {point.source_id: point.uid for point in self.points} for line_item in lines_data: line_id = int(line_item.get("id", self._next_line_id)) line = LedLine( id=line_id, name=str(line_item.get("name", f"Line {line_id}")), color=str(line_item.get("color", default_color(line_id))), ) for block_item in line_item.get("blocks", []): block_id = int(block_item.get("id", self._next_block_id)) line.blocks.append( LedBlock( id=block_id, name=str(block_item.get("name", f"Block {block_id}")), point_ids=self._normalize_loaded_point_ids( block_item.get("point_ids", []), source_to_uid, str(line_item.get("point_id_mode", block_item.get("point_id_mode", "auto"))), ), ) ) self._next_block_id = max(self._next_block_id, block_id + 1) self.lines.append(line) self._next_line_id = max(self._next_line_id, line_id + 1) def _rebuild_lines_from_points(self) -> None: for line_id in sorted({point.line_id for point in self.points if point.line_id is not None}): line = LedLine(int(line_id), f"Line {line_id}", default_color(int(line_id))) block_ids = sorted({ point.block_id for point in self.points if point.line_id == line_id and point.block_id is not None }) if not block_ids: ids = [ point.uid for point in sorted( self.points, key=lambda point: ( point.line_index if point.line_index is not None else 10**9, point.source_id, ), ) if point.line_id == line_id ] if ids: line.blocks.append(LedBlock(self._next_block_id, "Block 1", ids)) self._next_block_id += 1 for block_id in block_ids: ids = [ point.uid for point in sorted( self.points, key=lambda point: ( point.block_index if point.block_index is not None else 10**9, point.source_id, ), ) if point.line_id == line_id and point.block_id == block_id ] line.blocks.append(LedBlock(int(block_id), f"Block {block_id}", ids)) self._next_block_id = max(self._next_block_id, int(block_id) + 1) self.lines.append(line) self._next_line_id = max(self._next_line_id, int(line_id) + 1) def save(self, path: str | Path) -> None: path = Path(path) self.reindex_all() source_base_dir = self.path.parent if self.path is not None else path.parent background_for_save = self.background.with_embedded_image(source_base_dir) payload = self.to_payload(background=background_for_save) serialized = json.dumps(payload, ensure_ascii=False, indent=2, allow_nan=False) if json.loads(serialized) != payload: raise ValueError("Не удалось без потерь сериализовать схему.") temporary_path: Path | None = None try: with tempfile.NamedTemporaryFile( mode="w", encoding="utf-8", newline="\n", dir=path.parent, prefix=f".{path.name}.", suffix=".tmp", delete=False, ) as temporary_file: temporary_file.write(serialized) temporary_file.write("\n") temporary_file.flush() os.fsync(temporary_file.fileno()) temporary_path = Path(temporary_file.name) os.replace(temporary_path, path) finally: if temporary_path is not None and temporary_path.exists(): temporary_path.unlink() self.path = path self.background = background_for_save def to_payload( self, *, point_payloads: list[dict[str, Any]] | None = None, lines_payload: list[dict[str, Any]] | None = None, collection_name: str | None = None, background: BackgroundImage | None = None, ) -> dict[str, Any]: payload = dict(self.extra_payload) payload[collection_name or self.collection_name] = ( [self.point_to_payload(point) for point in self.points] if point_payloads is None else point_payloads ) payload["unfolding"] = self.describe_unfolding().to_payload() payload["coordinate_unit"] = self._coordinate_unit_for_save() payload["grid_step_mm"] = self.grid_step_mm payload["background"] = (self.background if background is None else background).to_payload() payload["lines"] = self.lines_to_payload() if lines_payload is None else lines_payload return payload def point_to_payload(self, point: LedPoint, *, id_value: int | None = None) -> dict[str, Any]: payload = dict(point.extra_payload) payload.update( { "id": point.source_id if id_value is None else id_value, "x": point.x, "y": point.y, "row": point.row, "col": point.col, "line_id": point.line_id, "block_id": point.block_id, "line_index": point.line_index, "block_index": point.block_index, "manual": point.manual, } ) return payload def lines_to_payload(self, *, point_id_map: dict[int, int] | None = None) -> list[dict[str, Any]]: return [ { "id": line.id, "name": line.name, "color": line.color, "blocks": [ { "id": block.id, "name": block.name, "point_ids": [ point_id_map[point_id] if point_id_map is not None else point_id for point_id in block.point_ids if point_id_map is None or point_id in point_id_map ], } for block in line.blocks ], "point_id_mode": "uid", } for line in self.lines ] def add_line(self) -> LedLine: line_limit = self._line_limit() if len(self.lines) >= line_limit: raise ValueError(f"Mapping supports no more than {line_limit} lines.") line = LedLine(self._next_line_id, f"Линия {self._next_line_id}", default_color(self._next_line_id)) self._next_line_id += 1 self.lines.append(line) return line def add_manual_point(self, x: float, y: float) -> LedPoint: uid = len(self.points) source_id = max((point.source_id for point in self.points), default=-1) + 1 row, col = self._grid_cell_for_coordinate(float(x), float(y)) point = LedPoint( uid=uid, source_id=source_id, x=float(x), y=float(y), row=row, col=col, coordinate_unit=self._coordinate_unit_for_save(), manual=True, ) self.points.append(point) return point def add_point_copies( self, templates: list[LedPoint], *, offset_x: float, offset_y: float, ) -> list[LedPoint]: if not templates: return [] next_source_id = max((point.source_id for point in self.points), default=-1) + 1 copies: list[LedPoint] = [] for template in list(templates): x = float(template.x) + float(offset_x) y = float(template.y) + float(offset_y) row, col = self._grid_cell_for_coordinate(x, y) point = LedPoint( uid=len(self.points), source_id=next_source_id, x=x, y=y, row=row, col=col, coordinate_unit=self._coordinate_unit_for_save(), manual=True, extra_payload=deepcopy(template.extra_payload), ) self.points.append(point) copies.append(point) next_source_id += 1 return copies def update_point_position(self, point_id: int, x: float | None = None, y: float | None = None) -> None: point = self.get_point(point_id) if x is not None: point.x = float(x) if y is not None: point.y = float(y) point.row, point.col = self._grid_cell_for_coordinate(point.x, point.y, exclude_uid=point.uid) def update_manual_point_position(self, point_id: int, x: float | None = None, y: float | None = None) -> None: self.update_point_position(point_id, x=x, y=y) def remove_points(self, point_ids: list[int]) -> int: point_id_set = {point_id for point_id in point_ids if 0 <= point_id < len(self.points)} if not point_id_set: return 0 old_to_new: dict[int, int] = {} kept_points: list[LedPoint] = [] for point in self.points: if point.uid in point_id_set: continue old_to_new[point.uid] = len(kept_points) point.uid = len(kept_points) kept_points.append(point) self.points = kept_points for line in self.lines: for block in line.blocks: block.point_ids = [ old_to_new[point_id] for point_id in block.point_ids if point_id in old_to_new ] self.reindex_all() return len(point_id_set) def add_block(self, line_id: int, point_ids: list[int], *, after_block_id: int | None = None) -> LedBlock: line = self.get_line(line_id) clean_point_ids = unique_in_order([ point_id for point_id in point_ids if 0 <= point_id < len(self.points) ]) self._detach_points_from_blocks(clean_point_ids) block = LedBlock(len(line.blocks) + 1, f"Блок {len(line.blocks) + 1}", clean_point_ids) insert_index = len(line.blocks) if after_block_id is not None: for index, existing_block in enumerate(line.blocks): if existing_block.id == after_block_id: insert_index = index + 1 break line.blocks.insert(insert_index, block) for point_id in block.point_ids: point = self.points[point_id] point.line_id = line.id point.block_id = block.id self.reindex_all() return block def append_points_to_block(self, line_id: int, block_id: int, point_ids: list[int]) -> int: block = self._get_block(line_id, block_id) existing = set(block.point_ids) clean_point_ids = unique_in_order([ point_id for point_id in point_ids if 0 <= point_id < len(self.points) and point_id not in existing ]) if not clean_point_ids: return 0 self._detach_points_from_blocks(clean_point_ids) block = self._get_block(line_id, block_id) block.point_ids.extend(clean_point_ids) self.reindex_all() return len(clean_point_ids) def merge_blocks(self, line_id: int, block_ids: list[int]) -> LedBlock: line = self.get_line(line_id) target_ids = set(block_ids) if len(target_ids) < 2: raise ValueError("At least two blocks are required for merging.") selected_blocks = [block for block in line.blocks if block.id in target_ids] if len(selected_blocks) != len(target_ids): missing = target_ids - {block.id for block in selected_blocks} raise KeyError(f"Unknown block ids: {sorted(missing)}") merged_block = selected_blocks[0] merged_block.point_ids = unique_in_order( point_id for block in selected_blocks for point_id in block.point_ids ) line.blocks = [ block for block in line.blocks if block.id not in target_ids or block.id == merged_block.id ] for point_id in merged_block.point_ids: point = self.points[point_id] point.line_id = line.id point.block_id = merged_block.id self.reindex_all() return merged_block def get_line(self, line_id: int) -> LedLine: for line in self.lines: if line.id == line_id: return line raise KeyError(f"Unknown line id: {line_id}") def get_point(self, point_id: int) -> LedPoint: return self.points[point_id] def _get_block(self, line_id: int, block_id: int) -> LedBlock: line = self.get_line(line_id) for block in line.blocks: if block.id == block_id: return block raise KeyError(block_id) def clear_point_assignments(self, point_ids: list[int]) -> int: point_id_set = set(point_ids) removed = 0 for line in self.lines: for block in line.blocks: before = len(block.point_ids) block.point_ids = [point_id for point_id in block.point_ids if point_id not in point_id_set] removed += before - len(block.point_ids) for point_id in point_id_set: point = self.points[point_id] if point.line_id is not None or point.block_id is not None: point.line_id = None point.block_id = None point.line_index = None point.block_index = None self.reindex_all() return removed def remove_block(self, line_id: int, block_id: int) -> int: return self.remove_blocks([(line_id, block_id)]) def remove_blocks(self, block_keys: list[tuple[int, int]]) -> int: target_keys = set(block_keys) if not target_keys: return 0 existing_keys = { (line.id, block.id) for line in self.lines for block in line.blocks } missing_keys = target_keys - existing_keys if missing_keys: raise KeyError(f"Unknown block keys: {sorted(missing_keys)}") removed_ids: list[int] = [] for line in self.lines: kept_blocks: list[LedBlock] = [] for block in line.blocks: if (line.id, block.id) in target_keys: removed_ids.extend(block.point_ids) else: kept_blocks.append(block) line.blocks = kept_blocks unique_removed_ids = unique_in_order(removed_ids) self._clear_points(unique_removed_ids) self.reindex_all() return len(unique_removed_ids) def remove_line(self, line_id: int) -> int: removed_ids: list[int] = [] kept_lines: list[LedLine] = [] for line in self.lines: if line.id == line_id: for block in line.blocks: removed_ids.extend(block.point_ids) else: kept_lines.append(line) self.lines = kept_lines self._clear_points(removed_ids) self.renumber_lines() self.reindex_all() return len(removed_ids) def clear_all_assignments(self) -> int: assigned = sum(1 for point in self.points if point.line_id is not None or point.block_id is not None) self.lines.clear() self._clear_points([point.uid for point in self.points]) self._next_line_id = 1 self._next_block_id = 1 return assigned def renumber_lines(self) -> None: self.order_lines_clockwise() id_map: dict[int, int] = {} for new_id, line in enumerate(self.lines, start=1): old_id = line.id id_map[old_id] = new_id line.id = new_id if not line.name or re.match(r"^(Линия|Line)\s+\d+$", line.name): line.name = f"Линия {new_id}" line.color = default_color(new_id) for point in self.points: if point.line_id in id_map: point.line_id = id_map[point.line_id] self._next_line_id = len(self.lines) + 1 def _validate_line_limit(self) -> None: line_limit = self._line_limit() if len(self.lines) > line_limit: raise ValueError(f"Mapping supports no more than {line_limit} lines.") def _line_limit(self) -> int: return MAX_LINES def _coordinate_unit_for_save(self) -> str: if self.coordinate_unit in {"m", "meter", "meters"}: return "m" if self.coordinate_unit in {"mm", "millimeter", "millimeters"}: return "mm" self._canonicalize_coordinate_unit() return self.coordinate_unit def _canonicalize_coordinate_unit(self) -> None: if self.coordinate_unit in {"m", "meter", "meters"}: unit = "m" elif self.coordinate_unit in {"mm", "millimeter", "millimeters"}: unit = "mm" else: unit = "m" if self.points and coordinate_display_scale(self.points) == 1000.0 else "mm" self.coordinate_unit = unit for point in self.points: point.coordinate_unit = unit def grid_step_in_document_units(self) -> float: scale = 1000.0 if self._coordinate_unit_for_save() == "m" else 1.0 return NominalGridDistance(self.grid_step_mm / scale, MAX_NEIGHBOR_TOLERANCE_MM / scale) def _grid_cell_for_coordinate(self, x: float, y: float, *, exclude_uid: int | None = None) -> tuple[int, int]: anchors = [point for point in self.points if point.uid != exclude_uid and not point.manual] if not anchors: anchors = [point for point in self.points if point.uid != exclude_uid] if not anchors: return 0, 0 step = self.grid_step_in_document_units() row_axis, row_sign, col_axis, col_sign = infer_grid_coordinate_axes(anchors) source_row_value = x if row_axis == "x" else y source_col_value = x if col_axis == "x" else y row_anchor = min(anchors, key=lambda point: (abs((point.x if row_axis == "x" else point.y) - source_row_value), point.uid)) col_anchor = min(anchors, key=lambda point: (abs((point.x if col_axis == "x" else point.y) - source_col_value), point.uid)) row_value = row_anchor.x if row_axis == "x" else row_anchor.y col_value = col_anchor.x if col_axis == "x" else col_anchor.y row = row_anchor.row + round((source_row_value - row_value) / step * row_sign) col = col_anchor.col + round((source_col_value - col_value) / step * col_sign) return int(row), int(col) def order_lines_clockwise(self) -> None: self.lines = clockwise_sort(self.lines, self._points_for_line) def order_blocks_clockwise(self, line_id: int, *, reorder_leds: bool = False) -> None: line = self.get_line(line_id) if reorder_leds: for block in line.blocks: start_point_id = block.point_ids[0] if block.point_ids else None ordered_ids = [ point.uid for point in order_points_like_grid_svg( self._points_for_block(block), start_point_id, min_neighbor_distance=self.grid_step_in_document_units(), ) ] if start_point_id is not None: ordered_ids = rotate_ids_to_start(ordered_ids, start_point_id) block.point_ids = ordered_ids line.blocks = clockwise_sort(line.blocks, self._points_for_block) def describe_unfolding(self) -> UnfoldingDescription: summary = UnfoldingDescription( max_lines=self._line_limit(), line_order="clockwise", block_order="clockwise", led_index_scope="line", block_size=self.unfolding.block_size, ) for line in self.lines: led_count = sum(len(block.point_ids) for block in line.blocks) block_count = len(line.blocks) or summary.expected_block_count(led_count) summary.block_counts.append(UnfoldingLineSummary(line.id, led_count, block_count)) return summary def _points_for_line(self, line: LedLine) -> list[LedPoint]: return [point for block in line.blocks for point in self._points_for_block(block)] def _points_for_block(self, block: LedBlock) -> list[LedPoint]: return [self.points[point_id] for point_id in block.point_ids if 0 <= point_id < len(self.points)] def _normalize_loaded_point_ids( self, values: Any, source_to_uid: dict[int, int], mode: str = "auto", ) -> list[int]: if not isinstance(values, list): return [] raw_ids = [int(value) for value in values] if mode in {"source", "source_id"}: return unique_in_order([ source_to_uid[value] for value in raw_ids if value in source_to_uid ]) valid_uid_ids = [point_id for point_id in raw_ids if 0 <= point_id < len(self.points)] if mode == "uid": return unique_in_order(valid_uid_ids) if mode == "auto" and len(valid_uid_ids) == len(raw_ids): return unique_in_order(valid_uid_ids) return unique_in_order([ source_to_uid[value] for value in raw_ids if value in source_to_uid ]) def _deduplicate_block_memberships(self) -> None: seen: set[int] = set() for line in self.lines: for block in line.blocks: unique_ids = [] for point_id in block.point_ids: if not 0 <= point_id < len(self.points): continue if point_id in seen: continue seen.add(point_id) unique_ids.append(point_id) block.point_ids = unique_ids def _detach_points_from_blocks(self, point_ids: list[int]) -> int: if not point_ids: return 0 point_id_set = set(point_ids) removed = 0 for line in self.lines: for block in line.blocks: before = len(block.point_ids) block.point_ids = [point_id for point_id in block.point_ids if point_id not in point_id_set] removed += before - len(block.point_ids) self._clear_points(list(point_id_set)) return removed def _clear_points(self, point_ids: list[int]) -> None: for point_id in point_ids: point = self.points[point_id] point.line_id = None point.block_id = None point.line_index = None point.block_index = None def auto_connect_line(self, line_id: int) -> None: line = self.get_line(line_id) self.order_blocks_clockwise(line.id, reorder_leds=True) self.reindex_line(line.id) def rebuild_line_blocks_by_route(self, line_id: int) -> int: line = self.get_line(line_id) point_ids = [point_id for block in line.blocks for point_id in block.point_ids] if not point_ids: return 0 start_ids = [block.point_ids[0] for block in line.blocks if block.point_ids] start_id = start_ids[0] if start_ids else point_ids[0] groups = self.route_groups_from_points(point_ids, start_id) line.blocks.clear() for block_number, group in enumerate(groups, start=1): line.blocks.append(LedBlock(block_number, f"Блок {block_number}", group)) self.reindex_all() return len(groups) def auto_blocks_from_selection(self, line_id: int, point_ids: list[int], groups: list[list[int]] | None = None) -> None: groups = self.auto_block_preview(point_ids) if groups is None else groups if not groups: return self._detach_points_from_blocks([point_id for group in groups for point_id in group]) line = self.get_line(line_id) for group_ids in groups: block_number = len(line.blocks) + 1 block = LedBlock(block_number, f"Блок {block_number}", unique_in_order(group_ids)) line.blocks.append(block) self.order_blocks_clockwise(line_id, reorder_leds=False) self.reindex_all() def route_groups_from_points(self, point_ids: list[int], start_point_id: int | None = None) -> list[list[int]]: points = [ self.points[point_id] for point_id in unique_in_order(point_ids) if 0 <= point_id < len(self.points) ] return [ [point.uid for point in route] for route in trace_route_paths_like_grid_svg( points, start_point_id, min_neighbor_distance=self.grid_step_in_document_units(), ) ] def add_clockwise_block_from_seed(self, line_id: int, start_point_id: int) -> LedBlock: if not 0 <= start_point_id < len(self.points): raise ValueError(f"Unknown LED point id: {start_point_id}") group_ids = self.clockwise_block_from_seed_preview(start_point_id) return self.add_clockwise_block_from_group(line_id, start_point_id, group_ids) def add_counterclockwise_block_from_seed(self, line_id: int, start_point_id: int) -> LedBlock: if not 0 <= start_point_id < len(self.points): raise ValueError(f"Unknown LED point id: {start_point_id}") group_ids = self.counterclockwise_block_from_seed_preview(start_point_id) return self.add_clockwise_block_from_group(line_id, start_point_id, group_ids) def add_clockwise_block_from_group(self, line_id: int, start_point_id: int, group_ids: list[int]) -> LedBlock: if not 0 <= start_point_id < len(self.points): raise ValueError(f"Unknown LED point id: {start_point_id}") block = self.add_block(line_id, group_ids) block.point_ids = rotate_ids_to_start(block.point_ids, start_point_id) self.reindex_line(line_id) return block def clockwise_block_from_seed_preview(self, start_point_id: int) -> list[int]: return self.route_block_from_seed_preview(start_point_id, direction="clockwise") def counterclockwise_block_from_seed_preview(self, start_point_id: int) -> list[int]: return self.route_block_from_seed_preview(start_point_id, direction="counterclockwise") def route_block_from_seed_preview(self, start_point_id: int, *, direction: str = "clockwise") -> list[int]: if not 0 <= start_point_id < len(self.points): return [] candidate_points = [ point for point in self.points if point.uid == start_point_id or (point.line_id is None and point.block_id is None) ] threshold = self.grid_step_in_document_units() adjacency = build_route_adjacency(candidate_points, threshold) route = trace_single_route_like_grid_svg( candidate_points, start_point_id, adjacency=adjacency, allowed_ids={point.uid for point in candidate_points}, min_neighbor_distance=threshold, direction=direction, ) if not route: return [start_point_id] return rotate_ids_to_start( [point.uid for point in normalize_route_direction(route, threshold, direction=direction)], start_point_id, ) def auto_block_preview(self, point_ids: list[int]) -> list[list[int]]: points = [ self.points[point_id] for point_id in sorted_unique(point_ids) if 0 <= point_id < len(self.points) ] return [ [point.uid for point in route] for route in trace_route_paths_like_grid_svg( points, min_neighbor_distance=self.grid_step_in_document_units(), ) ] def reindex_all(self) -> None: for point in self.points: point.line_id = None point.block_id = None point.line_index = None point.block_index = None for line in self.lines: self.reindex_line(line.id) def reindex_line(self, line_id: int) -> None: line = self.get_line(line_id) self.renumber_blocks(line) line_index = 0 for block in line.blocks: for block_index, point_id in enumerate(block.point_ids): point = self.points[point_id] point.line_id = line.id point.block_id = block.id point.line_index = line_index point.block_index = block_index line_index += 1 @staticmethod def renumber_blocks(line: LedLine) -> None: for block_number, block in enumerate(line.blocks, start=1): block.id = block_number if not block.name or re.match(r"^(Блок|Block)\s+\d+$", block.name): block.name = f"Блок {block_number}" def sorted_unique(values: list[int]) -> list[int]: return sorted(set(values)) def unique_in_order(values: list[int]) -> list[int]: seen: set[int] = set() result: list[int] = [] for value in values: if value in seen: continue seen.add(value) result.append(value) return result def rotate_ids_to_start(values: list[int], start_value: int) -> list[int]: if start_value not in values: return values[:] start_index = values.index(start_value) return [*values[start_index:], *values[:start_index]] def normalize_route_clockwise(route: list[LedPoint], min_neighbor_distance: float) -> list[LedPoint]: return normalize_route_direction(route, min_neighbor_distance, direction="clockwise") def normalize_route_direction( route: list[LedPoint], min_neighbor_distance: float, *, direction: str = "clockwise", ) -> list[LedPoint]: if len(route) < 3: return route[:] if not is_valid_route_neighbor(route[-1], route[0], min_neighbor_distance): return route[:] signed_area = route_signed_area(route) if direction == "counterclockwise": return list(reversed(route)) if signed_area < 0 else route[:] return list(reversed(route)) if signed_area > 0 else route[:] def route_signed_area(route: list[LedPoint]) -> float: area = 0.0 for current, next_point in zip(route, [*route[1:], route[0]]): area += current.x * next_point.y - next_point.x * current.y return area / 2.0 def _optional_int(value: Any) -> int | None: return None if value is None else int(value) def _optional_float(value: Any) -> float | None: try: return None if value is None else float(value) except (TypeError, ValueError): return None def _positive_finite_float(value: Any, fallback: float) -> float: parsed = _optional_float(value) if parsed is None or not math.isfinite(parsed) or parsed <= 0.0: return fallback return min(1000.0, max(1.0, parsed)) def infer_grid_coordinate_axes(points: list[LedPoint]) -> tuple[str, int, str, int]: row_x = _axis_correlation(points, "row", "x") row_y = _axis_correlation(points, "row", "y") col_x = _axis_correlation(points, "col", "x") col_y = _axis_correlation(points, "col", "y") standard_score = abs(row_y) + abs(col_x) transposed_score = abs(row_x) + abs(col_y) if transposed_score > standard_score: return "x", _correlation_sign(row_x), "y", _correlation_sign(col_y) return "y", _correlation_sign(row_y), "x", _correlation_sign(col_x) def _axis_correlation(points: list[LedPoint], grid_attr: str, coordinate_attr: str) -> float: if len(points) < 2: return 0.0 grid_values = [float(getattr(point, grid_attr)) for point in points] coordinate_values = [float(getattr(point, coordinate_attr)) for point in points] grid_mean = sum(grid_values) / len(grid_values) coordinate_mean = sum(coordinate_values) / len(coordinate_values) covariance = sum( (grid_value - grid_mean) * (coordinate_value - coordinate_mean) for grid_value, coordinate_value in zip(grid_values, coordinate_values) ) grid_variance = sum((grid_value - grid_mean) ** 2 for grid_value in grid_values) coordinate_variance = sum((coordinate_value - coordinate_mean) ** 2 for coordinate_value in coordinate_values) if grid_variance <= 0.0 or coordinate_variance <= 0.0: return 0.0 return covariance / math.sqrt(grid_variance * coordinate_variance) def _correlation_sign(value: float) -> int: return -1 if value < 0.0 else 1 def default_color(line_id: int) -> str: colors = ["#e11d48", "#2563eb", "#16a34a", "#f59e0b", "#7c3aed", "#0891b2"] return colors[(line_id - 1) % len(colors)] def distance(a: LedPoint, b: LedPoint) -> float: return math.hypot(a.x - b.x, a.y - b.y) def are_neighbors(a: LedPoint, b: LedPoint) -> bool: return abs(a.row - b.row) <= 1 and abs(a.col - b.col) <= 1 def connected_groups(points: list[LedPoint]) -> list[list[LedPoint]]: remaining = set(point.uid for point in points) by_id = {point.uid: point for point in points} groups: list[list[LedPoint]] = [] while remaining: start = remaining.pop() queue = [start] group = [by_id[start]] while queue: current_id = queue.pop() current = by_id[current_id] for candidate_id in list(remaining): if are_neighbors(current, by_id[candidate_id]): remaining.remove(candidate_id) queue.append(candidate_id) group.append(by_id[candidate_id]) groups.append(group) return sorted(groups, key=lambda group: min(point.uid for point in group)) def contour_groups_like_grid_svg(points: list[LedPoint]) -> list[list[LedPoint]]: points_by_cell: dict[tuple[int, int], list[LedPoint]] = {} for point in points: points_by_cell.setdefault((point.row, point.col), []).append(point) cells = set(points_by_cell) seen: set[tuple[int, int]] = set() groups: list[list[LedPoint]] = [] neighbors = ((1, 0), (-1, 0), (0, 1), (0, -1)) for start_cell in sorted(cells): if start_cell in seen: continue queue = deque([start_cell]) seen.add(start_cell) group_cells: list[tuple[int, int]] = [] while queue: row, col = queue.popleft() group_cells.append((row, col)) for d_row, d_col in neighbors: next_cell = (row + d_row, col + d_col) if next_cell in cells and next_cell not in seen: seen.add(next_cell) queue.append(next_cell) groups.append([ point for cell in group_cells for point in points_by_cell[cell] ]) return sorted(groups, key=lambda group: min(point_sort_key_for_model(point) for point in group)) def route_groups_like_grid_svg( points: list[LedPoint], min_neighbor_distance: float | None = None, ) -> list[list[LedPoint]]: points_by_cell: dict[tuple[int, int], list[LedPoint]] = {} for point in points: points_by_cell.setdefault((point.row, point.col), []).append(point) cells = set(points_by_cell) seen: set[tuple[int, int]] = set() groups: list[list[LedPoint]] = [] neighbors = ((1, 0), (-1, 0), (0, 1), (0, -1), (1, 1), (-1, 1), (-1, -1), (1, -1)) threshold = min_neighbor_distance or min_neighbor_distance_for_points(points) for start_cell in sorted(cells): if start_cell in seen: continue queue = deque([start_cell]) seen.add(start_cell) group_cells: list[tuple[int, int]] = [] while queue: row, col = queue.popleft() group_cells.append((row, col)) for d_row, d_col in neighbors: next_cell = (row + d_row, col + d_col) if next_cell not in cells or next_cell in seen: continue if not cells_have_valid_route_neighbor(points_by_cell[(row, col)], points_by_cell[next_cell], threshold): continue seen.add(next_cell) queue.append(next_cell) groups.append([ point for cell in group_cells for point in points_by_cell[cell] ]) return sorted(groups, key=lambda group: min(point_sort_key_for_model(point) for point in group)) def cells_have_valid_route_neighbor( first_points: list[LedPoint], second_points: list[LedPoint], threshold: float, ) -> bool: return any( distance(first, second) <= max_neighbor_distance(threshold) for first in first_points for second in second_points ) def order_group_nearest(points: list[LedPoint]) -> list[LedPoint]: if len(points) <= 1: return points[:] remaining = points[:] ordered = [min(remaining, key=lambda point: (point.row, point.col, point.source_id))] remaining.remove(ordered[0]) while remaining: previous = ordered[-1] next_point = min(remaining, key=lambda point: (distance(previous, point), point.row, point.col)) ordered.append(next_point) remaining.remove(next_point) return ordered def order_points_like_grid_svg( points: list[LedPoint], start_point_id: int | None = None, *, min_neighbor_distance: float | None = None, ) -> list[LedPoint]: if len(points) <= 1: return points[:] min_neighbor_distance = min_neighbor_distance or min_neighbor_distance_for_points(points) remaining = points[:] if start_point_id is None: start_point = min(remaining, key=point_sort_key_for_model) else: start_point = next((point for point in remaining if point.uid == start_point_id), None) if start_point is None: start_point = min(remaining, key=point_sort_key_for_model) ordered = [start_point] remaining.remove(start_point) while remaining: previous = ordered[-1] neighbors = [ point for point in remaining if is_valid_route_neighbor(previous, point, min_neighbor_distance) ] if neighbors: next_point = min( neighbors, key=lambda point: route_neighbor_preference(previous, point, min_neighbor_distance), ) else: next_point = min( remaining, key=lambda point: ( distance(previous, point), point_sort_key_for_model(point), ), ) ordered.append(next_point) remaining.remove(next_point) return ordered def trace_route_paths_like_grid_svg( points: list[LedPoint], start_point_id: int | None = None, *, min_neighbor_distance: float | None = None, ) -> list[list[LedPoint]]: remaining = points[:] remaining_ids = {point.uid for point in remaining} routes: list[list[LedPoint]] = [] preferred_start_id = start_point_id min_neighbor_distance = min_neighbor_distance or min_neighbor_distance_for_points(points) while remaining: if preferred_start_id is not None: start_point = next((point for point in remaining if point.uid == preferred_start_id), None) else: start_point = None if start_point is None: start_point = min(remaining, key=point_sort_key_for_model) adjacency = build_route_adjacency(remaining, min_neighbor_distance) route = trace_single_route_like_grid_svg( remaining, start_point.uid, adjacency=adjacency, allowed_ids=remaining_ids, min_neighbor_distance=min_neighbor_distance, ) if not route: break route_ids = {point.uid for point in route} routes.append(route) remaining_ids -= route_ids remaining = [point for point in remaining if point.uid in remaining_ids] preferred_start_id = None return routes def trace_single_route_like_grid_svg( points: list[LedPoint], start_point_id: int, adjacency: dict[int, list[LedPoint]] | None = None, allowed_ids: set[int] | None = None, min_neighbor_distance: float | None = None, direction: str = "clockwise", ) -> list[LedPoint]: if not points: return [] min_neighbor_distance = min_neighbor_distance or min_neighbor_distance_for_points(points) adjacency = adjacency or build_route_adjacency(points, min_neighbor_distance) remaining = points[:] remaining_ids = set(allowed_ids) if allowed_ids is not None else {point.uid for point in remaining} start_point = next((point for point in remaining if point.uid == start_point_id), None) if start_point is None: start_point = min(remaining, key=point_sort_key_for_model) route = [start_point] remaining.remove(start_point) remaining_ids.remove(start_point.uid) while remaining: previous = route[-1] neighbors = [ point for point in adjacency.get(previous.uid, []) if point.uid in remaining_ids ] if not neighbors: neighbors = fallback_route_neighbors( route, previous, remaining, remaining_ids, min_neighbor_distance, ) if not neighbors: break next_point = choose_next_route_point(route, neighbors, min_neighbor_distance, direction=direction) route.append(next_point) remaining.remove(next_point) remaining_ids.remove(next_point.uid) return route def fallback_route_neighbors( route: list[LedPoint], current: LedPoint, points: list[LedPoint], allowed_ids: set[int], min_neighbor_distance: float, ) -> list[LedPoint]: valid_neighbors = [ point for point in points if point.uid in allowed_ids and is_valid_route_neighbor(current, point, min_neighbor_distance) ] if valid_neighbors or len(route) < 2: return valid_neighbors trajectory_neighbors = [ point for point in points if point.uid in allowed_ids and is_trajectory_continuation(route[-2], current, point, min_neighbor_distance) ] return trajectory_neighbors def choose_next_route_point( route: list[LedPoint], neighbors: list[LedPoint], min_neighbor_distance: float, *, direction: str = "clockwise", ) -> LedPoint: if len(route) < 2: return min(neighbors, key=lambda point: route_path_preference(route, point, min_neighbor_distance, direction=direction)) neighbors = nearest_neighbor_band(route[-1], neighbors, min_neighbor_distance) previous = route[-2] current = route[-1] same_vector_direction = [ point for point in neighbors if same_direction_vector(previous, current, point) ] if same_vector_direction: return min( same_vector_direction, key=lambda point: route_path_preference(route, point, min_neighbor_distance, direction=direction), ) forward_direction = [ point for point in neighbors if forward_direction_vector(previous, current, point) ] if forward_direction: return min( forward_direction, key=lambda point: route_path_preference(route, point, min_neighbor_distance, direction=direction), ) return min(neighbors, key=lambda point: route_path_preference(route, point, min_neighbor_distance, direction=direction)) def nearest_neighbor_band( current: LedPoint, neighbors: list[LedPoint], min_neighbor_distance: float, ) -> list[LedPoint]: nearest_distance = min(distance(current, point) for point in neighbors) band_limit = min(nearest_distance + neighbor_tolerance(min_neighbor_distance), max_neighbor_distance(min_neighbor_distance)) band = [point for point in neighbors if distance(current, point) <= band_limit] return band or [min(neighbors, key=lambda point: distance(current, point))] def same_direction_vector(previous: LedPoint, current: LedPoint, candidate: LedPoint) -> bool: incoming_x = current.x - previous.x incoming_y = current.y - previous.y outgoing_x = candidate.x - current.x outgoing_y = candidate.y - current.y if incoming_x == 0 and incoming_y == 0: return False if outgoing_x == 0 and outgoing_y == 0: return False dot = incoming_x * outgoing_x + incoming_y * outgoing_y cross = incoming_x * outgoing_y - incoming_y * outgoing_x return dot > 0 and abs(cross) <= max(1e-6, 0.1 * math.hypot(incoming_x, incoming_y) * math.hypot(outgoing_x, outgoing_y)) def forward_direction_vector(previous: LedPoint, current: LedPoint, candidate: LedPoint) -> bool: incoming_x = current.x - previous.x incoming_y = current.y - previous.y outgoing_x = candidate.x - current.x outgoing_y = candidate.y - current.y incoming_length = math.hypot(incoming_x, incoming_y) outgoing_length = math.hypot(outgoing_x, outgoing_y) if incoming_length == 0 or outgoing_length == 0: return False dot = incoming_x * outgoing_x + incoming_y * outgoing_y cos_angle = dot / (incoming_length * outgoing_length) return cos_angle > -0.25 def is_trajectory_continuation( previous: LedPoint, current: LedPoint, candidate: LedPoint, min_neighbor_distance: float, ) -> bool: if not forward_direction_vector(previous, current, candidate): return False return distance(current, candidate) <= route_continuation_distance_limit(current, candidate, min_neighbor_distance) def route_continuation_distance_limit(current: LedPoint, candidate: LedPoint, min_neighbor_distance: float) -> float: return min_neighbor_distance * 2.0 + neighbor_tolerance(min_neighbor_distance) def build_route_adjacency( points: list[LedPoint], min_neighbor_distance: float, ) -> dict[int, list[LedPoint]]: search_radius = min_neighbor_distance * math.sqrt(2.0) + neighbor_tolerance(min_neighbor_distance) bucket_size = max(search_radius, 1e-9) points_by_bucket: dict[tuple[int, int], list[LedPoint]] = {} for point in points: bucket = (math.floor(point.x / bucket_size), math.floor(point.y / bucket_size)) points_by_bucket.setdefault(bucket, []).append(point) raw_adjacency: dict[int, list[LedPoint]] = {} for point in points: candidates: list[LedPoint] = [] bucket = (math.floor(point.x / bucket_size), math.floor(point.y / bucket_size)) for d_x in (-1, 0, 1): for d_y in (-1, 0, 1): for candidate in points_by_bucket.get((bucket[0] + d_x, bucket[1] + d_y), []): if candidate.uid == point.uid: continue if is_valid_route_neighbor(point, candidate, min_neighbor_distance): candidates.append(candidate) raw_adjacency[point.uid] = candidates adjacency: dict[int, list[LedPoint]] = {} for point in points: candidates = [ candidate for candidate in raw_adjacency[point.uid] if ( not route_edge_skips_intermediate_point( point, candidate, points_by_bucket, bucket_size, min_neighbor_distance, ) and not is_parallel_track_jump(point, candidate, raw_adjacency) ) ] adjacency[point.uid] = sorted( candidates, key=lambda candidate: route_neighbor_preference(point, candidate, min_neighbor_distance), ) return adjacency def route_edge_skips_intermediate_point( current: LedPoint, candidate: LedPoint, points_by_bucket: dict[tuple[int, int], list[LedPoint]], bucket_size: float, min_neighbor_distance: float, ) -> bool: segment_x = candidate.x - current.x segment_y = candidate.y - current.y segment_length_sq = segment_x * segment_x + segment_y * segment_y if segment_length_sq <= 0: return False segment_length = math.sqrt(segment_length_sq) line_tolerance = max(neighbor_tolerance(min_neighbor_distance), min_neighbor_distance * 0.25) end_tolerance = max(1e-6, line_tolerance * 0.25) min_bucket_x = math.floor((min(current.x, candidate.x) - line_tolerance) / bucket_size) max_bucket_x = math.floor((max(current.x, candidate.x) + line_tolerance) / bucket_size) min_bucket_y = math.floor((min(current.y, candidate.y) - line_tolerance) / bucket_size) max_bucket_y = math.floor((max(current.y, candidate.y) + line_tolerance) / bucket_size) nearby_points = ( point for bucket_x in range(min_bucket_x, max_bucket_x + 1) for bucket_y in range(min_bucket_y, max_bucket_y + 1) for point in points_by_bucket.get((bucket_x, bucket_y), []) ) for point in nearby_points: if point.uid in {current.uid, candidate.uid}: continue point_x = point.x - current.x point_y = point.y - current.y projection = (point_x * segment_x + point_y * segment_y) / segment_length_sq if projection <= 0.0 or projection >= 1.0: continue perpendicular = abs(point_x * segment_y - point_y * segment_x) / segment_length if perpendicular > line_tolerance: continue if distance(current, point) < segment_length - end_tolerance: return True return False def is_parallel_track_jump( current: LedPoint, candidate: LedPoint, raw_adjacency: dict[int, list[LedPoint]], ) -> bool: d_x = candidate.x - current.x d_y = candidate.y - current.y if abs(d_x) > abs(d_y): return ( route_neighbor_axis_count(current, raw_adjacency, "vertical") >= 2 and route_neighbor_axis_count(candidate, raw_adjacency, "vertical") >= 2 ) if abs(d_y) > abs(d_x): return ( route_neighbor_axis_count(current, raw_adjacency, "horizontal") >= 2 and route_neighbor_axis_count(candidate, raw_adjacency, "horizontal") >= 2 ) return False def route_neighbor_axis_count( point: LedPoint, raw_adjacency: dict[int, list[LedPoint]], axis: str, ) -> int: if axis == "horizontal": return sum(1 for candidate in raw_adjacency[point.uid] if abs(candidate.x - point.x) > abs(candidate.y - point.y)) return sum(1 for candidate in raw_adjacency[point.uid] if abs(candidate.y - point.y) > abs(candidate.x - point.x)) def trace_closed_route( points: list[LedPoint], start_point: LedPoint, min_neighbor_distance: float, adjacency: dict[int, list[LedPoint]] | None = None, allowed_ids: set[int] | None = None, ) -> list[LedPoint]: adjacency = adjacency or build_route_adjacency(points, min_neighbor_distance) allowed_ids = set(allowed_ids) if allowed_ids is not None else {point.uid for point in points} if len(adjacency.get(start_point.uid, [])) < 2: return [] route = [start_point] visited = {start_point.uid} stack: list[tuple[LedPoint, list[LedPoint], int]] = [( start_point, [point for point in adjacency[start_point.uid] if point.uid in allowed_ids], 0, )] attempts = 0 max_attempts = min(5000, max(1000, len(points) * 25)) while stack and attempts < max_attempts: current, candidates, next_index = stack[-1] attempts += 1 meaningful_continuations = [ point for point in adjacency.get(current.uid, []) if point.uid in allowed_ids and point.uid not in visited and len([next_point for next_point in adjacency.get(point.uid, []) if next_point.uid in allowed_ids]) > 1 ] if ( len(route) >= 4 and is_valid_route_neighbor(current, start_point, min_neighbor_distance) and not meaningful_continuations ): return route[:] if next_index >= len(candidates): stack.pop() removed = route.pop() visited.remove(removed.uid) continue candidate = candidates[next_index] stack[-1] = (current, candidates, next_index + 1) if candidate.uid in visited: continue visited.add(candidate.uid) route.append(candidate) next_candidates = [ point for point in adjacency.get(candidate.uid, []) if point.uid in allowed_ids and point.uid not in visited ] stack.append((candidate, next_candidates, 0)) return [] def point_sort_key_for_model(point: LedPoint) -> tuple[int, int, float, float, int]: return point.row, point.col, point.x, point.y, point.uid def too_close_preference(current: LedPoint, candidate: LedPoint, min_distance: float) -> int: return 1 if distance(current, candidate) < min_distance else 0 def route_neighbor_preference( current: LedPoint, candidate: LedPoint, min_distance: float, ) -> tuple[int, int, int, float, tuple[int, int, float, float, int]]: return ( too_close_preference(current, candidate, min_distance), diagonal_cell_preference(current, candidate), cardinal_direction_preference(current, candidate), distance(current, candidate), point_sort_key_for_model(candidate), ) def route_path_preference( route: list[LedPoint], candidate: LedPoint, min_distance: float, *, direction: str = "clockwise", ) -> tuple[int, float, int, int, float, tuple[int, int, float, float, int]]: current = route[-1] if len(route) < 2: return ( diagonal_cell_preference(current, candidate), 0.0, cardinal_direction_preference(current, candidate), too_close_preference(current, candidate, min_distance), 0, distance(current, candidate), point_sort_key_for_model(candidate), ) previous = route[-2] turn_class, turn_rank = route_turn_preference(previous, current, candidate, direction=direction) return ( turn_class, turn_rank, diagonal_cell_preference(current, candidate), too_close_preference(current, candidate, min_distance), cardinal_direction_preference(current, candidate), distance(current, candidate), point_sort_key_for_model(candidate), ) def route_clockwise_turn_preference(previous: LedPoint, current: LedPoint, candidate: LedPoint) -> tuple[int, float]: return route_turn_preference(previous, current, candidate, direction="clockwise") def route_turn_preference( previous: LedPoint, current: LedPoint, candidate: LedPoint, *, direction: str = "clockwise", ) -> tuple[int, float]: incoming_x = current.x - previous.x incoming_y = current.y - previous.y outgoing_x = candidate.x - current.x outgoing_y = candidate.y - current.y cross = incoming_x * outgoing_y - incoming_y * outgoing_x angle = route_turn_angle(previous, current, candidate) epsilon = 1e-6 if abs(cross) <= epsilon: return 0, 0.0 if angle < math.radians(35): return 0, angle if direction == "counterclockwise": if cross > epsilon: return 1, -angle return 2, angle if cross < -epsilon: return 1, -angle return 2, angle def route_turn_angle(previous: LedPoint, current: LedPoint, candidate: LedPoint) -> float: incoming_x = current.x - previous.x incoming_y = current.y - previous.y outgoing_x = candidate.x - current.x outgoing_y = candidate.y - current.y incoming_length = math.hypot(incoming_x, incoming_y) outgoing_length = math.hypot(outgoing_x, outgoing_y) if incoming_length == 0 or outgoing_length == 0: return 0.0 dot = incoming_x * outgoing_x + incoming_y * outgoing_y cos_angle = max(-1.0, min(1.0, dot / (incoming_length * outgoing_length))) return math.acos(cos_angle) def diagonal_cell_preference(current: LedPoint, candidate: LedPoint) -> int: return 1 if abs(current.row - candidate.row) == 1 and abs(current.col - candidate.col) == 1 else 0 def cardinal_direction_preference(current: LedPoint, candidate: LedPoint) -> int: d_x = candidate.x - current.x d_y = candidate.y - current.y if abs(d_y) >= abs(d_x) and d_y > 0: return 0 if abs(d_x) > abs(d_y) and d_x < 0: return 1 if abs(d_y) >= abs(d_x) and d_y < 0: return 2 if abs(d_x) > abs(d_y) and d_x > 0: return 3 return 4 def is_valid_route_neighbor(current: LedPoint, candidate: LedPoint, nominal_distance: float) -> bool: return distance(current, candidate) <= route_neighbor_distance_limit( current, candidate, nominal_distance, ) def route_neighbor_distance_limit(current: LedPoint, candidate: LedPoint, nominal_distance: float) -> float: if abs(current.row - candidate.row) == 1 and abs(current.col - candidate.col) == 1: return nominal_distance * math.sqrt(2.0) + neighbor_tolerance(nominal_distance) return max_neighbor_distance(nominal_distance) def max_neighbor_distance(nominal_distance: float) -> float: return nominal_distance + neighbor_tolerance(nominal_distance) def neighbor_tolerance(nominal_distance: float) -> float: explicit_tolerance = getattr(nominal_distance, "tolerance", None) if explicit_tolerance is not None: return float(explicit_tolerance) return MAX_NEIGHBOR_TOLERANCE_MM / (MIN_NEIGHBOR_DISTANCE_MM / nominal_distance) def min_neighbor_distance_for_points(points: list[LedPoint]) -> float: scale = coordinate_display_scale(points) return NominalGridDistance(MIN_NEIGHBOR_DISTANCE_MM / scale, MAX_NEIGHBOR_TOLERANCE_MM / scale) def coordinate_display_scale(points: list[LedPoint]) -> float: if not points: return 1.0 explicit_units = {point.coordinate_unit for point in points if point.coordinate_unit != "auto"} if explicit_units & {"m", "meter", "meters"}: return 1000.0 if explicit_units & {"mm", "millimeter", "millimeters"}: return 1.0 xs = [point.x for point in points] ys = [point.y for point in points] span = max(max(xs) - min(xs), max(ys) - min(ys)) return 1000.0 if span <= METER_LIKE_MAX_SPAN else 1.0 def clockwise_preference(current: LedPoint, candidate: LedPoint) -> int: d_row = candidate.row - current.row d_col = candidate.col - current.col clockwise_directions = [ (1, 0), (1, 1), (0, 1), (-1, 1), (-1, 0), (-1, -1), (0, -1), (1, -1), ] try: return clockwise_directions.index((sign(d_row), sign(d_col))) except ValueError: return len(clockwise_directions) def same_cell_preference(current: LedPoint, candidate: LedPoint) -> int: return 0 if current.row == candidate.row and current.col == candidate.col else 1 def upward_left_preference(current: LedPoint, candidate: LedPoint) -> int: if candidate.row < current.row and candidate.col <= current.col: return 0 if candidate.row < current.row: return 1 return 2 def sign(value: int) -> int: if value > 0: return 1 if value < 0: return -1 return 0