import numpy as np import asyncio from typing import List, Optional, Tuple from pydantic import BaseModel from app.services.terrain_service import terrain_service, TerrainService from app.services.los_service import los_service from app.services.buildings_service import buildings_service, Building from app.services.materials_service import materials_service from app.services.dominant_path_service import dominant_path_service from app.services.street_canyon_service import street_canyon_service, Street from app.services.reflection_service import reflection_service class CoveragePoint(BaseModel): lat: float lon: float rsrp: float # dBm distance: float # meters from site has_los: bool terrain_loss: float # dB building_loss: float # dB reflection_gain: float = 0.0 # dB (NEW) class CoverageSettings(BaseModel): radius: float = 10000 # meters resolution: float = 200 # meters min_signal: float = -120 # dBm threshold # Layer toggles 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 # Preset preset: Optional[str] = None # fast, standard, detailed, full # Propagation model presets PRESETS = { "fast": { "use_terrain": True, "use_buildings": False, "use_materials": False, "use_dominant_path": False, "use_street_canyon": False, "use_reflections": False, }, "standard": { "use_terrain": True, "use_buildings": True, "use_materials": True, "use_dominant_path": False, "use_street_canyon": False, "use_reflections": False, }, "detailed": { "use_terrain": True, "use_buildings": True, "use_materials": True, "use_dominant_path": True, "use_street_canyon": False, "use_reflections": False, }, "full": { "use_terrain": True, "use_buildings": True, "use_materials": True, "use_dominant_path": True, "use_street_canyon": True, "use_reflections": True, }, } def apply_preset(settings: CoverageSettings) -> CoverageSettings: """Apply preset configuration to settings""" if settings.preset and settings.preset in PRESETS: for key, value in PRESETS[settings.preset].items(): setattr(settings, key, value) return settings class SiteParams(BaseModel): lat: float lon: float height: float = 30 # antenna height meters power: float = 43 # dBm (20W) gain: float = 15 # dBi frequency: float = 1800 # MHz azimuth: Optional[float] = None # degrees, None = omni beamwidth: Optional[float] = 65 # degrees class CoverageService: """ RF Coverage calculation with terrain, buildings, materials, dominant path, street canyon, and reflections """ EARTH_RADIUS = 6371000 def __init__(self): self.terrain = terrain_service self.buildings = buildings_service self.los = los_service async def calculate_coverage( self, site: SiteParams, settings: CoverageSettings ) -> List[CoveragePoint]: """ Calculate coverage grid for a single site Returns list of CoveragePoint with RSRP values """ # Apply preset if specified settings = apply_preset(settings) points = [] # Generate grid grid = self._generate_grid( site.lat, site.lon, settings.radius, settings.resolution ) # Calculate bbox for data fetching lat_delta = settings.radius / 111000 lon_delta = settings.radius / (111000 * np.cos(np.radians(site.lat))) # Fetch buildings for coverage area (if enabled) buildings: List[Building] = [] if settings.use_buildings: buildings = await self.buildings.fetch_buildings( site.lat - lat_delta, site.lon - lon_delta, site.lat + lat_delta, site.lon + lon_delta ) # Fetch streets (if street canyon enabled) streets: List[Street] = [] if settings.use_street_canyon: streets = await street_canyon_service.fetch_streets( site.lat - lat_delta, site.lon - lon_delta, site.lat + lat_delta, site.lon + lon_delta ) # Calculate coverage for each point for lat, lon in grid: point = await self._calculate_point( site, lat, lon, settings, buildings, streets ) if point.rsrp >= settings.min_signal: points.append(point) return points async def calculate_multi_site_coverage( self, sites: List[SiteParams], settings: CoverageSettings ) -> List[CoveragePoint]: """ Calculate combined coverage from multiple sites Best server (strongest signal) wins at each point """ if not sites: return [] # Apply preset once settings = apply_preset(settings) # Get all individual coverages all_coverages = await asyncio.gather(*[ self.calculate_coverage(site, settings) for site in sites ]) # Combine by best signal point_map: dict[Tuple[float, float], CoveragePoint] = {} for coverage in all_coverages: for point in coverage: key = (round(point.lat, 6), round(point.lon, 6)) if key not in point_map or point.rsrp > point_map[key].rsrp: point_map[key] = point return list(point_map.values()) def _generate_grid( self, center_lat: float, center_lon: float, radius: float, resolution: float ) -> List[Tuple[float, float]]: """Generate coverage grid points""" points = [] # Convert resolution to degrees lat_step = resolution / 111000 lon_step = resolution / (111000 * np.cos(np.radians(center_lat))) # Calculate grid bounds lat_delta = radius / 111000 lon_delta = radius / (111000 * np.cos(np.radians(center_lat))) lat = center_lat - lat_delta while lat <= center_lat + lat_delta: lon = center_lon - lon_delta while lon <= center_lon + lon_delta: # Check if within radius (circular, not square) dist = TerrainService.haversine_distance(center_lat, center_lon, lat, lon) if dist <= radius: points.append((lat, lon)) lon += lon_step lat += lat_step return points async def _calculate_point( self, site: SiteParams, lat: float, lon: float, settings: CoverageSettings, buildings: List[Building], streets: List[Street] ) -> CoveragePoint: """Calculate RSRP at a single point with all propagation models""" # Distance distance = TerrainService.haversine_distance(site.lat, site.lon, lat, lon) if distance < 1: distance = 1 # Avoid division by zero # Base path loss (Okumura-Hata for urban) path_loss = self._okumura_hata( distance, site.frequency, site.height, 1.5 # 1.5m receiver height ) # Antenna pattern loss (if directional) antenna_loss = 0.0 if site.azimuth is not None and site.beamwidth: antenna_loss = self._antenna_pattern_loss( site.lat, site.lon, lat, lon, site.azimuth, site.beamwidth ) # Terrain loss (LoS check) terrain_loss = 0.0 has_los = True if settings.use_terrain: los_result = await self.los.check_line_of_sight( site.lat, site.lon, site.height, lat, lon, 1.5 # receiver at 1.5m ) has_los = los_result["has_los"] if not has_los: # Add diffraction loss based on clearance clearance = los_result["clearance"] terrain_loss = self._diffraction_loss(clearance, site.frequency) # Building loss (with optional material awareness) building_loss = 0.0 if settings.use_buildings and buildings: if settings.use_materials: # Material-aware building loss for building in buildings: intersection = self.buildings.line_intersects_building( site.lat, site.lon, site.height + await self.terrain.get_elevation(site.lat, site.lon), lat, lon, 1.5 + await self.terrain.get_elevation(lat, lon), building ) if intersection is not None: material = materials_service.detect_material(building.tags) building_loss += materials_service.get_penetration_loss( material, site.frequency ) has_los = False break # One building is enough else: # Simple building loss (legacy behavior) for building in buildings: intersection = self.buildings.line_intersects_building( site.lat, site.lon, site.height + await self.terrain.get_elevation(site.lat, site.lon), lat, lon, 1.5 + await self.terrain.get_elevation(lat, lon), building ) if intersection is not None: building_loss += 20.0 # Default concrete has_los = False break # Dominant path analysis (find best route) if settings.use_dominant_path and buildings: paths = await dominant_path_service.find_dominant_paths( site.lat, site.lon, site.height, lat, lon, 1.5, site.frequency, buildings ) if paths: best_path = paths[0] # Use best path's loss if it's better if best_path.is_valid and best_path.path_loss < (path_loss + terrain_loss + building_loss): path_loss = best_path.path_loss terrain_loss = 0 building_loss = 0 has_los = best_path.path_type == "direct" and not best_path.materials_crossed # Street canyon model if settings.use_street_canyon and streets: canyon_loss, street_path = await street_canyon_service.calculate_street_canyon_loss( site.lat, site.lon, site.height, lat, lon, 1.5, site.frequency, streets ) # Use canyon loss if better than current total if canyon_loss < (path_loss + terrain_loss + building_loss): path_loss = canyon_loss terrain_loss = 0 building_loss = 0 # Reflections reflection_gain = 0.0 if settings.use_reflections and buildings: reflection_paths = await reflection_service.find_reflection_paths( site.lat, site.lon, site.height, lat, lon, 1.5, site.frequency, buildings ) if reflection_paths: # Combine direct and reflected signals direct_rsrp = site.power + site.gain - path_loss - antenna_loss - terrain_loss - building_loss combined_rsrp = reflection_service.combine_paths( direct_rsrp, reflection_paths, site.power + site.gain ) reflection_gain = max(0, combined_rsrp - direct_rsrp) # Final RSRP rsrp = site.power + site.gain - path_loss - antenna_loss - terrain_loss - building_loss + reflection_gain return CoveragePoint( lat=lat, lon=lon, rsrp=rsrp, distance=distance, has_los=has_los, terrain_loss=terrain_loss, building_loss=building_loss, reflection_gain=reflection_gain ) def _okumura_hata( self, distance: float, # meters frequency: float, # MHz tx_height: float, # meters rx_height: float # meters ) -> float: """ Okumura-Hata path loss model (urban) Returns path loss in dB """ d_km = distance / 1000 if d_km < 0.1: d_km = 0.1 # Minimum distance # Mobile antenna height correction (urban) a_hm = (1.1 * np.log10(frequency) - 0.7) * rx_height - (1.56 * np.log10(frequency) - 0.8) # Path loss L = (69.55 + 26.16 * np.log10(frequency) - 13.82 * np.log10(tx_height) - a_hm + (44.9 - 6.55 * np.log10(tx_height)) * np.log10(d_km)) return L def _antenna_pattern_loss( self, site_lat: float, site_lon: float, point_lat: float, point_lon: float, azimuth: float, beamwidth: float ) -> float: """Calculate antenna pattern attenuation""" # Calculate bearing from site to point bearing = self._calculate_bearing(site_lat, site_lon, point_lat, point_lon) # Angle difference from main lobe angle_diff = abs(bearing - azimuth) if angle_diff > 180: angle_diff = 360 - angle_diff # Simple cosine pattern approximation # 3dB beamwidth = angle where power drops to half half_beamwidth = beamwidth / 2 if angle_diff <= half_beamwidth: # Within main lobe - minimal loss loss = 3 * (angle_diff / half_beamwidth) ** 2 else: # Outside main lobe - significant loss loss = 3 + 12 * ((angle_diff - half_beamwidth) / half_beamwidth) ** 2 loss = min(loss, 25) # Cap at 25dB (back lobe) return loss def _calculate_bearing( self, lat1: float, lon1: float, lat2: float, lon2: float ) -> float: """Calculate bearing from point 1 to point 2 (degrees)""" lat1, lon1, lat2, lon2 = map(np.radians, [lat1, lon1, lat2, lon2]) dlon = lon2 - lon1 x = np.sin(dlon) * np.cos(lat2) y = np.cos(lat1) * np.sin(lat2) - np.sin(lat1) * np.cos(lat2) * np.cos(dlon) bearing = np.degrees(np.arctan2(x, y)) return (bearing + 360) % 360 def _diffraction_loss(self, clearance: float, frequency: float) -> float: """ Knife-edge diffraction loss Args: clearance: Clearance in meters (negative = obstructed) frequency: Frequency in MHz Returns: Additional loss in dB """ if clearance >= 0: return 0.0 # No obstruction # Fresnel parameter approximation v = abs(clearance) / 10 # Normalize # Knife-edge loss approximation if v <= 0: loss = 0 elif v < 2.4: loss = 6.02 + 9.11 * v - 1.27 * v**2 else: loss = 13.0 + 20 * np.log10(v) return min(loss, 40) # Cap at 40dB # Singleton coverage_service = CoverageService()