294 lines
11 KiB
Python
294 lines
11 KiB
Python
"""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
|
|
|
|
|
|
MAX_ROUTE_DATA = 4096
|
|
MAX_REQUEST_ID = 80
|
|
MAX_ENDPOINT_TEXT = 80
|
|
_REQUEST_ID = re.compile(r"^[A-Za-z0-9_.:-]{1,80}$")
|
|
|
|
|
|
class NavigatorRequestError(ValueError):
|
|
"""The WebApp sent an unsupported or unsafe route request."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RouteEndpoint:
|
|
room: str
|
|
floor: str | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RouteRequest:
|
|
version: int
|
|
request_id: str
|
|
start: RouteEndpoint
|
|
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} должен быть строкой")
|
|
value = value.strip()
|
|
if not value or len(value) > limit or any(char in value for char in "<>\r\n"):
|
|
raise NavigatorRequestError(f"Некорректное поле {field}")
|
|
return value
|
|
|
|
|
|
def _endpoint(value: Any, field: str) -> RouteEndpoint:
|
|
if not isinstance(value, dict):
|
|
raise NavigatorRequestError(f"{field} должен быть объектом")
|
|
room = _text(value.get("room"), f"{field}.room")
|
|
floor_value = value.get("floor")
|
|
floor = None if floor_value is None else _text(floor_value, f"{field}.floor", 16)
|
|
return RouteEndpoint(room=room, floor=floor)
|
|
|
|
|
|
def parse_route_request(raw: str | bytes) -> RouteRequest:
|
|
if isinstance(raw, bytes):
|
|
if len(raw) > MAX_ROUTE_DATA:
|
|
raise NavigatorRequestError("Слишком большой запрос маршрута")
|
|
raw = raw.decode("utf-8", errors="strict")
|
|
elif not isinstance(raw, str) or len(raw.encode("utf-8")) > MAX_ROUTE_DATA:
|
|
raise NavigatorRequestError("Слишком большой запрос маршрута")
|
|
|
|
try:
|
|
payload = json.loads(raw)
|
|
except (UnicodeError, json.JSONDecodeError) as error:
|
|
raise NavigatorRequestError("Запрос маршрута не является JSON") from error
|
|
|
|
if not isinstance(payload, dict) or payload.get("v") != 1 or payload.get("type") != "route.request":
|
|
raise NavigatorRequestError("Неподдерживаемый формат запроса маршрута")
|
|
|
|
request_id = _text(payload.get("request_id"), "request_id", MAX_REQUEST_ID)
|
|
if not _REQUEST_ID.fullmatch(request_id):
|
|
raise NavigatorRequestError("Некорректный request_id")
|
|
start = _endpoint(payload.get("start"), "start")
|
|
end = _endpoint(payload.get("end"), "end")
|
|
if start.room.casefold() == end.room.casefold() and start.floor == end.floor:
|
|
raise NavigatorRequestError("Начальная и конечная аудитории совпадают")
|
|
return RouteRequest(version=1, request_id=request_id, start=start, end=end)
|
|
|
|
|
|
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
|
|
|
|
|
|
@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 pathfinder manifest shared with the WebView."""
|
|
|
|
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]])
|
|
if second_start is None:
|
|
return None
|
|
if end_floor.name == "2":
|
|
second_path = _bfs(second_floor, second_start, request.end.room)
|
|
if not second_path:
|
|
return None
|
|
return {"v": 1, "segments": [_segment(start_floor, start_path), _segment(second_floor, second_path)]}
|
|
|
|
second_end = _nearest_node(second_floor, end_floor.positions[end_path[-1]])
|
|
if 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."""
|
|
|
|
data = json.dumps(route, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
|
|
return base64.urlsafe_b64encode(data).decode("ascii").rstrip("=")
|
|
|
|
|
|
def decode_route(value: str) -> dict[str, Any] | None:
|
|
"""Decode a route in tests/tools; the WebApp performs the same operation."""
|
|
|
|
try:
|
|
padded = value + "=" * ((4 - len(value) % 4) % 4)
|
|
result = json.loads(base64.urlsafe_b64decode(padded).decode("utf-8"))
|
|
except (ValueError, TypeError, UnicodeError, binascii.Error, json.JSONDecodeError):
|
|
return None
|
|
return result if isinstance(result, dict) else None
|