first commit

api/app/services/terrain.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+from app.core.diffraction import deygout
+from app.core.fresnel import los_analysis
+from app.core.geo import GeoPoint, linestring_geojson, sample_path
+from app.core.surface import build_surface_profile
+from app.models.terrain import (
+    ElevationResponse,
+    LosRequest,
+    LosResponse,
+    Obstruction,
+    ProfileSample,
+    TerrainProfileRequest,
+    TerrainProfileResponse,
+)
+
+
+def elevation_at(lat: float, lon: float, surface: str) -> ElevationResponse:
+    raise NotImplementedError("DEM COG sampling is not configured yet")
+
+
+def terrain_profile(request: TerrainProfileRequest) -> TerrainProfileResponse:
+    points = sample_path(
+        GeoPoint(lat=request.start.lat, lon=request.start.lon),
+        GeoPoint(lat=request.end.lat, lon=request.end.lon),
+        request.samples,
+    )
+    profile = build_surface_profile(
+        points,
+        include_buildings=request.include_buildings,
+        include_canopy=request.include_canopy,
+    )
+    return TerrainProfileResponse(
+        distance_m=profile.distance_m,
+        samples=[ProfileSample(**sample.__dict__) for sample in profile.samples],
+        path_geojson=linestring_geojson(points),
+    )
+
+
+def los(request: LosRequest) -> LosResponse:
+    profile_request = TerrainProfileRequest(
+        start=request.tx,
+        end=request.rx,
+        samples=request.samples,
+        include_buildings=request.include_buildings,
+        include_canopy=request.include_canopy,
+        k_factor=request.k_factor,
+    )
+    points = sample_path(
+        GeoPoint(lat=request.tx.lat, lon=request.tx.lon),
+        GeoPoint(lat=request.rx.lat, lon=request.rx.lon),
+        request.samples,
+    )
+    surface_profile = build_surface_profile(
+        points,
+        include_buildings=request.include_buildings,
+        include_canopy=request.include_canopy,
+    )
+    result = los_analysis(
+        surface_profile,
+        request.tx.height_agl,
+        request.rx.height_agl,
+        request.frequency_mhz * 1_000_000,
+        request.fresnel_clearance,
+        request.k_factor,
+    )
+    diffraction_loss = deygout(
+        surface_profile,
+        request.tx.height_agl,
+        request.rx.height_agl,
+        request.frequency_mhz * 1_000_000,
+    )
+
+    def convert(sample: object) -> Obstruction:
+        return Obstruction(
+            distance_m=sample.distance_m,
+            clearance_m=sample.clearance_m,
+            type=sample.obstruction_type,
+        )
+
+    return LosResponse(
+        los_clear=result.los_clear,
+        geometric_los=result.geometric_los,
+        first_fresnel_clearance_pct=result.first_fresnel_clearance_pct,
+        worst_obstruction=convert(result.worst_sample) if result.worst_sample else None,
+        obstructions=[convert(sample) for sample in result.obstructions],
+        diffraction_loss_db=diffraction_loss,
+        profile_ref=f"inline:{profile_request.samples}",
+    )