25 KiB
25 KiB
RFCP Iteration 1.6: Enhanced Environment + Bugfixes
Date: January 31, 2025
Type: Backend Enhancement + Bugfixes
Estimated: 8-12 hours
Location: /opt/rfcp/backend/
🎯 Goal
Fix critical bugs from 1.5, add spatial indexing for performance, implement water/ground reflections and vegetation loss.
🐛 Bugfixes (Priority)
Bug 1: OSM building:levels Parsing Error
Error:
ValueError: invalid literal for int() with base 10: '1а'
Cause: OSM data contains non-numeric levels like "1а", "2-3", "5+", etc.
Location: app/services/buildings_service.py line ~160
Fix:
# Add to BuildingsService class
def _safe_int(self, value) -> Optional[int]:
"""Safely parse int from OSM tag (handles '1а', '2-3', '5+', etc.)"""
if not value:
return None
try:
return int(value)
except (ValueError, TypeError):
import re
match = re.search(r'\d+', str(value))
if match:
return int(match.group())
return None
# Replace line ~160
# OLD:
levels=int(tags.get("building:levels", 0)) or None,
# NEW:
levels=self._safe_int(tags.get("building:levels")),
Bug 2: Height Parsing Error
Similar issue with height tag:
# OSM can have: "10 m", "10m", "10.5", "~12"
def _safe_float(self, value) -> Optional[float]:
"""Safely parse float from OSM tag"""
if not value:
return None
try:
# Remove common suffixes
cleaned = str(value).lower().replace('m', '').replace('~', '').strip()
return float(cleaned)
except (ValueError, TypeError):
import re
match = re.search(r'[\d.]+', str(value))
if match:
return float(match.group())
return None
Bug 3: Request Timeout / Queue Blocking
Issue: Long calculations block other requests
Fix: Add timeout to coverage calculation
# app/api/routes/coverage.py
import asyncio
@router.post("/calculate")
async def calculate_coverage(request: CoverageRequest) -> CoverageResponse:
try:
# Add timeout (5 minutes max)
result = await asyncio.wait_for(
coverage_service.calculate_coverage(...),
timeout=300.0
)
return result
except asyncio.TimeoutError:
raise HTTPException(408, "Calculation timeout - try smaller radius or lower resolution")
✅ Enhancement Tasks
Task 1: Spatial Indexing with R-tree (3-4 hours)
Install:
pip install Rtree
app/services/spatial_index.py:
from rtree import index
from typing import List, Tuple, Optional
from app.services.buildings_service import Building
class SpatialIndex:
"""R-tree spatial index for fast building lookups"""
def __init__(self):
self._index: Optional[index.Index] = None
self._buildings: dict[int, Building] = {}
self._bounds: Optional[Tuple[float, float, float, float]] = None
def build(self, buildings: List[Building]):
"""Build spatial index from buildings list"""
self._index = index.Index()
self._buildings = {}
for i, building in enumerate(buildings):
# Get bounding box of building
lons = [p[0] for p in building.geometry]
lats = [p[1] for p in building.geometry]
bbox = (min(lons), min(lats), max(lons), max(lats))
self._index.insert(i, bbox)
self._buildings[i] = building
if buildings:
all_lons = [p[0] for b in buildings for p in b.geometry]
all_lats = [p[1] for b in buildings for p in b.geometry]
self._bounds = (min(all_lons), min(all_lats), max(all_lons), max(all_lats))
def query_point(self, lat: float, lon: float, buffer: float = 0.001) -> List[Building]:
"""Find buildings near a point"""
if not self._index:
return []
bbox = (lon - buffer, lat - buffer, lon + buffer, lat + buffer)
indices = list(self._index.intersection(bbox))
return [self._buildings[i] for i in indices]
def query_line(
self,
lat1: float, lon1: float,
lat2: float, lon2: float,
buffer: float = 0.001
) -> List[Building]:
"""Find buildings along a line (for LoS checks)"""
if not self._index:
return []
# Bounding box of line + buffer
min_lon = min(lon1, lon2) - buffer
max_lon = max(lon1, lon2) + buffer
min_lat = min(lat1, lat2) - buffer
max_lat = max(lat1, lat2) + buffer
bbox = (min_lon, min_lat, max_lon, max_lat)
indices = list(self._index.intersection(bbox))
return [self._buildings[i] for i in indices]
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._index:
return []
bbox = (min_lon, min_lat, max_lon, max_lat)
indices = list(self._index.intersection(bbox))
return [self._buildings[i] for i in indices]
# Global instance with caching
_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]
Update coverage_service.py:
from app.services.spatial_index import get_spatial_index
# In calculate_coverage():
if settings.use_buildings and buildings:
# Build spatial index once
cache_key = f"{min_lat:.3f},{min_lon:.3f},{max_lat:.3f},{max_lon:.3f}"
spatial_idx = get_spatial_index(cache_key, buildings)
# In _calculate_point():
# OLD: for building in buildings:
# NEW:
nearby_buildings = spatial_idx.query_point(lat, lon)
for building in nearby_buildings:
Task 2: Water Reflection (2-3 hours)
app/services/water_service.py:
import httpx
from typing import List, Tuple, Optional
from pydantic import BaseModel
import json
from pathlib import Path
class WaterBody(BaseModel):
"""Water body from OSM"""
id: int
geometry: List[Tuple[float, float]] # [(lon, lat), ...]
water_type: str # river, lake, pond, reservoir
name: Optional[str] = None
class WaterService:
"""OSM water bodies for reflection calculations"""
OVERPASS_URL = "https://overpass-api.de/api/interpreter"
# Reflection coefficients by water type
REFLECTION_COEFF = {
"lake": 0.8,
"reservoir": 0.8,
"river": 0.7,
"pond": 0.75,
"water": 0.7, # generic
}
def __init__(self, cache_dir: str = "/opt/rfcp/backend/data/water"):
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(exist_ok=True, parents=True)
self._cache: dict[str, List[WaterBody]] = {}
async def fetch_water_bodies(
self,
min_lat: float, min_lon: float,
max_lat: float, max_lon: float
) -> List[WaterBody]:
"""Fetch water bodies in bounding box"""
cache_key = f"{min_lat:.2f}_{min_lon:.2f}_{max_lat:.2f}_{max_lon:.2f}"
if cache_key in self._cache:
return self._cache[cache_key]
cache_file = self.cache_dir / f"{cache_key}.json"
if cache_file.exists():
with open(cache_file) as f:
data = json.load(f)
bodies = [WaterBody(**w) for w in data]
self._cache[cache_key] = bodies
return bodies
query = f"""
[out:json][timeout:30];
(
way["natural"="water"]({min_lat},{min_lon},{max_lat},{max_lon});
relation["natural"="water"]({min_lat},{min_lon},{max_lat},{max_lon});
way["waterway"]({min_lat},{min_lon},{max_lat},{max_lon});
);
out body;
>;
out skel qt;
"""
try:
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(self.OVERPASS_URL, data={"data": query})
response.raise_for_status()
data = response.json()
except Exception as e:
print(f"Water fetch error: {e}")
return []
bodies = self._parse_response(data)
# Cache
with open(cache_file, 'w') as f:
json.dump([w.model_dump() for w in bodies], f)
self._cache[cache_key] = bodies
return bodies
def _parse_response(self, data: dict) -> List[WaterBody]:
"""Parse Overpass response"""
nodes = {}
for element in data.get("elements", []):
if element["type"] == "node":
nodes[element["id"]] = (element["lon"], element["lat"])
bodies = []
for element in data.get("elements", []):
if element["type"] != "way":
continue
tags = element.get("tags", {})
# Determine water type
water_type = tags.get("water", tags.get("waterway", tags.get("natural", "water")))
geometry = []
for node_id in element.get("nodes", []):
if node_id in nodes:
geometry.append(nodes[node_id])
if len(geometry) < 3:
continue
bodies.append(WaterBody(
id=element["id"],
geometry=geometry,
water_type=water_type,
name=tags.get("name")
))
return bodies
def get_reflection_coefficient(self, water_type: str) -> float:
"""Get reflection coefficient for water type"""
return self.REFLECTION_COEFF.get(water_type, 0.7)
def point_over_water(self, lat: float, lon: float, water_bodies: List[WaterBody]) -> Optional[WaterBody]:
"""Check if point is over water"""
for body in water_bodies:
if self._point_in_polygon(lat, lon, body.geometry):
return body
return None
def _point_in_polygon(self, lat: float, lon: float, polygon: List[Tuple[float, float]]) -> bool:
"""Ray casting algorithm"""
n = len(polygon)
inside = False
j = n - 1
for i in range(n):
xi, yi = polygon[i]
xj, yj = polygon[j]
if ((yi > lat) != (yj > lat)) and (lon < (xj - xi) * (lat - yi) / (yj - yi) + xi):
inside = not inside
j = i
return inside
water_service = WaterService()
Task 3: Vegetation Loss (2-3 hours)
app/services/vegetation_service.py:
import httpx
from typing import List, Tuple, Optional
from pydantic import BaseModel
import json
from pathlib import Path
class VegetationArea(BaseModel):
"""Vegetation area from OSM"""
id: int
geometry: List[Tuple[float, float]]
vegetation_type: str # forest, wood, scrub, orchard
density: str # dense, sparse, mixed
class VegetationService:
"""OSM vegetation for signal attenuation"""
OVERPASS_URL = "https://overpass-api.de/api/interpreter"
# Attenuation dB per 100 meters
ATTENUATION_DB_PER_100M = {
"forest": 8.0, # Dense forest
"wood": 6.0, # Woods
"tree_row": 2.0, # Single row of trees
"scrub": 3.0, # Bushes
"orchard": 2.0, # Fruit trees (spaced)
"vineyard": 1.0, # Low vegetation
"meadow": 0.5, # Grass only
}
# Seasonal factor (summer = full foliage)
SEASONAL_FACTOR = {
"summer": 1.0,
"winter": 0.3, # Deciduous trees lose leaves
"spring": 0.6,
"autumn": 0.7,
}
def __init__(self, cache_dir: str = "/opt/rfcp/backend/data/vegetation"):
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(exist_ok=True, parents=True)
self._cache: dict[str, List[VegetationArea]] = {}
async def fetch_vegetation(
self,
min_lat: float, min_lon: float,
max_lat: float, max_lon: float
) -> List[VegetationArea]:
"""Fetch vegetation areas in bounding box"""
cache_key = f"{min_lat:.2f}_{min_lon:.2f}_{max_lat:.2f}_{max_lon:.2f}"
if cache_key in self._cache:
return self._cache[cache_key]
cache_file = self.cache_dir / f"{cache_key}.json"
if cache_file.exists():
with open(cache_file) as f:
data = json.load(f)
areas = [VegetationArea(**v) for v in data]
self._cache[cache_key] = areas
return areas
query = f"""
[out:json][timeout:30];
(
way["landuse"="forest"]({min_lat},{min_lon},{max_lat},{max_lon});
way["natural"="wood"]({min_lat},{min_lon},{max_lat},{max_lon});
way["landuse"="orchard"]({min_lat},{min_lon},{max_lat},{max_lon});
way["natural"="scrub"]({min_lat},{min_lon},{max_lat},{max_lon});
);
out body;
>;
out skel qt;
"""
try:
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(self.OVERPASS_URL, data={"data": query})
response.raise_for_status()
data = response.json()
except Exception as e:
print(f"Vegetation fetch error: {e}")
return []
areas = self._parse_response(data)
# Cache
with open(cache_file, 'w') as f:
json.dump([v.model_dump() for v in areas], f)
self._cache[cache_key] = areas
return areas
def _parse_response(self, data: dict) -> List[VegetationArea]:
"""Parse Overpass response"""
nodes = {}
for element in data.get("elements", []):
if element["type"] == "node":
nodes[element["id"]] = (element["lon"], element["lat"])
areas = []
for element in data.get("elements", []):
if element["type"] != "way":
continue
tags = element.get("tags", {})
veg_type = tags.get("landuse", tags.get("natural", "forest"))
geometry = []
for node_id in element.get("nodes", []):
if node_id in nodes:
geometry.append(nodes[node_id])
if len(geometry) < 3:
continue
# Determine density
leaf_type = tags.get("leaf_type", "mixed")
density = "dense" if leaf_type == "needleleaved" else "mixed"
areas.append(VegetationArea(
id=element["id"],
geometry=geometry,
vegetation_type=veg_type,
density=density
))
return areas
def calculate_vegetation_loss(
self,
lat1: float, lon1: float,
lat2: float, lon2: float,
vegetation_areas: List[VegetationArea],
season: str = "summer"
) -> float:
"""
Calculate signal loss through vegetation along path
Returns loss in dB
"""
from app.services.terrain_service import TerrainService
total_loss = 0.0
path_length = TerrainService.haversine_distance(lat1, lon1, lat2, lon2)
if path_length < 1:
return 0.0
# Sample points along path
num_samples = max(10, int(path_length / 50)) # Every 50m
vegetation_distance = 0.0
current_veg_type = None
for i in range(num_samples):
t = i / num_samples
lat = lat1 + t * (lat2 - lat1)
lon = lon1 + t * (lon2 - lon1)
# Check if point is in vegetation
veg = self._point_in_vegetation(lat, lon, vegetation_areas)
if veg:
segment_length = path_length / num_samples
vegetation_distance += segment_length
current_veg_type = veg.vegetation_type
if vegetation_distance > 0 and current_veg_type:
# Calculate loss
attenuation = self.ATTENUATION_DB_PER_100M.get(current_veg_type, 4.0)
seasonal = self.SEASONAL_FACTOR.get(season, 1.0)
total_loss = (vegetation_distance / 100) * attenuation * seasonal
return min(total_loss, 40.0) # Cap at 40 dB
def _point_in_vegetation(
self,
lat: float, lon: float,
areas: List[VegetationArea]
) -> Optional[VegetationArea]:
"""Check if point is in vegetation area"""
for area in areas:
if self._point_in_polygon(lat, lon, area.geometry):
return area
return None
def _point_in_polygon(self, lat: float, lon: float, polygon: List[Tuple[float, float]]) -> bool:
"""Ray casting algorithm"""
n = len(polygon)
inside = False
j = n - 1
for i in range(n):
xi, yi = polygon[i]
xj, yj = polygon[j]
if ((yi > lat) != (yj > lat)) and (lon < (xj - xi) * (lat - yi) / (yj - yi) + xi):
inside = not inside
j = i
return inside
vegetation_service = VegetationService()
Task 4: Ground Reflection (1-2 hours)
Update app/services/reflection_service.py:
# Add to ReflectionService class
# Ground types and reflection coefficients
GROUND_REFLECTION = {
"urban": 0.3, # Asphalt, concrete
"suburban": 0.4, # Mixed
"rural": 0.5, # Grass, soil
"water": 0.8, # Lakes, rivers
"desert": 0.6, # Sand
}
def calculate_ground_reflection(
self,
tx_lat: float, tx_lon: float, tx_height: float,
rx_lat: float, rx_lon: float, rx_height: float,
frequency_mhz: float,
ground_type: str = "rural",
water_body: Optional[WaterBody] = None
) -> ReflectionPath:
"""Calculate ground/water reflection path"""
from app.services.terrain_service import TerrainService
# Direct distance
direct_dist = TerrainService.haversine_distance(tx_lat, tx_lon, rx_lat, rx_lon)
# Reflection point (simplified - midpoint)
mid_lat = (tx_lat + rx_lat) / 2
mid_lon = (tx_lon + rx_lon) / 2
# Get ground elevation at reflection point
ground_elev = await self.terrain.get_elevation(mid_lat, mid_lon)
# Path lengths (TX -> ground -> RX)
d1 = direct_dist / 2
h1 = tx_height
h2 = rx_height
# Actual path length
path1 = np.sqrt(d1**2 + h1**2)
path2 = np.sqrt(d1**2 + h2**2)
total_path = path1 + path2
# Reflection coefficient
if water_body:
coeff = self.GROUND_REFLECTION["water"]
else:
coeff = self.GROUND_REFLECTION.get(ground_type, 0.4)
# Path loss
path_loss = self._free_space_loss(total_path, frequency_mhz)
reflection_loss = -10 * np.log10(coeff)
total_loss = path_loss + reflection_loss
return ReflectionPath(
points=[(tx_lat, tx_lon), (mid_lat, mid_lon), (rx_lat, rx_lon)],
total_distance=total_path,
total_loss=total_loss,
reflection_count=1,
materials=["ground" if not water_body else "water"]
)
Task 5: Integrate into Coverage Service (2 hours)
Update app/services/coverage_service.py:
# Add imports
from app.services.spatial_index import get_spatial_index
from app.services.water_service import water_service
from app.services.vegetation_service import vegetation_service
# Update CoverageSettings
class CoverageSettings(BaseModel):
# ... existing fields ...
use_water_reflection: bool = False
use_vegetation: bool = False
season: str = "summer" # For vegetation loss
# Update calculate_coverage()
async def calculate_coverage(self, site: SiteParams, settings: CoverageSettings):
# ... existing setup ...
# Fetch additional data if enabled
water_bodies = []
vegetation_areas = []
if settings.use_water_reflection:
water_bodies = await water_service.fetch_water_bodies(
min_lat, min_lon, max_lat, max_lon
)
if settings.use_vegetation:
vegetation_areas = await vegetation_service.fetch_vegetation(
min_lat, min_lon, max_lat, max_lon
)
# Build spatial index for buildings
spatial_idx = None
if settings.use_buildings and buildings:
cache_key = f"{min_lat:.3f},{min_lon:.3f},{max_lat:.3f},{max_lon:.3f}"
spatial_idx = get_spatial_index(cache_key, buildings)
# Calculate points...
# Update _calculate_point()
async def _calculate_point(self, ...):
# ... existing code ...
# Vegetation loss
vegetation_loss = 0.0
if settings.use_vegetation and vegetation_areas:
vegetation_loss = vegetation_service.calculate_vegetation_loss(
site.lat, site.lon, lat, lon,
vegetation_areas, settings.season
)
# Water reflection boost
water_reflection_gain = 0.0
if settings.use_water_reflection and water_bodies:
water = water_service.point_over_water(lat, lon, water_bodies)
if water:
# Better reflection over water
water_reflection_gain = 3.0 # ~3dB boost
# Final RSRP
rsrp = (site.power + site.gain - path_loss - antenna_loss
- terrain_loss - building_loss - vegetation_loss
+ reflection_gain + water_reflection_gain)
Task 6: Update API & Frontend Settings (1 hour)
Update presets in coverage.py:
PRESETS = {
"fast": {
"use_terrain": True,
"use_buildings": False,
"use_materials": False,
"use_dominant_path": False,
"use_street_canyon": False,
"use_reflections": False,
"use_water_reflection": False,
"use_vegetation": False,
},
# ... standard, detailed ...
"full": {
"use_terrain": True,
"use_buildings": True,
"use_materials": True,
"use_dominant_path": True,
"use_street_canyon": True,
"use_reflections": True,
"use_water_reflection": True,
"use_vegetation": True,
},
}
📁 Files Summary
New Files:
app/services/
├── spatial_index.py # R-tree for fast building lookup
├── water_service.py # OSM water bodies
└── vegetation_service.py # OSM forest/vegetation
data/
├── water/ # Water bodies cache
└── vegetation/ # Vegetation cache
Modified Files:
app/services/
├── buildings_service.py # Fix parsing bugs
├── coverage_service.py # Integrate new services
└── reflection_service.py # Add ground reflection
app/api/routes/
└── coverage.py # Timeout, new settings
requirements.txt # Add Rtree
🧪 Testing
# Install new dependency
cd /opt/rfcp/backend
source venv/bin/activate
pip install Rtree
# Restart
sudo systemctl restart rfcp-backend
# Test parsing fix
curl -X POST "https://api.rfcp.eliah.one/api/coverage/calculate" \
-H "Content-Type: application/json" \
-d '{"sites":[{"lat":48.46,"lon":35.05,"height":30,"power":43,"gain":15,"frequency":1800}],"settings":{"radius":1000,"resolution":100,"preset":"standard"}}'
# Test full preset with new features
curl -X POST "https://api.rfcp.eliah.one/api/coverage/calculate" \
-H "Content-Type: application/json" \
-d '{"sites":[{"lat":48.46,"lon":35.05,"height":30,"power":43,"gain":15,"frequency":1800}],"settings":{"radius":500,"resolution":50,"preset":"full"}}'
# Run test scripts
./rfcp-integration-test.sh
./rfcp-propagation-test.sh
✅ Success Criteria
building:levels = "1а"no longer crashes- R-tree spatial index speeds up building queries
- Water bodies fetched and cached
- Vegetation areas fetched and cached
- Coverage includes
vegetation_lossfield - Full preset uses all new features
- No timeout on reasonable requests (5km, 100m resolution)
- All tests pass (21/21 integration, 8/8 propagation)
📝 Notes
- R-tree requires
libspatialindexsystem library - Install on Ubuntu:
apt install libspatialindex-dev - Vegetation loss is seasonal — default to summer
- Water reflection most noticeable near large lakes/rivers
🔜 Next: 1.6.1 or 2.1
1.6.1 — Extra Factors (optional):
- Weather/rain attenuation
- Indoor penetration layer
- Time-of-day atmospheric effects
2.1 — Desktop Installer:
- Electron packaging
- Offline mode
- GPU acceleration
Ready for Claude Code 🚀