mytec
defa3ad440
Major refactoring of RFCP backend: - Modular propagation models (8 models) - SharedMemoryManager for terrain data - ProcessPoolExecutor parallel processing - WebSocket progress streaming - Building filtering pipeline (351k → 15k) - 82 unit tests Performance: Standard preset 38s → 5s (7.6x speedup) Known issue: Detailed preset timeout (fix in 3.1.0)
241 lines
7.7 KiB
Python
241 lines
7.7 KiB
Python
"""
|
|
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()
|