From b0a801a820aa65861bc74b0eea63e8b850a68ce3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D1=81=D0=B8=D0=BD=D0=B5=D1=87=D0=BA=D0=B0=20=E2=99=A1?= <63897347+ktnk-dev@users.noreply.github.com> Date: Wed, 9 Sep 2026 17:27:47 +0000 Subject: [PATCH] calculate navigator routes locally --- README.md | 10 +-- bot/navigator.py | 2 +- config.example.py | 4 +- models/navigator.py | 200 ++++++++++++++++++++++++++++++++++++++++---- web/map/js/map.js | 2 +- 5 files changed, 195 insertions(+), 23 deletions(-) diff --git a/README.md b/README.md index 378f849..8d05fc4 100644 --- a/README.md +++ b/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. diff --git a/bot/navigator.py b/bot/navigator.py index 8da5fe1..d397479 100644 --- a/bot/navigator.py +++ b/bot/navigator.py @@ -101,7 +101,7 @@ def _send_route_result(message: Message, request: RouteRequest, route: dict | No else: text = ( f"Маршрут не построен\n\n{start} - {end}\n" - "Для этой версии карты ещё не подключены скомпилированные точки." + "Не удалось найти путь между выбранными аудиториями." ) return send(message, text, reply_markup=_result_markup(_map_url(request, route))) diff --git a/config.example.py b/config.example.py index 966f2be..2a29ebd 100644 --- a/config.example.py +++ b/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] diff --git a/models/navigator.py b/models/navigator.py index 0daf492..c0c0d45 100644 --- a/models/navigator.py +++ b/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.""" diff --git a/web/map/js/map.js b/web/map/js/map.js index 0b1b125..e0e0ccd 100644 --- a/web/map/js/map.js +++ b/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 : []