mytec/rfcp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

@mytec: iter2.2 ready for testing

Browse Source
This commit is contained in:
mytec
2026-01-31 16:16:15 +02:00
parent baf57ad77f
commit f6a39df366
9 changed files with 901 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

113
backend/app/services/vegetation_service.py
View File

@@ -5,11 +5,13 @@ Forests and dense vegetation attenuate RF signals significantly.
Uses ITU-R P.833 approximations for foliage loss.
"""
import os
import httpx
import json
from typing import List, Tuple, Optional
from pydantic import BaseModel
import json
from pathlib import Path
from datetime import datetime, timedelta
class VegetationArea(BaseModel):
@@ -20,6 +22,62 @@ class VegetationArea(BaseModel):
density: str # dense, sparse, mixed
class VegetationCache:
"""Local cache for vegetation data with expiry"""
CACHE_EXPIRY_DAYS = 30
def __init__(self):
self.data_path = Path(os.environ.get('RFCP_DATA_PATH', './data'))
self.cache_path = self.data_path / 'osm' / 'vegetation'
self.cache_path.mkdir(parents=True, exist_ok=True)
def _get_cache_key(self, min_lat: float, min_lon: float, max_lat: float, max_lon: float) -> str:
return f"{min_lat:.2f}_{min_lon:.2f}_{max_lat:.2f}_{max_lon:.2f}"
def _get_cache_file(self, cache_key: str) -> Path:
return self.cache_path / f"{cache_key}.json"
def get(self, min_lat: float, min_lon: float, max_lat: float, max_lon: float) -> Optional[list]:
cache_key = self._get_cache_key(min_lat, min_lon, max_lat, max_lon)
cache_file = self._get_cache_file(cache_key)
if not cache_file.exists():
return None
try:
data = json.loads(cache_file.read_text())
cached_at = datetime.fromisoformat(data.get('_cached_at', '2000-01-01'))
if datetime.now() - cached_at > timedelta(days=self.CACHE_EXPIRY_DAYS):
return None
return data.get('data')
except Exception as e:
print(f"[VegetationCache] Failed to read cache: {e}")
return None
def set(self, min_lat: float, min_lon: float, max_lat: float, max_lon: float, data):
cache_key = self._get_cache_key(min_lat, min_lon, max_lat, max_lon)
cache_file = self._get_cache_file(cache_key)
try:
cache_data = {
'_cached_at': datetime.now().isoformat(),
'_bbox': [min_lat, min_lon, max_lat, max_lon],
'data': data
}
cache_file.write_text(json.dumps(cache_data))
except Exception as e:
print(f"[VegetationCache] Failed to write cache: {e}")
def clear(self):
for f in self.cache_path.glob("*.json"):
f.unlink()
def get_size_mb(self) -> float:
total = sum(f.stat().st_size for f in self.cache_path.glob("*.json"))
return total / (1024 * 1024)
class VegetationService:
"""OSM vegetation for signal attenuation"""
@@ -44,33 +102,33 @@ class VegetationService:
"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]] = {}
def __init__(self):
self.cache = VegetationCache()
self._memory_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"""
"""Fetch vegetation areas in bounding box, using cache if available"""
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]
# Memory cache
if cache_key in self._memory_cache:
return self._memory_cache[cache_key]
cache_file = self.cache_dir / f"{cache_key}.json"
if cache_file.exists():
try:
with open(cache_file) as f:
data = json.load(f)
areas = [VegetationArea(**v) for v in data]
self._cache[cache_key] = areas
return areas
except Exception:
pass
# Disk cache with expiry
cached = self.cache.get(min_lat, min_lon, max_lat, max_lon)
if cached is not None:
print(f"[Vegetation] Cache hit for bbox")
areas = [VegetationArea(**v) for v in cached]
self._memory_cache[cache_key] = areas
return areas
# Fetch from Overpass
print(f"[Vegetation] Fetching from Overpass API...")
query = f"""
[out:json][timeout:30];
@@ -91,17 +149,17 @@ class VegetationService:
response.raise_for_status()
data = response.json()
except Exception as e:
print(f"Vegetation fetch error: {e}")
print(f"[Vegetation] Fetch error: {e}")
return []
areas = self._parse_response(data)
# Cache
# Save to disk cache
if areas:
with open(cache_file, 'w') as f:
json.dump([v.model_dump() for v in areas], f)
self._cache[cache_key] = areas
self.cache.set(min_lat, min_lon, max_lat, max_lon,
[v.model_dump() for v in areas])
self._memory_cache[cache_key] = areas
return areas
def _parse_response(self, data: dict) -> List[VegetationArea]:
@@ -128,7 +186,6 @@ class VegetationService:
if len(geometry) < 3:
continue
# Determine density from leaf_type tag
leaf_type = tags.get("leaf_type", "mixed")
density = "dense" if leaf_type == "needleleaved" else "mixed"
@@ -151,7 +208,7 @@ class VegetationService:
"""
Calculate signal loss through vegetation along path.
Samples points along the TX→RX path and accumulates
Samples points along the TX->RX path and accumulates
attenuation for each segment inside vegetation.
Returns loss in dB (capped at 40 dB).
@@ -163,7 +220,6 @@ class VegetationService:
if path_length < 1:
return 0.0
# Sample points along path — every ~50m
num_samples = max(10, int(path_length / 50))
segment_length = path_length / num_samples
@@ -174,7 +230,6 @@ class VegetationService:
lat = lat1 + t * (lat2 - lat1)
lon = lon1 + t * (lon2 - lon1)
# Check if sample point is inside any vegetation area
veg = self._point_in_vegetation(lat, lon, vegetation_areas)
if veg:
@@ -182,7 +237,7 @@ class VegetationService:
seasonal = self.SEASONAL_FACTOR.get(season, 1.0)
total_loss += (segment_length / 100) * attenuation * seasonal
return min(total_loss, 40.0) # Cap at 40 dB
return min(total_loss, 40.0)
def _point_in_vegetation(
self,
@@ -199,7 +254,7 @@ class VegetationService:
def _point_in_polygon(
lat: float, lon: float, polygon: List[Tuple[float, float]]
) -> bool:
"""Ray casting algorithm — polygon coords are (lon, lat)"""
"""Ray casting algorithm -- polygon coords are (lon, lat)"""
n = len(polygon)
inside = False