Helldragon67
78f1dcaab4
Some checks failed
Build and deploy the backend to staging / Build and push image (pull_request) Successful in 2m28s
Run linting on the backend code / Build (pull_request) Successful in 26s
Run testing on the backend code / Build (pull_request) Failing after 1m51s
Build and deploy the backend to staging / Deploy to staging (pull_request) Successful in 24s
117 lines
4.4 KiB
Python
117 lines
4.4 KiB
Python
from typing import Literal, List
|
|
import urllib
|
|
import json
|
|
import xml.etree.ElementTree as ET
|
|
|
|
from .caching_strategy import get_cache_key, CachingStrategy
|
|
|
|
|
|
ElementTypes = List[Literal['way', 'node', 'relation']]
|
|
|
|
|
|
def build_query(area: tuple, element_types: ElementTypes, 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.
|
|
element_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(element_types, list) :
|
|
element_types = [element_types]
|
|
|
|
query = '('
|
|
|
|
# Round the radius to nearest 50 and coordinates to generate less queries
|
|
search_radius = round(area[0] / 50) * 50
|
|
loc = tuple((round(area[1], 2), round(area[2], 2)))
|
|
search_area = f"(around:{search_radius}, {str(loc[0])}, {str(loc[1])})"
|
|
|
|
if conditions :
|
|
conditions = '(if: ' + ' && '.join(conditions) + ')'
|
|
else :
|
|
conditions = ''
|
|
|
|
for elem in element_types :
|
|
query += elem + '[' + selector + ']' + conditions + search_area + ';'
|
|
|
|
query += ');' + f'out {out};'
|
|
|
|
return query
|
|
|
|
|
|
def send_overpass_query(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 = CachingStrategy.get(cache_key)
|
|
if cached_response is not None :
|
|
print("Cache hit!")
|
|
return cached_response
|
|
|
|
# Define the Overpass API endpoint
|
|
overpass_url = "https://overpass-api.de/api/interpreter"
|
|
|
|
# Prepare the data to be sent as POST request, encoded as bytes
|
|
data = urllib.parse.urlencode({'data': query}).encode('utf-8')
|
|
|
|
# Create a custom header with a User-Agent
|
|
headers = {
|
|
'User-Agent': 'Mozilla/5.0 (compatible; OverpassQuery/1.0; +http://example.com)',
|
|
}
|
|
|
|
try:
|
|
# Create a Request object with the specified URL, data, and headers
|
|
request = urllib.request.Request(overpass_url, data=data, headers=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
|
|
CachingStrategy.set(cache_key, root)
|
|
|
|
return root
|
|
|
|
except urllib.error.URLError as e:
|
|
print(f"Error connecting to Overpass API: {e}")
|
|
return None
|
|
except json.JSONDecodeError:
|
|
print("Error decoding the JSON response from Overpass API.")
|
|
return None
|