Danil Iarosh
f8f6a216f5
- Добавлена проверка и безопасная установка OTA-обновлений Windows с откатом. - Нумерация линий и блоков переведена на индексы с нуля. - Добавлена миграция старых JSON-схем с сохранением физической адресации. - Обновлены генератор, сборка, документация и smoke-проверки. - Подготовлена версия 3.3.0.
2008 lines
75 KiB
Python
2008 lines
75 KiB
Python
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",
|
||
"index_base",
|
||
"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 = 0
|
||
self._next_block_id = 0
|
||
|
||
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 = 0
|
||
self._next_block_id = 0
|
||
|
||
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._migrate_legacy_index_base(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 _migrate_legacy_index_base(self, data: dict[str, Any]) -> None:
|
||
"""Однократно переводит старые 1-based ID, сохраняя физические адреса лент."""
|
||
marker = data.get("index_base")
|
||
if marker == 0:
|
||
return
|
||
if marker not in (None, 1):
|
||
raise ValueError("Unsupported mapping index_base; expected 0 or 1.")
|
||
identifiers = [line.id for line in self.lines]
|
||
identifiers.extend(point.line_id for point in self.points if point.line_id is not None)
|
||
if marker is None and (not identifiers or min(identifiers) == 0):
|
||
return
|
||
if identifiers and min(identifiers) < 1:
|
||
return
|
||
for line in self.lines:
|
||
old_id = line.id
|
||
line.id -= 1
|
||
if re.fullmatch(r"(?:Линия|Line)\s+\d+", line.name):
|
||
line.name = f"Линия {line.id}"
|
||
if line.color == default_color(old_id):
|
||
line.color = default_color(line.id)
|
||
for block in line.blocks:
|
||
block.id -= 1
|
||
if re.fullmatch(r"(?:Блок|Block)\s+\d+", block.name):
|
||
block.name = f"Блок {block.id}"
|
||
for point in self.points:
|
||
if point.line_id is not None:
|
||
point.line_id -= 1
|
||
if point.block_id is not None:
|
||
point.block_id -= 1
|
||
self._next_line_id = max((line.id for line in self.lines), default=-1) + 1
|
||
self._next_block_id = max(
|
||
(block.id for line in self.lines for block in line.blocks),
|
||
default=-1,
|
||
) + 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, f"Блок {self._next_block_id}", 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["index_base"] = 0
|
||
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), f"Блок {len(line.blocks)}", 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 = 0
|
||
self._next_block_id = 0
|
||
return assigned
|
||
|
||
def renumber_lines(self) -> None:
|
||
self.order_lines_clockwise()
|
||
id_map: dict[int, int] = {}
|
||
for new_id, line in enumerate(self.lines):
|
||
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)
|
||
|
||
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):
|
||
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)
|
||
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):
|
||
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 % 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
|