from __future__ import annotations import json import math import re from collections import deque from dataclasses import dataclass, field 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 MAX_NEIGHBOR_TOLERANCE_MM = 2.0 METER_LIKE_MAX_SPAN = 10.0 _DOCUMENT_KNOWN_FIELDS = { "background", "circles", "coordinate_unit", "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", } 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 @classmethod def from_payload(cls, payload) -> "BackgroundImage": # noqa: ANN001 if not isinstance(payload, dict): return cls() 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)), ) def to_payload(self) -> dict[str, bool | float | str]: return { "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, } def is_enabled(self) -> bool: return bool(self.path) 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 = "auto" 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 = "auto" 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"))) 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._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() payload = self.to_payload() path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") self.path = path 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, ) -> 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["background"] = self.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 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: line = self.get_line(line_id) removed_ids: list[int] = [] kept_blocks: list[LedBlock] = [] for block in line.blocks: if block.id == block_id: removed_ids.extend(block.point_ids) else: kept_blocks.append(block) line.blocks = kept_blocks self._clear_points(removed_ids) self.reindex_all() return len(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" return "m" if coordinate_display_scale(self.points) == 1000.0 else "mm" 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 scale = coordinate_display_scale(anchors) step = MIN_NEIGHBOR_DISTANCE_MM / scale 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)] 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) ] 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 = min_neighbor_distance_for_points(candidate_points) 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) ] 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 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]) -> 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_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) -> list[LedPoint]: if len(points) <= 1: return points[:] min_neighbor_distance = 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, ) -> 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_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: return MAX_NEIGHBOR_TOLERANCE_MM / (MIN_NEIGHBOR_DISTANCE_MM / nominal_distance) def min_neighbor_distance_for_points(points: list[LedPoint]) -> float: return MIN_NEIGHBOR_DISTANCE_MM / coordinate_display_scale(points) 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