further cleanup

This commit is contained in:
Remy Moll 2024-07-08 11:55:00 +02:00
parent f9c86261cb
commit 8f23a4747d
4 changed files with 182 additions and 277 deletions

View File

@ -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'
)

View File

@ -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

View File

@ -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}&param2={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

View File

@ -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')
if __name__ == '__main__':
start = (48.847132, 2.312359) # Café Chez César
# start = (47.377859, 8.540585) # Zurich HB
main(start)