"""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) @classmethod def build_query(self, area: tuple, osm_types: osm_types, selector: str, conditions=[], out='center') -> str: """ 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) -> ET: """ 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 def get_base_info(elem: ET.Element, osm_type: osm_types, with_name=False) : """ Extracts base information (coordinates, OSM ID, and optionally a name) from an OSM element. This function retrieves the latitude and longitude coordinates, OSM ID, and optionally the name of a given OpenStreetMap (OSM) element. It handles different OSM types (e.g., 'node', 'way') by extracting coordinates either directly or from a center tag, depending on the element type. Args: elem (ET.Element): The XML element representing the OSM entity. osm_type (str): The type of the OSM entity (e.g., 'node', 'way'). If 'node', the coordinates are extracted directly from the element; otherwise, from the 'center' tag. with_name (bool): Whether to extract and return the name of the element. If True, it attempts to find the 'name' tag within the element and return its value. Defaults to False. Returns: tuple: A tuple containing: - osm_id (str): The OSM ID of the element. - coords (tuple): A tuple of (latitude, longitude) coordinates. - name (str, optional): The name of the element if `with_name` is True; otherwise, not included. """ # 1. extract coordinates if osm_type != 'node' : center = elem.find('center') lat = float(center.get('lat')) lon = float(center.get('lon')) else : lat = float(elem.get('lat')) lon = float(elem.get('lon')) coords = tuple((lat, lon)) # 2. Extract OSM id osm_id = elem.get('id') # 3. Extract name if specified and return if with_name : name = elem.find("tag[@k='name']").get('v') if elem.find("tag[@k='name']") is not None else None return osm_id, coords, name else : return osm_id, coords