@mytec: feat: Phase 3.0 Architecture Refactor

Major refactoring of RFCP backend:
- Modular propagation models (8 models)
- SharedMemoryManager for terrain data
- ProcessPoolExecutor parallel processing
- WebSocket progress streaming
- Building filtering pipeline (351k → 15k)
- 82 unit tests

Performance: Standard preset 38s → 5s (7.6x speedup)

Known issue: Detailed preset timeout (fix in 3.1.0)
This commit is contained in:
2026-02-01 23:12:26 +02:00
parent 1dde56705a
commit defa3ad440
71 changed files with 7134 additions and 256 deletions

View File

View File

View File

@@ -0,0 +1,60 @@
"""
Unit tests for knife-edge diffraction calculations.
"""
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..'))
from app.geometry.diffraction import knife_edge_loss
def freq_to_wl(freq_mhz):
return 300.0 / freq_mhz
class TestKnifeEdgeLoss:
def test_no_obstruction_low_loss(self):
"""Negative h means clearance above LOS — loss should be small."""
loss = knife_edge_loss(d1_m=500, d2_m=500, h_m=-10, wavelength_m=freq_to_wl(1800))
assert loss >= 0
assert loss < 3
def test_grazing_obstruction(self):
"""h=0 means exactly at LOS line — ~6 dB loss."""
loss = knife_edge_loss(d1_m=500, d2_m=500, h_m=0, wavelength_m=freq_to_wl(1800))
assert 5 < loss < 8
def test_obstruction_increases_loss(self):
wl = freq_to_wl(1800)
loss_low = knife_edge_loss(d1_m=500, d2_m=500, h_m=1, wavelength_m=wl)
loss_high = knife_edge_loss(d1_m=500, d2_m=500, h_m=10, wavelength_m=wl)
assert loss_high > loss_low
def test_higher_freq_more_loss(self):
"""Higher frequency = shorter wavelength = more diffraction loss."""
loss_low_f = knife_edge_loss(d1_m=500, d2_m=500, h_m=5, wavelength_m=freq_to_wl(450))
loss_high_f = knife_edge_loss(d1_m=500, d2_m=500, h_m=5, wavelength_m=freq_to_wl(1800))
assert loss_high_f > loss_low_f
def test_zero_distance_safe(self):
"""Should not crash on zero distances."""
loss = knife_edge_loss(d1_m=0, d2_m=500, h_m=5, wavelength_m=freq_to_wl(900))
assert loss >= 0
def test_large_clearance(self):
"""Very deep clearance (large negative h) should have near-zero loss."""
loss = knife_edge_loss(d1_m=500, d2_m=500, h_m=-50, wavelength_m=freq_to_wl(900))
assert loss < 1.0
if __name__ == "__main__":
instance = TestKnifeEdgeLoss()
for method_name in [m for m in dir(instance) if m.startswith("test_")]:
try:
getattr(instance, method_name)()
print(f" PASS {method_name}")
except Exception as e:
print(f" FAIL {method_name}: {e}")
print("\nAll tests completed.")

View File

@@ -0,0 +1,74 @@
"""
Unit tests for haversine distance calculations.
"""
import sys
import os
import numpy as np
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..'))
from app.geometry.haversine import haversine_distance, haversine_batch, points_to_local_coords
class TestHaversineDistance:
def test_same_point_is_zero(self):
d = haversine_distance(50.45, 30.52, 50.45, 30.52)
assert abs(d) < 1.0
def test_known_distance(self):
# Kyiv to Kharkiv ≈ 410 km
d = haversine_distance(50.45, 30.52, 49.99, 36.23)
assert 400000 < d < 420000
def test_short_distance(self):
# ~111m for 0.001 degree lat
d = haversine_distance(50.0, 30.0, 50.001, 30.0)
assert 100 < d < 120
class TestHaversineBatch:
def test_single_point(self):
lats = np.array([50.001])
lons = np.array([30.0])
distances = haversine_batch(50.0, 30.0, lats, lons)
assert len(distances) == 1
assert 100 < distances[0] < 120
def test_multiple_points(self):
lats = np.array([50.001, 50.01, 50.1])
lons = np.array([30.0, 30.0, 30.0])
distances = haversine_batch(50.0, 30.0, lats, lons)
assert len(distances) == 3
# Should be monotonically increasing
assert distances[0] < distances[1] < distances[2]
class TestLocalCoords:
def test_same_point_is_origin(self):
x, y = points_to_local_coords(50.0, 30.0, np.array([50.0]), np.array([30.0]))
assert abs(x[0]) < 1.0
assert abs(y[0]) < 1.0
def test_north_is_positive_y(self):
x, y = points_to_local_coords(50.0, 30.0, np.array([50.001]), np.array([30.0]))
assert y[0] > 0
assert abs(x[0]) < 1.0
def test_east_is_positive_x(self):
x, y = points_to_local_coords(50.0, 30.0, np.array([50.0]), np.array([30.001]))
assert x[0] > 0
assert abs(y[0]) < 1.0
if __name__ == "__main__":
for cls in [TestHaversineDistance, TestHaversineBatch, TestLocalCoords]:
instance = cls()
for method_name in [m for m in dir(instance) if m.startswith("test_")]:
try:
getattr(instance, method_name)()
print(f" PASS {cls.__name__}.{method_name}")
except Exception as e:
print(f" FAIL {cls.__name__}.{method_name}: {e}")
print("\nAll tests completed.")

View File

@@ -0,0 +1,77 @@
"""
Unit tests for line-segment intersection calculations.
These require NumPy, so use __main__ block with conditional import.
"""
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..'))
import numpy as np
from app.geometry.intersection import line_segments_intersect_batch
class TestLineSegmentIntersect:
def test_crossing_lines(self):
"""Two crossing segments should intersect."""
# Line from (0,0)→(1,1) and (0,1)→(1,0)
result = line_segments_intersect_batch(
p1=np.array([0.0, 0.0]),
p2=np.array([1.0, 1.0]),
seg_starts=np.array([[0.0, 1.0]]),
seg_ends=np.array([[1.0, 0.0]]),
)
assert result[0] == True
def test_parallel_lines(self):
"""Parallel lines should not intersect."""
result = line_segments_intersect_batch(
p1=np.array([0.0, 0.0]),
p2=np.array([1.0, 0.0]),
seg_starts=np.array([[0.0, 1.0]]),
seg_ends=np.array([[1.0, 1.0]]),
)
assert result[0] == False
def test_non_crossing(self):
"""Segments that don't reach each other."""
result = line_segments_intersect_batch(
p1=np.array([0.0, 0.0]),
p2=np.array([0.5, 0.5]),
seg_starts=np.array([[0.8, 0.0]]),
seg_ends=np.array([[0.8, 1.0]]),
)
assert result[0] == False
def test_multiple_segments(self):
"""Batch test with multiple segments."""
result = line_segments_intersect_batch(
p1=np.array([0.0, 0.0]),
p2=np.array([1.0, 1.0]),
seg_starts=np.array([
[0.0, 1.0], # crosses
[2.0, 0.0], # doesn't cross
[0.5, 0.0], # crosses
]),
seg_ends=np.array([
[1.0, 0.0], # crosses
[2.0, 1.0], # doesn't cross
[0.5, 1.0], # crosses
]),
)
assert result[0] == True
assert result[1] == False
assert result[2] == True
if __name__ == "__main__":
instance = TestLineSegmentIntersect()
for method_name in [m for m in dir(instance) if m.startswith("test_")]:
try:
getattr(instance, method_name)()
print(f" PASS {method_name}")
except Exception as e:
print(f" FAIL {method_name}: {e}")
print("\nAll tests completed.")

View File

@@ -0,0 +1,103 @@
"""
Unit tests for line-of-sight and Fresnel zone calculations.
"""
import sys
import os
import math
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..'))
from app.geometry.los import fresnel_radius, check_los_terrain
def freq_to_wavelength(freq_mhz):
return 300.0 / freq_mhz
class TestFresnelRadius:
def test_positive_result(self):
r = fresnel_radius(500, 500, freq_to_wavelength(1800))
assert r > 0
def test_symmetric(self):
wl = freq_to_wavelength(900)
r1 = fresnel_radius(300, 700, wl)
r2 = fresnel_radius(700, 300, wl)
assert abs(r1 - r2) < 0.001
def test_lower_freq_larger_radius(self):
r_high = fresnel_radius(500, 500, freq_to_wavelength(1800))
r_low = fresnel_radius(500, 500, freq_to_wavelength(900))
assert r_low > r_high
def test_center_is_maximum(self):
"""Fresnel radius is largest at the midpoint of the path."""
wl = freq_to_wavelength(900)
r_center = fresnel_radius(500, 500, wl)
r_offset = fresnel_radius(200, 800, wl)
assert r_center > r_offset
def test_known_value(self):
"""First Fresnel zone radius at midpoint of 1km path at 1GHz ~ 8.66m."""
# F1 = sqrt(lambda * d1 * d2 / (d1+d2))
# lambda = 0.3m at 1000MHz, d1=d2=500m
# F1 = sqrt(0.3 * 500 * 500 / 1000) = sqrt(75) ~ 8.66m
r = fresnel_radius(500, 500, freq_to_wavelength(1000))
assert 8.0 < r < 9.5
def test_zero_distance(self):
r = fresnel_radius(0, 500, freq_to_wavelength(900))
assert r == 0.0
class TestCheckLosTerrain:
def test_flat_terrain_has_los(self):
profile = [
{"elevation": 100, "distance": 0},
{"elevation": 100, "distance": 250},
{"elevation": 100, "distance": 500},
{"elevation": 100, "distance": 750},
{"elevation": 100, "distance": 1000},
]
result = check_los_terrain(profile, tx_height=30, rx_height=1.5)
assert result["has_los"] is True
assert result["clearance"] > 0
def test_hill_blocks_los(self):
profile = [
{"elevation": 100, "distance": 0},
{"elevation": 100, "distance": 250},
{"elevation": 200, "distance": 500}, # 100m hill
{"elevation": 100, "distance": 750},
{"elevation": 100, "distance": 1000},
]
result = check_los_terrain(profile, tx_height=10, rx_height=1.5)
assert result["has_los"] is False
assert result["blocked_at"] is not None
def test_empty_profile(self):
result = check_los_terrain([], tx_height=30, rx_height=1.5)
assert result["has_los"] is True
def test_high_antenna_clears_hill(self):
profile = [
{"elevation": 100, "distance": 0},
{"elevation": 110, "distance": 500},
{"elevation": 100, "distance": 1000},
]
# TX at 150m (100+50), RX at 101.5m. LOS at 500m ≈ 125.75m, terrain=110m → clear
result = check_los_terrain(profile, tx_height=50, rx_height=1.5)
assert result["has_los"] is True
if __name__ == "__main__":
for cls in [TestFresnelRadius, TestCheckLosTerrain]:
instance = cls()
for method_name in [m for m in dir(instance) if m.startswith("test_")]:
try:
getattr(instance, method_name)()
print(f" PASS {cls.__name__}.{method_name}")
except Exception as e:
print(f" FAIL {cls.__name__}.{method_name}: {e}")
print("\nAll tests completed.")

View File

@@ -0,0 +1 @@

View File

@@ -0,0 +1,127 @@
"""
Integration tests for the PointCalculator.
Verifies end-to-end point calculation with various
propagation models and environmental conditions.
"""
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..'))
from app.core.calculator import PointCalculator
from app.propagation.free_space import FreeSpaceModel
from app.propagation.okumura_hata import OkumuraHataModel
from app.propagation.cost231_hata import Cost231HataModel
class TestPointCalculatorFSPL:
def test_basic_calculation(self):
calc = PointCalculator(FreeSpaceModel())
result = calc.calculate_point(
site_lat=50.0, site_lon=30.0, site_height=30,
site_power=43, site_gain=18, site_frequency=1800,
point_lat=50.001, point_lon=30.0,
distance=111,
)
assert result.rsrp > -50 # Strong signal at short range
assert result.has_los is True
assert result.model_used == "Free-Space"
assert result.path_loss > 0
assert result.terrain_loss == 0
assert result.building_loss == 0
def test_signal_decreases_with_distance(self):
calc = PointCalculator(FreeSpaceModel())
near = calc.calculate_point(
site_lat=50.0, site_lon=30.0, site_height=30,
site_power=43, site_gain=18, site_frequency=1800,
point_lat=50.001, point_lon=30.0, distance=100,
)
far = calc.calculate_point(
site_lat=50.0, site_lon=30.0, site_height=30,
site_power=43, site_gain=18, site_frequency=1800,
point_lat=50.01, point_lon=30.0, distance=1000,
)
assert near.rsrp > far.rsrp
def test_terrain_obstruction(self):
calc = PointCalculator(FreeSpaceModel())
los = calc.calculate_point(
site_lat=50.0, site_lon=30.0, site_height=30,
site_power=43, site_gain=18, site_frequency=1800,
point_lat=50.01, point_lon=30.0, distance=1000,
)
nlos = calc.calculate_point(
site_lat=50.0, site_lon=30.0, site_height=30,
site_power=43, site_gain=18, site_frequency=1800,
point_lat=50.01, point_lon=30.0, distance=1000,
terrain_clearance=-10,
)
assert nlos.rsrp < los.rsrp
assert nlos.has_los is False
assert nlos.terrain_loss > 0
def test_building_loss_applied(self):
calc = PointCalculator(FreeSpaceModel())
no_building = calc.calculate_point(
site_lat=50.0, site_lon=30.0, site_height=30,
site_power=43, site_gain=18, site_frequency=1800,
point_lat=50.01, point_lon=30.0, distance=1000,
)
with_building = calc.calculate_point(
site_lat=50.0, site_lon=30.0, site_height=30,
site_power=43, site_gain=18, site_frequency=1800,
point_lat=50.01, point_lon=30.0, distance=1000,
building_loss=20,
)
assert abs(no_building.rsrp - with_building.rsrp - 20) < 0.1
class TestPointCalculatorAntenna:
def test_off_axis_reduces_signal(self):
calc = PointCalculator(FreeSpaceModel())
omni = calc.calculate_point(
site_lat=50.0, site_lon=30.0, site_height=30,
site_power=43, site_gain=18, site_frequency=1800,
point_lat=50.001, point_lon=30.0, distance=111,
)
directional = calc.calculate_point(
site_lat=50.0, site_lon=30.0, site_height=30,
site_power=43, site_gain=18, site_frequency=1800,
point_lat=50.001, point_lon=30.0, distance=111,
azimuth=90, beamwidth=65, # Pointing East, point is North
)
assert directional.rsrp < omni.rsrp
class TestPointCalculatorModelFallback:
def test_out_of_range_uses_fspl(self):
"""When Okumura-Hata is out of valid range, should fall back to FSPL."""
calc = PointCalculator(OkumuraHataModel())
# 50m distance is below Okumura-Hata minimum (1km)
result = calc.calculate_point(
site_lat=50.0, site_lon=30.0, site_height=30,
site_power=43, site_gain=18, site_frequency=900,
point_lat=50.0, point_lon=30.0001, distance=50,
)
# Should still return a valid result (via FSPL fallback)
assert result.rsrp != 0
assert result.path_loss > 0
if __name__ == "__main__":
for cls_name, cls in [
("FSPL", TestPointCalculatorFSPL),
("Antenna", TestPointCalculatorAntenna),
("Fallback", TestPointCalculatorModelFallback),
]:
instance = cls()
for method_name in [m for m in dir(instance) if m.startswith("test_")]:
try:
getattr(instance, method_name)()
print(f" PASS {cls_name}.{method_name}")
except Exception as e:
print(f" FAIL {cls_name}.{method_name}: {e}")
print("\nAll tests completed.")

View File

@@ -0,0 +1,115 @@
"""
Integration tests for the CoverageEngine orchestrator.
Tests model selection, available models API, and the
engine's coordination logic (without running actual
coverage calculations, which require terrain data).
"""
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..'))
from app.core.engine import CoverageEngine, BandType, PresetType, CoverageSettings
class TestEngineModelSelection:
def test_lte_urban_uses_cost231(self):
engine = CoverageEngine()
model = engine.select_model(BandType.LTE, "urban")
assert model.name == "COST-231-Hata"
def test_lte_suburban_uses_okumura(self):
engine = CoverageEngine()
model = engine.select_model(BandType.LTE, "suburban")
assert model.name == "Okumura-Hata"
def test_lte_open_uses_fspl(self):
engine = CoverageEngine()
model = engine.select_model(BandType.LTE, "open")
assert model.name == "Free-Space"
def test_uhf_urban_uses_okumura(self):
engine = CoverageEngine()
model = engine.select_model(BandType.UHF, "urban")
assert model.name == "Okumura-Hata"
def test_uhf_rural_uses_longley_rice(self):
engine = CoverageEngine()
model = engine.select_model(BandType.UHF, "rural")
assert model.name == "Longley-Rice"
def test_vhf_urban_uses_p1546(self):
engine = CoverageEngine()
model = engine.select_model(BandType.VHF, "urban")
assert model.name == "ITU-R-P.1546"
def test_vhf_rural_uses_longley_rice(self):
engine = CoverageEngine()
model = engine.select_model(BandType.VHF, "rural")
assert model.name == "Longley-Rice"
def test_unknown_band_falls_back(self):
engine = CoverageEngine()
model = engine.select_model(BandType.CUSTOM, "desert")
assert model is not None # Should not crash
class TestEngineModelsAPI:
def test_returns_dict(self):
engine = CoverageEngine()
models = engine.get_available_models()
assert isinstance(models, dict)
assert len(models) >= 5
def test_model_info_structure(self):
engine = CoverageEngine()
models = engine.get_available_models()
for name, info in models.items():
assert "frequency_range" in info
assert "distance_range" in info
assert "bands" in info
assert len(info["bands"]) > 0
def test_all_expected_models_present(self):
engine = CoverageEngine()
models = engine.get_available_models()
expected = {"COST-231-Hata", "Okumura-Hata", "Free-Space", "Longley-Rice", "ITU-R-P.1546"}
assert expected.issubset(set(models.keys()))
class TestCoverageSettings:
def test_default_settings(self):
s = CoverageSettings()
assert s.radius == 10000
assert s.resolution == 200
assert s.preset == PresetType.STANDARD
assert s.band_type == BandType.LTE
def test_preset_values(self):
assert PresetType.FAST.value == "fast"
assert PresetType.STANDARD.value == "standard"
assert PresetType.DETAILED.value == "detailed"
assert PresetType.FULL.value == "full"
def test_band_type_values(self):
assert BandType.LTE.value == "lte"
assert BandType.UHF.value == "uhf"
assert BandType.VHF.value == "vhf"
if __name__ == "__main__":
for cls_name, cls in [
("ModelSelection", TestEngineModelSelection),
("ModelsAPI", TestEngineModelsAPI),
("CoverageSettings", TestCoverageSettings),
]:
instance = cls()
for method_name in [m for m in dir(instance) if m.startswith("test_")]:
try:
getattr(instance, method_name)()
print(f" PASS {cls_name}.{method_name}")
except Exception as e:
print(f" FAIL {cls_name}.{method_name}: {e}")
print("\nAll tests completed.")

View File

View File

@@ -0,0 +1,90 @@
"""
Unit tests for COST-231 Hata and COST-231 Walfisch-Ikegami models.
"""
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..'))
from app.propagation.base import PropagationInput
from app.propagation.cost231_hata import Cost231HataModel
from app.propagation.cost231_wi import Cost231WIModel
def make_input(**kwargs) -> PropagationInput:
defaults = {
"frequency_mhz": 1800,
"distance_m": 5000,
"tx_height_m": 30,
"rx_height_m": 1.5,
"environment": "urban",
}
defaults.update(kwargs)
return PropagationInput(**defaults)
class TestCost231Hata:
def test_typical_range(self):
model = Cost231HataModel()
out = model.calculate(make_input())
assert 130 < out.path_loss_db < 170
def test_model_name(self):
model = Cost231HataModel()
assert model.name == "COST-231-Hata"
def test_frequency_range(self):
model = Cost231HataModel()
assert model.is_valid_for(make_input(frequency_mhz=1500))
assert model.is_valid_for(make_input(frequency_mhz=2000))
assert not model.is_valid_for(make_input(frequency_mhz=900))
def test_distance_increases_loss(self):
model = Cost231HataModel()
loss_2 = model.calculate(make_input(distance_m=2000)).path_loss_db
loss_10 = model.calculate(make_input(distance_m=10000)).path_loss_db
assert loss_10 > loss_2
def test_urban_vs_suburban(self):
model = Cost231HataModel()
urban = model.calculate(make_input(environment="urban")).path_loss_db
suburban = model.calculate(make_input(environment="suburban")).path_loss_db
assert suburban < urban
class TestCost231WI:
def test_typical_range(self):
model = Cost231WIModel()
out = model.calculate(make_input(distance_m=500))
assert 80 < out.path_loss_db < 160
def test_model_name(self):
model = Cost231WIModel()
assert model.name == "COST-231-WI"
def test_distance_increases_loss(self):
model = Cost231WIModel()
loss_200 = model.calculate(make_input(distance_m=200)).path_loss_db
loss_1000 = model.calculate(make_input(distance_m=1000)).path_loss_db
assert loss_1000 > loss_200
def test_frequency_range(self):
model = Cost231WIModel()
assert model.is_valid_for(make_input(frequency_mhz=800))
assert model.is_valid_for(make_input(frequency_mhz=2000))
assert not model.is_valid_for(make_input(frequency_mhz=400))
if __name__ == "__main__":
for cls_name, cls in [("COST231Hata", TestCost231Hata), ("COST231WI", TestCost231WI)]:
instance = cls()
for method_name in [m for m in dir(instance) if m.startswith("test_")]:
try:
getattr(instance, method_name)()
print(f" PASS {cls_name}.{method_name}")
except AssertionError as e:
print(f" FAIL {cls_name}.{method_name}: {e}")
except Exception as e:
print(f" ERROR {cls_name}.{method_name}: {e}")
print("\nAll tests completed.")

View File

@@ -0,0 +1,82 @@
"""
Detailed unit tests for the Free Space Path Loss model.
"""
import sys
import os
import math
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..'))
from app.propagation.base import PropagationInput
from app.propagation.free_space import FreeSpaceModel
def make_input(**kwargs) -> PropagationInput:
defaults = {
"frequency_mhz": 1800,
"distance_m": 1000,
"tx_height_m": 30,
"rx_height_m": 1.5,
"environment": "urban",
}
defaults.update(kwargs)
return PropagationInput(**defaults)
class TestFreeSpaceModel:
def test_formula_accuracy(self):
"""FSPL = 20*log10(d_km) + 20*log10(f_MHz) + 32.45"""
model = FreeSpaceModel()
# At 1km, 1000MHz: 20*0 + 20*60 + 32.45 = 92.45 dB
out = model.calculate(make_input(distance_m=1000, frequency_mhz=1000))
expected = 32.45 + 20 * math.log10(1.0) + 20 * math.log10(1000)
assert abs(out.path_loss_db - expected) < 0.1
def test_6db_per_distance_doubling(self):
model = FreeSpaceModel()
loss_1 = model.calculate(make_input(distance_m=1000)).path_loss_db
loss_2 = model.calculate(make_input(distance_m=2000)).path_loss_db
assert abs((loss_2 - loss_1) - 6.02) < 0.1
def test_6db_per_frequency_doubling(self):
model = FreeSpaceModel()
loss_1 = model.calculate(make_input(frequency_mhz=900)).path_loss_db
loss_2 = model.calculate(make_input(frequency_mhz=1800)).path_loss_db
assert abs((loss_2 - loss_1) - 6.02) < 0.1
def test_always_los(self):
model = FreeSpaceModel()
out = model.calculate(make_input())
assert out.is_los is True
def test_model_name(self):
model = FreeSpaceModel()
assert model.name == "Free-Space"
def test_wide_frequency_range(self):
model = FreeSpaceModel()
assert model.is_valid_for(make_input(frequency_mhz=1))
assert model.is_valid_for(make_input(frequency_mhz=100000))
def test_very_short_distance(self):
model = FreeSpaceModel()
out = model.calculate(make_input(distance_m=10))
assert out.path_loss_db > 0
assert out.path_loss_db < 80
def test_very_long_distance(self):
model = FreeSpaceModel()
out = model.calculate(make_input(distance_m=100000))
assert out.path_loss_db > 120
if __name__ == "__main__":
instance = TestFreeSpaceModel()
for method_name in [m for m in dir(instance) if m.startswith("test_")]:
try:
getattr(instance, method_name)()
print(f" PASS {method_name}")
except Exception as e:
print(f" FAIL {method_name}: {e}")
print("\nAll tests completed.")

View File

@@ -0,0 +1,93 @@
"""
Detailed unit tests for the Okumura-Hata model.
"""
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..'))
from app.propagation.base import PropagationInput
from app.propagation.okumura_hata import OkumuraHataModel
def make_input(**kwargs) -> PropagationInput:
defaults = {
"frequency_mhz": 900,
"distance_m": 5000,
"tx_height_m": 30,
"rx_height_m": 1.5,
"environment": "urban",
}
defaults.update(kwargs)
return PropagationInput(**defaults)
class TestOkumuraHata:
def test_urban_typical_range(self):
model = OkumuraHataModel()
out = model.calculate(make_input())
# 900MHz, 5km, urban: expect ~130-155 dB
assert 120 < out.path_loss_db < 160
def test_environment_ordering(self):
"""Urban > suburban > rural path loss."""
model = OkumuraHataModel()
urban = model.calculate(make_input(environment="urban")).path_loss_db
suburban = model.calculate(make_input(environment="suburban")).path_loss_db
rural = model.calculate(make_input(environment="rural")).path_loss_db
assert urban > suburban > rural
def test_distance_increases_loss(self):
model = OkumuraHataModel()
loss_1 = model.calculate(make_input(distance_m=2000)).path_loss_db
loss_5 = model.calculate(make_input(distance_m=5000)).path_loss_db
loss_10 = model.calculate(make_input(distance_m=10000)).path_loss_db
assert loss_1 < loss_5 < loss_10
def test_frequency_increases_loss(self):
model = OkumuraHataModel()
loss_450 = model.calculate(make_input(frequency_mhz=450)).path_loss_db
loss_900 = model.calculate(make_input(frequency_mhz=900)).path_loss_db
assert loss_900 > loss_450
def test_higher_tx_reduces_loss(self):
model = OkumuraHataModel()
loss_low = model.calculate(make_input(tx_height_m=10)).path_loss_db
loss_high = model.calculate(make_input(tx_height_m=50)).path_loss_db
assert loss_high < loss_low
def test_valid_frequency_range(self):
model = OkumuraHataModel()
assert model.is_valid_for(make_input(frequency_mhz=150))
assert model.is_valid_for(make_input(frequency_mhz=1500))
assert not model.is_valid_for(make_input(frequency_mhz=2000))
def test_valid_distance_range(self):
model = OkumuraHataModel()
assert model.is_valid_for(make_input(distance_m=500))
assert model.is_valid_for(make_input(distance_m=20000))
# Out of range
assert not model.is_valid_for(make_input(distance_m=50))
def test_model_name(self):
model = OkumuraHataModel()
assert model.name == "Okumura-Hata"
def test_open_environment(self):
"""Open environment should have even less loss than rural."""
model = OkumuraHataModel()
rural = model.calculate(make_input(environment="rural")).path_loss_db
open_area = model.calculate(make_input(environment="open")).path_loss_db
assert open_area < rural
if __name__ == "__main__":
instance = TestOkumuraHata()
for method_name in [m for m in dir(instance) if m.startswith("test_")]:
try:
getattr(instance, method_name)()
print(f" PASS {method_name}")
except Exception as e:
print(f" FAIL {method_name}: {e}")
print("\nAll tests completed.")

View File

@@ -0,0 +1,198 @@
"""
Unit tests for propagation models.
Run: cd backend && python -m pytest tests/test_models/test_propagation.py -v
"""
import math
import sys
import os
# Add backend to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..'))
from app.propagation.base import PropagationInput
from app.propagation.free_space import FreeSpaceModel
from app.propagation.okumura_hata import OkumuraHataModel
from app.propagation.cost231_hata import Cost231HataModel
from app.propagation.cost231_wi import Cost231WIModel
from app.propagation.itu_r_p1546 import ITUR_P1546Model
from app.propagation.longley_rice import LongleyRiceModel
from app.propagation.itu_r_p526 import KnifeEdgeDiffractionModel
def make_input(**kwargs) -> PropagationInput:
defaults = {
"frequency_mhz": 1800,
"distance_m": 1000,
"tx_height_m": 30,
"rx_height_m": 1.5,
"environment": "urban",
}
defaults.update(kwargs)
return PropagationInput(**defaults)
class TestFreeSpaceModel:
def test_basic_fspl(self):
model = FreeSpaceModel()
output = model.calculate(make_input(distance_m=1000, frequency_mhz=1800))
# FSPL at 1km, 1800MHz ≈ 97.5 dB
assert 95 < output.path_loss_db < 100
assert output.is_los is True
assert output.model_name == "Free-Space"
def test_distance_increases_loss(self):
model = FreeSpaceModel()
loss_1km = model.calculate(make_input(distance_m=1000)).path_loss_db
loss_2km = model.calculate(make_input(distance_m=2000)).path_loss_db
# Doubling distance adds ~6 dB
assert 5 < (loss_2km - loss_1km) < 7
def test_frequency_increases_loss(self):
model = FreeSpaceModel()
loss_900 = model.calculate(make_input(frequency_mhz=900)).path_loss_db
loss_1800 = model.calculate(make_input(frequency_mhz=1800)).path_loss_db
# Doubling frequency adds ~6 dB
assert 5 < (loss_1800 - loss_900) < 7
def test_valid_range(self):
model = FreeSpaceModel()
assert model.is_valid_for(make_input(distance_m=100))
assert model.is_valid_for(make_input(distance_m=100000))
class TestOkumuraHata:
def test_urban_loss(self):
model = OkumuraHataModel()
output = model.calculate(make_input(
frequency_mhz=900, distance_m=5000,
tx_height_m=30, environment="urban",
))
# Typical urban loss at 5km, 900MHz: 130-150 dB
assert 120 < output.path_loss_db < 160
assert output.model_name == "Okumura-Hata"
def test_suburban_less_than_urban(self):
model = OkumuraHataModel()
inp = make_input(frequency_mhz=900, distance_m=5000)
urban = model.calculate(PropagationInput(**{**inp.__dict__, "environment": "urban"}))
suburban = model.calculate(PropagationInput(**{**inp.__dict__, "environment": "suburban"}))
assert suburban.path_loss_db < urban.path_loss_db
def test_rural_less_than_suburban(self):
model = OkumuraHataModel()
inp = make_input(frequency_mhz=900, distance_m=5000)
suburban = model.calculate(PropagationInput(**{**inp.__dict__, "environment": "suburban"}))
rural = model.calculate(PropagationInput(**{**inp.__dict__, "environment": "rural"}))
assert rural.path_loss_db < suburban.path_loss_db
def test_valid_range(self):
model = OkumuraHataModel()
assert model.is_valid_for(make_input(frequency_mhz=900, distance_m=5000))
assert not model.is_valid_for(make_input(frequency_mhz=2000, distance_m=5000))
class TestCost231Hata:
def test_basic_loss(self):
model = Cost231HataModel()
output = model.calculate(make_input(
frequency_mhz=1800, distance_m=5000,
))
assert 130 < output.path_loss_db < 170
assert output.model_name == "COST-231-Hata"
def test_valid_range(self):
model = Cost231HataModel()
assert model.is_valid_for(make_input(frequency_mhz=1800, distance_m=5000))
assert not model.is_valid_for(make_input(frequency_mhz=900, distance_m=5000))
class TestCost231WI:
def test_basic_loss(self):
model = Cost231WIModel()
output = model.calculate(make_input(
frequency_mhz=1800, distance_m=500,
environment="urban",
))
assert 80 < output.path_loss_db < 160
assert output.model_name == "COST-231-WI"
class TestITUR_P1546:
def test_basic_loss(self):
model = ITUR_P1546Model()
output = model.calculate(make_input(
frequency_mhz=450, distance_m=10000,
))
assert 80 < output.path_loss_db < 160
assert output.model_name == "ITU-R-P.1546"
class TestLongleyRice:
def test_basic_loss(self):
model = LongleyRiceModel()
output = model.calculate(make_input(
frequency_mhz=150, distance_m=20000,
terrain_roughness_m=50,
))
assert 90 < output.path_loss_db < 160
assert output.model_name == "Longley-Rice"
def test_flat_terrain_is_los(self):
model = LongleyRiceModel()
output = model.calculate(make_input(
frequency_mhz=150, distance_m=5000,
terrain_roughness_m=5,
))
assert output.is_los is True
class TestKnifeEdgeDiffraction:
def test_no_obstruction(self):
loss = KnifeEdgeDiffractionModel.calculate_loss(
d1_m=500, d2_m=500, h_m=-5, wavelength_m=0.167,
)
assert loss >= 0
def test_obstruction_increases_loss(self):
loss_low = KnifeEdgeDiffractionModel.calculate_loss(
d1_m=500, d2_m=500, h_m=1, wavelength_m=0.167,
)
loss_high = KnifeEdgeDiffractionModel.calculate_loss(
d1_m=500, d2_m=500, h_m=10, wavelength_m=0.167,
)
assert loss_high > loss_low
def test_clearance_loss_positive_clearance(self):
loss = KnifeEdgeDiffractionModel.calculate_clearance_loss(5.0, 1800)
assert loss == 0.0
def test_clearance_loss_negative_clearance(self):
loss = KnifeEdgeDiffractionModel.calculate_clearance_loss(-10.0, 1800)
assert loss > 0
if __name__ == "__main__":
# Quick run without pytest
for cls_name, cls in [
("FreeSpace", TestFreeSpaceModel),
("OkumuraHata", TestOkumuraHata),
("COST231Hata", TestCost231Hata),
("COST231WI", TestCost231WI),
("ITU-R-P.1546", TestITUR_P1546),
("LongleyRice", TestLongleyRice),
("KnifeEdge", TestKnifeEdgeDiffraction),
]:
instance = cls()
methods = [m for m in dir(instance) if m.startswith("test_")]
for method_name in methods:
try:
getattr(instance, method_name)()
print(f" PASS {cls_name}.{method_name}")
except AssertionError as e:
print(f" FAIL {cls_name}.{method_name}: {e}")
except Exception as e:
print(f" ERROR {cls_name}.{method_name}: {e}")
print("\nAll tests completed.")

View File

@@ -0,0 +1 @@

View File

@@ -0,0 +1,126 @@
"""
Unit tests for the unified cache service.
"""
import sys
import os
import time
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..'))
from app.services.cache import MemoryCache, CacheManager
class TestMemoryCache:
def test_get_miss(self):
cache = MemoryCache("test", max_entries=10)
assert cache.get("nonexistent") is None
def test_put_get(self):
cache = MemoryCache("test", max_entries=10)
cache.put("key1", "value1", size_bytes=100)
assert cache.get("key1") == "value1"
def test_overwrite(self):
cache = MemoryCache("test", max_entries=10)
cache.put("key1", "v1", size_bytes=100)
cache.put("key1", "v2", size_bytes=200)
assert cache.get("key1") == "v2"
assert cache.size == 1
assert cache.size_bytes == 200
def test_eviction_by_entries(self):
cache = MemoryCache("test", max_entries=3)
cache.put("a", 1)
cache.put("b", 2)
cache.put("c", 3)
assert cache.size == 3
cache.put("d", 4) # Should evict 'a' (LRU)
assert cache.size == 3
assert cache.get("a") is None
assert cache.get("d") == 4
def test_eviction_by_size(self):
cache = MemoryCache("test", max_entries=100, max_size_bytes=300)
cache.put("a", 1, size_bytes=100)
cache.put("b", 2, size_bytes=100)
cache.put("c", 3, size_bytes=100)
assert cache.size_bytes == 300
cache.put("d", 4, size_bytes=100) # Should evict 'a'
assert cache.size_bytes == 300
assert cache.get("a") is None
def test_lru_access_order(self):
cache = MemoryCache("test", max_entries=3)
cache.put("a", 1)
cache.put("b", 2)
cache.put("c", 3)
# Access 'a' to make it recently used
cache.get("a")
# Add 'd' — should evict 'b' (now LRU)
cache.put("d", 4)
assert cache.get("a") == 1 # Still there
assert cache.get("b") is None # Evicted
def test_remove(self):
cache = MemoryCache("test", max_entries=10)
cache.put("key1", "val", size_bytes=50)
assert cache.remove("key1") is True
assert cache.get("key1") is None
assert cache.size_bytes == 0
def test_clear(self):
cache = MemoryCache("test", max_entries=10)
cache.put("a", 1, size_bytes=100)
cache.put("b", 2, size_bytes=100)
cache.clear()
assert cache.size == 0
assert cache.size_bytes == 0
def test_stats(self):
cache = MemoryCache("test", max_entries=10, max_size_bytes=1024)
cache.put("a", 1, size_bytes=100)
cache.get("a") # hit
cache.get("b") # miss
s = cache.stats()
assert s["name"] == "test"
assert s["entries"] == 1
assert s["hits"] == 1
assert s["misses"] == 1
assert s["hit_rate"] == 50.0
class TestCacheManager:
def test_singleton_structure(self):
mgr = CacheManager()
assert mgr.terrain is not None
assert mgr.buildings is not None
assert mgr.spatial is not None
assert mgr.osm_disk is not None
def test_stats(self):
mgr = CacheManager()
s = mgr.stats()
assert "terrain" in s
assert "buildings" in s
assert "total_memory_mb" in s
def test_clear_all(self):
mgr = CacheManager()
mgr.terrain.put("test", "data", 100)
mgr.buildings.put("test", "data", 100)
mgr.clear_all()
assert mgr.terrain.size == 0
assert mgr.buildings.size == 0
if __name__ == "__main__":
for cls_name, cls in [("MemoryCache", TestMemoryCache), ("CacheManager", TestCacheManager)]:
instance = cls()
for method_name in [m for m in dir(instance) if m.startswith("test_")]:
try:
getattr(instance, method_name)()
print(f" PASS {cls_name}.{method_name}")
except Exception as e:
print(f" FAIL {cls_name}.{method_name}: {e}")
print("\nAll tests completed.")