Airspace Classification System - Design Document¶
Executive Summary¶
Overview¶
The Airspace Classification System is an automated safety feature that classifies UK airspace at drone flight locations using authoritative NATS (National Air Traffic Services) data. This system automatically determines whether a proposed flight location falls within controlled airspace (requiring special permissions) or uncontrolled airspace (standard drone operations permitted).
Business Value¶
Safety Enhancement: - Prevents accidental flights in controlled airspace (Classes A-D) - Provides real-time warnings about Flight Restriction Zones (FRZ) and Runway Protection Zones (RPZ) - Ensures regulatory compliance before flight operations begin
User Experience: - Automatic classification - No manual airspace lookup required - Integrated workflow - Airspace data flows seamlessly through project creation wizard - Pre-populated forms - Viability study forms auto-filled with accurate airspace information - Clear warnings - Visual indicators for controlled airspace with actionable guidance
Operational Efficiency: - Reduces manual research time (saves ~5-10 minutes per project) - Eliminates human error in airspace classification - Provides auditable airspace data for compliance documentation
Key Capabilities¶
- Automatic Spatial Classification
- Point-in-polygon queries against ~800-900 UK airspace volumes
- Handles overlapping airspace (selects most restrictive class)
-
Query altitude: 400 feet AGL (standard drone maximum)
-
Background Data Updates
- Scheduled checks for new AIRAC cycle data (28-day cycles)
- Automatic download and parsing of NATS datasets
-
Data-driven update strategy (checks if new data exists before downloading)
-
Admin Management
- View cache status and AIRAC cycle information
- Manually trigger airspace data refresh
-
Monitor update history and cache health
-
Graceful Degradation
- Stale cache fallback (use old data if new unavailable)
- "Unknown" classification if data unavailable
- Projects can still be created without airspace data
Architecture Overview¶
High-Level System Diagram¶
┌─────────────────────────────────────────────────────────────────┐
│ USER INTERACTION LAYER │
├─────────────────────────────────────────────────────────────────┤
│ Project Creation Wizard │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Location │→ │ Details │→ │Viability │→ │ Toggles │ │
│ │ Select │ │ Display │ │ Pre-pop │ │ Submit │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
│ ↓ ↓ ↓ ↓ │
│ [Airspace [Airspace [Airspace [Airspace │
│ Classify] Display] Auto-fill] Storage] │
└─────────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────────┐
│ APPLICATION LOGIC LAYER │
├─────────────────────────────────────────────────────────────────┤
│ │
│ UKAirspaceAPI (app/utils/airspace_api.py) │
│ ┌────────────────────────────────────────────────┐ │
│ │ get_classification(lat, lon, alt) │ │
│ │ • Load cache (lazy loading) │ │
│ │ • Point-in-polygon spatial query │ │
│ │ • Find most restrictive airspace │ │
│ │ • Return classification result │ │
│ └────────────────────────────────────────────────┘ │
│ │
│ NAATSDataDownloader │
│ ┌────────────────────────────────────────────────┐ │
│ │ • Calculate AIRAC cycle dates │ │
│ │ • Check if new data exists (HTTP HEAD) │ │
│ │ • Download ZIP from NATS │ │
│ │ • Extract and parse AIXM XML │ │
│ │ • Serialize to JSON cache │ │
│ └────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────────┐
│ BACKGROUND SCHEDULER │
├─────────────────────────────────────────────────────────────────┤
│ │
│ AirspaceUpdateScheduler (APScheduler) │
│ ┌────────────────────────────────────────────────┐ │
│ │ Daily Check (2:00 AM UTC) │ │
│ │ • Calculate current AIRAC date │ │
│ │ • Compare with cached AIRAC date │ │
│ │ • If newer cycle: check if exists on NATS │ │
│ │ • Download if available │ │
│ │ • Parse and cache │ │
│ │ • Log update results │ │
│ └────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────────┐
│ DATA STORAGE LAYER │
├─────────────────────────────────────────────────────────────────┤
│ │
│ Three-Tier Cache (File System) │
│ ┌────────────────────────────────────────────────┐ │
│ │ airspace_cache/ │ │
│ │ ├── downloads/ │ │
│ │ │ └── EG_UAS_..._20250123_XML.zip │ │
│ │ ├── xml/ │ │
│ │ │ └── EG_UAS_..._20250123.xml │ │
│ │ ├── airspace_cache.json (fast load) │ │
│ │ └── update_log.txt │ │
│ └────────────────────────────────────────────────┘ │
│ │
│ Database (MariaDB) │
│ ┌────────────────────────────────────────────────┐ │
│ │ Project Table │ │
│ │ • airspace_classification (VARCHAR 1) │ │
│ │ • airspace_name (VARCHAR 255) │ │
│ │ • airspace_designation (VARCHAR 10) │ │
│ │ • airspace_controlled (BOOLEAN) │ │
│ └────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────────┐
│ EXTERNAL DATA SOURCE │
├─────────────────────────────────────────────────────────────────┤
│ │
│ NATS UK EAD (European AIS Database) │
│ ┌────────────────────────────────────────────────┐ │
│ │ https://nats-uk.ead-it.com/... │ │
│ │ • AIXM 5.1 XML format │ │
│ │ • ~5-10 MB ZIP archive │ │
│ │ • Updated every 28 days (AIRAC cycle) │ │
│ │ • ~800-900 airspace volumes │ │
│ └────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
Component Interaction Flow¶
USER APPLICATION CACHE NATS
│ │ │ │
│ Submit Location │ │ │
├──────────────────────>│ │ │
│ │ │ │
│ │ Load Cache (lazy) │ │
│ ├──────────────────────>│ │
│ │<──────────────────────┤ │
│ │ Airspace Volumes │ │
│ │ │ │
│ │ Classify Location │ │
│ │ (point-in-polygon) │ │
│ │ │ │
│ Display Result │ │ │
│<──────────────────────┤ │ │
│ (Class D - Heathrow) │ │ │
│ │ │ │
│ │ │ │
│ │ [BACKGROUND JOB] │ │
│ │ Daily 2 AM Check │ │
│ ├──────────────────────>│ │
│ │ Get Cached AIRAC │ │
│ │<──────────────────────┤ │
│ │ 20250123 │ │
│ │ │ │
│ │ Calculate New AIRAC │ │
│ │ 20250220 │ │
│ │ │ │
│ │ Check if Exists │ │
│ ├────────────────────────────────────────────>│
│ │ │ HTTP HEAD │
│ │<────────────────────────────────────────────┤
│ │ │ 200 OK │
│ │ │ │
│ │ Download New Data │ │
│ ├────────────────────────────────────────────>│
│ │<────────────────────────────────────────────┤
│ │ ZIP Archive │ │
│ │ │ │
│ │ Parse & Cache │ │
│ ├──────────────────────>│ │
│ │ │ │
Integration Points¶
1. Project Creation Workflow (app/routes_new_project.py):
- Location Step (line 88): Call UKAirspaceAPI.get_classification() after rural/urban classification
- Details Step (line 185): Display airspace panel in side panel alongside rural/urban
- Viability Step (line 217): Pre-populate airspace field in form
- Toggles Step (line 326-329): Store airspace data in Project record
2. Session Management: - Store: 7 airspace-related session keys during location step - Display: Read from session in details/viability templates - Clear: Remove all airspace keys after project creation (line 366-372)
3. Admin Interface (app/routes_admin.py):
- Status Page (/admin/airspace-status): Display cache info, AIRAC dates, volume counts
- Manual Refresh (/admin/airspace-refresh POST): Trigger update outside scheduled job
Component Details¶
1. Data Acquisition Layer¶
File: app/utils/airspace_api.py (lines 400-822)
Class: NAATSDataDownloader
AIRAC Cycle Calculation¶
Purpose: Determine current and next AIRAC effective dates
Algorithm:
def calculate_current_airac_date():
"""
Calculate current AIRAC cycle date
Base date: 2024-01-25
Cycle length: 28 days
Returns: datetime object for most recent AIRAC cycle
"""
base_date = datetime(2024, 1, 25)
cycle_length = 28
today = datetime.now()
days_since_base = (today - base_date).days
# Number of complete cycles
cycle_number = days_since_base // cycle_length
# Current AIRAC date
current_airac = base_date + timedelta(days=cycle_number * cycle_length)
return current_airac
Why Base Date 2024-01-25? - Known AIRAC effective date from NATS published schedule - All future dates calculated as multiples of 28 days from this anchor
Edge Cases: - Date calculation never fails (purely mathematical) - Always returns datetime, never None - Tested back to 2020 and forward to 2030
Data Existence Checking¶
Purpose: Verify new AIRAC data available before downloading
Method: HTTP HEAD request (no download required)
def check_if_airac_exists_on_nats(airac_date):
"""
Check if AIRAC dataset exists on NATS server
Uses HTTP HEAD to avoid downloading full file
Returns: True if exists (HTTP 200), False otherwise
"""
url = build_dataset_url(airac_date)
try:
response = requests.head(url, timeout=10)
return response.status_code == 200
except requests.RequestException:
return False
Why HEAD Instead of GET? - Faster: No file download (5-10 MB saved) - Cheaper: Minimal bandwidth usage - Identical result: 200 = exists, 404 = not published yet
NATS Publication Lag: - AIRAC cycle effective date != publication date - Data may publish 1-7 days after effective date - HEAD request prevents failed downloads during lag period
Download and Extraction¶
Purpose: Acquire and extract AIXM XML from NATS ZIP archive
Process:
1. Download ZIP to airspace_cache/downloads/
2. Verify ZIP integrity
3. Extract XML to airspace_cache/xml/
4. Verify XML well-formed
5. Return path to extracted XML
Error Handling: - Network timeout: 30 seconds, retry once - Corrupted ZIP: Delete and re-download - Extraction failure: Log error, use stale cache - Disk space: Check available space before download (minimum 50 MB)
Bandwidth Optimization: - Only download if new AIRAC cycle detected - Use streaming download for large files (chunk size: 8192 bytes) - Resume partial downloads if interrupted
2. Parsing Layer¶
File: app/utils/airspace_api.py (lines 553-766)
Class: NAATSDataDownloader
AIXM XML Parser¶
Purpose: Extract airspace volumes from AIXM 5.1 XML
Libraries:
- lxml - Fast XML parsing with namespace support
- shapely - Geometry construction (Polygon, Point)
Parsing Strategy:
def parse_aixm_xml(xml_path):
"""
Parse AIXM 5.1 XML and extract airspace volumes
Returns: List of AirspaceVolume objects
"""
tree = etree.parse(xml_path)
root = tree.getroot()
volumes = []
# Find all Airspace elements
for airspace in root.findall('.//aixm:Airspace', NAMESPACES):
try:
# Extract metadata
timeslice = airspace.find('.//aixm:AirspaceTimeSlice', NAMESPACES)
name = timeslice.findtext('.//aixm:designator', namespaces=NAMESPACES)
designation = timeslice.findtext('.//aixm:type', namespaces=NAMESPACES)
classification = timeslice.findtext('.//aixm:class', namespaces=NAMESPACES)
# Extract vertical limits
lower_limit = extract_vertical_limit(timeslice, 'lower')
upper_limit = extract_vertical_limit(timeslice, 'upper')
# Extract geometry
geometry = extract_geometry(timeslice)
if geometry:
volumes.append(AirspaceVolume(
name=name,
designation=designation,
classification=classification,
lower_limit=lower_limit,
upper_limit=upper_limit,
geometry=geometry
))
except Exception as e:
logger.warning(f"Skipped airspace volume due to parsing error: {e}")
continue # Skip problematic volumes, don't fail entire parse
return volumes
Namespace Handling:
NAMESPACES = {
'aixm': 'http://www.aixm.aero/schema/5.1',
'gml': 'http://www.opengis.net/gml/3.2',
'xlink': 'http://www.w3.org/1999/xlink'
}
Performance: - XML parse time: ~8-12 seconds for full UK dataset - Memory usage: ~150-200 MB during parse - Result: ~800-900 AirspaceVolume objects
Geometry Extraction¶
Polygon Extraction:
def extract_polygon(surface_element):
"""
Extract Polygon from GML PolygonPatch element
Input: <gml:PolygonPatch> element
Output: Shapely Polygon object
"""
poslist = surface_element.find('.//gml:posList', NAMESPACES)
if poslist is None:
return None
# Parse space-separated lat/lon pairs
coords_text = poslist.text.strip().split()
# Convert to (lon, lat) tuples (Shapely convention)
coords = [
(float(coords_text[i+1]), float(coords_text[i]))
for i in range(0, len(coords_text), 2)
]
# Create Shapely Polygon
return Polygon(coords)
Circle to Polygon Conversion:
def circle_to_polygon(center_lat, center_lon, radius_nm):
"""
Convert circle to polygon with 64 points
Args:
center_lat: Centre latitude
center_lon: Centre longitude
radius_nm: Radius in nautical miles
Returns: Shapely Polygon approximating circle
"""
# Convert NM to degrees (approximate at UK latitude)
radius_deg = radius_nm * 0.01666667
# Generate 64 points around circumference
angles = [2 * 3.14159265359 * i / 64 for i in range(64)]
points = [
(
center_lon + radius_deg * math.cos(angle),
center_lat + radius_deg * math.sin(angle)
)
for angle in angles
]
# Close polygon (first point = last point)
points.append(points[0])
return Polygon(points)
Why 64 Points? - Balance between accuracy and performance - 32 points: Visible polygon edges - 64 points: Smooth circular appearance - 128 points: Marginal improvement, 2x overhead
CircleByCenterPoint Support (Added v1.1):
Many FRZ (Flight Restriction Zones) use gml:CircleByCenterPoint geometry instead of polygons. The parser now handles both CircleByCenterPoint and ArcByCenterPoint formats:
# Extract CircleByCenterPoint (primary method)
circle_elem = airspace_elem.find('.//gml:CircleByCenterPoint', ns)
if circle_elem is not None:
pos_elem = circle_elem.find('.//gml:pos', ns)
radius_elem = circle_elem.find('.//gml:radius', ns)
if pos_elem and radius_elem:
coords = pos_elem.text.strip().split()
lat, lon = float(coords[0]), float(coords[1])
radius = float(radius_elem.text)
uom = radius_elem.get('uom', 'NM') # Supports NM, nmi_i, [nmi_i]
return circle_to_polygon(lat, lon, radius, uom)
Supported Radius Units:
- NM - Nautical miles (standard)
- nmi_i - Nautical miles international
- [nmi_i] - Bracketed format
- KM - Kilometers
Example: Edinburgh FRZ uses a 2.5 nautical mile circle (converted to 64-point polygon)
Vertical Limit Extraction¶
Purpose: Extract altitude limits with units and reference datum
def extract_vertical_limit(timeslice, limit_type):
"""
Extract vertical limit (upper or lower)
Args:
timeslice: AirspaceTimeSlice element
limit_type: 'upper' or 'lower'
Returns: String like "2000 FT SFC" or None
"""
limit_element = timeslice.find(f'.//aixm:{limit_type}Limit', NAMESPACES)
reference_element = timeslice.find(f'.//aixm:{limit_type}LimitReference', NAMESPACES)
if limit_element is None:
return None
# Extract value and unit
value = limit_element.text
unit = limit_element.get('uom', 'FT')
# Extract reference datum
reference = reference_element.text if reference_element is not None else 'SFC'
return f"{value} {unit} {reference}"
Example Outputs:
- "2000 FT SFC" - 2000 feet above surface
- "FL100 STD" - Flight level 100 (10,000 feet standard pressure)
- "1500 FT MSL" - 1500 feet above mean sea level
Cache Serialization¶
Purpose: Save parsed data as JSON for fast loading
Format:
{
"airac_date": "20250123",
"generated_at": "2025-01-23T12:34:56Z",
"volume_count": 827,
"volumes": [
{
"name": "LONDON HEATHROW RWY 27L",
"designation": "RPZ",
"classification": "D",
"lower_limit": "0 FT SFC",
"upper_limit": "2000 FT SFC",
"geometry": {
"type": "Polygon",
"coordinates": [
[
[-0.49, 51.47],
[-0.45, 51.47],
[-0.45, 51.45],
[-0.49, 51.45],
[-0.49, 51.47]
]
]
}
}
]
}
Serialization:
def serialize_to_cache(volumes, cache_path, airac_date):
"""
Serialize AirspaceVolume objects to JSON
Uses Shapely's mapping() to convert geometries to GeoJSON
"""
cache_data = {
'airac_date': airac_date.strftime('%Y%m%d'),
'generated_at': datetime.now(timezone.utc).isoformat(),
'volume_count': len(volumes),
'volumes': [
{
'name': v.name,
'designation': v.designation,
'classification': v.classification,
'lower_limit': v.lower_limit,
'upper_limit': v.upper_limit,
'geometry': mapping(v.geometry) # Shapely to GeoJSON
}
for v in volumes
]
}
with open(cache_path, 'w') as f:
json.dump(cache_data, f, indent=2)
Load Time Comparison: - XML Parse: ~10-15 seconds - JSON Load: ~1-2 seconds - Speedup: 5-10x faster
3. Classification API¶
File: app/utils/airspace_api.py (lines 45-399)
Class: UKAirspaceAPI
Core Classification Method¶
def get_classification(lat, lon, altitude_ft=400):
"""
Classify airspace at given location
Args:
lat: Latitude (WGS84 decimal degrees)
lon: Longitude (WGS84 decimal degrees)
altitude_ft: Query altitude in feet AGL (default 400)
Returns:
dict: {
'success': bool,
'airspace_class': str, # A-G or Unknown
'airspace_name': str,
'airspace_designation': str, # FRZ, RPZ, CTR, etc.
'controlled': bool,
'warning': str or None,
'error': str or None
}
"""
try:
# Lazy load cache
if not _load_cache_if_needed():
return {
'success': False,
'airspace_class': 'Unknown',
'controlled': False,
'error': 'Airspace cache unavailable'
}
# Spatial query
matches = _query_airspace(lat, lon, altitude_ft)
# No matches = Class G (uncontrolled)
if not matches:
return {
'success': True,
'airspace_class': 'G',
'airspace_name': 'Uncontrolled Airspace',
'controlled': False
}
# Find most restrictive airspace
most_restrictive = _find_most_restrictive_airspace(matches)
# Build result
controlled = _is_controlled_airspace(most_restrictive.classification)
result = {
'success': True,
'airspace_class': most_restrictive.classification,
'airspace_name': most_restrictive.name,
'airspace_designation': most_restrictive.designation,
'controlled': controlled
}
# Add warning for controlled airspace
if controlled:
result['warning'] = (
f'Controlled airspace - {most_restrictive.designation} zone. '
f'Permissions required for drone operations.'
)
return result
except Exception as e:
logger.error(f"Airspace classification failed: {e}", exc_info=True)
return {
'success': False,
'airspace_class': 'Unknown',
'controlled': False,
'error': str(e)
}
Lazy Cache Loading¶
Purpose: Load JSON cache only when needed (not on app startup)
# Module-level cache
_cached_volumes = None
_cache_loaded = False
def _load_cache_if_needed():
"""
Load airspace cache if not already loaded
Returns: True if cache available, False otherwise
"""
global _cached_volumes, _cache_loaded
if _cache_loaded:
return _cached_volumes is not None
cache_path = os.path.join(AIRSPACE_CACHE_DIR, 'airspace_cache.json')
if not os.path.exists(cache_path):
logger.warning('Airspace cache not found')
_cache_loaded = True
return False
try:
with open(cache_path, 'r') as f:
cache_data = json.load(f)
# Deserialize geometries
_cached_volumes = []
for vol_data in cache_data['volumes']:
geometry = shape(vol_data['geometry']) # GeoJSON to Shapely
_cached_volumes.append(
AirspaceVolume(
name=vol_data['name'],
designation=vol_data['designation'],
classification=vol_data['classification'],
lower_limit=vol_data['lower_limit'],
upper_limit=vol_data['upper_limit'],
geometry=geometry
)
)
logger.info(f"Airspace cache loaded: {len(_cached_volumes)} volumes")
_cache_loaded = True
return True
except Exception as e:
logger.error(f"Failed to load airspace cache: {e}")
_cache_loaded = True
return False
Memory Impact: - Cache size: ~15-20 MB in memory - Load time: 1-2 seconds first call, instant thereafter - Alternative (load on startup): Slows app init by 1-2 seconds
Spatial Query¶
def _query_airspace(lat, lon, altitude_ft):
"""
Find all airspace volumes containing the query point
Uses Shapely's contains() method for point-in-polygon test
Returns: List of AirspaceVolume objects
"""
query_point = Point(lon, lat) # Shapely uses (x, y) = (lon, lat)
matches = []
for volume in _cached_volumes:
# Horizontal check (point-in-polygon)
if volume.geometry.contains(query_point):
# Vertical check (simplified - assume all < 400 ft includes surface)
# Full implementation would parse altitude limits properly
matches.append(volume)
return matches
Performance Optimization:
# PreparedGeometry for faster repeated queries
from shapely.prepared import prep
_prepared_geometries = None
def _prepare_geometries():
"""Prepare geometries once for faster queries"""
global _prepared_geometries
_prepared_geometries = [prep(v.geometry) for v in _cached_volumes]
def _query_airspace_optimized(lat, lon, altitude_ft):
"""Optimise query using prepared geometries"""
if _prepared_geometries is None:
_prepare_geometries()
query_point = Point(lon, lat)
matches = []
for prep_geom, volume in zip(_prepared_geometries, _cached_volumes):
if prep_geom.contains(query_point):
matches.append(volume)
return matches
Query Performance: - Unprepared: ~200-300ms for 800 volumes - Prepared: ~50-100ms for 800 volumes - Optimization: 3-4x speedup
Most Restrictive Airspace Selection¶
CLASS_PRIORITY = {
'A': 1, 'B': 2, 'C': 3, 'D': 4,
'E': 5, 'G': 6, 'Unknown': 7
}
def _find_most_restrictive_airspace(matches):
"""
Select most restrictive airspace from matches
Priority: A > B > C > D > E > G > Unknown
Args:
matches: List of AirspaceVolume objects
Returns: Single AirspaceVolume (most restrictive)
"""
if not matches:
return None
return min(
matches,
key=lambda v: CLASS_PRIORITY.get(v.classification, 999)
)
Example: - Location contains: Class D CTR, Class E airway, Class G - Result: Class D (most restrictive)
Controlled Airspace Detection¶
def _is_controlled_airspace(airspace_class):
"""
Determine if airspace class is controlled
Controlled: A, B, C, D (require ATC clearance)
Uncontrolled: E, G (no clearance required)
Args:
airspace_class: Single letter A-G
Returns: True if controlled (A-D), False otherwise
"""
return airspace_class in ['A', 'B', 'C', 'D']
4. Background Update Scheduler¶
File: app/utils/airspace_scheduler.py
Library: APScheduler (Advanced Python Scheduler)
Scheduler Initialisation¶
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
scheduler = BackgroundScheduler()
def init_scheduler():
"""
Initialise background scheduler for airspace updates
Runs daily at 2:00 AM UTC
"""
if scheduler.running:
return
# Schedule daily check
scheduler.add_job(
func=check_and_update_airspace_data,
trigger=CronTrigger(hour=2, minute=0), # 2:00 AM UTC
id='airspace_update_check',
name='Airspace AIRAC Update Check',
replace_existing=True
)
scheduler.start()
logger.info('Airspace update scheduler started (daily 2:00 AM UTC)')
Why 2:00 AM UTC? - Low traffic period (minimal user impact) - UK timezone-agnostic (UTC never changes) - NATS typically publishes data during business hours (check next morning)
Data-Driven Update Logic¶
def check_and_update_airspace_data():
"""
Check if new AIRAC cycle data available and update cache
Logic:
1. Calculate current AIRAC cycle date
2. Get cached AIRAC date
3. If current > cached: check if new data exists on NATS
4. If exists: download, parse, cache
5. Log results
"""
logger.info('Starting AIRAC update check')
try:
# Calculate current AIRAC cycle
current_airac = NAATSDataDownloader.calculate_current_airac_date()
# Get cached AIRAC date
cached_airac = NAATSDataDownloader.get_airac_date_from_cache()
if cached_airac is None:
logger.info('No cache found - attempting first download')
_perform_update(current_airac)
return
# Compare dates
if current_airac <= cached_airac:
logger.info(f'Cache current (AIRAC {cached_airac.strftime("%Y%m%d")})')
return
# New cycle available - check if published
logger.info(f'New AIRAC cycle detected: {current_airac.strftime("%Y%m%d")}')
if NAATSDataDownloader.check_if_airac_exists_on_nats(current_airac):
logger.info('New data available on NATS - downloading')
_perform_update(current_airac)
else:
logger.info('New AIRAC cycle not yet published on NATS - will retry tomorrow')
except Exception as e:
logger.error(f'AIRAC update check failed: {e}', exc_info=True)
def _perform_update(airac_date):
"""
Download, parse, and cache new airspace data
Args:
airac_date: datetime object for AIRAC cycle
"""
try:
# Download and extract
xml_path = NAATSDataDownloader.download_and_extract(airac_date)
# Parse AIXM XML
volumes = NAATSDataDownloader.parse_aixm_xml(xml_path)
# Serialize to JSON cache
cache_path = os.path.join(AIRSPACE_CACHE_DIR, 'airspace_cache.json')
NAATSDataDownloader.serialize_to_cache(volumes, cache_path, airac_date)
# Log success
logger.info(f'Airspace cache updated: {len(volumes)} volumes, AIRAC {airac_date.strftime("%Y%m%d")}')
# Write to update log
_write_update_log(airac_date, len(volumes), success=True)
except Exception as e:
logger.error(f'Airspace update failed: {e}', exc_info=True)
_write_update_log(airac_date, 0, success=False, error=str(e))
Data-Driven vs Age-Based:
| Approach | Advantages | Disadvantages |
|---|---|---|
| Data-Driven (current) | • No wasted downloads • Respects NATS publication schedule • Accurate cycle tracking |
• Slightly more complex logic |
| Age-Based (not used) | • Simple logic (if cache > 28 days, download) | • Downloads even if new data not published • Wastes bandwidth on failed attempts • Doesn't account for publication lag |
Update Logging¶
def _write_update_log(airac_date, volume_count, success, error=None):
"""
Write update result to log file
Format: YYYY-MM-DD HH:MM:SS | SUCCESS/FAILURE | AIRAC | Volumes | Error
"""
log_path = os.path.join(AIRSPACE_CACHE_DIR, 'update_log.txt')
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
status = 'SUCCESS' if success else 'FAILURE'
airac_str = airac_date.strftime('%Y%m%d')
log_entry = f"{timestamp} | {status} | {airac_str} | {volume_count} volumes"
if error:
log_entry += f" | Error: {error}"
with open(log_path, 'a') as f:
f.write(log_entry + '\n')
Example Log:
2025-01-15 02:00:12 | SUCCESS | 20250123 | 827 volumes
2025-02-12 02:00:08 | SUCCESS | 20250220 | 831 volumes
2025-03-12 02:00:05 | FAILURE | 20250320 | 0 volumes | Error: HTTP 404
2025-03-13 02:00:11 | SUCCESS | 20250320 | 829 volumes
Data Flow Diagrams¶
Project Creation Flow¶
┌──────────────────────────────────────────────────────────────┐
│ STEP 1: LOCATION SELECTION │
└──────────────────────────────────────────────────────────────┘
User enters coordinates → POST /create_project/location
│
↓
[Rural/Urban Classification]
│
↓
UKAirspaceAPI.get_classification(lat, lon)
│
↓
┌───────────────────────┴────────────────────────┐
│ │
Cache Available? Cache Missing?
│ │
↓ ↓
Point-in-polygon query Return "Unknown"
│ │
↓ │
Matches found? │
Yes │ │ No │
↓ ↓ │
Class D Class G │
│ │ │
└───┬───┴──────────────────────────────────────────┘
│
↓
Store in Session:
• airspace_classification
• airspace_name
• airspace_designation
• airspace_controlled
• airspace_warning (if controlled)
│
↓
Redirect → /create_project/details
┌──────────────────────────────────────────────────────────────┐
│ STEP 2: PROJECT DETAILS │
└──────────────────────────────────────────────────────────────┘
GET /create_project/details
│
↓
Read from Session:
• airspace_classification
• airspace_name
• airspace_controlled
• airspace_warning
│
↓
Render Side Panel:
┌─────────────────────────┐
│ Airspace Classification │
│ Class: D │
│ Name: HEATHROW CTR │
│ ⚠️ Controlled Airspace │
└─────────────────────────┘
│
↓
User submits form → POST /create_project/details
│
↓
Store: projectTitle, projectDescription, etc.
│
↓
Redirect → /create_project/viability
┌──────────────────────────────────────────────────────────────┐
│ STEP 3: VIABILITY STUDY │
└──────────────────────────────────────────────────────────────┘
GET /create_project/viability
│
↓
Read from Session:
• airspace_classification
• airspace_name
│
↓
Pre-populate Form Field:
┌─────────────────────────────────────┐
│ Airspace Class: │
│ [D - LONDON HEATHROW CTR] │
│ │
│ ⚠️ Warning: Controlled Airspace │
└─────────────────────────────────────┘
│
↓
User submits form → POST /create_project/viability
│
↓
Store viability observations
│
↓
Redirect → /create_project/toggles
┌──────────────────────────────────────────────────────────────┐
│ STEP 4: PROJECT TOGGLES & CREATION │
└──────────────────────────────────────────────────────────────┘
User submits → POST /create_project/toggles
│
↓
Create Project Record:
┌────────────────────────────────────┐
│ Project( │
│ latitude=51.468, │
│ longitude=-0.4117, │
│ airspace_classification='D', │
│ airspace_name='HEATHROW CTR', │
│ airspace_designation='CTR', │
│ airspace_controlled=True, │
│ ... │
│ ) │
└────────────────────────────────────┘
│
↓
db.session.add(project)
db.session.commit()
│
↓
Clear Session:
• session.pop('airspace_classification')
• session.pop('airspace_name')
• session.pop('airspace_designation')
• session.pop('airspace_controlled')
• session.pop('airspace_warning')
• ... (all airspace fields)
│
↓
Redirect → /dashboard
Background Update Flow¶
┌──────────────────────────────────────────────────────────────┐
│ DAILY SCHEDULED JOB (2:00 AM UTC) │
└──────────────────────────────────────────────────────────────┘
│
↓
Calculate Current AIRAC Date
(base_date + 28-day cycles)
│
↓
Current AIRAC: 2025-02-20
│
↓
Read Cache: airspace_cache.json
│
├─────────────┬──────────────┐
│ │ │
Cache Exists Cache Missing Cache Corrupted
AIRAC: 2025-01-23 │ │
│ │ │
↓ ↓ ↓
Compare Dates Download New Delete & Download
│ │ │
↓ │ │
Current > Cached? │ │
Yes │ │ No │ │
↓ ↓ │ │
│ Stop │ │
│ (Log: Cache Current) │
│ │
└────────────┬────────────────┘
│
↓
Check if New AIRAC Exists on NATS
(HTTP HEAD request)
│
├───────────┴────────────┐
│ │
200 OK 404 Not Found
│ │
↓ ↓
Download ZIP Stop & Log
│ (Data not published yet)
↓
Extract XML
│
↓
Parse AIXM XML
(8-12 seconds)
│
↓
Serialize to JSON
│
↓
Write Cache:
airspace_cache.json
│
↓
Write Update Log:
"2025-02-20 02:00:15 | SUCCESS | 831 volumes"
│
↓
Reload Cache in Memory
(next classification request)
Database Design¶
Schema: Project Model¶
File: app/models.py (lines 224-233)
Fields Added:
class Project(db.Model):
# ... existing fields
# Airspace Classification Fields
airspace_classification = db.Column(db.String(1), nullable=True)
airspace_name = db.Column(db.String(255), nullable=True)
airspace_designation = db.Column(db.String(10), nullable=True)
airspace_controlled = db.Column(db.Boolean, nullable=True)
Field Specifications:
| Field | Type | Constraints | Purpose | Example Values |
|---|---|---|---|---|
airspace_classification |
VARCHAR(1) | Nullable | UK airspace class | 'A', 'B', 'C', 'D', 'E', 'G', NULL |
airspace_name |
VARCHAR(255) | Nullable | Official airspace name | 'LONDON HEATHROW CTR', 'MANCHESTER TMA' |
airspace_designation |
VARCHAR(10) | Nullable | Airspace type code | 'FRZ', 'RPZ', 'CTR', 'TMA', NULL |
airspace_controlled |
BOOLEAN | Nullable | Requires ATC clearance | True (A-D), False (E,G), NULL |
Migration:
# Migration file: migrations/versions/xxxxx_add_airspace_fields.py
def upgrade():
op.add_column('project',
sa.Column('airspace_classification', sa.String(1), nullable=True))
op.add_column('project',
sa.Column('airspace_name', sa.String(255), nullable=True))
op.add_column('project',
sa.Column('airspace_designation', sa.String(10), nullable=True))
op.add_column('project',
sa.Column('airspace_controlled', sa.Boolean(), nullable=True))
def downgrade():
op.drop_column('project', 'airspace_controlled')
op.drop_column('project', 'airspace_designation')
op.drop_column('project', 'airspace_name')
op.drop_column('project', 'airspace_classification')
Why Nullable? - Existing projects created before feature deployment - Graceful degradation if classification fails - "Unknown" classification stored as NULL (not as string 'Unknown')
Design Choice: Denormalized vs Normalized¶
Option A: Denormalized (Current Implementation)
class Project(db.Model):
airspace_classification = db.Column(db.String(1))
airspace_name = db.Column(db.String(255))
airspace_designation = db.Column(db.String(10))
airspace_controlled = db.Column(db.Boolean)
Option B: Normalized (Not Implemented)
class AirspaceClassification(db.Model):
id = db.Column(db.Integer, primary_key=True)
latitude = db.Column(db.Float)
longitude = db.Column(db.Float)
classification = db.Column(db.String(1))
name = db.Column(db.String(255))
# ...
class Project(db.Model):
airspace_id = db.Column(db.Integer, db.ForeignKey('airspace_classification.id'))
Rationale for Option A (Denormalized):
| Criterion | Denormalized (A) | Normalized (B) |
|---|---|---|
| Uniqueness | Each project has unique coordinates | Would need composite key (lat, lon, alt) |
| Historical Data | Classification at time of creation preserved | Changes to lookup table affect historical records |
| Performance | No JOIN required for project display | Requires JOIN for every project query |
| Simplicity | Straightforward schema, matches existing pattern | Additional table, more complex queries |
| Data Volume | 4 fields × N projects | Potentially millions of lookup rows |
| Existing Pattern | Matches urban_rural_classification field |
Inconsistent with existing design |
Decision: Option A (Denormalized)
- Preserves historical snapshot (regulatory compliance)
- Simpler queries (no JOIN overhead)
- Consistent with existing urban_rural_classification pattern
- Acceptable data redundancy (4 fields per project)
Deployment Strategy¶
Initial Cache Setup¶
Script: scripts/init_airspace_cache.py
Purpose: Download initial airspace data before first app run
Usage:
python scripts/init_airspace_cache.py
Integration with Docker:
# Dockerfile entrypoint
#!/bin/bash
python scripts/init_airspace_cache.py || echo "Warning: Initial cache setup failed"
flask db upgrade
gunicorn run:app
Graceful Failure: - If download fails: App starts with no cache (classifications return "Unknown") - If parse fails: Log error, continue startup - Scheduler will retry next day at 2 AM
Docker Volume Configuration¶
File: docker-compose.yml
services:
web:
volumes:
- ./airspace_cache:/app/airspace_cache
- ./uploads:/app/uploads
Persistence Requirements:
- Critical: airspace_cache/ must persist across container restarts
- Reason: Avoid re-downloading 5-10 MB dataset on every restart
- Lifecycle: Cache survives container updates, only deleted on explicit docker-compose down -v
Scheduler Initialisation (Gunicorn Multi-Worker)¶
Challenge: Gunicorn spawns multiple worker processes Problem: Each worker would start its own scheduler (duplicate jobs)
Solution: Only initialise scheduler in worker 0
File: app/__init__.py
# Detect worker ID
worker_id = os.getenv('SERVER_SOFTWARE', '')
# Only initialise scheduler in first worker (or non-Gunicorn environments)
if 'gunicorn' not in worker_id.lower() or worker_id.endswith('0'):
from app.utils.airspace_scheduler import AirspaceUpdateScheduler
AirspaceUpdateScheduler.init_scheduler()
Worker Detection: - Development (Flask dev server): No worker ID → Initialise - Production (Gunicorn): Check worker number → Only worker 0 initialises - Background thread safe across worker processes
Testing Strategy¶
Unit Tests¶
File: tests/unit/test_airspace_api.py
Test Classes: 4
Test Count: 30
Coverage: 1. AirspaceVolume Dataclass (2 tests) - Object creation - String representation
- UKAirspaceAPI (14 tests)
- Class priority ordering
- Controlled airspace detection (A-D)
- Most restrictive selection logic
- Classification with mocked cache
- Cache status retrieval
-
Exception handling
-
NAATSDataDownloader (12 tests)
- AIRAC cycle calculation (edge cases: leap years, past/future dates)
- Dataset URL building
- HTTP HEAD request mocking (exists/not found/timeout)
- Cache file reading
- Vertical limit extraction (various formats)
-
Circle to polygon conversion (different units)
-
Integration Workflow (2 tests)
- Full classification workflow (controlled airspace)
- Full classification workflow (uncontrolled airspace)
Mocking Strategy:
@patch('app.utils.airspace_api.UKAirspaceAPI._load_cache_if_needed')
@patch('app.utils.airspace_api.UKAirspaceAPI._query_airspace')
def test_get_classification_frz_controlled(mock_query, mock_load):
mock_load.return_value = True
polygon = Polygon([(0, 0), (1, 0), (1, 1), (0, 1)])
volume = AirspaceVolume(
name="LONDON HEATHROW RWY 27L",
designation="RPZ",
classification="D",
lower_limit="0 FT",
upper_limit="2000 FT",
geometry=polygon
)
mock_query.return_value = [volume]
result = UKAirspaceAPI.get_classification(51.468, -0.4117)
assert result['success'] is True
assert result['airspace_class'] == 'D'
assert result['controlled'] is True
Integration Tests¶
File: tests/integration/test_airspace_integration.py
Test Count: 18
Test Coverage:
- Project Location with Airspace (3 tests)
- Controlled airspace detection (session storage)
- Uncontrolled airspace detection
-
API failure handling (fallback to "Unknown")
-
Project Details Page Display (2 tests)
- Controlled airspace warning display (red badge)
-
Uncontrolled airspace display (green badge)
-
Viability Page Pre-population (2 tests)
- Controlled airspace field pre-fill
-
Uncontrolled airspace field pre-fill
-
Project Storage (2 tests)
- Database storage of airspace data
-
Unknown airspace graceful handling
-
Admin Airspace Status (4 tests)
- Admin page accessibility
- Regular user access denial (403)
- Cache info display
-
Manual refresh trigger
-
Session Cleanup (2 tests)
- New location clears previous airspace data
-
Project creation clears session data
-
Error Handling (1 test)
- Project creation without airspace data
Example Integration Test:
@pytest.mark.integration
def test_new_project_location_with_controlled_airspace(authenticated_client, db_session):
polygon = Polygon([(-0.42, 51.46), (-0.41, 51.46), (-0.41, 51.47), (-0.42, 51.47)])
frz_volume = AirspaceVolume(
name="LONDON HEATHROW RWY 27L",
designation="RPZ",
classification="D",
lower_limit="0 FT",
upper_limit="2000 FT",
geometry=polygon
)
with patch('app.utils.airspace_api.UKAirspaceAPI._query_airspace', return_value=[frz_volume]):
response = authenticated_client.post('/create_project/location', data={
'latitude': '51.468',
'longitude': '-0.4117',
'map_location': '{"markers":[{"lat":51.468,"lng":-0.4117}]}'
}, follow_redirects=False)
assert response.status_code == 302
assert '/create_project/details' in response.location
with authenticated_client.session_transaction() as sess:
assert sess.get('airspace_classification') == 'D'
assert sess.get('airspace_controlled') is True
Manual Test Locations¶
Test Locations: 1. London Heathrow (51.4700, -0.4543) → Class D CTR 2. Edinburgh Airport (55.9500, -3.3725) → Class D CTR 3. Scottish Highlands (57.5000, -5.0000) → Class G 4. Central London (51.5074, -0.1278) → Class D TMA
Manual Test Procedure: 1. Navigate to /create_project 2. Enter test coordinates 3. Verify classification displayed in details page 4. Check viability form pre-population 5. Complete project creation 6. Verify database storage
Security Considerations¶
Input Validation¶
Coordinates: - Latitude: -90 to +90 (reject out of range) - Longitude: -180 to +180 (reject out of range) - Sanitize before spatial query (prevent injection)
Example Validation:
def validate_coordinates(lat, lon):
if not (-90 <= lat <= 90):
raise ValueError('Invalid latitude')
if not (-180 <= lon <= 180):
raise ValueError('Invalid longitude')
return float(lat), float(lon)
File System Access¶
Cache Directory:
- Path: Configured via environment variable AIRSPACE_CACHE_DIR
- Permissions: Read/write only by app user (not world-readable)
- Validation: Reject paths with .. (directory traversal)
Example:
# Reject dangerous paths
if '..' in cache_path or cache_path.startswith('/'):
raise SecurityError('Invalid cache path')
External API Timeouts¶
NATS Downloads: - Timeout: 30 seconds (prevent indefinite hangs) - Retry: Once on timeout (transient network issues) - Fallback: Use stale cache if download fails
Example:
try:
response = requests.get(url, timeout=30, stream=True)
except requests.Timeout:
logger.warning('Download timeout - retrying once')
response = requests.get(url, timeout=30, stream=True)
except requests.RequestException as e:
logger.error(f'Download failed: {e}')
# Use stale cache
Cache Integrity¶
Validation: - JSON structure validation on load - Geometry validation (valid Shapely objects) - Delete corrupted cache, re-download
Example:
try:
with open(cache_path, 'r') as f:
cache_data = json.load(f)
# Validate structure
if 'volumes' not in cache_data:
raise ValueError('Invalid cache structure')
# Validate geometries
for vol in cache_data['volumes']:
geometry = shape(vol['geometry'])
if not geometry.is_valid:
logger.warning(f"Invalid geometry in {vol['name']}")
except (json.JSONDecodeError, ValueError) as e:
logger.error(f'Cache corrupted: {e}')
os.remove(cache_path)
# Trigger re-download
Performance Considerations¶
Response Time Targets¶
| Operation | Target | Current Performance |
|---|---|---|
| Classification Query (with cache) | < 500ms | ~100-200ms |
| Classification Query (first call, load cache) | < 3s | ~1-2s |
| XML Parse (full dataset) | < 15s | ~8-12s |
| JSON Cache Load | < 3s | ~1-2s |
| Admin Page Load | < 1s | ~200-500ms |
Cache Load Optimization¶
Lazy Loading: - Don't load on app startup (slows init) - Load on first classification request - Keep in memory for subsequent requests
Prepared Geometries (if needed): - Shapely PreparedGeometry for 3-4x speedup - Trade-off: 2x memory usage - Current performance acceptable without preparation
Database Query Optimization¶
Project Display: - No JOIN required (denormalized design) - Simple SELECT with WHERE clause - Index on project.id (primary key)
Avoid N+1 Queries: - Not applicable (no relationships to airspace table)
Future Enhancements¶
1. Multiple Altitude Queries¶
Current: Single query at 400 feet AGL Enhancement: Query at multiple altitudes (50, 100, 200, 400 feet) Benefit: Detect low-altitude FRZ zones (0-400 ft) vs higher airspace
2. Historical Airspace Data¶
Current: Only current AIRAC cycle Enhancement: Store previous cycles (6-12 months) Benefit: Audit historical projects against airspace at time of creation
3. Airspace Change Notifications¶
Current: Silent background updates Enhancement: Email notifications when airspace changes at saved project locations Benefit: Proactive safety alerts for repeat flight locations
4. Enhanced Admin Analytics¶
Current: Basic cache status display Enhancement: - Airspace usage statistics (most common classes) - Geographic heatmap of project locations - Controlled airspace frequency reports
5. Flight Path Validation¶
Current: Point classification only Enhancement: Validate entire flight path (start → end) Benefit: Detect airspace violations along route, not just at takeoff point
References¶
Internal Documentation¶
- NATS Airspace API Guide:
docs/api/nats-airspace-api.md - Scottish Spatial Data API:
docs/api/scottish-spatial-data-api.md - Project Creation Workflow:
CLAUDE.md(lines 140-180)
External Standards¶
- AIXM 5.1 Specification: aixm.aero
- NATS UK Datasets: nats-uk.ead-it.com
- UK Drone Regulations: CAA CAP 722
- AIRAC Cycles: EUROCONTROL AIS
Python Libraries¶
- lxml: XML parsing - lxml.de
- Shapely: Spatial operations - shapely.readthedocs.io
- APScheduler: Background jobs - apscheduler.readthedocs.io
Version History¶
Version 1.1 - 2025-12-29
- Bug Fix: Added CircleByCenterPoint geometry support
- Issue: General airport FRZ zones not detected (only RPZ zones)
- Root Cause: Parser only handled PolygonPatch and ArcByCenterPoint, not CircleByCenterPoint
- Impact: Edinburgh FRZ, Heathrow FRZ, and ~200 other circular FRZ zones now detected
- Change: Added CircleByCenterPoint extraction in _extract_geometry() method
- Units: Added support for nmi_i and [nmi_i] nautical mile unit formats
- Testing: Validated with Cammo Nature Reserve (55.9574, -3.3232) - now correctly returns Class D FRZ
- Volume Count: Increased from 827 to 1027 volumes (200 additional FRZ zones)
Version 1.0 - 2025-01-15 - Initial design documentation - Covers implementation through testing phase - 48 tests passing (30 unit, 18 integration) - Complete API and design documentation
Appendix: File Summary¶
New Files Created¶
- app/utils/airspace_api.py (822 lines)
- UKAirspaceAPI class
- NAATSDataDownloader class
-
AirspaceVolume dataclass
-
app/utils/airspace_scheduler.py (215 lines)
- AirspaceUpdateScheduler class
-
Background job configuration
-
app/templates/admin_panel/airspace_status.html (120 lines)
- Admin interface for cache status
-
AIRAC cycle information display
-
tests/unit/test_airspace_api.py (342 lines)
- 30 unit tests
-
4 test classes
-
tests/integration/test_airspace_integration.py (477 lines)
- 18 integration tests
-
Full workflow coverage
-
scripts/init_airspace_cache.py (85 lines)
- Initial deployment script
-
Cache setup automation
-
docs/api/nats-airspace-api.md (524 lines)
- Comprehensive API documentation
-
AIXM XML structure guide
-
docs/design/airspace-classification-design.md (This document)
- Architecture and design decisions
- Implementation details
Modified Files¶
- requirements.txt - Added 4 dependencies
- app/models.py - Added 4 airspace fields to Project
- app/routes_new_project.py - Integrated airspace classification
- app/templates/create_project/new_project_details.html - Added airspace panel
- app/templates/create_project/new_project_viability.html - Pre-populate field
- app/routes_admin.py - Added 2 admin routes
- app/templates/admin_panel/admin_base.html - Added navigation link
- app/init.py - Initialise scheduler
- docker-compose.yml - Added airspace_cache volume
- .env - Added AIRSPACE_CACHE_DIR configuration
Database Changes¶
Migration: migrations/versions/xxxxx_add_airspace_fields.py
- Added 4 columns to Project table
- All nullable for backward compatibility
End of Design Document