calculate navigator routes locally

README.md

@@ -25,8 +25,8 @@ uv run main.py
 `route.request` через `Telegram.WebApp.sendData`. Обработчик находится в
 `bot/navigator.py`, а протокол — в `models/navigator.py`.
 
-Скомпилированный граф пока не подключён: `calculate_route` намеренно
-возвращает `None`, поэтому бот отвечает «Маршрут не построен» и открывает
-карту с выбранными аудиториями. После появления данных нужно заменить только
-этот адаптер; UI уже умеет отрисовывать сегменты с индексами `#Точки` или
-координатами.
+Скомпилированный граф хранится локально в игнорируемом файле
+`web/map/data/navigator.json`. `calculate_route` строит путь по нему без
+сетевых запросов; внешний API навигатора используется только как эталон при
+проверке. UI получает номера этажей и индексы точек и рисует маршрут по
+координатам из того же manifest.

bot/navigator.py

@@ -101,7 +101,7 @@ def _send_route_result(message: Message, request: RouteRequest, route: dict | No
     else:
         text = (
             f"<b>Маршрут не построен</b>\n\n{start}  -  {end}\n"
-            "Для этой версии карты ещё не подключены скомпилированные точки."
+            "Не удалось найти путь между выбранными аудиториями."
         )
     return send(message, text, reply_markup=_result_markup(_map_url(request, route)))
 

config.example.py

@@ -13,8 +13,8 @@ WEB_BASE_URL: str = 'https://zatups.example.com'
 # Base URL for the schedule (should be runned separately)
 SCHEDULE_BASE_URL: str = 'http://10.9.8.3:18822'
 
-# The map navigator is mocked until the compiled points/graph are supplied.
-# Route requests are still validated and delivered back to the bot.
+# The navigator reads its local graph from web/map/data/navigator.json.
+# That generated file is ignored by Git and must be supplied at deployment.
 
 # Telegram user IDs
 ADMINS: list[int] = [5016590523, 5001115363]

models/navigator.py

@@ -1,17 +1,16 @@
-"""Validated protocol used by the map WebApp and the bot.
-
-The current map bundle intentionally does not contain the compiled graph yet.
-Keeping parsing and the route adapter here means the real navigator can be
-connected later without trusting nodes or edges supplied by the WebApp.
-"""
+"""Validated WebApp protocol and local pathfinder implementation."""
 
 from __future__ import annotations
 
 import base64
 import binascii
 import json
+import math
 import re
+from collections import deque
 from dataclasses import dataclass
+from functools import lru_cache
+from pathlib import Path
 from typing import Any
 
 
@@ -39,6 +38,18 @@ class RouteRequest:
     end: RouteEndpoint
 
 
+@dataclass(frozen=True)
+class CompiledFloor:
+    name: str
+    floor_id: str
+    neighbors: tuple[tuple[str, ...], ...]
+    graph: tuple[tuple[int, ...], ...]
+    positions: tuple[tuple[float, float], ...]
+
+
+_MANIFEST_PATH = Path(__file__).resolve().parents[1] / "web" / "map" / "data" / "navigator.json"
+
+
 def _text(value: Any, field: str, limit: int = MAX_ENDPOINT_TEXT) -> str:
     if not isinstance(value, str):
         raise NavigatorRequestError(f"{field} должен быть строкой")
@@ -83,19 +94,180 @@ def parse_route_request(raw: str | bytes) -> RouteRequest:
     return RouteRequest(version=1, request_id=request_id, start=start, end=end)
 
 
-def calculate_route(request: RouteRequest) -> dict[str, Any] | None:
-    """Return a trusted route or ``None`` until compiled graph data is added.
+def _decode_pathfinder(value: Any) -> dict[str, Any] | None:
+    for _ in range(2):
+        if not isinstance(value, str):
+            break
+        try:
+            value = json.loads(value)
+        except json.JSONDecodeError:
+            return None
+    return value if isinstance(value, dict) else None
 
-    This is an explicit mock rather than a proximity-based fallback.  A real
-    implementation should load the server-side compiled ``neighbors`` /
-    ``graph`` data, validate both room IDs against its manifest, and return a
-    payload with ``segments`` containing node indexes or coordinates.
-    """
 
-    del request
+@lru_cache(maxsize=4)
+def _load_floors(mtime_ns: int) -> dict[str, CompiledFloor]:
+    del mtime_ns  # cache key; contents are read again whenever the file changes
+    manifest = json.loads(_MANIFEST_PATH.read_text(encoding="utf-8"))
+    raw_floors = manifest.get("floors") if isinstance(manifest, dict) else None
+    if not isinstance(raw_floors, dict):
+        return {}
+
+    floors: dict[str, CompiledFloor] = {}
+    for key, raw_floor in raw_floors.items():
+        if not isinstance(raw_floor, dict):
+            continue
+        name = str(raw_floor.get("floor_name", key)).strip()
+        pathfinder = _decode_pathfinder(raw_floor.get("pathfinder_data", raw_floor))
+        if not pathfinder:
+            continue
+        raw_neighbors = pathfinder.get("neighbors")
+        raw_graph = pathfinder.get("graph")
+        raw_positions = raw_floor.get("positions", pathfinder.get("positions"))
+        if not all(isinstance(value, list) for value in (raw_neighbors, raw_graph, raw_positions)):
+            continue
+        if not (len(raw_neighbors) == len(raw_graph) == len(raw_positions)):
+            continue
+
+        try:
+            neighbors = tuple(tuple(str(item) for item in items) for items in raw_neighbors)
+            graph = tuple(tuple(int(item) for item in items) for items in raw_graph)
+            positions = tuple((float(item[0]), float(item[1])) for item in raw_positions)
+        except (TypeError, ValueError, IndexError):
+            continue
+        node_count = len(graph)
+        if any(node < 0 or node >= node_count for edges in graph for node in edges):
+            continue
+        floors[name] = CompiledFloor(
+            name=name,
+            floor_id=str(raw_floor.get("id", name)),
+            neighbors=neighbors,
+            graph=graph,
+            positions=positions,
+        )
+    return floors
+
+
+def _floors() -> dict[str, CompiledFloor]:
+    try:
+        return _load_floors(_MANIFEST_PATH.stat().st_mtime_ns)
+    except (OSError, ValueError, TypeError, json.JSONDecodeError):
+        return {}
+
+
+def _contains_room(floor: CompiledFloor, room: str) -> bool:
+    return any(room in names for names in floor.neighbors)
+
+
+def _resolve_floor(floors: dict[str, CompiledFloor], endpoint: RouteEndpoint) -> CompiledFloor | None:
+    if endpoint.floor and endpoint.floor in floors:
+        floor = floors[endpoint.floor]
+        return floor if _contains_room(floor, endpoint.room) else None
+    matches = [floor for floor in floors.values() if _contains_room(floor, endpoint.room)]
+    return matches[0] if len(matches) == 1 else None
+
+
+def _bfs(floor: CompiledFloor, start: str | int, end: str | int) -> list[int] | None:
+    if isinstance(start, int):
+        start_index = start if 0 <= start < len(floor.graph) else None
+    else:
+        start_index = next((index for index, names in enumerate(floor.neighbors) if start in names), None)
+    if start_index is None:
+        return None
+
+    def reached(index: int) -> bool:
+        return index == end if isinstance(end, int) else end in floor.neighbors[index]
+
+    if reached(start_index):
+        return [start_index]
+
+    queue = deque([start_index])
+    visited = {start_index}
+    parents: dict[int, int] = {}
+    while queue:
+        node = queue.popleft()
+        for neighbor in floor.graph[node]:
+            if neighbor in visited:
+                continue
+            parents[neighbor] = node
+            if reached(neighbor):
+                path = [neighbor]
+                while path[-1] != start_index:
+                    path.append(parents[path[-1]])
+                return path[::-1]
+            visited.add(neighbor)
+            queue.append(neighbor)
     return None
 
 
+def _xy_search(floor: CompiledFloor, x: float, y: float, *, safe: bool = True) -> int | None:
+    closest_index: int | None = None
+    closest_distance = math.inf
+    for index, (point_x, point_y) in enumerate(floor.positions):
+        if safe and "лест" not in ",".join(floor.neighbors[index]).casefold():
+            continue
+        distance = math.hypot(point_x - x, point_y - y)
+        if distance < closest_distance:
+            closest_distance = distance
+            closest_index = index
+    return closest_index if closest_index is not None and closest_distance < 2000 else None
+
+
+def _nearest_node(floor: CompiledFloor, position: tuple[float, float]) -> int | None:
+    safe_index = _xy_search(floor, *position, safe=True)
+    return safe_index if safe_index is not None else _xy_search(floor, *position, safe=False)
+
+
+def _segment(floor: CompiledFloor, path: list[int]) -> dict[str, Any]:
+    # The WebView only needs the display floor number.  Keeping the UUID out of
+    # the payload also makes route URLs shorter.
+    return {"array": path, "floor_id": floor.name, "floor_name": floor.name}
+
+
+def calculate_route(request: RouteRequest) -> dict[str, Any] | None:
+    """Calculate a route from the ignored local pathfinder manifest."""
+
+    floors = _floors()
+    start_floor = _resolve_floor(floors, request.start)
+    end_floor = _resolve_floor(floors, request.end)
+    if not start_floor or not end_floor:
+        return None
+
+    if start_floor.name == end_floor.name:
+        path = _bfs(start_floor, request.start.room, request.end.room)
+        return {"v": 1, "segments": [_segment(start_floor, path)]} if path else None
+
+    end_path = _bfs(end_floor, request.end.room, "Лестница")
+    if not end_path:
+        return None
+
+    if start_floor.name == "2":
+        destination = _nearest_node(start_floor, end_floor.positions[end_path[-1]])
+        start_path = _bfs(start_floor, request.start.room, destination) if destination is not None else None
+        if not start_path:
+            return None
+        segments = [_segment(start_floor, start_path), _segment(end_floor, end_path)]
+        return {"v": 1, "segments": segments}
+
+    start_path = _bfs(start_floor, request.start.room, "Лестница")
+    second_floor = floors.get("2")
+    if not start_path or not second_floor:
+        return None
+    second_start = _nearest_node(second_floor, start_floor.positions[start_path[-1]])
+    second_end = _nearest_node(second_floor, end_floor.positions[end_path[-1]])
+    if second_start is None or second_end is None:
+        return None
+    second_path = _bfs(second_floor, second_start, second_end)
+    if not second_path:
+        return None
+    segments = [
+        _segment(start_floor, start_path),
+        _segment(second_floor, second_path),
+        _segment(end_floor, end_path),
+    ]
+    return {"v": 1, "segments": segments}
+
+
 def encode_route(route: dict[str, Any]) -> str:
     """Encode a route for a URL without putting credentials in the URL."""
 

web/map/js/map.js

@@ -649,7 +649,7 @@ function renderRouteCard() {
 
     if (failed) {
         title.textContent = 'Маршрут не построен'
-        steps.append(createRouteStep('Попробуй ещё раз', 'Для этой версии карты скомпилированные точки маршрута ещё не подключены.'))
+        steps.append(createRouteStep('Попробуй ещё раз', 'Не удалось найти путь между выбранными аудиториями.'))
     } else {
         title.textContent = 'Маршрут построен'
         const segments = Array.isArray(state.route.segments) ? state.route.segments : []