from __future__ import annotations import json import math import zipfile from dataclasses import asdict, dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any from shapely import affinity from shapely.geometry import LineString, MultiLineString, Polygon, mapping, shape from .io import write_json EARTH_METRES_PER_DEGREE_LATITUDE = 110_574.0 @dataclass(frozen=True) class MissionParameters: altitude_agl_m: float = 30.0 flight_speed_m_s: float = 2.0 line_spacing_m: float = 7.2 edge_buffer_m: float = 5.0 heading_deg: float = 0.0 front_overlap_pct: int = 85 side_overlap_pct: int = 80 rth_height_m: float = 45.0 def validate(self) -> None: if not 20 <= self.altitude_agl_m <= 60: raise ValueError("altitude_agl_m must be within the study's conservative envelope 20-60 m") if not 0.5 <= self.flight_speed_m_s <= 10: raise ValueError("flight_speed_m_s must be within 0.5-10 m/s") if not 2 <= self.line_spacing_m <= 30: raise ValueError("line_spacing_m must be within 2-30 m") if not 0 <= self.edge_buffer_m <= 20: raise ValueError("edge_buffer_m must be within 0-20 m") if not 60 <= self.front_overlap_pct <= 95 or not 60 <= self.side_overlap_pct <= 95: raise ValueError("front and side overlap must be within 60-95 percent") if self.rth_height_m < self.altitude_agl_m: raise ValueError("rth_height_m must not be lower than mapping altitude") def _local_scale(latitude: float) -> tuple[float, float]: return (111_320.0 * math.cos(math.radians(latitude)), EARTH_METRES_PER_DEGREE_LATITUDE) def _to_local(coordinates: list[tuple[float, float]], origin: tuple[float, float]) -> list[tuple[float, float]]: lon_scale, lat_scale = _local_scale(origin[1]) return [((lon - origin[0]) * lon_scale, (lat - origin[1]) * lat_scale) for lon, lat in coordinates] def _to_wgs84(coordinates: list[tuple[float, float]], origin: tuple[float, float]) -> list[tuple[float, float]]: lon_scale, lat_scale = _local_scale(origin[1]) return [(origin[0] + x / lon_scale, origin[1] + y / lat_scale) for x, y in coordinates] def load_field_polygon(path: str | Path) -> Polygon: value = json.loads(Path(path).read_text(encoding="utf-8")) geometry = value["geometry"] if value.get("type") == "Feature" else value polygon = shape(geometry) if not isinstance(polygon, Polygon) or not polygon.is_valid: raise ValueError("Field input must be one valid GeoJSON Polygon") if polygon.area <= 0: raise ValueError("Field polygon has no area") return polygon def plan_lawnmower_route(field_wgs84: Polygon, parameters: MissionParameters) -> dict[str, Any]: parameters.validate() centroid = field_wgs84.centroid origin = (centroid.x, centroid.y) shell = _to_local([(float(x), float(y)) for x, y in field_wgs84.exterior.coords], origin) local_field = Polygon(shell) safe_field = local_field.buffer(-parameters.edge_buffer_m) if safe_field.is_empty or not isinstance(safe_field, Polygon): raise ValueError("edge_buffer_m removes the usable field area") rotated = affinity.rotate(safe_field, -parameters.heading_deg, origin="centroid") min_x, min_y, max_x, max_y = rotated.bounds scan_lines: list[LineString] = [] y = min_y + parameters.line_spacing_m / 2.0 while y <= max_y: clipped = rotated.intersection(LineString([(min_x - 10, y), (max_x + 10, y)])) parts = list(clipped.geoms) if isinstance(clipped, MultiLineString) else [clipped] scan_lines.extend(part for part in parts if isinstance(part, LineString) and part.length >= 2.0) y += parameters.line_spacing_m if not scan_lines: raise ValueError("No route segments fit inside the buffered field") points_rotated: list[tuple[float, float]] = [] for index, segment in enumerate(sorted(scan_lines, key=lambda line: line.centroid.y)): endpoints = [tuple(segment.coords[0]), tuple(segment.coords[-1])] endpoints.sort(key=lambda point: point[0], reverse=bool(index % 2)) points_rotated.extend(endpoints) route_local = LineString(points_rotated) route_unrotated = affinity.rotate(route_local, parameters.heading_deg, origin=rotated.centroid) waypoints = _to_wgs84([(float(x), float(y)) for x, y in route_unrotated.coords], origin) for lon, lat in waypoints: point_local = _to_local([(lon, lat)], origin)[0] if not safe_field.buffer(0.05).covers(shape({"type": "Point", "coordinates": point_local})): raise AssertionError("Planner generated a waypoint outside the safe field") distance_m = float(route_unrotated.length) estimated_seconds = distance_m / parameters.flight_speed_m_s + len(scan_lines) * 2.0 return { "schemaVersion": 1, "planner": "m3m_scout.lawnmower-v1", "planningOnly": True, "executionAuthorized": False, "coordinateSystem": "WGS84 input; local equirectangular planning for a small field", "parameters": asdict(parameters), "fieldAreaM2": round(float(local_field.area), 2), "safeAreaM2": round(float(safe_field.area), 2), "routeDistanceM": round(distance_m, 2), "estimatedFlightSecondsExcludingTakeoffLanding": round(estimated_seconds, 1), "scanLineCount": len(scan_lines), "waypoints": [ {"index": index, "longitude": round(lon, 8), "latitude": round(lat, 8), "altitudeAglM": parameters.altitude_agl_m} for index, (lon, lat) in enumerate(waypoints) ], } def route_geojson(route: dict[str, Any]) -> dict[str, Any]: coordinates = [[point["longitude"], point["latitude"]] for point in route["waypoints"]] return { "type": "FeatureCollection", "features": [{ "type": "Feature", "properties": { "planningOnly": True, "routeDistanceM": route["routeDistanceM"], "altitudeAglM": route["parameters"]["altitude_agl_m"], }, "geometry": {"type": "LineString", "coordinates": coordinates}, }], } def evaluate_preflight_gate(checklist: dict[str, Any], now: datetime | None = None) -> dict[str, Any]: now = now or datetime.now(timezone.utc) required_truthy = [ "landholderPermissionReference", "pilotCredentialReference", "aircraftRegistrationReference", "daylightOperation", "visualLineOfSightPlanned", "peopleAndRoadExclusionConfirmed", "weatherWithinManualLimits", "pilot2MissionPreviewCompleted", ] failures = [key for key in required_truthy if not checklist.get(key)] checked_at = checklist.get("airspaceCheckedAt") if not checked_at: failures.append("airspaceCheckedAt") else: try: timestamp = datetime.fromisoformat(str(checked_at).replace("Z", "+00:00")) age_hours = (now - timestamp.astimezone(timezone.utc)).total_seconds() / 3600 if age_hours < 0 or age_hours > 24: failures.append("airspaceCheckedAtWithin24Hours") except ValueError: failures.append("airspaceCheckedAtValidIso8601") return { "schemaVersion": 1, "passed": not failures, "failures": sorted(set(failures)), "decision": "ELIGIBLE_FOR_OPERATOR_FINAL_REVIEW" if not failures else "NO_GO", "note": "Passing this software gate is not regulatory approval and does not command an aircraft.", } def write_planning_bundle(route: dict[str, Any], output_dir: str | Path) -> list[Path]: output = Path(output_dir) output.mkdir(parents=True, exist_ok=True) summary = write_json(output / "mission-summary.json", route) geojson = write_json(output / "planned-route.geojson", route_geojson(route)) coordinates = " ".join(f"{p['longitude']},{p['latitude']},{p['altitudeAglM']}" for p in route["waypoints"]) kml = f''' M3M planning preview - NOT FLIGHT EXECUTABLE Operator preview only. Recreate and validate the mapping mission in DJI Pilot 2. planned routerelativeToGround{coordinates} ''' kmz = output / "route-preview-not-flight-executable.kmz" def _stable_zip_entry(name: str, content: str) -> zipfile.ZipInfo: entry = zipfile.ZipInfo(name, date_time=(2026, 8, 4, 0, 0, 0)) entry.compress_type = zipfile.ZIP_DEFLATED entry.external_attr = 0o644 << 16 return entry with zipfile.ZipFile(kmz, "w", compression=zipfile.ZIP_DEFLATED) as archive: archive.writestr(_stable_zip_entry("doc.kml", kml), kml) notice = "Planning preview only. It contains no DJI WPML waylines.wpml and cannot authorize or command a flight.\n" archive.writestr(_stable_zip_entry("SAFETY-NOTICE.txt", notice), notice) return [summary, geojson, kmz]