diff --git a/atak/__init__.py b/atak/__init__.py new file mode 100644 index 0000000..a49cfc0 --- /dev/null +++ b/atak/__init__.py @@ -0,0 +1,3 @@ +"""ATAK bridge helpers.""" + +from __future__ import annotations diff --git a/atak/bridge.py b/atak/bridge.py new file mode 100644 index 0000000..2ea9268 --- /dev/null +++ b/atak/bridge.py @@ -0,0 +1,22 @@ +"""ATAK bridge entry points.""" + +from __future__ import annotations + +import json +from pathlib import Path + +from atak.cot import PlanMapping, write_plan_kml + + +def export_plan_kml(plan: PlanMapping, output_path: str | Path) -> Path: + """Export a plan response to a KML file for manual ATAK import.""" + return write_plan_kml(plan, output_path) + + +def export_plan_json_kml(plan_json: str, output_path: str | Path) -> Path: + """Export a JSON-encoded plan response to KML.""" + decoded = json.loads(plan_json) + if not isinstance(decoded, dict): + msg = "plan JSON must decode to an object" + raise ValueError(msg) + return export_plan_kml(decoded, output_path) diff --git a/atak/cot.py b/atak/cot.py new file mode 100644 index 0000000..88f227a --- /dev/null +++ b/atak/cot.py @@ -0,0 +1,147 @@ +"""KML and CoT-adjacent helpers for ATAK route export.""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from pathlib import Path +from xml.etree import ElementTree + +KML_NS = "http://www.opengis.net/kml/2.2" +ElementTree.register_namespace("", KML_NS) + +JsonScalar = str | int | float | bool | None +JsonValue = JsonScalar | Mapping[str, "JsonValue"] | Sequence["JsonValue"] +PlanMapping = Mapping[str, JsonValue] + + +class PlanExportError(ValueError): + """Raised when a plan response cannot be exported for ATAK.""" + + +def plan_to_kml(plan: PlanMapping, *, document_name: str = "TERA route") -> str: + """Serialize a PlanResponse-like mapping to a KML document.""" + coordinates = _extract_linestring_coordinates(plan) + waypoints = _extract_waypoints(plan) + + kml = ElementTree.Element(_kml_tag("kml")) + document = ElementTree.SubElement(kml, _kml_tag("Document")) + ElementTree.SubElement(document, _kml_tag("name")).text = document_name + + route_placemark = ElementTree.SubElement(document, _kml_tag("Placemark")) + ElementTree.SubElement(route_placemark, _kml_tag("name")).text = "TERA route" + rationale = plan.get("rationale") + if isinstance(rationale, str): + ElementTree.SubElement(route_placemark, _kml_tag("description")).text = rationale + + line = ElementTree.SubElement(route_placemark, _kml_tag("LineString")) + ElementTree.SubElement(line, _kml_tag("tessellate")).text = "1" + ElementTree.SubElement(line, _kml_tag("coordinates")).text = " ".join( + _format_kml_coordinate(lon, lat, alt) for lon, lat, alt in coordinates + ) + + for index, waypoint in enumerate(waypoints, start=1): + placemark = ElementTree.SubElement(document, _kml_tag("Placemark")) + ElementTree.SubElement(placemark, _kml_tag("name")).text = ( + waypoint.label or f"Waypoint {index}" + ) + point = ElementTree.SubElement(placemark, _kml_tag("Point")) + ElementTree.SubElement(point, _kml_tag("coordinates")).text = _format_kml_coordinate( + waypoint.lon, + waypoint.lat, + None, + ) + + ElementTree.indent(kml, space=" ") + xml = ElementTree.tostring(kml, encoding="utf-8", xml_declaration=True) + return xml.decode("utf-8") + + +def write_plan_kml( + plan: PlanMapping, + output_path: str | Path, + *, + document_name: str = "TERA route", +) -> Path: + """Write a PlanResponse-like mapping as a KML file and return the path.""" + path = Path(output_path) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(plan_to_kml(plan, document_name=document_name), encoding="utf-8") + return path + + +class _Waypoint: + def __init__(self, *, lat: float, lon: float, label: str | None) -> None: + self.lat = lat + self.lon = lon + self.label = label + + +def _kml_tag(name: str) -> str: + return f"{{{KML_NS}}}{name}" + + +def _extract_linestring_coordinates(plan: PlanMapping) -> list[tuple[float, float, float | None]]: + route = _mapping_field(plan, "route") + geometry = _mapping_field(route, "geometry") + geometry_type = geometry.get("type") + if geometry_type != "LineString": + raise PlanExportError("plan route geometry must be a GeoJSON LineString") + + raw_coordinates = geometry.get("coordinates") + if not isinstance(raw_coordinates, Sequence) or isinstance(raw_coordinates, str): + raise PlanExportError("plan route geometry coordinates must be a sequence") + + coordinates: list[tuple[float, float, float | None]] = [] + for raw_coord in raw_coordinates: + if not isinstance(raw_coord, Sequence) or isinstance(raw_coord, str): + raise PlanExportError("each route coordinate must be a sequence") + if len(raw_coord) not in {2, 3}: + raise PlanExportError( + "each route coordinate must contain lon, lat, and optional altitude" + ) + + lon = _number(raw_coord[0], "route longitude") + lat = _number(raw_coord[1], "route latitude") + alt = _number(raw_coord[2], "route altitude") if len(raw_coord) == 3 else None + coordinates.append((lon, lat, alt)) + + if len(coordinates) < 2: + raise PlanExportError("route must contain at least two coordinates") + + return coordinates + + +def _extract_waypoints(plan: PlanMapping) -> list[_Waypoint]: + raw_waypoints = plan.get("waypoints", []) + if not isinstance(raw_waypoints, Sequence) or isinstance(raw_waypoints, str): + raise PlanExportError("plan waypoints must be a sequence") + + waypoints: list[_Waypoint] = [] + for raw_waypoint in raw_waypoints: + if not isinstance(raw_waypoint, Mapping): + raise PlanExportError("each waypoint must be an object") + lat = _number(raw_waypoint.get("lat"), "waypoint latitude") + lon = _number(raw_waypoint.get("lon"), "waypoint longitude") + raw_label = raw_waypoint.get("label") + label = raw_label if isinstance(raw_label, str) else None + waypoints.append(_Waypoint(lat=lat, lon=lon, label=label)) + + return waypoints + + +def _mapping_field(source: Mapping[str, JsonValue], name: str) -> Mapping[str, JsonValue]: + value = source.get(name) + if not isinstance(value, Mapping): + raise PlanExportError(f"plan field {name!r} must be an object") + return value + + +def _number(value: JsonValue, label: str) -> float: + if isinstance(value, bool) or not isinstance(value, int | float): + raise PlanExportError(f"{label} must be numeric") + return float(value) + + +def _format_kml_coordinate(lon: float, lat: float, alt: float | None) -> str: + altitude = 0.0 if alt is None else alt + return f"{lon:.7f},{lat:.7f},{altitude:.2f}" diff --git a/data/README.md b/data/README.md new file mode 100644 index 0000000..1744c5f --- /dev/null +++ b/data/README.md @@ -0,0 +1,49 @@ +# Data Pipeline + +The Phase 1/2 data build produces four local artifacts: + +- `data/extracts/sf.osm.pbf` +- `data/extracts/austere.osm.pbf` +- `data/dem/sf.tif` +- `data/dem/austere.tif` + +Install lane tools: + +```bash +brew install osmium-tool gdal +``` + +Fetch and verify the full Phase 1 data bundle: + +```bash +data/scripts/fetch_all.sh +``` + +This downloads a small San Francisco PBF from BBBike, queries Overpass for the +austere AO, downloads public Copernicus GLO-30 DEM source tiles, crops both AOIs, +then writes and verifies `data/manifest.sha256`. + +Clip OSM extracts from a larger source PBF: + +```bash +data/scripts/clip_osm.sh /path/to/california-latest.osm.pbf +``` + +Crop DEM GeoTIFFs from one or more source DEM tiles: + +```bash +data/scripts/build_dem.sh /path/to/source-dem-1.tif /path/to/source-dem-2.tif +``` + +Write and verify the artifact manifest: + +```bash +data/scripts/write_manifest.sh +data/scripts/verify_manifest.sh +``` + +`data/aois.yml` is the source of truth for AOI bounding boxes and output paths. + +Root-level `make data-fetch` / `make data-verify` are documented lane entry +points, but the root `Makefile` is owned by P2. Until P2 wires those targets, +use `data/scripts/fetch_all.sh` and `data/scripts/verify_manifest.sh` directly. diff --git a/data/aois.yml b/data/aois.yml new file mode 100644 index 0000000..140b081 --- /dev/null +++ b/data/aois.yml @@ -0,0 +1,30 @@ +aois: + - name: sf + label: San Francisco demo AO + bbox: + west: -122.535 + south: 37.690 + east: -122.340 + north: 37.835 + route_profiles: + - foot + - foot_covered + osm_extract: data/extracts/sf.osm.pbf + dem: data/dem/sf.tif + cesium_cache: data/cache/cesium/sf + notes: Ferry Building to local freshwater fallback demo. + - name: austere + label: MWTC Bridgeport austere demo AO + bbox: + west: -119.720 + south: 38.160 + east: -119.360 + north: 38.460 + route_profiles: + - foot + - foot_covered + - vehicle_mrap + osm_extract: data/extracts/austere.osm.pbf + dem: data/dem/austere.tif + cesium_cache: data/cache/cesium/austere + notes: Mountain Warfare Training Center / Pickel Meadows proxy near Bridgeport. diff --git a/data/manifest.sha256 b/data/manifest.sha256 new file mode 100644 index 0000000..09b3a8a --- /dev/null +++ b/data/manifest.sha256 @@ -0,0 +1 @@ +# Placeholder. Regenerate with data/scripts/write_manifest.sh after OSM/DEM artifacts are built. diff --git a/data/scripts/build_dem.sh b/data/scripts/build_dem.sh new file mode 100755 index 0000000..bd0eebd --- /dev/null +++ b/data/scripts/build_dem.sh @@ -0,0 +1,70 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" +AOIS_FILE="${AOIS_FILE:-$ROOT_DIR/data/aois.yml}" +WORK_DIR="${WORK_DIR:-$ROOT_DIR/data/runtime/dem-build}" + +if [[ "$#" -lt 1 ]]; then + echo "usage: data/scripts/build_dem.sh [source-dem-2.tif ...]" >&2 + exit 2 +fi + +if ! command -v gdalbuildvrt >/dev/null 2>&1; then + echo "error: gdalbuildvrt is required. macOS: brew install gdal" >&2 + exit 127 +fi + +if ! command -v gdalwarp >/dev/null 2>&1; then + echo "error: gdalwarp is required. macOS: brew install gdal" >&2 + exit 127 +fi + +mkdir -p "$WORK_DIR" +VRT="$WORK_DIR/source-dem.vrt" + +gdalbuildvrt -overwrite "$VRT" "$@" + +"$ROOT_DIR/.venv/bin/python" - "$ROOT_DIR" "$AOIS_FILE" "$VRT" <<'PY' +from __future__ import annotations + +import subprocess +import sys +from pathlib import Path + +import yaml + +root = Path(sys.argv[1]) +aois_file = Path(sys.argv[2]) +vrt = Path(sys.argv[3]) +config = yaml.safe_load(aois_file.read_text(encoding="utf-8")) + +for aoi in config["aois"]: + bbox = aoi["bbox"] + output = root / aoi["dem"] + output.parent.mkdir(parents=True, exist_ok=True) + command = [ + "gdalwarp", + "-overwrite", + "-of", + "GTiff", + "-t_srs", + "EPSG:4326", + "-te", + str(bbox["west"]), + str(bbox["south"]), + str(bbox["east"]), + str(bbox["north"]), + "-r", + "bilinear", + "-co", + "COMPRESS=DEFLATE", + "-co", + "PREDICTOR=2", + str(vrt), + str(output), + ] + print(f"[data] building DEM {aoi['name']} -> {output}") + subprocess.run(command, check=True) +PY + diff --git a/data/scripts/clip_osm.sh b/data/scripts/clip_osm.sh new file mode 100755 index 0000000..30ec2b8 --- /dev/null +++ b/data/scripts/clip_osm.sh @@ -0,0 +1,56 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" +AOIS_FILE="${AOIS_FILE:-$ROOT_DIR/data/aois.yml}" +SOURCE_PBF="${1:-}" + +if [[ -z "$SOURCE_PBF" ]]; then + echo "usage: data/scripts/clip_osm.sh " >&2 + exit 2 +fi + +if ! command -v osmium >/dev/null 2>&1; then + echo "error: osmium is required. macOS: brew install osmium-tool" >&2 + exit 127 +fi + +if [[ ! -f "$SOURCE_PBF" ]]; then + echo "error: source PBF not found: $SOURCE_PBF" >&2 + exit 1 +fi + +"$ROOT_DIR/.venv/bin/python" - "$ROOT_DIR" "$AOIS_FILE" "$SOURCE_PBF" <<'PY' +from __future__ import annotations + +import subprocess +import sys +from pathlib import Path + +import yaml + +root = Path(sys.argv[1]) +aois_file = Path(sys.argv[2]) +source_pbf = Path(sys.argv[3]) +config = yaml.safe_load(aois_file.read_text(encoding="utf-8")) + +for aoi in config["aois"]: + bbox = aoi["bbox"] + output = root / aoi["osm_extract"] + output.parent.mkdir(parents=True, exist_ok=True) + bbox_arg = f"{bbox['west']},{bbox['south']},{bbox['east']},{bbox['north']}" + command = [ + "osmium", + "extract", + "--bbox", + bbox_arg, + "--strategy", + "complete_ways", + "--overwrite", + "--output", + str(output), + str(source_pbf), + ] + print(f"[data] clipping {aoi['name']} -> {output}") + subprocess.run(command, check=True) +PY diff --git a/data/scripts/fetch_all.sh b/data/scripts/fetch_all.sh new file mode 100755 index 0000000..45c8728 --- /dev/null +++ b/data/scripts/fetch_all.sh @@ -0,0 +1,9 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" + +"$ROOT_DIR/data/scripts/fetch_osm.sh" +"$ROOT_DIR/data/scripts/fetch_dem.sh" +"$ROOT_DIR/data/scripts/write_manifest.sh" +"$ROOT_DIR/data/scripts/verify_manifest.sh" diff --git a/data/scripts/fetch_dem.sh b/data/scripts/fetch_dem.sh new file mode 100755 index 0000000..f725d06 --- /dev/null +++ b/data/scripts/fetch_dem.sh @@ -0,0 +1,71 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" +AOIS_FILE="${AOIS_FILE:-$ROOT_DIR/data/aois.yml}" +SOURCE_DIR="${SOURCE_DIR:-$ROOT_DIR/data/runtime/dem-source}" +COPERNICUS_BASE_URL="${COPERNICUS_BASE_URL:-https://copernicus-dem-30m.s3.amazonaws.com}" + +if ! command -v curl >/dev/null 2>&1; then + echo "error: curl is required" >&2 + exit 127 +fi + +if ! command -v gdalbuildvrt >/dev/null 2>&1; then + echo "error: gdalbuildvrt is required. macOS: brew install gdal" >&2 + exit 127 +fi + +if ! command -v gdalwarp >/dev/null 2>&1; then + echo "error: gdalwarp is required. macOS: brew install gdal" >&2 + exit 127 +fi + +mkdir -p "$SOURCE_DIR" + +mapfile -t tile_urls < <( + "$ROOT_DIR/.venv/bin/python" - "$AOIS_FILE" "$COPERNICUS_BASE_URL" <<'PY' +from __future__ import annotations + +import math +import sys +from pathlib import Path + +import yaml + +aois_file = Path(sys.argv[1]) +base_url = sys.argv[2].rstrip("/") +config = yaml.safe_load(aois_file.read_text(encoding="utf-8")) + +tiles: set[tuple[int, int]] = set() +for aoi in config["aois"]: + bbox = aoi["bbox"] + west = math.floor(float(bbox["west"])) + east = math.floor(float(bbox["east"])) + south = math.floor(float(bbox["south"])) + north = math.floor(float(bbox["north"])) + for lat in range(south, north + 1): + for lon in range(west, east + 1): + tiles.add((lat, lon)) + +for lat, lon in sorted(tiles): + ns = "N" if lat >= 0 else "S" + ew = "E" if lon >= 0 else "W" + dirname = f"Copernicus_DSM_COG_10_{ns}{abs(lat):02d}_00_{ew}{abs(lon):03d}_00_DEM" + print(f"{base_url}/{dirname}/{dirname}.tif") +PY +) + +source_files=() +for url in "${tile_urls[@]}"; do + output="$SOURCE_DIR/$(basename "$url")" + source_files+=("$output") + if [[ -f "$output" ]]; then + echo "[data] DEM source exists: $output" + continue + fi + echo "[data] downloading DEM source: $url" + curl --fail --location --retry 3 --connect-timeout 10 --output "$output" "$url" +done + +"$ROOT_DIR/data/scripts/build_dem.sh" "${source_files[@]}" diff --git a/data/scripts/fetch_osm.sh b/data/scripts/fetch_osm.sh new file mode 100755 index 0000000..752b7eb --- /dev/null +++ b/data/scripts/fetch_osm.sh @@ -0,0 +1,84 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" +AOIS_FILE="${AOIS_FILE:-$ROOT_DIR/data/aois.yml}" +SOURCE_DIR="${SOURCE_DIR:-$ROOT_DIR/data/runtime/osm-source}" +SF_SOURCE_URL="${SF_SOURCE_URL:-https://download.bbbike.org/osm/bbbike/SanFrancisco/SanFrancisco.osm.pbf}" +OVERPASS_URL="${OVERPASS_URL:-https://overpass-api.de/api/interpreter}" + +if ! command -v curl >/dev/null 2>&1; then + echo "error: curl is required" >&2 + exit 127 +fi + +if ! command -v osmium >/dev/null 2>&1; then + echo "error: osmium is required. macOS: brew install osmium-tool" >&2 + exit 127 +fi + +mkdir -p "$SOURCE_DIR" + +sf_source="$SOURCE_DIR/sf-source.osm.pbf" +if [[ ! -f "$sf_source" ]]; then + echo "[data] downloading SF OSM source: $SF_SOURCE_URL" + curl --fail --location --retry 3 --connect-timeout 10 --output "$sf_source" "$SF_SOURCE_URL" +else + echo "[data] SF OSM source exists: $sf_source" +fi + +"$ROOT_DIR/data/scripts/clip_osm.sh" "$sf_source" + +mapfile -t overpass_jobs < <( + "$ROOT_DIR/.venv/bin/python" - "$AOIS_FILE" <<'PY' +from __future__ import annotations + +import sys +from pathlib import Path + +import yaml + +config = yaml.safe_load(Path(sys.argv[1]).read_text(encoding="utf-8")) +for aoi in config["aois"]: + if aoi["name"] == "sf": + continue + bbox = aoi["bbox"] + print( + "\t".join( + [ + aoi["name"], + str(aoi["osm_extract"]), + str(bbox["south"]), + str(bbox["west"]), + str(bbox["north"]), + str(bbox["east"]), + ] + ) + ) +PY +) + +for job in "${overpass_jobs[@]}"; do + IFS=$'\t' read -r name output_rel south west north east <<<"$job" + output="$ROOT_DIR/$output_rel" + xml_source="$SOURCE_DIR/$name.osm" + mkdir -p "$(dirname "$output")" + + if [[ ! -f "$xml_source" ]]; then + query="[out:xml][timeout:180][maxsize:1073741824];(node($south,$west,$north,$east);way($south,$west,$north,$east);relation($south,$west,$north,$east););(._;>;);out meta;" + echo "[data] querying OSM for $name via Overpass" + curl \ + --fail \ + --location \ + --retry 3 \ + --connect-timeout 10 \ + --data-urlencode "data=$query" \ + --output "$xml_source" \ + "$OVERPASS_URL" + else + echo "[data] OSM XML source exists: $xml_source" + fi + + echo "[data] converting $name OSM XML -> $output" + osmium cat --overwrite --output "$output" "$xml_source" +done diff --git a/data/scripts/verify_manifest.sh b/data/scripts/verify_manifest.sh new file mode 100755 index 0000000..af3b4f7 --- /dev/null +++ b/data/scripts/verify_manifest.sh @@ -0,0 +1,19 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" +MANIFEST="${MANIFEST:-$ROOT_DIR/data/manifest.sha256}" + +cd "$ROOT_DIR" + +if [[ ! -f "$MANIFEST" ]]; then + echo "error: manifest not found: $MANIFEST" >&2 + exit 1 +fi + +if ! grep -Eq "^[0-9a-f]{64} data/" "$MANIFEST"; then + echo "error: manifest has no data artifact checksums: $MANIFEST" >&2 + exit 1 +fi + +shasum -a 256 -c "$MANIFEST" diff --git a/data/scripts/write_manifest.sh b/data/scripts/write_manifest.sh new file mode 100755 index 0000000..0694b71 --- /dev/null +++ b/data/scripts/write_manifest.sh @@ -0,0 +1,22 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" +MANIFEST="${MANIFEST:-$ROOT_DIR/data/manifest.sha256}" + +cd "$ROOT_DIR" +mkdir -p data + +artifact_list="$(mktemp)" +find data/extracts data/dem -type f ! -name ".gitkeep" -print | LC_ALL=C sort > "$artifact_list" + +if [[ ! -s "$artifact_list" ]]; then + rm -f "$artifact_list" + echo "error: no data artifacts found under data/extracts or data/dem" >&2 + exit 1 +fi + +xargs shasum -a 256 < "$artifact_list" > "$MANIFEST" +rm -f "$artifact_list" + +echo "[data] wrote $MANIFEST" diff --git a/routing/valhalla_client.py b/routing/valhalla_client.py new file mode 100644 index 0000000..e0903a8 --- /dev/null +++ b/routing/valhalla_client.py @@ -0,0 +1,184 @@ +"""Local Valhalla CLI client for offline route generation.""" + +from __future__ import annotations + +import json +import subprocess +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from pathlib import Path +from typing import Protocol, cast + + +class ValhallaRouteError(RuntimeError): + """Raised when Valhalla cannot return a usable route.""" + + +@dataclass(frozen=True) +class Coord: + lat: float + lon: float + + +@dataclass(frozen=True) +class RouteResult: + coordinates: list[tuple[float, float]] + distance_m: float + time_s: float + + def as_geojson_feature(self) -> dict[str, object]: + return { + "type": "Feature", + "geometry": { + "type": "LineString", + "coordinates": [[lon, lat] for lat, lon in self.coordinates], + }, + "properties": { + "source": "valhalla", + "distance_m": self.distance_m, + "time_s": self.time_s, + }, + } + + +class CommandRunner(Protocol): + def __call__(self, command: Sequence[str]) -> str: + """Run a local command and return stdout.""" + + +class SubprocessRunner: + def __call__(self, command: Sequence[str]) -> str: + # Valhalla is a local binary in the offline path; callers pass argv, never shell text. + completed = subprocess.run( # noqa: S603 + list(command), + check=True, + capture_output=True, + text=True, + ) + return completed.stdout + + +class ValhallaClient: + def __init__( + self, + config_path: str | Path, + *, + executable: str = "valhalla_run_route", + runner: CommandRunner | None = None, + ) -> None: + self.config_path = Path(config_path) + self.executable = executable + self.runner = runner or SubprocessRunner() + + def route( + self, + origin: Coord, + destination: Coord, + *, + costing: str = "pedestrian", + ) -> RouteResult: + request = { + "locations": [ + {"lat": origin.lat, "lon": origin.lon}, + {"lat": destination.lat, "lon": destination.lon}, + ], + "costing": costing, + "directions_options": {"units": "kilometers"}, + } + output = self.runner( + [ + self.executable, + str(self.config_path), + json.dumps(request, separators=(",", ":")), + ] + ) + return parse_valhalla_route(output) + + +def parse_valhalla_route(output_json: str) -> RouteResult: + decoded: object = json.loads(output_json) + if not isinstance(decoded, dict): + raise ValhallaRouteError("Valhalla output must be a JSON object") + + trip = _dict_field(decoded, "trip") + legs = _list_field(trip, "legs") + if not legs: + raise ValhallaRouteError("Valhalla output did not include any legs") + + summary = _dict_field(trip, "summary") + distance_m = _number(summary.get("length"), "trip summary length") * 1000.0 + time_s = _number(summary.get("time"), "trip summary time") + + coordinates: list[tuple[float, float]] = [] + for raw_leg in legs: + if not isinstance(raw_leg, dict): + raise ValhallaRouteError("Valhalla leg must be an object") + leg = cast(dict[str, object], raw_leg) + shape = leg.get("shape") + if not isinstance(shape, str) or not shape: + raise ValhallaRouteError("Valhalla leg did not include an encoded shape") + leg_coordinates = decode_valhalla_polyline(shape) + if coordinates and leg_coordinates: + coordinates.extend(leg_coordinates[1:]) + continue + coordinates.extend(leg_coordinates) + + if len(coordinates) < 2: + raise ValhallaRouteError("Valhalla route must decode to at least two coordinates") + + return RouteResult(coordinates=coordinates, distance_m=distance_m, time_s=time_s) + + +def decode_valhalla_polyline(shape: str, *, precision: int = 6) -> list[tuple[float, float]]: + factor = float(10**precision) + lat = 0 + lon = 0 + index = 0 + coordinates: list[tuple[float, float]] = [] + + while index < len(shape): + lat_delta, index = _decode_polyline_value(shape, index) + lon_delta, index = _decode_polyline_value(shape, index) + lat += lat_delta + lon += lon_delta + coordinates.append((lat / factor, lon / factor)) + + return coordinates + + +def _decode_polyline_value(shape: str, index: int) -> tuple[int, int]: + result = 1 + shift = 0 + + while True: + if index >= len(shape): + raise ValhallaRouteError("encoded shape ended unexpectedly") + value = ord(shape[index]) - 63 - 1 + index += 1 + result += value << shift + shift += 5 + if value < 0x1F: + break + + delta = ~(result >> 1) if result & 1 else result >> 1 + return delta, index + + +def _dict_field(source: Mapping[str, object], name: str) -> dict[str, object]: + value = source.get(name) + if not isinstance(value, dict): + raise ValhallaRouteError(f"Valhalla field {name!r} must be an object") + return cast(dict[str, object], value) + + +def _list_field(source: dict[str, object], name: str) -> list[object]: + value = source.get(name) + if not isinstance(value, list): + raise ValhallaRouteError(f"Valhalla field {name!r} must be a list") + return value + + +def _number(value: object, label: str) -> float: + if isinstance(value, bool) or not isinstance(value, int | float): + raise ValhallaRouteError(f"{label} must be numeric") + return float(value) diff --git a/tests/test_atak_cot.py b/tests/test_atak_cot.py new file mode 100644 index 0000000..01192f3 --- /dev/null +++ b/tests/test_atak_cot.py @@ -0,0 +1,76 @@ +"""Tests for ATAK route export helpers.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from atak.cot import PlanExportError, plan_to_kml, write_plan_kml + + +def _sample_plan() -> dict[str, object]: + return { + "request_id": "req-1", + "route": { + "type": "Feature", + "geometry": { + "type": "LineString", + "coordinates": [ + [-122.3937, 37.7955], + [-122.3927, 37.7965], + [-122.3917, 37.7975], + ], + }, + "properties": {}, + }, + "waypoints": [ + {"lat": 37.7975, "lon": -122.3917, "label": "Freshwater source"}, + ], + "rationale": "Route follows covered terrain to the nearest water source.", + "cost_breakdown": {"distance_m": 150.0, "time_s": 120.0, "elevation_gain_m": 0.0}, + } + + +def test_plan_to_kml_exports_route_and_waypoint() -> None: + kml = plan_to_kml(_sample_plan(), document_name="Demo route") + + assert "Demo route" in kml + assert "TERA route" in kml + assert "Freshwater source" in kml + assert "-122.3937000,37.7955000,0.00" in kml + assert "-122.3917000,37.7975000,0.00" in kml + + +def test_write_plan_kml_creates_parent_directory(tmp_path: Path) -> None: + output_path = tmp_path / "exports" / "route.kml" + + written = write_plan_kml(_sample_plan(), output_path) + + assert written == output_path + assert output_path.exists() + assert "Freshwater source" in output_path.read_text(encoding="utf-8") + + +def test_plan_to_kml_rejects_non_linestring() -> None: + plan = _sample_plan() + route = plan["route"] + assert isinstance(route, dict) + geometry = route["geometry"] + assert isinstance(geometry, dict) + geometry["type"] = "Point" + + with pytest.raises(PlanExportError, match="LineString"): + plan_to_kml(plan) + + +def test_plan_to_kml_rejects_single_coordinate_route() -> None: + plan = _sample_plan() + route = plan["route"] + assert isinstance(route, dict) + geometry = route["geometry"] + assert isinstance(geometry, dict) + geometry["coordinates"] = [[-122.3937, 37.7955]] + + with pytest.raises(PlanExportError, match="at least two"): + plan_to_kml(plan) diff --git a/tests/test_valhalla_client.py b/tests/test_valhalla_client.py new file mode 100644 index 0000000..c9870f5 --- /dev/null +++ b/tests/test_valhalla_client.py @@ -0,0 +1,102 @@ +"""Tests for local Valhalla route parsing.""" + +from __future__ import annotations + +import json +from collections.abc import Sequence +from pathlib import Path + +import pytest + +from routing.valhalla_client import ( + Coord, + ValhallaClient, + ValhallaRouteError, + decode_valhalla_polyline, + parse_valhalla_route, +) + + +def _encode_polyline(coordinates: Sequence[tuple[float, float]], *, precision: int = 6) -> str: + factor = 10**precision + last_lat = 0 + last_lon = 0 + encoded = [] + + for lat, lon in coordinates: + next_lat = round(lat * factor) + next_lon = round(lon * factor) + encoded.append(_encode_polyline_value(next_lat - last_lat)) + encoded.append(_encode_polyline_value(next_lon - last_lon)) + last_lat = next_lat + last_lon = next_lon + + return "".join(encoded) + + +def _encode_polyline_value(value: int) -> str: + shifted = ~(value << 1) if value < 0 else value << 1 + chunks = [] + while shifted >= 0x20: + chunks.append(chr((0x20 | (shifted & 0x1F)) + 63)) + shifted >>= 5 + chunks.append(chr(shifted + 63)) + return "".join(chunks) + + +def test_decode_valhalla_polyline_round_trips_coordinates() -> None: + coordinates = [(37.7955, -122.3937), (37.7965, -122.3927)] + shape = _encode_polyline(coordinates) + + assert decode_valhalla_polyline(shape) == coordinates + + +def test_parse_valhalla_route_returns_geojson_ready_result() -> None: + coordinates = [(37.7955, -122.3937), (37.7965, -122.3927)] + output = { + "trip": { + "summary": {"length": 0.15, "time": 120.0}, + "legs": [{"shape": _encode_polyline(coordinates)}], + } + } + + result = parse_valhalla_route(json.dumps(output)) + geojson = result.as_geojson_feature() + + assert result.distance_m == 150.0 + assert result.time_s == 120.0 + assert geojson["type"] == "Feature" + assert geojson["geometry"] == { + "type": "LineString", + "coordinates": [[-122.3937, 37.7955], [-122.3927, 37.7965]], + } + + +def test_parse_valhalla_route_rejects_missing_shape() -> None: + output = {"trip": {"summary": {"length": 0.15, "time": 120.0}, "legs": [{}]}} + + with pytest.raises(ValhallaRouteError, match="encoded shape"): + parse_valhalla_route(json.dumps(output)) + + +def test_valhalla_client_invokes_local_cli(tmp_path: Path) -> None: + shape = _encode_polyline([(37.7955, -122.3937), (37.7965, -122.3927)]) + output = json.dumps( + {"trip": {"summary": {"length": 0.15, "time": 120.0}, "legs": [{"shape": shape}]}} + ) + seen_command: list[str] = [] + + def runner(command: Sequence[str]) -> str: + seen_command.extend(command) + return output + + client = ValhallaClient(tmp_path / "valhalla.json", runner=runner) + result = client.route( + Coord(lat=37.7955, lon=-122.3937), + Coord(lat=37.7965, lon=-122.3927), + ) + + assert result.distance_m == 150.0 + assert seen_command[0] == "valhalla_run_route" + assert "valhalla.json" in seen_command[1] + assert '"costing":"pedestrian"' in seen_command[2]