Helldragon67
577ee232fc
Some checks failed
Build and deploy the backend to staging / Build and push image (pull_request) Successful in 2m4s
Run linting on the backend code / Build (pull_request) Successful in 29s
Run testing on the backend code / Build (pull_request) Failing after 4m39s
Build and deploy the backend to staging / Deploy to staging (pull_request) Successful in 24s
127 lines
5.3 KiB
Python
127 lines
5.3 KiB
Python
"""Module allowing connexion to overpass api and fectch data from OSM."""
|
|
from typing import Literal, List
|
|
import urllib
|
|
import logging
|
|
import xml.etree.ElementTree as ET
|
|
|
|
from .caching_strategy import get_cache_key, CachingStrategy
|
|
from ..constants import OSM_CACHE_DIR
|
|
|
|
logger = logging.getLogger('Overpass')
|
|
osm_types = List[Literal['way', 'node', 'relation']]
|
|
|
|
|
|
class Overpass :
|
|
"""
|
|
Overpass class to manage the query building and sending to overpass api.
|
|
The caching strategy is a part of this class and initialized upon creation of the Overpass object.
|
|
"""
|
|
def __init__(self, caching_strategy: str = 'XML', cache_dir: str = OSM_CACHE_DIR) :
|
|
"""
|
|
Initialize the Overpass instance with the url, headers and caching strategy.
|
|
"""
|
|
self.overpass_url = "https://overpass-api.de/api/interpreter"
|
|
self.headers = {'User-Agent': 'Mozilla/5.0 (compatible; OverpassQuery/1.0; +http://example.com)',}
|
|
self.caching_strategy = CachingStrategy.use(caching_strategy, cache_dir=cache_dir)
|
|
|
|
|
|
def build_query(self, area: tuple, osm_types: osm_types,
|
|
selector: str, conditions=[], out='center'):
|
|
"""
|
|
Constructs a query string for the Overpass API to retrieve OpenStreetMap (OSM) data.
|
|
|
|
Args:
|
|
area (tuple): A tuple representing the geographical search area, typically in the format
|
|
(radius, latitude, longitude). The first element is a string like "around:2000"
|
|
specifying the search radius, and the second and third elements represent
|
|
the latitude and longitude as floats or strings.
|
|
osm_types (list[str]): A list of OSM element types to search for. Must be one or more of
|
|
'Way', 'Node', or 'Relation'.
|
|
selector (str): The key or tag to filter the OSM elements (e.g., 'amenity', 'highway', etc.).
|
|
conditions (list, optional): A list of conditions to apply as additional filters for the
|
|
selected OSM elements. The conditions should be written in
|
|
the Overpass QL format, and they are combined with '&&' if
|
|
multiple are provided. Defaults to an empty list.
|
|
out (str, optional): Specifies the output type, such as 'center', 'body', or 'tags'.
|
|
Defaults to 'center'.
|
|
|
|
Returns:
|
|
str: The constructed Overpass QL query string.
|
|
|
|
Notes:
|
|
- If no conditions are provided, the query will just use the `selector` to filter the OSM
|
|
elements without additional constraints.
|
|
- The search area must always formatted as "(radius, lat, lon)".
|
|
"""
|
|
if not isinstance(conditions, list) :
|
|
conditions = [conditions]
|
|
if not isinstance(osm_types, list) :
|
|
osm_types = [osm_types]
|
|
|
|
query = '('
|
|
|
|
# Round the radius to nearest 50 and coordinates to generate less queries
|
|
if area[0] > 500 :
|
|
search_radius = round(area[0] / 50) * 50
|
|
loc = tuple((round(area[1], 2), round(area[2], 2)))
|
|
else :
|
|
search_radius = round(area[0] / 25) * 25
|
|
loc = tuple((round(area[1], 3), round(area[2], 3)))
|
|
|
|
search_area = f"(around:{search_radius}, {str(loc[0])}, {str(loc[1])})"
|
|
|
|
if conditions :
|
|
conditions = '(if: ' + ' && '.join(conditions) + ')'
|
|
else :
|
|
conditions = ''
|
|
|
|
for elem in osm_types :
|
|
query += elem + '[' + selector + ']' + conditions + search_area + ';'
|
|
|
|
query += ');' + f'out {out};'
|
|
|
|
return query
|
|
|
|
|
|
def send_query(self, query: str) -> dict:
|
|
"""
|
|
Sends the Overpass QL query to the Overpass API and returns the parsed JSON response.
|
|
|
|
Args:
|
|
query (str): The Overpass QL query to be sent to the Overpass API.
|
|
|
|
Returns:
|
|
dict: The parsed JSON response from the Overpass API, or None if the request fails.
|
|
"""
|
|
|
|
# Generate a cache key for the current query
|
|
cache_key = get_cache_key(query)
|
|
|
|
# Try to fetch the result from the cache
|
|
cached_response = self.caching_strategy.get(cache_key)
|
|
if cached_response is not None :
|
|
logger.debug("Cache hit.")
|
|
return cached_response
|
|
|
|
# Prepare the data to be sent as POST request, encoded as bytes
|
|
data = urllib.parse.urlencode({'data': query}).encode('utf-8')
|
|
|
|
try:
|
|
# Create a Request object with the specified URL, data, and headers
|
|
request = urllib.request.Request(self.overpass_url, data=data, headers=self.headers)
|
|
|
|
# Send the request and read the response
|
|
with urllib.request.urlopen(request) as response:
|
|
# Read and decode the response
|
|
response_data = response.read().decode('utf-8')
|
|
root = ET.fromstring(response_data)
|
|
|
|
# Cache the response data as an ElementTree root
|
|
self.caching_strategy.set(cache_key, root)
|
|
logger.debug("Response data added to cache.")
|
|
|
|
return root
|
|
|
|
except urllib.error.URLError as e:
|
|
raise ConnectionError(f"Error connecting to Overpass API: {e}") from e
|