From b9356dc4eefd47666c17efb80176005d395e9c93 Mon Sep 17 00:00:00 2001
From: Helldragon67 <kilian.scheidecker@orange.fr>
Date: Thu, 23 Jan 2025 11:28:41 +0100
Subject: [PATCH] linting
---
backend/src/overpass/caching_strategy.py | 72 ++++++++++++---
backend/src/overpass/overpass.py | 9 +-
backend/src/sandbox/overpass_test.py | 4 +-
backend/src/utils/cluster_manager.py | 26 +++---
backend/src/utils/landmarks_manager.py | 32 +++----
backend/src/utils/optimizer.py | 6 +-
backend/src/utils/refiner.py | 7 +-
backend/src/utils/toilets_manager.py | 108 +++++++++++++++--------
8 files changed, 173 insertions(+), 91 deletions(-)
diff --git a/backend/src/overpass/caching_strategy.py b/backend/src/overpass/caching_strategy.py
index e53b228..02f6b0a 100644
--- a/backend/src/overpass/caching_strategy.py
+++ b/backend/src/overpass/caching_strategy.py
@@ -1,3 +1,4 @@
+"""Module defining the caching strategy for overpass requests."""
import os
import xml.etree.ElementTree as ET
import hashlib
@@ -15,18 +16,41 @@ def get_cache_key(query: str) -> str:
class CachingStrategyBase:
+ """
+ Base class for implementing caching strategies.
+
+ This class defines the structure for a caching strategy with basic methods
+ that must be implemented by subclasses. Subclasses should define how to
+ retrieve, store, and close the cache.
+ """
def get(self, key):
+ """Retrieve the cached data associated with the provided key."""
raise NotImplementedError('Subclass should implement get')
- def set(self, key, data):
+ def set(self, key, value):
+ """Store data in the cache with the specified key."""
raise NotImplementedError('Subclass should implement set')
def close(self):
- pass
+ """Clean up or close any resources used by the caching strategy."""
-# For later use if xml does not suit well
class JSONCache(CachingStrategyBase):
+ """
+ A caching strategy that stores and retrieves data in JSON format.
+
+ This class provides methods to cache data as JSON files in a specified directory.
+ The directory is automatically suffixed with '_JSON' to distinguish it from other
+ caching strategies. The data is stored and retrieved using JSON serialization.
+
+ Args:
+ cache_dir (str): The base directory where JSON cache files will be stored.
+ Defaults to 'OSM_CACHE_DIR' with a '_JSON' suffix.
+
+ Methods:
+ get(key): Retrieve cached data from a JSON file associated with the given key.
+ set(key, value): Store data in a JSON file with the specified key.
+ """
def __init__(self, cache_dir=OSM_CACHE_DIR):
# Add the class name as a suffix to the directory
self._cache_dir = f'{cache_dir}_JSON'
@@ -39,16 +63,31 @@ class JSONCache(CachingStrategyBase):
def get(self, key):
filename = self._filename(key)
if os.path.exists(filename):
- with open(filename, 'r') as file:
+ with open(filename, 'r', encoding='utf-8') as file:
return ujson.load(file)
return None
def set(self, key, value):
- with open(self._filename(key), 'w') as file:
+ with open(self._filename(key), 'w', encoding='utf-8') as file:
ujson.dump(value, file)
class XMLCache(CachingStrategyBase):
+ """
+ A caching strategy that stores and retrieves data in XML format.
+
+ This class provides methods to cache data as XML files in a specified directory.
+ The directory is automatically suffixed with '_XML' to distinguish it from other
+ caching strategies. The data is stored and retrieved using XML serialization.
+
+ Args:
+ cache_dir (str): The base directory where XML cache files will be stored.
+ Defaults to 'OSM_CACHE_DIR' with a '_XML' suffix.
+
+ Methods:
+ get(key): Retrieve cached data from a XML file associated with the given key.
+ set(key, value): Store data in a XML file with the specified key.
+ """
def __init__(self, cache_dir=OSM_CACHE_DIR):
# Add the class name as a suffix to the directory
self._cache_dir = f'{cache_dir}_XML'
@@ -84,13 +123,22 @@ class XMLCache(CachingStrategyBase):
class CachingStrategy:
- __strategy = XMLCache() # Default caching strategy
+ """
+ A class to manage different caching strategies.
- # Dictionary to map string identifiers to caching strategy classes
+ This class provides an interface to switch between different caching strategies
+ (e.g., XMLCache, JSONCache) dynamically. It allows caching data in different formats,
+ depending on the strategy being used. By default, it uses the XMLCache strategy.
+
+ Attributes:
+ __strategy (CachingStrategyBase): The currently active caching strategy.
+ __strategies (dict): A mapping between strategy names (as strings) and their corresponding
+ classes, allowing dynamic selection of caching strategies.
+ """
+ __strategy = XMLCache() # Default caching strategy
__strategies = {
'XML': XMLCache,
'JSON': JSONCache,
- # Add more strategies here if needed
}
@classmethod
@@ -105,13 +153,13 @@ class CachingStrategy:
# If a previous strategy exists, close it
if cls.__strategy:
cls.__strategy.close()
-
+
# Retrieve the strategy class based on the strategy name
strategy_class = cls.__strategies.get(strategy_name)
-
+
if not strategy_class:
raise ValueError(f"Unknown caching strategy: {strategy_name}")
-
+
# Instantiate the new strategy with the provided arguments
cls.__strategy = strategy_class(**kwargs)
return cls.__strategy
@@ -129,5 +177,3 @@ class CachingStrategy:
if not cls.__strategy:
raise RuntimeError("Caching strategy has not been set.")
cls.__strategy.set(key, value)
-
-
diff --git a/backend/src/overpass/overpass.py b/backend/src/overpass/overpass.py
index f61eac8..54c5ce9 100644
--- a/backend/src/overpass/overpass.py
+++ b/backend/src/overpass/overpass.py
@@ -1,3 +1,4 @@
+"""Module allowing connexion to overpass api and fectch data from OSM."""
from typing import Literal, List
import urllib
import json
@@ -9,8 +10,8 @@ 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'):
+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.
@@ -62,7 +63,7 @@ def build_query(area: tuple, element_types: ElementTypes, selector: str,
return query
-def send_overpass_query(query: str) -> dict:
+def send_query(query: str) -> dict:
"""
Sends the Overpass QL query to the Overpass API and returns the parsed JSON response.
@@ -96,7 +97,7 @@ def send_overpass_query(query: str) -> dict:
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
diff --git a/backend/src/sandbox/overpass_test.py b/backend/src/sandbox/overpass_test.py
index a3e5c4e..b0aefea 100644
--- a/backend/src/sandbox/overpass_test.py
+++ b/backend/src/sandbox/overpass_test.py
@@ -122,7 +122,7 @@ def dict_to_selector_list(d: dict) -> list:
return return_list
-def send_overpass_query(query: str) -> dict:
+def send_query(query: str) -> dict:
"""
Sends the Overpass QL query to the Overpass API and returns the parsed JSON response.
@@ -280,7 +280,7 @@ for sel in dict_to_selector_list(amenity_selector):
out='center')
print(query + '\n')
- root = send_overpass_query(query)
+ root = send_query(query)
landmarks += parse_result(root, 'nature')
diff --git a/backend/src/utils/cluster_manager.py b/backend/src/utils/cluster_manager.py
index 156e82a..ec8c2c7 100644
--- a/backend/src/utils/cluster_manager.py
+++ b/backend/src/utils/cluster_manager.py
@@ -5,11 +5,9 @@ from typing import Literal
import numpy as np
from sklearn.cluster import DBSCAN
from pydantic import BaseModel
-# from OSMPythonTools.overpass import Overpass, overpassQueryBuilder
-# from OSMPythonTools.cachingStrategy import CachingStrategy, JSON
-from ..overpass.overpass import build_query, send_overpass_query
-from ..overpass.caching_strategy import CachingStrategy
+from ..overpass.overpass import build_query, send_query
+from ..overpass.caching_strategy import CachingStrategy
from ..structs.landmark import Landmark
from .get_time_distance import get_distance
from ..constants import OSM_CACHE_DIR
@@ -81,8 +79,6 @@ class ClusterManager:
Args:
bbox: The bounding box coordinates (around:radius, center_lat, center_lon).
"""
- # self.overpass = Overpass()
- # CachingStrategy.use(JSON, cacheDir=OSM_CACHE_DIR)
CachingStrategy.use('XML', cache_dir=OSM_CACHE_DIR)
self.cluster_type = cluster_type
@@ -94,6 +90,8 @@ class ClusterManager:
osm_types = ['way']
sel = '"historic"~"^(monument|building|yes)$"'
out = 'ids center'
+ else :
+ raise NotImplementedError("Please choose only an available option for cluster detection")
# Initialize the points for cluster detection
query = build_query(
@@ -105,25 +103,25 @@ class ClusterManager:
self.logger.debug(f"Cluster query: {query}")
try:
- result = send_overpass_query(query)
+ result = send_query(query)
except Exception as e:
self.logger.error(f"Error fetching landmarks: {e}")
if result is None :
self.logger.error(f"Error fetching {cluster_type} clusters, overpass query returned None.")
self.valid = False
-
+
else :
points = []
for osm_type in osm_types :
for elem in result.findall(osm_type):
- center = elem.find('center')
if osm_type != 'node' :
+ center = elem.find('center')
lat = float(center.get('lat'))
lon = float(center.get('lon'))
points.append(tuple((lat, lon)))
-
+
else :
lat = float(elem.get('lat'))
lon = float(elem.get('lon'))
@@ -136,7 +134,7 @@ class ClusterManager:
if self.cluster_type == 'shopping' and len(self.all_points) > 200 :
dbscan = DBSCAN(eps=0.00118, min_samples=15, algorithm='kd_tree') # for large cities
elif self.cluster_type == 'sightseeing' :
- dbscan = DBSCAN(eps=0.003, min_samples=10, algorithm='kd_tree') # for historic neighborhoods
+ dbscan = DBSCAN(eps=0.0025, min_samples=15, algorithm='kd_tree') # for historic neighborhoods
else :
dbscan = DBSCAN(eps=0.00075, min_samples=10, algorithm='kd_tree') # for small cities
@@ -249,7 +247,7 @@ class ClusterManager:
)
try:
- result = send_overpass_query(query)
+ result = send_query(query)
except Exception as e:
self.logger.error(f"Error fetching landmarks: {e}")
continue
@@ -270,7 +268,7 @@ class ClusterManager:
if osm_type != 'node' :
lat = float(center.get('lat'))
lon = float(center.get('lon'))
-
+
else :
lat = float(elem.get('lat'))
lon = float(elem.get('lon'))
@@ -290,7 +288,7 @@ class ClusterManager:
return Landmark(
name=new_name,
type=self.cluster_type,
- location=cluster.centroid, # TODO: use the fact the we can also recognize streets.
+ location=cluster.centroid, # later: use the fact the we can also recognize streets.
attractiveness=cluster.importance,
n_tags=0,
osm_id=osm_id,
diff --git a/backend/src/utils/landmarks_manager.py b/backend/src/utils/landmarks_manager.py
index 7b34db2..a31af62 100644
--- a/backend/src/utils/landmarks_manager.py
+++ b/backend/src/utils/landmarks_manager.py
@@ -1,14 +1,14 @@
"""Module used to import data from OSM and arrange them in categories."""
import logging
-import yaml
import xml.etree.ElementTree as ET
+import yaml
from ..structs.preferences import Preferences
from ..structs.landmark import Landmark
from .take_most_important import take_most_important
from .cluster_manager import ClusterManager
-from ..overpass.overpass import build_query, send_overpass_query
+from ..overpass.overpass import build_query, send_query
from ..overpass.caching_strategy import CachingStrategy
from ..constants import AMENITY_SELECTORS_PATH, LANDMARK_PARAMETERS_PATH, OPTIMIZER_PARAMETERS_PATH, OSM_CACHE_DIR
@@ -205,11 +205,11 @@ class LandmarkManager:
self.logger.debug(f"Query: {query}")
try:
- result = send_overpass_query(query)
+ result = send_query(query)
except Exception as e:
self.logger.error(f"Error fetching landmarks: {e}")
continue
-
+
return_list += self.xml_to_landmarks(result, landmarktype, preference_level)
self.logger.debug(f"Fetched {len(return_list)} landmarks of type {landmarktype} in {bbox}")
@@ -240,23 +240,27 @@ class LandmarkManager:
for osm_type in ['node', 'way', 'relation'] :
for elem in root.findall(osm_type):
name = elem.find("tag[@k='name']").get('v') if elem.find("tag[@k='name']") is not None else None
- center = elem.find('center')
tags = elem.findall('tag')
- # Extract the center latitude and longitude if available.
- if name is not None and center is not None:
+ if osm_type != 'node' :
+ center = elem.find('center')
lat = float(center.get('lat'))
lon = float(center.get('lon'))
coords = tuple((lat, lon))
+
else :
+ lat = float(elem.get('lat'))
+ lon = float(elem.get('lon'))
+ coords = tuple((lat, lon))
+
+ if name is None or coords is None :
continue
-
# Convert this to Landmark object
landmark = Landmark(name=name,
type=landmarktype,
location=coords,
- osm_id=elem.get('id'),
+ osm_id=elem.get('id'),
osm_type=osm_type,
attractiveness=0,
n_tags=len(tags))
@@ -277,7 +281,7 @@ class LandmarkManager:
break
# if value == 'apartments' :
# break
-
+
# Fill in the other attributes.
if key == 'image' :
landmark.image_url = value
@@ -291,7 +295,7 @@ class LandmarkManager:
landmark.name_en = value
if 'building:' in key or 'pay' in key :
landmark.n_tags -= 1
-
+
# Set the duration.
if value in ['museum', 'aquarium', 'planetarium'] :
landmark.duration = 60
@@ -304,12 +308,10 @@ class LandmarkManager:
else :
landmark.duration = 5
- else:
- # add them to cache here before setting the score
- # name should be : 'osm_type + str(osm_id) + 'json'
+ else:
self.set_landmark_score(landmark, landmarktype, preference_level)
landmarks.append(landmark)
- # self.logger.debug('new landmark added')
+
continue
return landmarks
diff --git a/backend/src/utils/optimizer.py b/backend/src/utils/optimizer.py
index b2f0c73..3bbe703 100644
--- a/backend/src/utils/optimizer.py
+++ b/backend/src/utils/optimizer.py
@@ -153,10 +153,10 @@ class Optimizer:
up_ind_y = upper_ind[1]
# Loop over the upper triangular indices, excluding diagonal elements
- for i in range(len(up_ind_x)):
- if up_ind_x[i] != up_ind_y[i]:
+ for i, up_ind in enumerate(up_ind_x):
+ if up_ind != up_ind_y[i]:
# Add (L*L-L)/2 constraints to break symmetry
- prob += (x[up_ind_x[i]*L + up_ind_y[i]] + x[up_ind_y[i]*L + up_ind_x[i]] <= 1)
+ prob += (x[up_ind*L + up_ind_y[i]] + x[up_ind_y[i]*L + up_ind] <= 1)
def init_eq_not_stay(self, prob: pl.LpProblem, x: pl.LpVariable, L: int):
diff --git a/backend/src/utils/refiner.py b/backend/src/utils/refiner.py
index 9de9709..013d9fc 100644
--- a/backend/src/utils/refiner.py
+++ b/backend/src/utils/refiner.py
@@ -6,7 +6,8 @@ from shapely import buffer, LineString, Point, Polygon, MultiPoint, concave_hull
from ..structs.landmark import Landmark
-from . import get_time_distance, take_most_important
+from .get_time_distance import get_time
+from .take_most_important import take_most_important
from .optimizer import Optimizer
from ..constants import OPTIMIZER_PARAMETERS_PATH
@@ -195,7 +196,7 @@ class Refiner :
# Step 4: Use nearest neighbor heuristic to visit all landmarks
while unvisited_landmarks:
- nearest_landmark = min(unvisited_landmarks, key=lambda lm: get_time_distance.get_time(current_landmark.location, lm.location))
+ nearest_landmark = min(unvisited_landmarks, key=lambda lm: get_time(current_landmark.location, lm.location))
path.append(nearest_landmark)
coordinates.append(nearest_landmark.location)
current_landmark = nearest_landmark
@@ -238,7 +239,7 @@ class Refiner :
if self.is_in_area(area, landmark.location) and landmark.name not in visited_names:
second_order_landmarks.append(landmark)
- return take_most_important.take_most_important(second_order_landmarks, int(self.max_landmarks_refiner*0.75))
+ return take_most_important(second_order_landmarks, int(self.max_landmarks_refiner*0.75))
# Try fix the shortest path using shapely
diff --git a/backend/src/utils/toilets_manager.py b/backend/src/utils/toilets_manager.py
index bc1082f..f27ac77 100644
--- a/backend/src/utils/toilets_manager.py
+++ b/backend/src/utils/toilets_manager.py
@@ -1,8 +1,9 @@
"""Module for finding public toilets around given coordinates."""
import logging
-from OSMPythonTools.overpass import Overpass, overpassQueryBuilder
-from OSMPythonTools.cachingStrategy import CachingStrategy, JSON
+import xml.etree.ElementTree as ET
+from ..overpass.overpass import build_query, send_query
+from ..overpass.caching_strategy import CachingStrategy
from ..structs.landmark import Toilets
from ..constants import OSM_CACHE_DIR
@@ -39,8 +40,7 @@ class ToiletsManager:
self.radius = radius
self.location = location
- self.overpass = Overpass()
- CachingStrategy.use(JSON, cacheDir=OSM_CACHE_DIR)
+ CachingStrategy.use('XML', cacheDir=OSM_CACHE_DIR)
def generate_toilet_list(self) -> list[Toilets] :
@@ -53,49 +53,83 @@ class ToiletsManager:
about the toilets found around the given coordinates.
"""
bbox = tuple((f"around:{self.radius}", str(self.location[0]), str(self.location[1])))
+ osm_types = ['node', 'way', 'relation']
toilets_list = []
- query = overpassQueryBuilder(
- bbox = bbox,
- elementType = ['node', 'way', 'relation'],
- # selector can in principle be a list already,
- # but it generates the intersection of the queries
- # we want the union
+ query = build_query(
+ area = bbox,
+ element_types = osm_types,
selector = ['"amenity"="toilets"'],
- includeCenter = True,
- out = 'center'
+ out = 'ids center tags'
)
self.logger.debug(f"Query: {query}")
try:
- result = self.overpass.query(query)
+ result = send_query(query)
except Exception as e:
self.logger.error(f"Error fetching landmarks: {e}")
return None
- for elem in result.elements():
- location = (elem.centerLat(), elem.centerLon())
-
- # handle unprecise and no-name locations
- if location[0] is None:
- location = (elem.lat(), elem.lon())
- else :
- continue
-
- toilets = Toilets(location=location)
-
- if 'wheelchair' in elem.tags().keys() and elem.tag('wheelchair') == 'yes':
- toilets.wheelchair = True
-
- if 'changing_table' in elem.tags().keys() and elem.tag('changing_table') == 'yes':
- toilets.changing_table = True
-
- if 'fee' in elem.tags().keys() and elem.tag('fee') == 'yes':
- toilets.fee = True
-
- if 'opening_hours' in elem.tags().keys() :
- toilets.opening_hours = elem.tag('opening_hours')
-
- toilets_list.append(toilets)
+ toilets_list = self.xml_to_toilets(result)
+
+ return toilets_list
+
+
+ def xml_to_toilets(self, root: ET.Element) -> list[Toilets]:
+ """
+ Parse the Overpass API result and extract landmarks.
+
+ This method processes the XML root element returned by the Overpass API and
+ extracts landmarks of types 'node', 'way', and 'relation'. It retrieves
+ relevant information such as name, coordinates, and tags, and converts them
+ into Landmark objects.
+
+ Args:
+ root (ET.Element): The root element of the XML response from Overpass API.
+ elem_type (str): The type of landmark (e.g., node, way, relation).
+
+ Returns:
+ list[Landmark]: A list of Landmark objects extracted from the XML data.
+ """
+ if root is None :
+ return []
+
+ toilets_list = []
+ for osm_type in ['node', 'way', 'relation'] :
+ for elem in root.findall(osm_type):
+ center = elem.find('center')
+
+ # Extract the center latitude and longitude if available.
+ if osm_type != 'node' :
+ lat = float(center.get('lat'))
+ lon = float(center.get('lon'))
+ location = tuple((lat, lon))
+
+ else :
+ lat = float(elem.get('lat'))
+ lon = float(elem.get('lon'))
+ location = tuple((lat, lon))
+
+ if location is None :
+ continue
+
+ toilets = Toilets(location=location)
+
+ # Extract tags as a dictionary
+ tags = {tag.get('k'): tag.get('v') for tag in elem.findall('tag')}
+
+ if 'wheelchair' in tags().keys() and tags['wheelchair'] == 'yes':
+ toilets.wheelchair = True
+
+ if 'changing_table' in tags().keys() and tags['changing_table'] == 'yes':
+ toilets.changing_table = True
+
+ if 'fee' in tags().keys() and tags['fee'] == 'yes':
+ toilets.fee = True
+
+ if 'opening_hours' in tags().keys() :
+ toilets.opening_hours = elem.tag('opening_hours')
+
+ toilets_list.append(toilets)
return toilets_list