switch to osmnx

backend/Dockerfile

@@ -12,5 +12,6 @@ EXPOSE 8000
 
 # Set environment variables used by the deployment. These can be overridden by the user using this image.
 ENV NUM_WORKERS=1
+ENV OSM_CACHE_DIR=/cache
 
 CMD ["pipenv", "run", "fastapi", "run", "src/main.py", '--port 8000', '--workers $NUM_WORKERS']

backend/Pipfile

@@ -7,8 +7,9 @@ name = "pypi"
 numpy = "*"
 scipy = "*"
 fastapi = "*"
-osmpythontools = "*"
 pydantic = "*"
 shapely = "*"
+osmnx = "*"
+networkx = "*"
 
 [dev-packages]

backend/src/amenities/nature.am

@@ -1,11 +0,0 @@
-'leisure'='park'
-geological
-'natural'='geyser'
-'natural'='hot_spring'
-'natural'='arch'
-'natural'='volcano'
-'natural'='stone'
-'tourism'='alpine_hut'
-'tourism'='viewpoint'
-'tourism'='zoo'
-'waterway'='waterfall'

backend/src/amenities/shopping.am

@@ -1,2 +0,0 @@
-'shop'='department_store'
-'shop'='mall'

backend/src/amenities/sightseeing.am

@@ -1,8 +0,0 @@
-'tourism'='museum'
-'tourism'='attraction'
-'tourism'='gallery'
-historic
-'amenity'='planetarium'
-'amenity'='place_of_worship'
-'amenity'='fountain'
-'water'='reflecting_pool'

backend/src/constants.py

@@ -1,9 +1,12 @@
 from pathlib import Path
-
+import os
 
 PARAMETERS_DIR = Path('src/parameters')
 AMENITY_SELECTORS_PATH = PARAMETERS_DIR / 'amenity_selectors.yaml'
 LANDMARK_PARAMETERS_PATH = PARAMETERS_DIR / 'landmark_parameters.yaml'
 OPTIMIZER_PARAMETERS_PATH = PARAMETERS_DIR / 'optimizer_parameters.yaml'
 
-OSM_CACHE_DIR = Path('cache')
+
+
+cache_dir_string = os.getenv('OSM_CACHE_DIR', './cache')
+OSM_CACHE_DIR = Path(cache_dir_string)

backend/src/landmarks_manager.py

@@ -1,150 +1,71 @@
-import math as m
-import json, os
 import yaml
+import os
+import osmnx as ox
+from shapely.geometry import Point, Polygon, LineString, MultiPolygon
 
-from typing import List, Tuple, Optional
-from OSMPythonTools.overpass import Overpass, overpassQueryBuilder
-
-import constants
 from structs.landmarks import Landmark, LandmarkType
 from structs.preferences import Preferences, Preference
+import constants
 
 
 SIGHTSEEING = LandmarkType(landmark_type='sightseeing')
 NATURE = LandmarkType(landmark_type='nature')
 SHOPPING = LandmarkType(landmark_type='shopping')
 
-
+ox.config(cache_folder=constants.OSM_CACHE_DIR)
 
 # Include the json here
 # Create a list of all things to visit given some preferences and a city. Ready for the optimizer
-def generate_landmarks(preferences: Preferences, coordinates: Tuple[float, float]) :
+def generate_landmarks(preferences: Preferences, center_coordinates: tuple[float, float]) :
     with constants.AMENITY_SELECTORS_PATH.open('r') as f:
         amenity_selectors = yaml.safe_load(f)
     
     with constants.LANDMARK_PARAMETERS_PATH.open('r') as f:
         # even though we don't use the parameters here, we already load them to avoid unnecessary io operations
         parameters = yaml.safe_load(f)
-
+        max_distance = parameters['city_bbox_side']
     L = []
 
     # List for sightseeing
-    if preferences.sightseeing.score != 0 :
+    if preferences.sightseeing.score != 0:
-        L1 = get_landmarks(amenity_selectors['sightseeing'], SIGHTSEEING, coordinates, parameters)
+        score_func = lambda loc, n_tags: int((count_elements_within_radius(loc, parameters['radius_close_to']) + n_tags * parameters['tag_coeff']) * parameters['church_coeff'])  
+        L1 = get_landmarks(amenity_selectors['sightseeing'], SIGHTSEEING, center_coordinates, max_distance, score_func)
         correct_score(L1, preferences.sightseeing)
         L += L1
     
     # List for nature
-    if preferences.nature.score != 0 :
+    if preferences.nature.score != 0:
-        L2 = get_landmarks(amenity_selectors['nature'], NATURE, coordinates, parameters)
+        score_func = lambda loc, n_tags: int((count_elements_within_radius(loc, parameters['radius_close_to']) + n_tags * parameters['tag_coeff']) * parameters['park_coeff'])  
+        L2 = get_landmarks(amenity_selectors['nature'], NATURE, center_coordinates, max_distance, score_func)
         correct_score(L2, preferences.nature)
         L += L2
     
     # List for shopping
-    if preferences.shopping.score != 0 :
+    if preferences.shopping.score != 0:
-        L3 = get_landmarks(amenity_selectors['shopping'], SHOPPING, coordinates, parameters)
+        score_func = lambda loc, n_tags: count_elements_within_radius(loc, parameters['radius_close_to']) + n_tags * parameters['tag_coeff']
+        L3 = get_landmarks(amenity_selectors['shopping'], SHOPPING, center_coordinates, max_distance, score_func)
         correct_score(L3, preferences.shopping)
         L += L3
 
-    L = remove_duplicates(L)
+    # remove duplicates
-
+    L = list(set(L))
-    return L, take_most_important(L, parameters)
+    print(len(L))
+    L_constrained = take_most_important(L, parameters['N_important'])
+    print(len(L_constrained))
+    return L, L_constrained
 
 
-"""def generate_landmarks(preferences: Preferences, city_country: str = None, coordinates: Tuple[float, float] = None) -> Tuple[List[Landmark], List[Landmark]] :
 
-    l_sights, l_nature, l_shop = get_amenities()
-    L = []
-
-    # List for sightseeing
-    if preferences.sightseeing.score != 0 :
-        L1 = get_landmarks(l_sights, SIGHTSEEING, city_country=city_country, coordinates=coordinates)
-        correct_score(L1, preferences.sightseeing)
-        L += L1
-    
-    # List for nature
-    if preferences.nature.score != 0 :
-        L2 = get_landmarks(l_nature, NATURE, city_country=city_country, coordinates=coordinates)
-        correct_score(L2, preferences.nature)
-        L += L2
-    
-    # List for shopping
-    if preferences.shopping.score != 0 :
-        L3 = get_landmarks(l_shop, SHOPPING, city_country=city_country, coordinates=coordinates)
-        correct_score(L3, preferences.shopping)
-        L += L3
-
-    return remove_duplicates(L), take_most_important(L)
-"""
-# Helper function to gather the amenities list
 # Take the most important landmarks from the list
-def take_most_important(L: List[Landmark], parameters: dict, N: int = 0) -> List[Landmark]:
+def take_most_important(landmarks: list[Landmark], n_max: int) -> list[Landmark]:
-    L_copy = []
-    L_clean = []
-    scores = [0]*len(L)
-    names = []
-    name_id = {}
 
-    for i, elem in enumerate(L) :
+    landmarks_sorted = sorted(landmarks, key=lambda x: x.attractiveness, reverse=True)
-        if elem.name not in names :
+    return landmarks_sorted[:n_max]
-            names.append(elem.name)
-            name_id[elem.name] = [i]
-            L_copy.append(elem)
-        else :
-            name_id[elem.name] += [i]
-            scores = []
-            for j in name_id[elem.name] :
-                scores.append(L[j].attractiveness)
-            best_id = max(range(len(scores)), key=scores.__getitem__)
-            t = name_id[elem.name][best_id]
-            if t == i :
-                for old in L_copy :
-                    if old.name == elem.name :
-                        old.attractiveness = L[t].attractiveness
 
-    scores = [0]*len(L_copy)
-    for i, elem in enumerate(L_copy) :
-        scores[i] = elem.attractiveness
-
-    
-    res = sorted(range(len(scores)), key = lambda sub: scores[sub])[-(parameters['N_important']-N):]
-
-    for i, elem in enumerate(L_copy) :
-        if i in res :
-            L_clean.append(elem)
-
-    return L_clean
-
-
-# Remove duplicate elements and elements with low score
-def remove_duplicates(L: List[Landmark]) -> List[Landmark] :
-    """
-    Removes duplicate landmarks based on their names from the given list.
-
-    Parameters:
-    L (List[Landmark]): A list of Landmark objects.
-
-    Returns:
-    List[Landmark]: A list of unique Landmark objects based on their names.
-    """
-
-    L_clean = []
-    names = []
-
-    for landmark in L :
-        if landmark.name in names: 
-            continue  
-
-
-        else :
-            names.append(landmark.name)
-            L_clean.append(landmark)
-    
-    return L_clean
 
 
 # Correct the score of a list of landmarks by taking into account preference settings
-def correct_score(L: List[Landmark], preference: Preference) :
+def correct_score(L: list[Landmark], preference: Preference) :
 
     if len(L) == 0 :
         return
@@ -157,195 +78,111 @@ def correct_score(L: List[Landmark], preference: Preference) :
 
 
 # Function to count elements within a certain radius of a location
-def count_elements_within_radius(coordinates: Tuple[float, float], radius: int) -> int:
+def count_elements_within_radius(point: Point, radius: int) -> int:
     
-    lat = coordinates[0]
+    center_coordinates = (point.x, point.y)
-    lon = coordinates[1]
+    try:
-
+        landmarks = ox.features_from_point(
-    alpha = (180*radius)/(6371000*m.pi)
+            center_point = center_coordinates,
-    bbox = {'latLower':lat-alpha,'lonLower':lon-alpha,'latHigher':lat+alpha,'lonHigher': lon+alpha}
+            dist = radius,
-        
+            tags = {'building': True} # this is a common tag to give an estimation of the number of elements in the area
-    # Build the query to find elements within the radius
+        )
-    radius_query = overpassQueryBuilder(bbox=[bbox['latLower'],bbox['lonLower'],bbox['latHigher'],bbox['lonHigher']],
+        return len(landmarks)
-                             elementType=['node', 'way', 'relation'])
+    except ox._errors.InsufficientResponseError:
-
+        return 0    
-    try : 
-        overpass = Overpass()
-        radius_result = overpass.query(radius_query)
-        return radius_result.countElements()
-    
-    except :
-        return None
 
 
-# Creates a bounding box around given coordinates
-def create_bbox(coordinates: Tuple[float, float], side_length: int) -> Tuple[float, float, float, float]:
-
-    lat = coordinates[0]
-    lon = coordinates[1]
-
-    # Half the side length in km (since it's a square bbox)
-    half_side_length_km = side_length / 2.0
-
-    # Convert distance to degrees
-    lat_diff = half_side_length_km / 111  # 1 degree latitude is approximately 111 km
-    lon_diff = half_side_length_km / (111 * m.cos(m.radians(lat)))  # Adjust for longitude based on latitude
-
-    # Calculate bbox
-    min_lat = lat - lat_diff
-    max_lat = lat + lat_diff
-    min_lon = lon - lon_diff
-    max_lon = lon + lon_diff
-
-    return min_lat, min_lon, max_lat, max_lon
 
 
 def get_landmarks(
-    list_amenity: list,
+    amenity_selectors: list[dict],
     landmarktype: LandmarkType,
-    coordinates: Tuple[float, float],
+    center_coordinates: tuple[float, float],
-    parameters: dict
+    distance: int,
-) -> List[Landmark]:
+    score_function: callable
+) -> list[Landmark]:
+    
+    landmarks = ox.features_from_point(
+        center_point = center_coordinates,
+        dist = distance,
+        tags = amenity_selectors
+    )
+
+    # cleanup the list
+    # remove rows where name is None
+    landmarks = landmarks[landmarks['name'].notna()]
+    # TODO: remove rows that are part of another building
+
+    ret_landmarks = []
+    for element, description in landmarks.iterrows():
+        osm_type = element[0]
+        osm_id = element[1]
+        location = description['geometry']
+        n_tags = len(description['nodes']) if type(description['nodes']) == list else 1
+
+        # print(description['nodes'])
+        print(description['name'])
+        # print(location, type(location))
+        if type(location) == Point:
+            location = location
+        elif type(location) == Polygon or type(location) == MultiPolygon:
+            location = location.centroid
+        elif type(location) == LineString:
+            location = location.interpolate(location.length/2)
+
+        score = score_function(location, n_tags)
+        print(score)
+        landmark = Landmark(
+            name = description['name'],
+            type = landmarktype,
+            location = (location.x, location.y),
+            osm_type = osm_type,
+            osm_id = osm_id,
+            attractiveness = score,
+            must_do = False,
+            n_tags = n_tags
+        )
+        ret_landmarks.append(landmark)
+
+    return ret_landmarks
+    # for elem in G.iterrows():
+    #     print(elem)
+    #     print(elem.name)
+    #     print(elem.address)
+    #     name = elem.tag('name')                             # Add name
+    #     location = (elem.centerLat(), elem.centerLon())     # Add coordinates (lat, lon)
+
+    #     # skip if unprecise location
+    #     if name is None or location[0] is None:
+    #         continue
+
+    #     # skip if unused
+    #     if 'disused:leisure' in elem.tags().keys():
+    #         continue
+        
+    #     # skip if part of another building
+    #     if 'building:part' in elem.tags().keys() and elem.tag('building:part') == 'yes':
+    #         continue
+        
+    #     else :
+    #         osm_type = elem.type()              # Add type : 'way' or 'relation'
+    #         osm_id = elem.id()                  # Add OSM id 
+    #         elem_type = landmarktype            # Add the landmark type as 'sightseeing
+    #         n_tags = len(elem.tags().keys())    # Add number of tags
+            
+    #         # Add score of given landmark based on the number of surrounding elements. Penalty for churches as there are A LOT
+    #         if amenity == "'amenity'='place_of_worship'" :
+    #             score = int((count_elements_within_radius(location, parameters['radius_close_to']) + n_tags*parameters['tag_coeff'] )*parameters['church_coeff'])  
+    #         elif amenity == "'leisure'='park'" :
+    #             score = int((count_elements_within_radius(location, parameters['radius_close_to']) + n_tags*parameters['tag_coeff'] )*parameters['park_coeff'])  
+    #         else :
+    #             score = count_elements_within_radius(location, parameters['radius_close_to']) + n_tags*parameters['tag_coeff']
+
+    #         if score is not None :
+    #             # Generate the landmark and append it to the list
+    #             landmark = Landmark(name=name, type=elem_type, location=location, osm_type=osm_type, osm_id=osm_id, attractiveness=score, must_do=False, n_tags=n_tags)
+    #             L.append(landmark)
+
+    # return L
 
 
-    # # Read the parameters from the file
-    # with open (os.path.dirname(os.path.abspath(__file__)) + '/parameters/landmarks_manager.params', "r") as f :
-    #     parameters = json.loads(f.read())
-    #     tag_coeff = parameters['tag coeff']
-    #     park_coeff = parameters['park coeff']
-    #     church_coeff = parameters['church coeff']
-    #     radius = parameters['radius close to']
-    #     bbox_side = parameters['city bbox side']
-    
-    # Create bbox around start location
-    bbox = create_bbox(coordinates, parameters['city_bbox_side'])
-
-    # Initialize some variables
-    N = 0
-    L = []
-    overpass = Overpass()
-
-    for amenity in list_amenity :
-        query = overpassQueryBuilder(bbox=bbox, elementType=['way', 'relation'], selector=amenity, includeCenter=True, out='body')
-        result = overpass.query(query)
-        N += result.countElements()
-
-        for elem in result.elements():
-
-            name = elem.tag('name')                             # Add name
-            location = (elem.centerLat(), elem.centerLon())     # Add coordinates (lat, lon)
-
-            # skip if unprecise location
-            if name is None or location[0] is None:
-                continue
-
-            # skip if unused
-            if 'disused:leisure' in elem.tags().keys():
-                continue
-            
-            # skip if part of another building
-            if 'building:part' in elem.tags().keys() and elem.tag('building:part') == 'yes':
-                continue
-            
-            else :
-                osm_type = elem.type()              # Add type : 'way' or 'relation'
-                osm_id = elem.id()                  # Add OSM id 
-                elem_type = landmarktype            # Add the landmark type as 'sightseeing
-                n_tags = len(elem.tags().keys())    # Add number of tags
-                
-                # Add score of given landmark based on the number of surrounding elements. Penalty for churches as there are A LOT
-                if amenity == "'amenity'='place_of_worship'" :
-                    score = int((count_elements_within_radius(location, parameters['radius_close_to']) + n_tags*parameters['tag_coeff'] )*parameters['church_coeff'])  
-                elif amenity == "'leisure'='park'" :
-                    score = int((count_elements_within_radius(location, parameters['radius_close_to']) + n_tags*parameters['tag_coeff'] )*parameters['park_coeff'])  
-                else :
-                    score = count_elements_within_radius(location, parameters['radius_close_to']) + n_tags*parameters['tag_coeff']
-
-                if score is not None :
-                    # Generate the landmark and append it to the list
-                    landmark = Landmark(name=name, type=elem_type, location=location, osm_type=osm_type, osm_id=osm_id, attractiveness=score, must_do=False, n_tags=n_tags)
-                    L.append(landmark)
-
-    return L
-
-
-
-"""def get_landmarks(list_amenity: list, landmarktype: LandmarkType, city_country: str = None, coordinates: Tuple[float, float] = None) -> List[Landmark] :
-    
-    if city_country is None and coordinates is None :
-        raise ValueError("Either one of 'city_country' and 'coordinates' arguments must be specified")
-
-    if city_country is not None and coordinates is not None :
-        raise ValueError("Cannot specify both 'city_country' and 'coordinates' at the same time, please choose either one")
-
-    # Read the parameters from the file
-    with open (os.path.dirname(os.path.abspath(__file__)) + '/parameters/landmarks_manager.params', "r") as f :
-        parameters = json.loads(f.read())
-        tag_coeff = parameters['tag coeff']
-        park_coeff = parameters['park coeff']
-        church_coeff = parameters['church coeff']
-        radius = parameters['radius close to']
-        bbox_side = parameters['city bbox side']
-    
-    # If city_country is specified :
-    if city_country is not None :
-        nominatim = Nominatim()
-        areaId = nominatim.query(city_country).areaId()
-        bbox = None
-
-    # If coordinates are specified :
-    elif coordinates is not None :
-        bbox = create_bbox(coordinates, bbox_side)
-        areaId = None
-
-    else :
-        raise ValueError("Argument number is not corresponding.")
-    
-    # Initialize some variables
-    N = 0
-    L = []
-    overpass = Overpass()
-
-    for amenity in list_amenity :
-        query = overpassQueryBuilder(area=areaId, bbox=bbox, elementType=['way', 'relation'], selector=amenity, includeCenter=True, out='body')
-        result = overpass.query(query)
-        N += result.countElements()
-
-        for elem in result.elements():
-
-            name = elem.tag('name')                             # Add name
-            location = (elem.centerLat(), elem.centerLon())     # Add coordinates (lat, lon)
-
-            # skip if unprecise location
-            if name is None or location[0] is None:
-                continue
-
-            # skip if unused
-            if 'disused:leisure' in elem.tags().keys():
-                continue
-            
-            # skip if part of another building
-            if 'building:part' in elem.tags().keys() and elem.tag('building:part') == 'yes':
-                continue
-            
-            else :
-                osm_type = elem.type()              # Add type : 'way' or 'relation'
-                osm_id = elem.id()                  # Add OSM id 
-                elem_type = landmarktype            # Add the landmark type as 'sightseeing
-                n_tags = len(elem.tags().keys())    # Add number of tags
-                
-                # Add score of given landmark based on the number of surrounding elements. Penalty for churches as there are A LOT
-                if amenity == "'amenity'='place_of_worship'" :
-                    score = int((count_elements_within_radius(location, radius) + n_tags*tag_coeff )*church_coeff)  
-                elif amenity == "'leisure'='park'" :
-                    score = int((count_elements_within_radius(location, radius) + n_tags*tag_coeff )*park_coeff)  
-                else :
-                    score = count_elements_within_radius(location, radius) + n_tags*tag_coeff
-
-                if score is not None :
-                    # Generate the landmark and append it to the list
-                    landmark = Landmark(name=name, type=elem_type, location=location, osm_type=osm_type, osm_id=osm_id, attractiveness=score, must_do=False, n_tags=n_tags)
-                    L.append(landmark)
-
-    return L
-"""

backend/src/optimizer.py

@@ -1,5 +1,4 @@
 import numpy as np
-import json, os
 import yaml
 
 from typing import List, Tuple

backend/src/parameters/amenity_selectors.yaml

@@ -1,26 +1,32 @@
 nature:
-  - "'leisure'='park'"
+  leisure: park
-  - "geological"
+  geological: ''
-  - "'natural'='geyser'"
+  natural:
-  - "'natural'='hot_spring'"
+    - geyser
-  - "'natural'='arch'"
+    - hot_spring
-  - "'natural'='volcano'"
+    - arch
-  - "'natural'='stone'"
+    - volcano
-  - "'tourism'='alpine_hut'"
+    - stone
-  - "'tourism'='viewpoint'"
+  tourism:
-  - "'tourism'='zoo'"
+    - alpine_hut
-  - "'waterway'='waterfall'"
+    - viewpoint
+    - zoo
+  waterway: waterfall
 
 shopping:
-  - "'shop'='department_store'"
+  shop:
-  - "'shop'='mall'"
+    - department_store
+    - mall
 
 sightseeing:
-  - "'tourism'='museum'"
+  tourism:
-  - "'tourism'='attraction'"
+    - museum
-  - "'tourism'='gallery'"
+    - attraction
-  - "historic"
+    - gallery
-  - "'amenity'='planetarium'"
+  historic: ''
-  - "'amenity'='place_of_worship'"
+  amenity:
-  - "'amenity'='fountain'"
+    - planetarium
-  - "'water'='reflecting_pool'"
+    - place_of_worship
+    - fountain
+  water:
+    - reflecting_pool

backend/src/parameters/landmark_parameters.yaml

@@ -1,5 +1,5 @@
-city_bbox_side: 10
+city_bbox_side: 1500 #m
-radius_close_to: 27.5
+radius_close_to: 30
 church_coeff: 0.6
 park_coeff: 1.5
 tag_coeff: 100

backend/src/structs/landmarks.py

@@ -1,6 +1,5 @@
 from typing import Optional
 from pydantic import BaseModel, Field
-
 from .landmarktype import LandmarkType
 
 from uuid import uuid4
@@ -28,3 +27,6 @@ class Landmark(BaseModel) :
     
     time_to_reach_next : Optional[int] = 0                      # TODO fix this in existing code
     next_uuid : Optional[str] = None                            # TODO implement this ASAP
+
+    def __hash__(self) -> int:
+        return self.uuid.int

backend/src/tester.py

@@ -90,7 +90,7 @@ def test4(coordinates: tuple[float, float]) -> List[Landmark]:
     #finish = Landmark(name='finish', type=LandmarkType(landmark_type='finish'), location=(48.847132, 2.312359), osm_type='finish', osm_id=0, attractiveness=0, must_do=True, n_tags = 0)
     
     # Generate the landmarks from the start location
-    landmarks, landmarks_short = generate_landmarks(preferences=preferences, coordinates=start.location)
+    landmarks, landmarks_short = generate_landmarks(preferences=preferences, center_coordinates=start.location)
     #write_data(landmarks, "landmarks.txt")
 
     # Insert start and finish to the landmarks list
@@ -98,7 +98,7 @@ def test4(coordinates: tuple[float, float]) -> List[Landmark]:
     landmarks_short.append(finish)
 
     # TODO use these parameters in another way
-    max_walking_time = 2    # hours
+    max_walking_time = 3    # hours
     detour = 30             # minutes
 
     # First stage optimization