From 83ca6bc12ce830d57675e3543f5baf84c95620bf Mon Sep 17 00:00:00 2001 From: grigo Date: Fri, 26 Jun 2026 12:23:46 +0300 Subject: [PATCH] added formulas --- API.md | 55 +++- FORMULAS.md | 34 ++- api/app/core/coverage.py | 276 +++++++++++++++++- .../__pycache__/coverage.cpython-313.pyc | Bin 2663 -> 2708 bytes api/app/models/coverage.py | 1 + api/app/workers/tasks.py | 9 +- api/pyproject.toml | 1 + api/tests/test_jobs_and_propagation.py | 103 ++++++- 8 files changed, 456 insertions(+), 23 deletions(-) diff --git a/API.md b/API.md index 8734410..d72ffb0 100644 --- a/API.md +++ b/API.md @@ -7,8 +7,8 @@ ## Состояние Реализации -- Работает: healthcheck, OpenAPI, DEM elevation/profile, OSM buildings query, landcover/canopy path sampling, terrain LOS/Fresnel/diffraction с учётом DEM, buildings и canopy surface layer, link budget (`manual`, `itm`, `p452`) с P.676 atmospheric loss, P.833 vegetation и canopy препятствиями, antenna omni/sector/file, async coverage/viewshed jobs через Redis + Celery. -- Частично работает: coverage raster export (`format=geotiff|png`), canopy bootstrap script (данные можно качать автоматически, но объём большой). +- Работает: healthcheck, OpenAPI, DEM elevation/profile, OSM buildings query, landcover/canopy path sampling, terrain LOS/Fresnel/diffraction с учётом DEM, buildings и canopy surface layer, link budget (`manual`, `itm`, `p452`) с P.676 atmospheric loss, P.833 vegetation и canopy препятствиями, antenna omni/sector/file, async coverage/viewshed jobs через Redis + Celery, coverage GeoJSON/GeoTIFF/PNG export. +- Частично работает: canopy bootstrap script (данные можно качать автоматически, но объём большой). - Пока не реализовано: полный P.1812 с ITU digital maps (используется terrain-based ITM + clutter), WhiteboxTools viewshed (используется `gdal_viewshed`). ## Общие Правила @@ -532,7 +532,24 @@ curl -s -X POST http://localhost:5603/api/v1/viewshed \ ### `POST /api/v1/coverage` -Ставит async job в Celery. Radial coverage с моделями `fspl`, `itm`, `p1812` и antenna gain (omni/sector/file). Результат — GeoJSON contour polygons. +Ставит async job в Celery. Coverage поддерживает модели `fspl`, `itm`, `p1812`, +antenna gain (omni/sector/file), vegetation loss и surface obstruction от +buildings/canopy. + +Форматы: + +- `geojson`: radial contour polygons по `levels_dbm`. +- `geotiff`: регулярная UTM-сетка `rx_power_dbm`, float32, NoData `-9999`. +- `png`: 8-bit preview той же UTM-сетки; геопривязка пишется через GDAL metadata. + +Для raster formats размер пикселя берётся из `range_step_m`, а расчёт ограничен +кругом `radius_m`. + +Surface obstruction: + +- для `model=fspl` добавляется полный Bullington/P.526 loss по surface profile; +- для `model=itm|p1812` добавляется только excess от buildings/canopy поверх + terrain-only профиля, чтобы не дублировать terrain diffraction внутри модели. ```bash curl -s -X POST http://localhost:5603/api/v1/coverage \ @@ -547,6 +564,7 @@ curl -s -X POST http://localhost:5603/api/v1/coverage \ "azimuth_step_deg": 1, "range_step_m": 30, "include_buildings": true, + "include_canopy": true, "include_vegetation": false, "levels_dbm": [-90, -100, -110], "format": "geojson" @@ -561,6 +579,37 @@ curl -s -X POST http://localhost:5603/api/v1/coverage \ } ``` +Для `format=geojson` результат job содержит `data.type=FeatureCollection`. +Для `format=geotiff|png` результат job содержит: + +```json +{ + "kind": "coverage", + "model": "fspl", + "format": "geotiff", + "data": { + "kind": "coverage_raster", + "format": "geotiff", + "uri": "/data/output/jobs/coverage-fspl-....tif", + "metadata": { + "crs": "EPSG:32636", + "width": 400, + "height": 400, + "resolution_m": 30, + "radius_m": 20000, + "nodata": -9999.0, + "value_units": "dBm", + "include_buildings": true, + "include_canopy": true, + "surface_obstruction": true, + "valid_pixels": 12345, + "min_dbm": -118.2, + "max_dbm": -36.4 + } + } +} +``` + ## Jobs ### `GET /api/v1/jobs/{job_id}` diff --git a/FORMULAS.md b/FORMULAS.md index 1e028da..3997f48 100644 --- a/FORMULAS.md +++ b/FORMULAS.md @@ -472,11 +472,24 @@ EIRP = P_{tx} + G_{ant}(azimuth, 0^\circ) $$ $$ -P_{rx} = EIRP - L_{path} - L_{veg} + G_{rx} +P_{rx} = EIRP - L_{path} - L_{veg} - L_{surface} + G_{rx} $$ `L_path` выбирается по `model`: FSPL, ITM или p1812-like. +Surface obstruction (`L_surface`) считается через Bullington/P.526 по профилю +`ground + max(building, canopy)`: + +- для `fspl`: добавляется полный Bullington loss по surface profile; +- для `itm` и `p1812`: добавляется только excess от buildings/canopy поверх + terrain-only профиля: + +$$ +L_{surface} = \max(0,\ L_{surface\_profile} - L_{terrain\_profile}) +$$ + +Это снижает риск двойного учёта terrain diffraction, уже входящей в ITM/p1812-like path loss. + Алгоритм контура: 1. Для каждого азимута от 0 до 360 градусов. @@ -484,6 +497,23 @@ $$ 3. Пока $P_{rx} \ge level\_dbm$, точка считается покрытой. 4. Последняя покрытая точка образует радиальный контур. +Raster export: + +1. Вокруг TX строится UTM-сетка с пикселем `range_step_m`. +2. Размер bbox: $2R \times 2R$, где $R = radius_m$. +3. Для центра каждой ячейки считается расстояние и азимут от TX. +4. Если расстояние больше `radius_m`, пишется NoData `-9999`. +5. Иначе в ячейку пишется $P_{rx}$ в дБм. + +GeoTIFF хранит float32 `rx_power_dbm` с CRS/transform. PNG хранит 8-bit preview, +масштабированный по min/max конечных значений: + +$$ +png = 1 + \frac{P_{rx} - P_{min}}{P_{max} - P_{min}} \cdot 254 +$$ + +Нулевое значение PNG зарезервировано под NoData. + API: `/coverage` → Celery job → `/jobs/{id}`. ## 14. Viewshed @@ -563,7 +593,7 @@ API: `/buildings/query`, косвенно terrain/link. 4. P.452, ITM и P.676 делегированы внешним библиотекам. 5. `deygout()` оставлен для совместимости, но новый код использует явное `bullington_equivalent_loss()`. 6. `link_viable` зависит только от `fade_margin_db > 0`, не от качества Френеля. -7. Coverage строится радиальными контурами, не полноценной 2D-сеткой. +7. Coverage с buildings/canopy требует surface profile на лучах/ячейках и может быть дорогим на мелком `range_step_m`. 8. Viewshed не использует `k_factor`. 9. `diffraction_db` в `/link/budget` оставлен для обратной совместимости; для новой интеграции лучше читать `propagation_excess_db` и `knife_edge_diffraction_db`. diff --git a/api/app/core/coverage.py b/api/app/core/coverage.py index 7914d7e..7b2a400 100644 --- a/api/app/core/coverage.py +++ b/api/app/core/coverage.py @@ -1,17 +1,28 @@ from __future__ import annotations +from pathlib import Path from typing import Any +from uuid import uuid4 +from affine import Affine +import numpy as np +import rasterio from pyproj import Geod +from pyproj import Transformer +from rasterio.transform import from_origin +from sqlalchemy.orm import Session from app.core.antenna import AntennaPattern, gain +from app.core.diffraction import bullington_equivalent_loss from app.core.geo import GeoPoint, sample_path from app.core.propagation import fspl, itm_loss, p1812_field from app.models.antenna import AntennaPatternSpec from app.models.coverage import CoverageRequest +from app.services.terrain import surface_profile_from_points from app.services.vegetation import vegetation_loss_along _GEOD = Geod(ellps="WGS84") +_NODATA_DBM = -9999.0 def _antenna_pattern(spec: AntennaPatternSpec) -> AntennaPattern: @@ -66,10 +77,13 @@ def _rx_power_dbm( distance_km: float, azimuth_deg: float, elevation_profile: list[float] | None = None, + db: Session | None = None, + rx: GeoPoint | None = None, ) -> float: tx = GeoPoint(lat=request.tx.lat, lon=request.tx.lon) - rx_point = _GEOD.fwd(request.tx.lon, request.tx.lat, azimuth_deg, distance_km * 1000) - rx = GeoPoint(lat=rx_point[1], lon=rx_point[0]) + if rx is None: + rx_point = _GEOD.fwd(request.tx.lon, request.tx.lat, azimuth_deg, distance_km * 1000) + rx = GeoPoint(lat=rx_point[1], lon=rx_point[0]) pattern = _antenna_pattern(request.antenna) antenna_gain = gain(pattern, azimuth_deg, 0.0) path_loss = _path_loss_db(request, tx, rx, distance_km, azimuth_deg, elevation_profile) @@ -81,14 +95,251 @@ def _rx_power_dbm( freq_hz=request.tx.frequency_mhz * 1_000_000, include_vegetation=True, ) + surface_obstruction_db = _surface_obstruction_loss_db(request, tx, rx, db) eirp = request.tx.power_dbm + antenna_gain - return eirp - path_loss - vegetation_db + request.rx.gain_dbi + return eirp - path_loss - vegetation_db - surface_obstruction_db + request.rx.gain_dbi + + +def _surface_obstruction_loss_db( + request: CoverageRequest, + tx: GeoPoint, + rx: GeoPoint, + db: Session | None, +) -> float: + if not request.include_buildings and not request.include_canopy: + return 0.0 + + points = sample_path(tx, rx, 64) + surface_profile = surface_profile_from_points( + points, + include_buildings=request.include_buildings, + include_canopy=request.include_canopy, + db=db, + ) + surface_loss = bullington_equivalent_loss( + surface_profile, + tx_height_agl=request.tx.height_agl, + rx_height_agl=request.rx.height_agl, + freq_hz=request.tx.frequency_mhz * 1_000_000, + ) + if request.model == "fspl": + return surface_loss + + terrain_profile = surface_profile_from_points( + points, + include_buildings=False, + include_canopy=False, + db=None, + ) + terrain_loss = bullington_equivalent_loss( + terrain_profile, + tx_height_agl=request.tx.height_agl, + rx_height_agl=request.rx.height_agl, + freq_hz=request.tx.frequency_mhz * 1_000_000, + ) + return max(0.0, surface_loss - terrain_loss) + + +def _utm_epsg(lon: float, lat: float) -> int: + zone = int((lon + 180) // 6) + 1 + return 32600 + zone if lat >= 0 else 32700 + zone + + +def _rx_power_at_point( + request: CoverageRequest, + lon: float, + lat: float, + distance_m: float, + azimuth_deg: float, + dem_path: str | None, + db: Session | None, +) -> float: + distance_km = max(distance_m, 1.0) / 1000.0 + elevation_profile = None + if request.model in {"itm", "p1812"} and dem_path is not None: + tx = GeoPoint(lat=request.tx.lat, lon=request.tx.lon) + rx = GeoPoint(lat=lat, lon=lon) + try: + from app.core.dem import elevations_along + + points = sample_path(tx, rx, 64) + elevation_profile = elevations_along(points, dem_path=dem_path).tolist() + except Exception: + elevation_profile = None + rx = GeoPoint(lat=lat, lon=lon) + return _rx_power_dbm(request, distance_km, azimuth_deg, elevation_profile, db=db, rx=rx) + + +def _coverage_grid( + request: CoverageRequest, + *, + dem_path: str | None = None, + db: Session | None = None, +) -> tuple[np.ndarray, dict[str, Any]]: + resolution_m = request.range_step_m + radius_m = request.radius_m + width = max(1, int(np.ceil((radius_m * 2) / resolution_m))) + height = width + epsg = _utm_epsg(request.tx.lon, request.tx.lat) + to_utm = Transformer.from_crs("EPSG:4326", f"EPSG:{epsg}", always_xy=True) + to_wgs84 = Transformer.from_crs(f"EPSG:{epsg}", "EPSG:4326", always_xy=True) + tx_x, tx_y = to_utm.transform(request.tx.lon, request.tx.lat) + west = tx_x - (width * resolution_m) / 2 + north = tx_y + (height * resolution_m) / 2 + data = np.full((height, width), _NODATA_DBM, dtype="float32") + + for row in range(height): + y = north - (row + 0.5) * resolution_m + for col in range(width): + x = west + (col + 0.5) * resolution_m + dx = x - tx_x + dy = y - tx_y + distance_m = float(np.hypot(dx, dy)) + if distance_m > radius_m: + continue + lon, lat = to_wgs84.transform(x, y) + azimuth, _, geodesic_distance_m = _GEOD.inv( + request.tx.lon, + request.tx.lat, + lon, + lat, + ) + data[row, col] = _rx_power_at_point( + request, + lon, + lat, + geodesic_distance_m, + azimuth % 360, + dem_path, + db, + ) + + transform = from_origin(west, north, resolution_m, resolution_m) + finite = data[data != _NODATA_DBM] + metadata: dict[str, Any] = { + "crs": f"EPSG:{epsg}", + "width": width, + "height": height, + "resolution_m": resolution_m, + "radius_m": radius_m, + "nodata": _NODATA_DBM, + "bounds": { + "west": west, + "south": north - height * resolution_m, + "east": west + width * resolution_m, + "north": north, + }, + "transform": [transform.a, transform.b, transform.c, transform.d, transform.e, transform.f], + "value_units": "dBm", + "valid_pixels": int(finite.size), + "include_buildings": request.include_buildings, + "include_canopy": request.include_canopy, + "surface_obstruction": request.include_buildings or request.include_canopy, + } + if finite.size: + metadata["min_dbm"] = float(np.min(finite)) + metadata["max_dbm"] = float(np.max(finite)) + return data, metadata + + +def _write_geotiff( + data: np.ndarray, + metadata: dict[str, Any], + path: Path, +) -> None: + transform = Affine(*metadata["transform"]) + with rasterio.open( + path, + "w", + driver="GTiff", + height=data.shape[0], + width=data.shape[1], + count=1, + dtype="float32", + crs=metadata["crs"], + transform=transform, + nodata=_NODATA_DBM, + compress="deflate", + ) as dataset: + dataset.write(data, 1) + dataset.update_tags( + model=metadata["model"], + frequency_mhz=str(metadata["frequency_mhz"]), + value_units="dBm", + ) + + +def _write_png_preview(data: np.ndarray, metadata: dict[str, Any], path: Path) -> dict[str, Any]: + finite_mask = data != _NODATA_DBM + preview = np.zeros(data.shape, dtype="uint8") + if finite_mask.any(): + finite = data[finite_mask] + min_value = float(np.min(finite)) + max_value = float(np.max(finite)) + if max_value > min_value: + scaled = 1 + ((data[finite_mask] - min_value) / (max_value - min_value) * 254) + preview[finite_mask] = scaled.astype("uint8") + else: + preview[finite_mask] = 255 + else: + min_value = None + max_value = None + + with rasterio.open( + path, + "w", + driver="PNG", + height=preview.shape[0], + width=preview.shape[1], + count=1, + dtype="uint8", + crs=metadata["crs"], + transform=Affine(*metadata["transform"]), + ) as dataset: + dataset.write(preview, 1) + return {"png_min_dbm": min_value, "png_max_dbm": max_value} + + +def _raster_export( + request: CoverageRequest, + *, + dem_path: str | None, + output_dir: Path, + db: Session | None, +) -> dict[str, Any]: + output_dir.mkdir(parents=True, exist_ok=True) + data, metadata = _coverage_grid(request, dem_path=dem_path, db=db) + metadata.update( + { + "model": request.model, + "frequency_mhz": request.tx.frequency_mhz, + "tx": {"lat": request.tx.lat, "lon": request.tx.lon}, + "format": request.format, + } + ) + stem = f"coverage-{request.model}-{uuid4().hex[:12]}" + if request.format == "geotiff": + path = output_dir / f"{stem}.tif" + _write_geotiff(data, metadata, path) + elif request.format == "png": + path = output_dir / f"{stem}.png" + metadata.update(_write_png_preview(data, metadata, path)) + else: + raise ValueError(f"Unsupported raster coverage format: {request.format}") + + return { + "kind": "coverage_raster", + "format": request.format, + "uri": str(path), + "metadata": metadata, + } def _contour_points( request: CoverageRequest, level_dbm: float, dem_path: str | None = None, + db: Session | None = None, ) -> list[list[float]]: coords: list[list[float]] = [] azimuth = 0.0 @@ -111,7 +362,7 @@ def _contour_points( elevation_profile = elevations_along(points, dem_path=dem_path).tolist() except Exception: elevation_profile = None - rx_power = _rx_power_dbm(request, distance_km, azimuth, elevation_profile) + rx_power = _rx_power_dbm(request, distance_km, azimuth, elevation_profile, db=db) if rx_power >= level_dbm: lon, lat, _ = _GEOD.fwd(request.tx.lon, request.tx.lat, azimuth, distance_m) last_good = [lon, lat] @@ -130,10 +381,12 @@ def compute_coverage( request: CoverageRequest, *, dem_path: str | None = None, + output_dir: str | Path | None = None, + db: Session | None = None, ) -> dict[str, Any]: features: list[dict[str, Any]] = [] for level in request.levels_dbm: - ring = _contour_points(request, level, dem_path=dem_path) + ring = _contour_points(request, level, dem_path=dem_path, db=db) if len(ring) < 4: continue features.append( @@ -143,6 +396,8 @@ def compute_coverage( "level_dbm": level, "model": request.model, "frequency_mhz": request.tx.frequency_mhz, + "include_buildings": request.include_buildings, + "include_canopy": request.include_canopy, }, "geometry": {"type": "Polygon", "coordinates": [ring]}, } @@ -156,11 +411,12 @@ def compute_coverage( "model": request.model, "radius_m": request.radius_m, "levels_dbm": request.levels_dbm, + "include_buildings": request.include_buildings, + "include_canopy": request.include_canopy, + "surface_obstruction": request.include_buildings or request.include_canopy, }, } - return { - "format": request.format, - "feature_count": len(features), - "message": "Raster coverage export is not implemented; use format=geojson", - } + if output_dir is None: + raise ValueError("output_dir is required for raster coverage export") + return _raster_export(request, dem_path=dem_path, output_dir=Path(output_dir), db=db) diff --git a/api/app/models/__pycache__/coverage.cpython-313.pyc b/api/app/models/__pycache__/coverage.cpython-313.pyc index b2e4165e27e4104ecbae890f649d328c9d5b8295..37b7514c517b9a121035ec19763d0e9024627beb 100644 GIT binary patch delta 265 zcmaDZGDVd4GcPX}0}$MBw#(|?$a{gA@z>=0O!AZeGS71g7A+Ev5sYCCmQV+(h~ZXZ z2$qy$2$r&BEE0(kjp1bliLwSugJgqcio{~Xfjn6-Pp(KJU0zdR@_Cl`{Ct^t$vLGd zsqx8)dHDsEllQW=Fe*%TV)JB_oZQ2fC};uHUSta-Zn5NK7MB#+PyWfK&A527Dtig1 zw0wi#1eeP!@{?z5xF(6L{%#$q=PnXk_pM00){p35W&5ZJs6WBZ%B_^+8OXRTu zD!;{&lUZC+WH(ucU7K;{WDoX|$;a4R8I>pda0GEV0TmR12<6EuI20J2CZFU8W7L?e zz!}8l3X%d58k1`{6&PJ7&*dyq0jmeg2!f;~L4*{LxW!?Uo1apelWJGwJ=vJciid|$ RbwcJB1|aoCZgM4;GytfXI`aSk diff --git a/api/app/models/coverage.py b/api/app/models/coverage.py index 4e3e346..1ab7f6e 100644 --- a/api/app/models/coverage.py +++ b/api/app/models/coverage.py @@ -27,6 +27,7 @@ class CoverageRequest(BaseModel): azimuth_step_deg: float = Field(default=1.0, gt=0, le=360) range_step_m: float = Field(default=30, gt=0) include_buildings: bool = True + include_canopy: bool = True include_vegetation: bool = True levels_dbm: list[float] = Field(default_factory=lambda: [-90, -100, -110]) format: OutputFormat = "geojson" diff --git a/api/app/workers/tasks.py b/api/app/workers/tasks.py index eefaebf..7aaea0d 100644 --- a/api/app/workers/tasks.py +++ b/api/app/workers/tasks.py @@ -3,6 +3,7 @@ from __future__ import annotations from app.config import get_settings from app.core.coverage import compute_coverage as run_coverage from app.core.viewshed import compute_viewshed as run_viewshed +from app.deps import SessionLocal from app.models.coverage import CoverageRequest from app.models.viewshed import ViewshedRequest from app.services.jobs import get_job_record, update_job @@ -19,7 +20,13 @@ def compute_coverage(job_id: str) -> dict[str, str]: try: update_job(job_id, status="running") request = CoverageRequest.model_validate(record["payload"]) - result = run_coverage(request, dem_path=str(settings.dem_path)) + with SessionLocal() as db: + result = run_coverage( + request, + dem_path=str(settings.dem_path), + output_dir=settings.jobs_output_path, + db=db, + ) update_job( job_id, status="done", diff --git a/api/pyproject.toml b/api/pyproject.toml index c5dd8d2..802c30e 100644 --- a/api/pyproject.toml +++ b/api/pyproject.toml @@ -4,6 +4,7 @@ version = "0.1.0" description = "RF propagation HTTP API" requires-python = ">=3.11" dependencies = [ + "affine", "astropy", "celery", "fastapi", diff --git a/api/tests/test_jobs_and_propagation.py b/api/tests/test_jobs_and_propagation.py index f8fa8a7..a0245e6 100644 --- a/api/tests/test_jobs_and_propagation.py +++ b/api/tests/test_jobs_and_propagation.py @@ -3,9 +3,12 @@ from __future__ import annotations from pathlib import Path import pytest +import rasterio +from app.core import coverage as coverage_core from app.core.antenna import AntennaPattern, gain from app.core.coverage import compute_coverage +from app.core.surface import SurfaceProfile, SurfaceSample from app.core.vegetation import WORLDCOVER_P833, p833_coefficients_for_class from app.models.coverage import CoverageRequest from app.services import jobs @@ -22,7 +25,94 @@ def test_jobs_store_roundtrip() -> None: def test_coverage_fspl_geojson() -> None: - request = CoverageRequest( + request = _coverage_request(format="geojson", radius_m=5000, range_step_m=500) + result = compute_coverage(request) + assert result["type"] == "FeatureCollection" + assert len(result["features"]) >= 1 + + +def test_coverage_geotiff_export(tmp_path: Path) -> None: + request = _coverage_request(format="geotiff") + + result = compute_coverage(request, output_dir=tmp_path) + + path = Path(result["uri"]) + assert path.exists() + assert result["kind"] == "coverage_raster" + assert result["format"] == "geotiff" + assert result["metadata"]["value_units"] == "dBm" + assert result["metadata"]["valid_pixels"] > 0 + with rasterio.open(path) as dataset: + assert dataset.driver == "GTiff" + assert dataset.count == 1 + assert dataset.crs is not None + assert dataset.nodata == -9999.0 + data = dataset.read(1) + assert data.shape == (2, 2) + assert (data != -9999.0).any() + + +def test_coverage_png_export(tmp_path: Path) -> None: + request = _coverage_request(format="png") + + result = compute_coverage(request, output_dir=tmp_path) + + path = Path(result["uri"]) + assert path.exists() + assert result["kind"] == "coverage_raster" + assert result["format"] == "png" + assert result["metadata"]["png_min_dbm"] is not None + with rasterio.open(path) as dataset: + assert dataset.driver == "PNG" + assert dataset.count == 1 + assert dataset.read(1).shape == (2, 2) + + +def test_coverage_surface_obstruction_reduces_rx_power(monkeypatch) -> None: + def fake_surface_profile(points, *, include_buildings, include_canopy, db=None): + midpoint = len(points) // 2 + samples = [ + SurfaceSample( + i=index, + lat=point.lat, + lon=point.lon, + distance_m=point.distance_m, + ground_m=0.0, + building_m=0.0, + canopy_m=80.0 if include_canopy and index == midpoint else 0.0, + surface_m=80.0 if include_canopy and index == midpoint else 0.0, + ) + for index, point in enumerate(points) + ] + return SurfaceProfile(distance_m=points[-1].distance_m, samples=samples) + + monkeypatch.setattr(coverage_core, "surface_profile_from_points", fake_surface_profile) + obstructed = _coverage_request( + format="geojson", + include_buildings=False, + include_canopy=True, + ) + clear = _coverage_request( + format="geojson", + include_buildings=False, + include_canopy=False, + ) + + obstructed_power = coverage_core._rx_power_dbm(obstructed, 1.0, 90.0) + clear_power = coverage_core._rx_power_dbm(clear, 1.0, 90.0) + + assert obstructed_power < clear_power + + +def _coverage_request( + *, + format: str, + radius_m: float = 1000, + range_step_m: float = 1000, + include_buildings: bool = False, + include_canopy: bool = False, +) -> CoverageRequest: + return CoverageRequest( tx={ "lat": 59.935, "lon": 30.305, @@ -33,16 +123,15 @@ def test_coverage_fspl_geojson() -> None: antenna={"pattern": "omni", "gain_dbi": 8}, rx={"height_agl": 2, "sensitivity_dbm": -110, "gain_dbi": 2}, model="fspl", - radius_m=5000, + radius_m=radius_m, azimuth_step_deg=90, - range_step_m=500, + range_step_m=range_step_m, + include_buildings=include_buildings, + include_canopy=include_canopy, include_vegetation=False, levels_dbm=[-90], - format="geojson", + format=format, ) - result = compute_coverage(request) - assert result["type"] == "FeatureCollection" - assert len(result["features"]) >= 1 def test_worldcover_p833_mapping() -> None: