import numpy as np from typing import List, Tuple, Optional from dataclasses import dataclass from app.services.buildings_service import Building from app.services.materials_service import materials_service @dataclass class ReflectionPath: """A reflection path with one or more bounces""" points: List[Tuple[float, float]] # [TX, reflection1, reflection2, ..., RX] total_distance: float total_loss: float reflection_count: int materials: List[str] class ReflectionService: """ Calculate reflection paths for RF propagation - Single bounce (most common) - Double bounce (around corners) - Ground reflection - Water surface reflection """ MAX_BOUNCES = 2 GROUND_REFLECTION_COEFF = 0.3 # Depends on surface # Ground types and reflection coefficients GROUND_REFLECTION = { "urban": 0.3, "suburban": 0.4, "rural": 0.5, "water": 0.8, "desert": 0.6, } async def find_reflection_paths( self, tx_lat: float, tx_lon: float, tx_height: float, rx_lat: float, rx_lon: float, rx_height: float, frequency_mhz: float, buildings: List[Building], include_ground: bool = True ) -> List[ReflectionPath]: """Find all viable reflection paths""" paths = [] # Single-bounce building reflections for building in buildings: path = self._find_single_bounce( tx_lat, tx_lon, tx_height, rx_lat, rx_lon, rx_height, frequency_mhz, building ) if path: paths.append(path) # Ground reflection if include_ground: ground_path = self._calculate_ground_reflection( tx_lat, tx_lon, tx_height, rx_lat, rx_lon, rx_height, frequency_mhz ) if ground_path: paths.append(ground_path) # Sort by loss (best first) paths.sort(key=lambda p: p.total_loss) return paths[:5] # Return top 5 def _find_single_bounce( self, tx_lat, tx_lon, tx_height, rx_lat, rx_lon, rx_height, frequency_mhz, building: Building ) -> Optional[ReflectionPath]: """Find single-bounce reflection off building""" # Find reflection point on building walls geometry = building.geometry for i in range(len(geometry) - 1): wall_start = geometry[i] wall_end = geometry[i + 1] ref_point = self._specular_reflection_point( (tx_lon, tx_lat), (rx_lon, rx_lat), wall_start, wall_end ) if not ref_point: continue ref_lat, ref_lon = ref_point[1], ref_point[0] # Calculate distances from app.services.terrain_service import TerrainService d1 = TerrainService.haversine_distance(tx_lat, tx_lon, ref_lat, ref_lon) d2 = TerrainService.haversine_distance(ref_lat, ref_lon, rx_lat, rx_lon) total_dist = d1 + d2 # Direct distance check - reflection shouldn't be much longer direct_dist = TerrainService.haversine_distance(tx_lat, tx_lon, rx_lat, rx_lon) if total_dist > direct_dist * 1.5: continue # Path loss path_loss = self._free_space_loss(total_dist, frequency_mhz) # Reflection loss material = materials_service.detect_material(building.tags) reflection_loss = materials_service.get_reflection_loss(material) total_loss = path_loss + reflection_loss return ReflectionPath( points=[(tx_lat, tx_lon), (ref_lat, ref_lon), (rx_lat, rx_lon)], total_distance=total_dist, total_loss=total_loss, reflection_count=1, materials=[material.value] ) return None def _calculate_ground_reflection( self, tx_lat, tx_lon, tx_height, rx_lat, rx_lon, rx_height, frequency_mhz, is_water: bool = False ) -> Optional[ReflectionPath]: """Calculate ground/water reflection path""" from app.services.terrain_service import TerrainService # Reflection point (simplified - midpoint for flat ground) mid_lat = (tx_lat + rx_lat) / 2 mid_lon = (tx_lon + rx_lon) / 2 # Path lengths d1 = TerrainService.haversine_distance(tx_lat, tx_lon, mid_lat, mid_lon) d2 = TerrainService.haversine_distance(mid_lat, mid_lon, rx_lat, rx_lon) # Actual path length considering heights path1 = np.sqrt(d1**2 + tx_height**2) path2 = np.sqrt(d2**2 + rx_height**2) total_dist = path1 + path2 # Path loss path_loss = self._free_space_loss(total_dist, frequency_mhz) # Reflection coefficient: water is much more reflective coeff = self.GROUND_REFLECTION.get("water" if is_water else "rural", 0.4) reflection_loss = -10 * np.log10(coeff) total_loss = path_loss + reflection_loss surface_type = "water" if is_water else "ground" return ReflectionPath( points=[(tx_lat, tx_lon), (mid_lat, mid_lon), (rx_lat, rx_lon)], total_distance=total_dist, total_loss=total_loss, reflection_count=1, materials=[surface_type] ) def _specular_reflection_point( self, tx: Tuple[float, float], # (lon, lat) rx: Tuple[float, float], wall_start: List[float], # [lon, lat] wall_end: List[float] ) -> Optional[Tuple[float, float]]: """Calculate specular reflection point on wall""" # Wall vector wx = wall_end[0] - wall_start[0] wy = wall_end[1] - wall_start[1] wall_len = np.sqrt(wx**2 + wy**2) if wall_len < 1e-10: return None # Normalize wx /= wall_len wy /= wall_len # Wall normal (perpendicular) nx = -wy ny = wx # Vector from wall start to TX tx_rel_x = tx[0] - wall_start[0] tx_rel_y = tx[1] - wall_start[1] # Distance from TX to wall line dist_to_wall = tx_rel_x * nx + tx_rel_y * ny # Mirror TX across wall mirror_x = tx[0] - 2 * dist_to_wall * nx mirror_y = tx[1] - 2 * dist_to_wall * ny # Line from mirror to RX dx = rx[0] - mirror_x dy = rx[1] - mirror_y # Find intersection with wall # Parametric: wall_start + t * wall_dir # Parametric: mirror + s * (rx - mirror) denom = dx * wy - dy * wx if abs(denom) < 1e-10: return None t = ((wall_start[0] - mirror_x) * wy - (wall_start[1] - mirror_y) * wx) / denom s = ((wall_start[0] - mirror_x) * dy - (wall_start[1] - mirror_y) * dx) / (-denom) if denom != 0 else 0 # Check if on wall segment and between mirror and RX if 0 <= s <= 1 and 0 <= t <= 1: ref_x = mirror_x + t * dx ref_y = mirror_y + t * dy return (ref_x, ref_y) return None def _free_space_loss(self, distance: float, frequency_mhz: float) -> float: """Free space path loss (dB)""" if distance <= 0: distance = 1 # FSPL = 20*log10(d) + 20*log10(f) + 20*log10(4*pi/c) # Simplified: FSPL = 32.45 + 20*log10(f_MHz) + 20*log10(d_km) d_km = distance / 1000 return 32.45 + 20 * np.log10(frequency_mhz) + 20 * np.log10(d_km + 0.001) def combine_paths( self, direct_power_dbm: float, reflection_paths: List[ReflectionPath], tx_power_dbm: float ) -> float: """ Combine direct and reflected signals (power sum) Returns total received power in dBm """ # Convert to linear power powers = [] if direct_power_dbm > -150: # Valid direct signal powers.append(10 ** (direct_power_dbm / 10)) for path in reflection_paths: reflected_power_dbm = tx_power_dbm - path.total_loss if reflected_power_dbm > -150: powers.append(10 ** (reflected_power_dbm / 10)) if not powers: return -150.0 # No signal # Sum powers (incoherent addition - conservative estimate) total_power = sum(powers) return 10 * np.log10(total_power) def find_reflection_paths_sync( self, tx_lat: float, tx_lon: float, tx_height: float, rx_lat: float, rx_lon: float, rx_height: float, frequency_mhz: float, buildings: List[Building], include_ground: bool = True ) -> List[ReflectionPath]: """Sync version (no I/O in the async original)""" paths = [] for building in buildings: path = self._find_single_bounce( tx_lat, tx_lon, tx_height, rx_lat, rx_lon, rx_height, frequency_mhz, building ) if path: paths.append(path) if include_ground: ground_path = self._calculate_ground_reflection( tx_lat, tx_lon, tx_height, rx_lat, rx_lon, rx_height, frequency_mhz ) if ground_path: paths.append(ground_path) paths.sort(key=lambda p: p.total_loss) return paths[:5] reflection_service = ReflectionService()