This guide explains how to create new country adapters for the GPX Route Roadworks Checker. Adapters allow the application to fetch roadworks data from different national or regional APIs and data sources.
- Architecture Overview
- Creating a New Adapter
- Base Adapter Interface
- Implementation Examples
- Data Format Guidelines
- Testing Your Adapter
- Registration and Integration
- Best Practices
- Troubleshooting
The application uses a plugin-style architecture where each country has its own adapter that:
- Fetches roadworks data from national/regional APIs or data sources
- Parses the data into a standardized format
- Caches results to improve performance
- Provides location and metadata information
Route Upload → Country Detection → Adapter Selection → Data Fetch → Parse → Cache → Display
Each adapter is responsible for:
- Knowing its coverage area (bounding box)
- Fetching data within date ranges
- Converting data to standard format
- Handling API errors gracefully
Before creating an adapter, identify your data source:
- API Endpoint: REST API, SOAP service, GraphQL
- Data Format: JSON, XML, DATEX II, CSV
- Authentication: API keys, OAuth, none
- Rate Limits: Requests per minute/hour
- Coverage: Geographic bounds
- Update Frequency: Real-time, hourly, daily
Create a new file in the adapters/ directory:
touch adapters/your_country_adapter.py# adapters/your_country_adapter.py
import requests
from datetime import datetime
from typing import List, Dict, Any, Optional
import logging
from .base_adapter import CountryAdapter
logger = logging.getLogger(__name__)
class YourCountryAdapter(CountryAdapter):
"""Your Country roadworks adapter"""
def __init__(self, cache_manager=None):
super().__init__(cache_manager)
self.api_base_url = "https://api.yourcountry.gov/roadworks"
self.timeout = 30
def get_country_code(self) -> str:
"""Return ISO country code"""
return 'YC'
def get_country_name(self) -> str:
"""Return human-readable country name"""
return 'Your Country'
def get_supported_bbox(self) -> str:
"""Return bounding box as 'min_lon,min_lat,max_lon,max_lat'"""
# Example for a fictional country
return "2.0,50.0,8.0,54.0"
def _fetch_roadworks_uncached(self, start_date: str, end_date: str, bbox: str = None) -> List[Dict[str, Any]]:
"""Fetch roadworks from your country's API"""
# Implementation goes here
pass
def parse_work_location(self, work: Dict[str, Any]) -> List[List[float]]:
"""Parse work location coordinates"""
# Implementation goes here
pass
def get_work_info(self, work: Dict[str, Any]) -> Dict[str, str]:
"""Extract work information for display"""
# Implementation goes here
passEvery adapter must implement these abstract methods:
Return the ISO 3166-1 alpha-2 country code (e.g., 'DE', 'FR', 'NL').
Return the human-readable country name for display.
Return the bounding box as a string: "min_longitude,min_latitude,max_longitude,max_latitude"
def get_supported_bbox(self) -> str:
# Germany example
return "5.98,47.30,15.02,54.98"Fetch roadworks data from your source. This is the core method that:
- Makes API requests or downloads data files
- Filters by date range and optionally by bounding box
- Returns list of roadwork dictionaries in standard format
Extract coordinates from a roadwork entry.
def parse_work_location(self, work: Dict[str, Any]) -> List[List[float]]:
# Return format: [[lat1, lon1], [lat2, lon2], ...]
coordinates = work.get('coordinates', [])
return coordinatesExtract display information from a roadwork entry.
def get_work_info(self, work: Dict[str, Any]) -> Dict[str, str]:
return {
'id': work.get('id', ''),
'description': work.get('description', ''),
'start_date': work.get('start_date', ''),
'end_date': work.get('end_date', ''),
'source': work.get('source', ''),
'owner': work.get('owner', ''),
'status': work.get('status', '')
}These methods are provided by the base class:
fetch_roadworks()- Main entry point with caching_is_in_date_range()- Date filtering helper_is_in_bbox()- Geographic filtering helper_parse_date()- Date parsing utility
def _fetch_roadworks_uncached(self, start_date: str, end_date: str, bbox: str = None) -> List[Dict[str, Any]]:
"""Fetch from REST API returning JSON"""
try:
params = {
'start_date': start_date,
'end_date': end_date,
'format': 'json'
}
if bbox:
params['bbox'] = bbox
response = requests.get(
f"{self.api_base_url}/roadworks",
params=params,
timeout=self.timeout,
headers={'Accept': 'application/json'}
)
response.raise_for_status()
data = response.json()
roadworks = []
for item in data.get('roadworks', []):
work = self._convert_api_response(item)
if work:
roadworks.append(work)
logger.info(f"Fetched {len(roadworks)} roadworks from Your Country API")
return roadworks
except requests.RequestException as e:
logger.error(f"Error fetching roadworks: {e}")
return []
def _convert_api_response(self, item: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""Convert API response to standard format"""
try:
# Extract coordinates
location = item.get('location', {})
coordinates = []
if 'coordinates' in location:
coords = location['coordinates']
if isinstance(coords[0], list):
# Multiple coordinates
coordinates = [[coord[1], coord[0]] for coord in coords] # [lat, lon]
else:
# Single coordinate
coordinates = [[coords[1], coords[0]]]
return {
'id': str(item.get('id', '')),
'description': item.get('description', 'Roadworks'),
'start_date': self._parse_date(item.get('start_date', '')),
'end_date': self._parse_date(item.get('end_date', '')),
'coordinates': coordinates,
'owner': item.get('authority', ''),
'status': item.get('status', 'active'),
'source': 'Your Country API',
'raw_data': item
}
except Exception as e:
logger.debug(f"Error converting roadwork item: {e}")
return Noneimport xml.etree.ElementTree as ET
def _fetch_roadworks_uncached(self, start_date: str, end_date: str, bbox: str = None) -> List[Dict[str, Any]]:
"""Fetch DATEX II XML data"""
try:
response = requests.get(
f"{self.api_base_url}/datex2/roadworks.xml",
timeout=self.timeout,
headers={'Accept': 'application/xml'}
)
response.raise_for_status()
# Parse XML
root = ET.fromstring(response.content)
roadworks = self._parse_datex_xml(root, start_date, end_date, bbox)
logger.info(f"Parsed {len(roadworks)} roadworks from DATEX II XML")
return roadworks
except (requests.RequestException, ET.ParseError) as e:
logger.error(f"Error fetching/parsing XML: {e}")
return []
def _parse_datex_xml(self, root: ET.Element, start_date: str, end_date: str, bbox: str = None) -> List[Dict[str, Any]]:
"""Parse DATEX II XML structure"""
roadworks = []
# Extract namespaces
namespaces = self._extract_namespaces(root)
# Find situation elements
for situation in root.findall('.//situation', namespaces):
try:
work = self._parse_situation_element(situation, namespaces)
if work and self._is_in_date_range(work, start_date, end_date):
if not bbox or self._is_in_bbox(work, bbox):
roadworks.append(work)
except Exception as e:
logger.debug(f"Error parsing situation: {e}")
continue
return roadworksimport gzip
import csv
from io import StringIO
def _fetch_roadworks_uncached(self, start_date: str, end_date: str, bbox: str = None) -> List[Dict[str, Any]]:
"""Download and parse compressed CSV file"""
try:
# Download compressed file
response = requests.get(
f"{self.api_base_url}/roadworks.csv.gz",
timeout=60, # Longer timeout for file download
stream=True
)
response.raise_for_status()
# Decompress and parse
csv_content = gzip.decompress(response.content).decode('utf-8')
roadworks = self._parse_csv_content(csv_content, start_date, end_date, bbox)
logger.info(f"Parsed {len(roadworks)} roadworks from CSV file")
return roadworks
except Exception as e:
logger.error(f"Error downloading/parsing CSV: {e}")
return []
def _parse_csv_content(self, csv_content: str, start_date: str, end_date: str, bbox: str = None) -> List[Dict[str, Any]]:
"""Parse CSV content into roadworks"""
roadworks = []
reader = csv.DictReader(StringIO(csv_content))
for row in reader:
try:
work = self._convert_csv_row(row)
if work and self._is_in_date_range(work, start_date, end_date):
if not bbox or self._is_in_bbox(work, bbox):
roadworks.append(work)
except Exception as e:
logger.debug(f"Error parsing CSV row: {e}")
continue
return roadworksEvery adapter should return roadworks in this standardized format:
{
'id': str, # Unique identifier
'description': str, # Human-readable description
'start_date': str, # ISO date format (YYYY-MM-DD) or empty
'end_date': str, # ISO date format (YYYY-MM-DD) or empty
'coordinates': List[List[float]], # [[lat1, lon1], [lat2, lon2], ...]
'owner': str, # Authority/organization responsible
'status': str, # active, planned, completed, etc.
'source': str, # Data source name
'raw_data': Dict[str, Any] # Original data for debugging/extension
}Coordinates should always be in [latitude, longitude] format:
# Single point
coordinates = [[52.5200, 13.4050]]
# Multiple points (line/polygon)
coordinates = [
[52.5200, 13.4050], # Point 1
[52.5210, 13.4060], # Point 2
[52.5220, 13.4070] # Point 3
]
# For polygon data, extract representative points or centroidUse the inherited _parse_date() method for consistent date parsing:
def _parse_date(self, date_str: str) -> str:
"""Parse various date formats to YYYY-MM-DD"""
# Handles: ISO dates, European dates, timestamps, etc.
# Returns empty string if parsing failsCreate tests in tests/test_adapters/test_your_country.py:
import unittest
from unittest.mock import patch, Mock
from adapters.your_country_adapter import YourCountryAdapter
class TestYourCountryAdapter(unittest.TestCase):
def setUp(self):
self.adapter = YourCountryAdapter()
def test_country_info(self):
"""Test basic country information"""
self.assertEqual(self.adapter.get_country_code(), 'YC')
self.assertEqual(self.adapter.get_country_name(), 'Your Country')
self.assertIsInstance(self.adapter.get_supported_bbox(), str)
@patch('requests.get')
def test_fetch_roadworks_success(self, mock_get):
"""Test successful roadworks fetch"""
mock_response = Mock()
mock_response.json.return_value = {
'roadworks': [
{
'id': '123',
'description': 'Test roadwork',
'location': {'coordinates': [13.4050, 52.5200]},
'start_date': '2025-01-01',
'end_date': '2025-02-01'
}
]
}
mock_response.raise_for_status.return_value = None
mock_get.return_value = mock_response
roadworks = self.adapter._fetch_roadworks_uncached('2025-01-01', '2025-12-31')
self.assertEqual(len(roadworks), 1)
self.assertEqual(roadworks[0]['id'], '123')
self.assertEqual(roadworks[0]['coordinates'], [[52.5200, 13.4050]])
@patch('requests.get')
def test_fetch_roadworks_api_error(self, mock_get):
"""Test API error handling"""
mock_get.side_effect = requests.RequestException("API Error")
roadworks = self.adapter._fetch_roadworks_uncached('2025-01-01', '2025-12-31')
self.assertEqual(len(roadworks), 0)
def test_parse_work_location(self):
"""Test coordinate parsing"""
work = {
'coordinates': [[52.5200, 13.4050], [52.5210, 13.4060]]
}
coords = self.adapter.parse_work_location(work)
self.assertEqual(len(coords), 2)
self.assertEqual(coords[0], [52.5200, 13.4050])
if __name__ == '__main__':
unittest.main()Test your adapter manually:
# Test script: test_adapter.py
from adapters.your_country_adapter import YourCountryAdapter
def test_adapter():
adapter = YourCountryAdapter()
print(f"Country: {adapter.get_country_name()} ({adapter.get_country_code()})")
print(f"Bbox: {adapter.get_supported_bbox()}")
# Test with current date range
from datetime import datetime, timedelta
start = datetime.now().strftime('%Y-%m-%d')
end = (datetime.now() + timedelta(days=30)).strftime('%Y-%m-%d')
print(f"Fetching roadworks from {start} to {end}...")
roadworks = adapter.fetch_roadworks(start, end)
print(f"Found {len(roadworks)} roadworks")
if roadworks:
work = roadworks[0]
print(f"Sample work: {work['description']}")
print(f"Coordinates: {work['coordinates']}")
info = adapter.get_work_info(work)
print(f"Work info: {info}")
if __name__ == '__main__':
test_adapter()Test with the full application:
# 1. Add your adapter to the registry (see next section)
# 2. Run the Flask app
python app.py
# 3. Upload a GPX file that crosses your country
# 4. Check logs for your adapter being called:
# INFO:adapters.your_country:Fetching roadworks for Your Country...Add your adapter to adapters/__init__.py:
from .belgium_gipod import BelgiumGIPODAdapter
from .france_bison_fute import FranceBisonFuteAdapter
from .netherlands_ndw import NetherlandsNDWAdapter
from .uk_streetmanager import UKStreetManagerAdapter
from .your_country_adapter import YourCountryAdapter # Add this line
def get_all_adapters():
"""Return list of all available country adapters"""
return [
BelgiumGIPODAdapter,
FranceBisonFuteAdapter,
NetherlandsNDWAdapter,
UKStreetManagerAdapter,
YourCountryAdapter, # Add this line
]Add your country to the README.md:
### Your Country
- **Your Agency Name**: Description of data source
- API: `https://api.yourcountry.gov/roadworks`
- Coverage: National road network
- Update Frequency: Real-time updatesIf your adapter needs configuration, add to .env.example:
# Your Country API Configuration
YOUR_COUNTRY_API_KEY=your-api-key-here
YOUR_COUNTRY_BASE_URL=https://api.yourcountry.govThen use in your adapter:
import os
class YourCountryAdapter(CountryAdapter):
def __init__(self, cache_manager=None):
super().__init__(cache_manager)
self.api_key = os.getenv('YOUR_COUNTRY_API_KEY')
self.api_base_url = os.getenv('YOUR_COUNTRY_BASE_URL', 'https://api.yourcountry.gov')- Graceful Degradation: Always return empty list on errors
- Specific Exceptions: Catch specific exceptions when possible
- Logging: Use appropriate log levels (DEBUG, INFO, WARNING, ERROR)
def _fetch_roadworks_uncached(self, start_date: str, end_date: str, bbox: str = None) -> List[Dict[str, Any]]:
try:
response = requests.get(url, timeout=self.timeout)
response.raise_for_status()
return self._parse_response(response)
except requests.Timeout:
logger.warning(f"Timeout fetching roadworks from {self.get_country_name()}")
return []
except requests.HTTPError as e:
logger.error(f"HTTP error from {self.get_country_name()} API: {e}")
return []
except requests.RequestException as e:
logger.error(f"Request error from {self.get_country_name()}: {e}")
return []
except Exception as e:
logger.error(f"Unexpected error in {self.get_country_name()} adapter: {e}")
return []- Efficient Parsing: Parse only what you need
- Memory Management: Don't load huge datasets into memory
- Timeouts: Set appropriate timeouts for requests
- Streaming: Use streaming for large downloads
# Stream large files
response = requests.get(url, stream=True, timeout=60)
for chunk in response.iter_content(chunk_size=8192):
# Process chunk by chunk
pass- Validation: Validate coordinates and dates
- Filtering: Filter out invalid or irrelevant data
- Normalization: Normalize text descriptions
- Deduplication: Remove duplicate entries
def _validate_coordinates(self, coords: List[List[float]]) -> bool:
"""Validate coordinate format and ranges"""
for coord in coords:
if len(coord) != 2:
return False
lat, lon = coord
if not (-90 <= lat <= 90) or not (-180 <= lon <= 180):
return False
return True- API Keys: Never hardcode API keys
- Input Validation: Validate all inputs
- Rate Limiting: Respect API rate limits
- HTTPS Only: Only use secure connections
def _make_request(self, url: str, params: Dict[str, str]) -> requests.Response:
"""Make secure API request with rate limiting"""
if not url.startswith('https://'):
raise ValueError("Only HTTPS URLs are allowed")
# Add rate limiting
time.sleep(self.rate_limit_delay)
headers = {
'User-Agent': 'GPX-Roadworks-Checker/1.0',
'Accept': 'application/json'
}
if self.api_key:
headers['Authorization'] = f'Bearer {self.api_key}'
return requests.get(url, params=params, headers=headers, timeout=self.timeout)- Check API endpoint: Verify URL is correct
- Check date range: Ensure dates are in correct format
- Check bounding box: Verify bbox covers your test area
- Check authentication: Ensure API keys are valid
- Wrong coordinate order: Ensure [lat, lon] not [lon, lat]
- Wrong coordinate system: Convert to WGS84 if needed
- Invalid coordinates: Validate ranges and format
- Large datasets: Implement pagination or streaming
- Slow APIs: Increase timeout or implement retries
- Memory usage: Process data in chunks
- Multiple formats: Handle various date formats
- Timezone issues: Convert to UTC or local time consistently
- Invalid dates: Validate date ranges
- Enable debug logging:
import logging
logging.basicConfig(level=logging.DEBUG)- Test with curl:
curl -v "https://api.yourcountry.gov/roadworks?start_date=2025-01-01"-
Use browser dev tools to inspect API calls
-
Test with minimal data first, then scale up
-
Check raw_data field in returned roadworks for debugging
- Check existing adapters for similar patterns
- Review API documentation thoroughly
- Test with API provider's examples first
- Ask in project issues if you're stuck
- Test thoroughly with various date ranges and locations
- Add unit tests with good coverage
- Update documentation (README, this guide)
- Follow code style (use black, flake8)
- Check performance with realistic data volumes
- Fork the repository
- Create feature branch:
git checkout -b add-yourcountry-adapter - Implement adapter following this guide
- Add tests and documentation
- Test integration with full application
- Submit pull request with:
- Clear description
- Test results
- Documentation updates
- Example usage
## New Country Adapter: [Country Name]
### Description
Brief description of the adapter and data source.
### Data Source
- **API/URL**: https://api.example.com
- **Format**: JSON/XML/DATEX II/CSV
- **Coverage**: National/Regional
- **Update Frequency**: Real-time/Daily/Weekly
- **Authentication**: None/API Key/OAuth
### Testing
- [ ] Unit tests pass
- [ ] Integration tests pass
- [ ] Manual testing with GPX files
- [ ] Performance testing with realistic data
### Documentation
- [ ] Updated README.md
- [ ] Added example configuration
- [ ] Documented any special requirements
### Changes
- Added `YourCountryAdapter` class
- Registered adapter in `__init__.py`
- Added unit tests
- Updated documentationThis comprehensive guide should help you or other contributors create robust, reliable adapters for new countries and data sources!