""" CoverageEngine — main orchestrator for coverage calculations. Coordinates data loading, model selection, parallel computation, and result aggregation. Does NOT implement propagation physics (delegated to models) or handle HTTP (delegated to API layer). """ import time import asyncio from enum import Enum from dataclasses import dataclass from typing import List, Optional, Callable, Awaitable from app.propagation.base import PropagationModel, PropagationInput from app.propagation.free_space import FreeSpaceModel from app.propagation.okumura_hata import OkumuraHataModel from app.propagation.cost231_hata import Cost231HataModel from app.propagation.cost231_wi import Cost231WIModel from app.propagation.itu_r_p1546 import ITUR_P1546Model from app.propagation.longley_rice import LongleyRiceModel from app.propagation.itu_r_p526 import KnifeEdgeDiffractionModel from app.core.result import CoverageResult, PointResult, compute_stats class BandType(Enum): LTE = "lte" # 700-2600 MHz UHF = "uhf" # 400-520 MHz VHF = "vhf" # 136-174 MHz CUSTOM = "custom" # User-defined class PresetType(Enum): FAST = "fast" STANDARD = "standard" DETAILED = "detailed" FULL = "full" @dataclass class Site: id: str lat: float lon: float height: float # meters AGL power: float # dBm gain: float # dBi frequency: float # MHz band_type: BandType = BandType.LTE azimuth: Optional[float] = None beamwidth: float = 65 tilt: float = 0 environment: str = "urban" @dataclass class CoverageSettings: radius: float = 10000 resolution: float = 200 min_signal: float = -120 preset: PresetType = PresetType.STANDARD band_type: BandType = BandType.LTE environment: str = "urban" terrain_enabled: bool = True buildings_enabled: bool = True diffraction_enabled: bool = True reflection_enabled: bool = False # Legacy toggles (backward compat) use_terrain: bool = True use_buildings: bool = True use_materials: bool = True use_dominant_path: bool = False use_street_canyon: bool = False use_reflections: bool = False use_water_reflection: bool = False use_vegetation: bool = False season: str = "summer" rain_rate: float = 0.0 indoor_loss_type: str = "none" use_atmospheric: bool = False temperature_c: float = 15.0 humidity_percent: float = 50.0 ProgressCallback = Callable[[str, float, Optional[float]], Awaitable[None]] class CoverageEngine: """ Main orchestrator for coverage calculations. Selects the appropriate propagation model based on band type and environment, then delegates to the existing coverage pipeline. """ _model_registry = { (BandType.LTE, "urban"): Cost231HataModel, (BandType.LTE, "suburban"): OkumuraHataModel, (BandType.LTE, "rural"): OkumuraHataModel, (BandType.LTE, "open"): FreeSpaceModel, (BandType.UHF, "urban"): OkumuraHataModel, (BandType.UHF, "suburban"): OkumuraHataModel, (BandType.UHF, "rural"): LongleyRiceModel, (BandType.VHF, "urban"): ITUR_P1546Model, (BandType.VHF, "suburban"): ITUR_P1546Model, (BandType.VHF, "rural"): LongleyRiceModel, } def __init__(self): self._models = {} self._init_models() self.free_space = FreeSpaceModel() self.diffraction = KnifeEdgeDiffractionModel() def _init_models(self): for key, model_cls in self._model_registry.items(): self._models[key] = model_cls() def select_model(self, band: BandType, environment: str) -> PropagationModel: key = (band, environment) if key in self._models: return self._models[key] if (band, "urban") in self._models: return self._models[(band, "urban")] return OkumuraHataModel() def get_available_models(self) -> dict: models = {} seen = set() for (band, env), model in self._models.items(): if model.name not in seen: seen.add(model.name) models[model.name] = { "frequency_range": model.frequency_range, "distance_range": model.distance_range, "bands": [], } models[model.name]["bands"].append(f"{band.value}/{env}") return models async def calculate( self, sites: List[Site], settings: CoverageSettings, progress_callback: Optional[ProgressCallback] = None, ) -> CoverageResult: """ Main calculation entry point. Delegates actual per-point work to the legacy coverage_service pipeline, wrapping it with the new clean interface. """ start_time = time.time() model = self.select_model(settings.band_type, settings.environment) if progress_callback: await progress_callback("init", 0.05, None) # Import legacy system from app.services.coverage_service import ( coverage_service, SiteParams, CoverageSettings as LegacySettings, ) from app.services.parallel_coverage_service import CancellationToken legacy_settings = LegacySettings( radius=settings.radius, resolution=settings.resolution, min_signal=settings.min_signal, use_terrain=settings.use_terrain, use_buildings=settings.use_buildings, use_materials=settings.use_materials, use_dominant_path=settings.use_dominant_path, use_street_canyon=settings.use_street_canyon, use_reflections=settings.use_reflections, use_water_reflection=settings.use_water_reflection, use_vegetation=settings.use_vegetation, season=settings.season, rain_rate=settings.rain_rate, indoor_loss_type=settings.indoor_loss_type, use_atmospheric=settings.use_atmospheric, temperature_c=settings.temperature_c, humidity_percent=settings.humidity_percent, preset=settings.preset.value if isinstance(settings.preset, PresetType) else settings.preset, ) cancel_token = CancellationToken() if progress_callback: await progress_callback("calculating", 0.25, None) legacy_sites = [ SiteParams( lat=s.lat, lon=s.lon, height=s.height, power=s.power, gain=s.gain, frequency=s.frequency, azimuth=s.azimuth, beamwidth=s.beamwidth, ) for s in sites ] if len(legacy_sites) == 1: points = await coverage_service.calculate_coverage( legacy_sites[0], legacy_settings, cancel_token, ) else: points = await coverage_service.calculate_multi_site_coverage( legacy_sites, legacy_settings, cancel_token, ) if progress_callback: await progress_callback("done", 1.0, None) result_points = [ PointResult( lat=p.lat, lon=p.lon, rsrp=p.rsrp, distance=p.distance, path_loss=0.0, terrain_loss=p.terrain_loss, building_loss=p.building_loss, diffraction_loss=0.0, has_los=p.has_los, model_used=model.name, ) for p in points ] computation_time = time.time() - start_time return CoverageResult( points=result_points, stats=compute_stats(result_points), computation_time=computation_time, models_used=[model.name], ) # Singleton engine = CoverageEngine()