""" R-tree spatial index for fast building and geometry lookups. Uses a simple grid-based approach (no external dependency) for O(1) amortised lookups instead of O(n) linear scans. """ from typing import List, Tuple, Optional, Dict from collections import defaultdict from app.services.buildings_service import Building class SpatialIndex: """Grid-based spatial index for fast building lookups""" def __init__(self, cell_size: float = 0.001): """ Args: cell_size: Grid cell size in degrees (~111m at equator) """ self.cell_size = cell_size self._grid: Dict[Tuple[int, int], List[Building]] = defaultdict(list) self._buildings: List[Building] = [] def _cell_key(self, lat: float, lon: float) -> Tuple[int, int]: """Convert lat/lon to grid cell key""" return (int(lat / self.cell_size), int(lon / self.cell_size)) def build(self, buildings: List[Building]): """Build spatial index from buildings list""" self._grid.clear() self._buildings = buildings for building in buildings: # Get bounding box of building lons = [p[0] for p in building.geometry] lats = [p[1] for p in building.geometry] min_lon, max_lon = min(lons), max(lons) min_lat, max_lat = min(lats), max(lats) # Insert into all overlapping grid cells min_cell_lat = int(min_lat / self.cell_size) max_cell_lat = int(max_lat / self.cell_size) min_cell_lon = int(min_lon / self.cell_size) max_cell_lon = int(max_lon / self.cell_size) for clat in range(min_cell_lat, max_cell_lat + 1): for clon in range(min_cell_lon, max_cell_lon + 1): self._grid[(clat, clon)].append(building) def query_point(self, lat: float, lon: float, buffer_cells: int = 1) -> List[Building]: """Find buildings near a point""" if not self._grid: return self._buildings # Fallback to linear scan center = self._cell_key(lat, lon) results = set() for dlat in range(-buffer_cells, buffer_cells + 1): for dlon in range(-buffer_cells, buffer_cells + 1): key = (center[0] + dlat, center[1] + dlon) for b in self._grid.get(key, []): results.add(b.id) # Return buildings by id (deduped) id_set = results return [b for b in self._buildings if b.id in id_set] def query_line( self, lat1: float, lon1: float, lat2: float, lon2: float, buffer_cells: int = 1 ) -> List[Building]: """Find buildings along a line (for LoS checks)""" if not self._grid: return self._buildings # Get bounding box cells of the line min_lat = min(lat1, lat2) max_lat = max(lat1, lat2) min_lon = min(lon1, lon2) max_lon = max(lon1, lon2) min_clat = int(min_lat / self.cell_size) - buffer_cells max_clat = int(max_lat / self.cell_size) + buffer_cells min_clon = int(min_lon / self.cell_size) - buffer_cells max_clon = int(max_lon / self.cell_size) + buffer_cells results = set() for clat in range(min_clat, max_clat + 1): for clon in range(min_clon, max_clon + 1): for b in self._grid.get((clat, clon), []): results.add(b.id) id_set = results return [b for b in self._buildings if b.id in id_set] def query_bbox( self, min_lat: float, min_lon: float, max_lat: float, max_lon: float ) -> List[Building]: """Find all buildings in bounding box""" if not self._grid: return self._buildings min_clat = int(min_lat / self.cell_size) max_clat = int(max_lat / self.cell_size) min_clon = int(min_lon / self.cell_size) max_clon = int(max_lon / self.cell_size) results = set() for clat in range(min_clat, max_clat + 1): for clon in range(min_clon, max_clon + 1): for b in self._grid.get((clat, clon), []): results.add(b.id) id_set = results return [b for b in self._buildings if b.id in id_set] # Global cache of spatial indices _spatial_indices: dict[str, SpatialIndex] = {} def get_spatial_index(cache_key: str, buildings: List[Building]) -> SpatialIndex: """Get or create spatial index for buildings""" if cache_key not in _spatial_indices: idx = SpatialIndex() idx.build(buildings) _spatial_indices[cache_key] = idx # Limit cache size if len(_spatial_indices) > 20: oldest = next(iter(_spatial_indices)) del _spatial_indices[oldest] return _spatial_indices[cache_key]