Performance Tips

🚀 v1.0.1 Performance Improvements

Benchmark Results

Key Optimizations

Table of Contents

Understanding Performance

Where Time Is Spent

Timing Benchmarks

Quick Wins

1. Reuse PredictionEngine Instances

for region in regions:
 engine = PredictionEngine() # DON'T DO THIS
 engine.predict(...)

engine = PredictionEngine() # Create once
for region in regions:
 engine.predict(...) # Reuse for all predictions

2. Batch Frequencies Together

for freq in [7.0, 14.0, 21.0]:
 engine.predict(rx_location, utc_time, frequencies=[freq])

engine.predict(rx_location, utc_time, frequencies=[7.0, 14.0, 21.0])

3. Reduce Number of Predictions

# Instead of 10 regions
TARGET_REGIONS = {
 'EU': ...,
 'JA': ...,
 'VK': ...,
 # Remove regions you don't work
}
# Instead of 7 bands
BANDS = {
 '40m': 7.150,
 '20m': 14.150,
 '15m': 21.200,
 # Remove bands you don't use
}
# Instead of 12 time points, use 8
utc_hours = [0, 3, 6, 9, 12, 15, 18, 21] # Every 3 hours instead of 2

4. Use Reasonable Solar Conditions

# ✅ Good - typical values
engine.params.ssn = 100.0 # Moderate solar cycle
# ❌ Slow - extreme values take longer to converge
engine.params.ssn = 300.0 # Very high (edge case)

Configuration Optimization

Reduce Path Resolution

# Default: High accuracy, slower
engine.params.min_angle = np.deg2rad(3.0)
# Faster: Slightly lower accuracy
engine.params.min_angle = np.deg2rad(5.0)

Adjust Raytracing Parameters

# In muf_calculator.py (internal settings)
MAX_ITERATIONS = 50 # Default: 100 (faster but less precise)
CONVERGENCE_TOLERANCE = 1e-3 # Default: 1e-4 (looser tolerance)

Skip Unnecessary Calculations

# If you only need MUF (not full signal predictions)
# Use muf_calculator directly instead of full PredictionEngine
from dvoacap.muf_calculator import MufCalculator
calc = MufCalculator()
# ... configure and run MUF calculation only
# Much faster than full prediction

Code-Level Optimization

Use NumPy Vectorization

results = []
for i in range(len(frequencies)):
 result = calculate_loss(frequencies[i])
 results.append(result)

import numpy as np
results = calculate_loss(np.array(frequencies)) # Vectorized

Avoid Repeated Calculations

class OptimizedEngine:
 def __init__(self):
 self._solar_cache = {}
 def get_solar_params(self, month, ssn, utc_time):
 """Cache solar calculations"""
 cache_key = (month, ssn, utc_time)
 if cache_key not in self._solar_cache:
 self._solar_cache[cache_key] = compute_solar_params(month, ssn, utc_time)
 return self._solar_cache[cache_key]

Minimize Object Creation

for i in range(1000):
 point = GeoPoint.from_degrees(lat, lon) # Creates new object each time
 # ...

point = GeoPoint.from_degrees(lat, lon) # Create once
for i in range(1000):
 # Reuse point
 # ...

Caching Strategies

Result Caching

import hashlib
import json
import pickle
from pathlib import Path
class CachedPredictionEngine:
 """Prediction engine with result caching"""
 def __init__(self, cache_dir='cache'):
 self.engine = PredictionEngine()
 self.cache_dir = Path(cache_dir)
 self.cache_dir.mkdir(exist_ok=True)
 def _cache_key(self, rx_location, frequencies, month, ssn, utc_time):
 """Generate cache key from parameters"""
 rx_lat, rx_lon = rx_location.to_degrees()
 params = {
 'rx_lat': round(rx_lat, 4),
 'rx_lon': round(rx_lon, 4),
 'frequencies': frequencies,
 'month': month,
 'ssn': round(ssn, 1),
 'utc_time': round(utc_time, 2)
 }
 key_str = json.dumps(params, sort_keys=True)
 return hashlib.md5(key_str.encode()).hexdigest()
 def predict(self, rx_location, frequencies, month, ssn, utc_time):
 """Predict with caching"""
 cache_key = self._cache_key(rx_location, frequencies, month, ssn, utc_time)
 cache_file = self.cache_dir / f"{cache_key}.pkl"
 # Check cache
 if cache_file.exists():
 with open(cache_file, 'rb') as f:
 return pickle.load(f)
 # Run prediction
 self.engine.params.month = month
 self.engine.params.ssn = ssn
 self.engine.predict(
 rx_location=rx_location,
 utc_time=utc_time / 24.0,
 frequencies=frequencies
 )
 # Cache results
 result = {
 'muf': self.engine.muf_calculator.muf,
 'predictions': self.engine.predictions
 }
 with open(cache_file, 'wb') as f:
 pickle.dump(result, f)
 return result
# Usage
engine = CachedPredictionEngine(cache_dir='prediction_cache')
result = engine.predict(...) # First call: runs prediction
result = engine.predict(...) # Second call: instant (from cache)

CCIR Map Caching

# ✅ Good - maps loaded once
maps = FourierMaps()
for month in range(1, 13):
 maps.set_conditions(month=month, ssn=100, utc_fraction=0.5)
 # Maps are reused
# ❌ Bad - reloads maps each time
for month in range(1, 13):
 maps = FourierMaps() # DON'T DO THIS
 maps.set_conditions(month=month, ssn=100, utc_fraction=0.5)

Parallel Processing

Multi-Processing for Multiple Paths

from multiprocessing import Pool
from dvoacap import PredictionEngine
from dvoacap.path_geometry import GeoPoint
def predict_region(args):
 """Worker function for parallel processing"""
 region_name, region_location, frequencies, month, ssn, utc_time = args
 # Each process gets its own engine
 engine = PredictionEngine()
 engine.params.month = month
 engine.params.ssn = ssn
 engine.params.tx_location = GeoPoint.from_degrees(44.374, -64.300)
 engine.predict(
 rx_location=region_location,
 utc_time=utc_time / 24.0,
 frequencies=frequencies
 )
 return {
 'region': region_name,
 'muf': engine.muf_calculator.muf,
 'predictions': engine.predictions
 }
# Parallel execution
if __name__ == '__main__':
 regions = {
 'EU': GeoPoint.from_degrees(50.0, 10.0),
 'JA': GeoPoint.from_degrees(36.0, 138.0),
 'VK': GeoPoint.from_degrees(-33.87, 151.21),
 # ... more regions
 }
 tasks = [
 (name, location, [7.0, 14.0, 21.0], 6, 100, 12)
 for name, location in regions.items()
 ]
 # Use 4 processes
 with Pool(processes=4) as pool:
 results = pool.map(predict_region, tasks)
 for result in results:
 print(f"{result['region']}: MUF {result['muf']:.2f} MHz")

Async I/O for Dashboard

import asyncio
from dvoacap import PredictionEngine
async def async_predict(region, location, frequencies):
 """Async wrapper for prediction"""
 loop = asyncio.get_event_loop()
 # Run prediction in thread pool
 result = await loop.run_in_executor(
 None,
 run_prediction,
 region, location, frequencies
 )
 return result
def run_prediction(region, location, frequencies):
 """Synchronous prediction function"""
 engine = PredictionEngine()
 # ... configure and run prediction
 return result
# Run predictions concurrently
async def main():
 tasks = [
 async_predict('EU', eu_location, [14.0]),
 async_predict('JA', ja_location, [14.0]),
 async_predict('VK', vk_location, [14.0]),
 ]
 results = await asyncio.gather(*tasks)
 return results
# Execute
results = asyncio.run(main())

Profiling

Identify Bottlenecks

python3 -m cProfile -o profile.stats generate_predictions.py

import pstats
stats = pstats.Stats('profile.stats')
stats.sort_stats('cumulative')
stats.print_stats(20) # Top 20 slowest functions

Line-by-Line Profiling

# Install
pip install line_profiler
# Add @profile decorator to function
@profile
def my_slow_function():
 # ...
# Run profiler
kernprof -l -v my_script.py

Memory Profiling

pip install memory_profiler
# Run
python3 -m memory_profiler generate_predictions.py

Future Improvements

Planned Optimizations

Experimental: Numba Acceleration

from numba import jit
import numpy as np
@jit(nopython=True)
def fast_path_loss_calculation(frequency, distance, layer_params):
 """JIT-compiled path loss calculation"""
 # ... computation
 return loss
# First call: compilation overhead (~1 second)
result = fast_path_loss_calculation(14.0, 5000, params)
# Subsequent calls: very fast (10-100x speedup)
result = fast_path_loss_calculation(21.0, 5000, params)

Benchmark Comparison

Optimization Impact Summary

Practical Examples

Optimized Dashboard Generation

for region in regions:
 for band in bands:
 for hour in range(24):
 engine = PredictionEngine() # ❌ Recreates each time
 engine.predict(region, [band], hour)

engine = PredictionEngine() # ✅ Create once
# Batch by region
for region in regions:
 # Batch frequencies
 all_bands = list(bands.values())
 # Batch time points
 for hour in [0, 3, 6, 9, 12, 15, 18, 21]: # Every 3 hours
 engine.predict(region, all_bands, hour)

Optimized Multi-Region Prediction

from concurrent.futures import ProcessPoolExecutor
from functools import partial
def optimized_multi_region_predict(regions, bands, hours, month, ssn):
 """Optimized multi-region prediction"""
 def predict_one(region_name, region_location):
 """Predict for one region (runs in separate process)"""
 engine = PredictionEngine()
 engine.params.month = month
 engine.params.ssn = ssn
 engine.params.tx_location = tx_location
 results = []
 for hour in hours:
 engine.predict(
 rx_location=region_location,
 utc_time=hour / 24.0,
 frequencies=bands
 )
 results.append({
 'hour': hour,
 'muf': engine.muf_calculator.muf,
 'predictions': engine.predictions
 })
 return region_name, results
 # Run in parallel
 with ProcessPoolExecutor(max_workers=4) as executor:
 futures = {
 executor.submit(predict_one, name, location): name
 for name, location in regions.items()
 }
 all_results = {}
 for future in futures:
 region_name, results = future.result()
 all_results[region_name] = results
 return all_results
# Usage
results = optimized_multi_region_predict(
 regions=TARGET_REGIONS,
 bands=[7.0, 14.0, 21.0],
 hours=[0, 6, 12, 18],
 month=6,
 ssn=100
)

Monitoring Performance

Add Timing to Your Code

import time
from functools import wraps
def timing_decorator(func):
 """Decorator to measure function execution time"""
 @wraps(func)
 def wrapper(*args, **kwargs):
 start = time.time()
 result = func(*args, **kwargs)
 elapsed = time.time() - start
 print(f"{func.__name__} took {elapsed:.2f} seconds")
 return result
 return wrapper
@timing_decorator
def generate_predictions():
 # ... prediction code
 pass
# Output: generate_predictions took 45.23 seconds

Next Steps