Compare commits

..

No commits in common. "25cc0fa300a8e0a4e39f19f6ecd22617c36612fa" and "f9c86261cbf8935868a01751378a837aa896691a" have entirely different histories.

10 changed files with 277 additions and 275 deletions

View File

@ -1,6 +1,5 @@
from pathlib import Path
import os
import logging
PARAMETERS_DIR = Path('src/parameters')
AMENITY_SELECTORS_PATH = PARAMETERS_DIR / 'amenity_selectors.yaml'
@ -11,9 +10,3 @@ 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 logging
import os
import osmnx as ox
from shapely.geometry import Point, Polygon, LineString, MultiPolygon
@ -7,145 +7,182 @@ 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)
class LandmarkManager:
logger = logging.getLogger(__name__)
# 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 = []
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']
# 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 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
# 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
# Take the most important landmarks from the list
def take_most_important(self, landmarks: list[Landmark], n_max: int) -> list[Landmark]:
# Take the most important landmarks from the list
def take_most_important(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(self, L: list[Landmark], preference: Preference):
# Correct the score of a list of landmarks by taking into account preference settings
def correct_score(L: list[Landmark], preference: Preference) :
if len(L) == 0 :
return
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 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
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(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]:
# 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:
landmarks = ox.features_from_point(
center_point = center_coordinates,
dist = distance,
tags = amenity_selectors
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
# 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)
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)
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
return ret_landmarks

View File

@ -1,45 +1,51 @@
from optimizer import solve_optimization
# from refiner import refine_optimization
from landmarks_manager import LandmarkManager
from refiner import refine_optimization
from landmarks_manager import generate_landmarks
from structs.landmarks import Landmark
from structs.landmarktype import LandmarkType
from structs.preferences import Preferences
from structs.preferences import Preferences, Preference
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 = []
@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
'''
# 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]:
if preferences is None :
raise ValueError("Please provide preferences in the form of a 'Preference' BaseModel class.")
if start is None:
raise ValueError("Please provide the starting coordinates as a tuple of floats.")
if end is None:
end = start
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")
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)
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)
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 = LandmarkManager.get_landmark_lists(preferences=preferences, coordinates=start.location)
print([l.name for l in landmarks_short])
landmarks, landmarks_short = generate_landmarks(preferences=preferences, coordinates=start.location)
# insert start and finish to the landmarks list
landmarks_short.insert(0, start_landmark)
landmarks_short.append(end_landmark)
landmarks_short.insert(0, start)
landmarks_short.append(finish)
# TODO infer these parameters from the preferences
# TODO use these parameters in another way
max_walking_time = 4 # hours
detour = 30 # minutes
@ -47,11 +53,32 @@ def main1(preferences: Preferences, start: tuple[float, float], end: tuple[float
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)
# linked_tour = ...
# return linked_tour[0].uuid
return base_tour[0].uuid
# 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)
@ -59,3 +86,5 @@ def main1(preferences: Preferences, start: tuple[float, float], end: tuple[float
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 LandmarkManager
from landmarks_manager import generate_landmarks
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,37 +23,74 @@ def write_data(L: List[Landmark], file_name: str):
data.to_json(file_name, indent = 2, force_ascii=False)
def main(coordinates: tuple[float, float]) -> List[Landmark]:
def test3(city_country: str) -> 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 = 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 = 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))
# 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 = manager.get_landmark_lists(preferences=preferences, center_coordinates=start.location)
print([l.name for l in landmarks_short])
landmarks, landmarks_short = generate_landmarks(preferences=preferences, center_coordinates=start.location)
#write_data(landmarks, "landmarks.txt")
# Insert start and finish to the landmarks list
@ -68,13 +105,12 @@ def main(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 base_tour
return refined_tour
if __name__ == '__main__':
start = (48.847132, 2.312359) # Café Chez César
# start = (47.377859, 8.540585) # Zurich HB
main(start)
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')

View File

@ -1,37 +0,0 @@
apiVersion: apps/v1
kind: Deployment
metadata:
name: nav-backend
spec:
replicas: 3
selector:
matchLabels:
app: nav-backend
template:
metadata:
labels:
app: nav-backend
spec:
containers:
- name: worker
image: backend-image
ports:
- containerPort: 8000
env:
- name: NUM_WORKERS
value: "3"
- name: OSM_CACHE_DIR
value: "/osm-cache"
volumeMounts:
- name: osm-cache
mountPath: /osm-cache
resources:
requests:
cpu: 100m
memory: 100Mi
limits:
cpu: 4
memory: 8Gi
volumes:
- name: osm-cache
emptyDir: {}

View File

@ -1,15 +0,0 @@
kind: IngressRoute
apiVersion: traefik.io/v1alpha1
metadata:
name: nav-ingress
spec:
entryPoints:
- websecure
routes:
- match: Host(`nav.kluster.moll.re`)
kind: Rule
services:
- name: nav-service
port: 8000
tls:
certResolver: default-tls

View File

@ -1,15 +0,0 @@
apiVersion: kustomize.config.k8s.io/v1beta1
kind: Kustomization
namespace: nav
resources:
- namespace.yaml
- deployment.yaml
- service.yaml
- ingress.yaml
images:
- name: backend-image
newName: git.kluster.moll.re/remoll/fast_network_navigation/backend
newTag: latest

View File

@ -1,4 +0,0 @@
apiVersion: v1
kind: Namespace
metadata:
name: placeholder

View File

@ -1,11 +0,0 @@
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
name: cache
spec:
storageClassName: "nfs-client"
accessModes:
- ReadWriteOnce
resources:
requests:
storage: "5Gi"

View File

@ -1,11 +0,0 @@
apiVersion: v1
kind: Service
metadata:
name: nav-service
spec:
selector:
app: nav-backend
ports:
- protocol: TCP
port: 8000
targetPort: 8000