diff --git a/backend/src/constants.py b/backend/src/constants.py index c60bf84..d5b7b5a 100644 --- a/backend/src/constants.py +++ b/backend/src/constants.py @@ -1,5 +1,6 @@ from pathlib import Path import os +import logging PARAMETERS_DIR = Path('src/parameters') AMENITY_SELECTORS_PATH = PARAMETERS_DIR / 'amenity_selectors.yaml' @@ -10,3 +11,9 @@ OPTIMIZER_PARAMETERS_PATH = PARAMETERS_DIR / 'optimizer_parameters.yaml' cache_dir_string = os.getenv('OSM_CACHE_DIR', './cache') OSM_CACHE_DIR = Path(cache_dir_string) + +logger = logging.getLogger(__name__) +logging.basicConfig( + level = logging.INFO, + format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s' +) diff --git a/backend/src/landmarks_manager.py b/backend/src/landmarks_manager.py index 8dcaa8c..690fae8 100644 --- a/backend/src/landmarks_manager.py +++ b/backend/src/landmarks_manager.py @@ -1,5 +1,5 @@ import yaml -import os +import logging import osmnx as ox from shapely.geometry import Point, Polygon, LineString, MultiPolygon @@ -7,182 +7,145 @@ 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, 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 = [] +class LandmarkManager: + logger = logging.getLogger(__name__) - # List for sightseeing - if preferences.sightseeing.score != 0: - 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: - 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: - 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 + def __init__(self) -> None: + ox.config(cache_folder=constants.OSM_CACHE_DIR) + with constants.AMENITY_SELECTORS_PATH.open('r') as f: + self.amenity_selectors = yaml.safe_load(f) + with constants.LANDMARK_PARAMETERS_PATH.open('r') as f: + self.parameters = yaml.safe_load(f) + # max_distance = parameters['city_bbox_side'] - # remove duplicates - L = list(set(L)) - print(len(L)) - L_constrained = take_most_important(L, parameters['N_important']) - print(len(L_constrained)) - return L, L_constrained + + def get_landmark_lists(self, preferences: Preferences, center_coordinates: tuple[float, float]) -> tuple[list[Landmark], list[Landmark]]: + ''' + Generate a list of landmarks based on the preferences of the user and the center (ie. start) coordinates. + The list is then used by the pathfinding algorithm to generate a path that goes through the most interesting landmarks. + :param preferences: the preferences specified by the user + :param center_coordinates: the coordinates of the starting point + ''' + + L = [] + + # List for sightseeing + if preferences.sightseeing.score != 0: + score_func = lambda loc, n_tags: int((self.count_elements_within_radius(loc, self.parameters['radius_close_to']) + n_tags * self.parameters['tag_coeff']) * self.parameters['church_coeff']) + L1 = self.fetch_landmarks(self.amenity_selectors['sightseeing'], SIGHTSEEING, center_coordinates, self.parameters['city_bbox_side'], score_func) + self.correct_score(L1, preferences.sightseeing) + L += L1 + + # List for nature + if preferences.nature.score != 0: + score_func = lambda loc, n_tags: int((self.count_elements_within_radius(loc, self.parameters['radius_close_to']) + n_tags * self.parameters['tag_coeff']) * self.parameters['park_coeff']) + L2 = self.fetch_landmarks(self.amenity_selectors['nature'], NATURE, center_coordinates, self.parameters['city_bbox_side'], score_func) + self.correct_score(L2, preferences.nature) + L += L2 + + # List for shopping + if preferences.shopping.score != 0: + score_func = lambda loc, n_tags: self.count_elements_within_radius(loc, self.parameters['radius_close_to']) + n_tags * self.parameters['tag_coeff'] + L3 = self.fetch_landmarks(self.amenity_selectors['shopping'], SHOPPING, center_coordinates, self.parameters['city_bbox_side'], score_func) + self.correct_score(L3, preferences.shopping) + L += L3 + + # remove duplicates + L = list(set(L)) + L_constrained = self.take_most_important(L, self.parameters['N_important']) + self.logger.info(f'Generated {len(L)} landmarks around {center_coordinates}, and constrained to {len(L_constrained)} most important ones.') + return L, L_constrained -# Take the most important landmarks from the list -def take_most_important(landmarks: list[Landmark], n_max: int) -> list[Landmark]: + # Take the most important landmarks from the list + def take_most_important(self, landmarks: list[Landmark], n_max: int) -> list[Landmark]: - landmarks_sorted = sorted(landmarks, key=lambda x: x.attractiveness, reverse=True) - return landmarks_sorted[:n_max] + landmarks_sorted = sorted(landmarks, key=lambda x: x.attractiveness, reverse=True) + return landmarks_sorted[:n_max] -# Correct the score of a list of landmarks by taking into account preference settings -def correct_score(L: list[Landmark], preference: Preference) : + # Correct the score of a list of landmarks by taking into account preference settings + def correct_score(self, L: list[Landmark], preference: Preference): - if len(L) == 0 : - return - - if L[0].type != preference.type : - raise TypeError(f"LandmarkType {preference.type} does not match the type of Landmark {L[0].name}") + if len(L) == 0 : + return - for elem in L : - elem.attractiveness = int(elem.attractiveness*preference.score/500) # arbitrary computation + if L[0].type != preference.type : + raise TypeError(f"LandmarkType {preference.type} does not match the type of Landmark {L[0].name}") + + for elem in L : + elem.attractiveness = int(elem.attractiveness*preference.score/500) # arbitrary computation -# Function to count elements within a certain radius of a location -def count_elements_within_radius(point: Point, radius: int) -> int: - - center_coordinates = (point.x, point.y) - try: + # Function to count elements within a certain radius of a location + def count_elements_within_radius(self, point: Point, radius: int) -> int: + + center_coordinates = (point.x, point.y) + try: + landmarks = ox.features_from_point( + center_point = center_coordinates, + dist = radius, + tags = {'building': True} # this is a common tag to give an estimation of the number of elements in the area + ) + return len(landmarks) + except ox._errors.InsufficientResponseError: + return 0 + + + + + def fetch_landmarks( + self, + amenity_selectors: list[dict], + landmarktype: LandmarkType, + center_coordinates: tuple[float, float], + distance: int, + score_function: callable + ) -> list[Landmark]: + landmarks = ox.features_from_point( center_point = center_coordinates, - dist = radius, - tags = {'building': True} # this is a common tag to give an estimation of the number of elements in the area + dist = distance, + tags = amenity_selectors ) - return len(landmarks) - except ox._errors.InsufficientResponseError: - return 0 + # 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 + 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) -def get_landmarks( - amenity_selectors: list[dict], - landmarktype: LandmarkType, - center_coordinates: tuple[float, float], - distance: int, - 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 - + score = score_function(location, n_tags) + 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 diff --git a/backend/src/main.py b/backend/src/main.py index 89b49fa..a1ff712 100644 --- a/backend/src/main.py +++ b/backend/src/main.py @@ -1,51 +1,45 @@ from optimizer import solve_optimization -from refiner import refine_optimization -from landmarks_manager import generate_landmarks +# from refiner import refine_optimization +from landmarks_manager import LandmarkManager from structs.landmarks import Landmark from structs.landmarktype import LandmarkType -from structs.preferences import Preferences, Preference +from structs.preferences import Preferences from fastapi import FastAPI, Query, Body -from typing import List app = FastAPI() - +manager = LandmarkManager() # TODO: needs a global variable to store the landmarks accross function calls # linked_tour = [] -# Assuming frontend is calling like this : -#"http://127.0.0.1:8000/process?param1={param1}¶m2={param2}" -@app.post("/optimizer_coords/{start_lat}/{start_lon}/{finish_lat}/{finish_lon}") -def main1(start_lat: float, start_lon: float, preferences: Preferences = Body(...), finish_lat: float = None, finish_lon: float = None) -> List[Landmark]: - +@app.post("/route/new") +def main1(preferences: Preferences, start: tuple[float, float], end: tuple[float, float] = None) -> str: + ''' + Main function to call the optimizer. + :param preferences: the preferences specified by the user as the post body + :param start: the coordinates of the starting point as a tuple of floats (as url query parameters) + :param end: the coordinates of the finishing point as a tuple of floats (as url query parameters) + :return: the uuid of the first landmark in the optimized route + ''' if preferences is None : raise ValueError("Please provide preferences in the form of a 'Preference' BaseModel class.") - if bool(start_lat) ^ bool(start_lon) : - raise ValueError("Please provide both latitude and longitude for the starting point") - if bool(finish_lat) ^ bool(finish_lon) : - raise ValueError("Please provide both latitude and longitude for the finish point") + if start is None: + raise ValueError("Please provide the starting coordinates as a tuple of floats.") + if end is None: + end = start - start = Landmark(name='start', type=LandmarkType(landmark_type='start'), location=(start_lat, start_lon), osm_type='start', osm_id=0, attractiveness=0, must_do=True, n_tags = 0) + start_landmark = Landmark(name='start', type=LandmarkType(landmark_type='start'), location=(start[0], start[1]), osm_type='start', osm_id=0, attractiveness=0, must_do=True, n_tags = 0) + end_landmark = Landmark(name='end', type=LandmarkType(landmark_type='end'), location=(end[0], end[1]), osm_type='end', osm_id=0, attractiveness=0, must_do=True, n_tags = 0) - if bool(finish_lat) and bool(finish_lon) : - finish = Landmark(name='finish', type=LandmarkType(landmark_type='finish'), location=(finish_lat, finish_lon), osm_type='finish', osm_id=0, attractiveness=0, must_do=True, n_tags = 0) - else : - finish = Landmark(name='finish', type=LandmarkType(landmark_type='finish'), location=(start_lat, start_lon), osm_type='finish', osm_id=0, attractiveness=0, must_do=True, n_tags = 0) - - - start = Landmark(name='start', type=LandmarkType(landmark_type='start'), location=(48.8375946, 2.2949904), osm_type='start', osm_id=0, attractiveness=0, must_do=True, n_tags = 0) - finish = Landmark(name='finish', type=LandmarkType(landmark_type='finish'), location=(48.8375946, 2.2949904), 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 = LandmarkManager.get_landmark_lists(preferences=preferences, coordinates=start.location) + print([l.name for l in landmarks_short]) # insert start and finish to the landmarks list - landmarks_short.insert(0, start) - landmarks_short.append(finish) + landmarks_short.insert(0, start_landmark) + landmarks_short.append(end_landmark) - - # TODO use these parameters in another way + # TODO infer these parameters from the preferences max_walking_time = 4 # hours detour = 30 # minutes @@ -53,32 +47,11 @@ def main1(start_lat: float, start_lon: float, preferences: Preferences = Body(.. base_tour = solve_optimization(landmarks_short, max_walking_time*60, True) # Second stage optimization - refined_tour = refine_optimization(landmarks, base_tour, max_walking_time*60+detour, True) + # refined_tour = refine_optimization(landmarks, base_tour, max_walking_time*60+detour, True) - - # TODO: should look something like this - # # set time to reach and transform into fully functional linked list - # linked_tour += link(refined_tour) - # return { - # 'city_name': 'Paris', - # 'n_stops': len(linked_tour), - # 'first_landmark_uuid': linked_tour[0].uuid, - # } - - return refined_tour - - - - -# input city, country in the form of 'Paris, France' -@app.post("/test2/{city_country}") -def test2(city_country: str, preferences: Preferences = Body(...)) -> List[Landmark]: - - landmarks = generate_landmarks(city_country, preferences) - - max_steps = 9000000 - - visiting_order = solve_optimization(landmarks, max_steps, True) + # linked_tour = ... + # return linked_tour[0].uuid + return base_tour[0].uuid @@ -86,5 +59,3 @@ def test2(city_country: str, preferences: Preferences = Body(...)) -> List[Landm def get_landmark(landmark_uuid: str) -> Landmark: #cherche dans linked_tour et retourne le landmark correspondant pass - - diff --git a/backend/src/tester.py b/backend/src/tester.py index fe7220d..57692d2 100644 --- a/backend/src/tester.py +++ b/backend/src/tester.py @@ -1,11 +1,11 @@ import pandas as pd from typing import List -from landmarks_manager import generate_landmarks +from landmarks_manager import LandmarkManager from fastapi.encoders import jsonable_encoder from optimizer import solve_optimization -from refiner import refine_optimization +# from refiner import refine_optimization from structs.landmarks import Landmark from structs.landmarktype import LandmarkType from structs.preferences import Preferences, Preference @@ -23,74 +23,37 @@ def write_data(L: List[Landmark], file_name: str): data.to_json(file_name, indent = 2, force_ascii=False) -def test3(city_country: str) -> List[Landmark]: +def main(coordinates: tuple[float, float]) -> List[Landmark]: + manager = LandmarkManager() preferences = Preferences( - sightseeing=Preference( - name='sightseeing', - type=LandmarkType(landmark_type='sightseeing'), - score = 5), - nature=Preference( - name='nature', - type=LandmarkType(landmark_type='nature'), - score = 0), - shopping=Preference( - name='shopping', - type=LandmarkType(landmark_type='shopping'), - score = 5)) - - coordinates = None - - landmarks, landmarks_short = generate_landmarks(preferences=preferences, city_country=city_country, coordinates=coordinates) - - #write_data(landmarks) - - start = Landmark(name='start', type=LandmarkType(landmark_type='start'), location=(48.2044576, 16.3870242), osm_type='start', osm_id=0, attractiveness=0, must_do=True, n_tags = 0) - finish = Landmark(name='finish', type=LandmarkType(landmark_type='finish'), location=(48.2044576, 16.3870242), osm_type='finish', osm_id=0, attractiveness=0, must_do=True, n_tags = 0) - - - test = landmarks_short - - test.insert(0, start) - test.append(finish) - - max_walking_time = 2 # hours - - visiting_list = solve_optimization(test, max_walking_time*60, True) - - - - - -def test4(coordinates: tuple[float, float]) -> List[Landmark]: - - - preferences = Preferences( - sightseeing=Preference( - name='sightseeing', - type=LandmarkType(landmark_type='sightseeing'), - score = 5), - nature=Preference( - name='nature', - type=LandmarkType(landmark_type='nature'), - score = 5), - shopping=Preference( - name='shopping', - type=LandmarkType(landmark_type='shopping'), - score = 5)) + sightseeing=Preference( + name='sightseeing', + type=LandmarkType(landmark_type='sightseeing'), + score = 5 + ), + nature=Preference( + name='nature', + type=LandmarkType(landmark_type='nature'), + score = 5 + ), + shopping=Preference( + name='shopping', + type=LandmarkType(landmark_type='shopping'), + score = 5 + ) + ) # Create start and finish start = Landmark(name='start', type=LandmarkType(landmark_type='start'), location=coordinates, osm_type='start', osm_id=0, attractiveness=0, must_do=True, n_tags = 0) finish = Landmark(name='finish', type=LandmarkType(landmark_type='finish'), location=coordinates, osm_type='finish', osm_id=0, attractiveness=0, must_do=True, n_tags = 0) - #finish = Landmark(name='finish', type=LandmarkType(landmark_type='finish'), location=(48.8777055, 2.3640967), osm_type='finish', osm_id=0, attractiveness=0, must_do=True, n_tags = 0) - #start = Landmark(name='start', type=LandmarkType(landmark_type='start'), location=(48.847132, 2.312359), osm_type='start', osm_id=0, attractiveness=0, must_do=True, n_tags = 0) - #finish = Landmark(name='finish', type=LandmarkType(landmark_type='finish'), location=(48.843185, 2.344533), osm_type='finish', osm_id=0, attractiveness=0, must_do=True, n_tags = 0) - #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, center_coordinates=start.location) + landmarks, landmarks_short = manager.get_landmark_lists(preferences=preferences, center_coordinates=start.location) + print([l.name for l in landmarks_short]) + #write_data(landmarks, "landmarks.txt") # Insert start and finish to the landmarks list @@ -105,12 +68,13 @@ def test4(coordinates: tuple[float, float]) -> List[Landmark]: base_tour = solve_optimization(landmarks_short, max_walking_time*60, True) # Second stage optimization - refined_tour = refine_optimization(landmarks, base_tour, max_walking_time*60+detour, True) + # refined_tour = refine_optimization(landmarks, base_tour, max_walking_time*60+detour, True) - return refined_tour + return base_tour -test4(tuple((48.8344400, 2.3220540))) # Café Chez César -#test4(tuple((48.8375946, 2.2949904))) # Point random -#test4(tuple((47.377859, 8.540585))) # Zurich HB -#test3('Vienna, Austria') \ No newline at end of file + +if __name__ == '__main__': + start = (48.847132, 2.312359) # Café Chez César + # start = (47.377859, 8.540585) # Zurich HB + main(start)