Merge modifications for more separate backend functions #69
3
backend/.gitignore
vendored
3
backend/.gitignore
vendored
@ -12,6 +12,9 @@ __pycache__/
|
|||||||
# C extensions
|
# C extensions
|
||||||
*.so
|
*.so
|
||||||
|
|
||||||
|
# Pytest reports
|
||||||
|
report.html
|
||||||
|
|
||||||
# Distribution / packaging
|
# Distribution / packaging
|
||||||
.Python
|
.Python
|
||||||
build/
|
build/
|
||||||
|
File diff suppressed because one or more lines are too long
@ -146,7 +146,7 @@ class ClusterManager:
|
|||||||
self.valid = False
|
self.valid = False
|
||||||
|
|
||||||
else :
|
else :
|
||||||
self.logger.debug(f"Detected 0 {cluster_type} clusters.")
|
self.logger.debug(f"Found 0 {cluster_type} clusters.")
|
||||||
self.valid = False
|
self.valid = False
|
||||||
|
|
||||||
|
|
||||||
|
@ -55,7 +55,12 @@ class LandmarkManager:
|
|||||||
self.logger.info('LandmakManager successfully initialized.')
|
self.logger.info('LandmakManager successfully initialized.')
|
||||||
|
|
||||||
|
|
||||||
def generate_landmarks_list(self, center_coordinates: tuple[float, float], preferences: Preferences) -> tuple[list[Landmark], list[Landmark]]:
|
def generate_landmarks_list(
|
||||||
|
self,
|
||||||
|
center_coordinates: tuple[float, float],
|
||||||
|
preferences: Preferences,
|
||||||
|
allow_clusters: bool = True
|
||||||
|
) -> tuple[list[Landmark], list[Landmark]]:
|
||||||
"""
|
"""
|
||||||
Generate and prioritize a list of landmarks based on user preferences.
|
Generate and prioritize a list of landmarks based on user preferences.
|
||||||
|
|
||||||
@ -63,16 +68,17 @@ class LandmarkManager:
|
|||||||
and current location. It scores and corrects these landmarks, removes duplicates, and then selects the most important
|
and current location. It scores and corrects these landmarks, removes duplicates, and then selects the most important
|
||||||
landmarks based on a predefined criterion.
|
landmarks based on a predefined criterion.
|
||||||
|
|
||||||
Args:
|
Parameters :
|
||||||
center_coordinates (tuple[float, float]): The latitude and longitude of the center location around which to search.
|
center_coordinates (tuple[float, float]): The latitude and longitude of the center location around which to search.
|
||||||
preferences (Preferences): The user's preference settings that influence the landmark selection.
|
preferences (Preferences): The user's preference settings that influence the landmark selection.
|
||||||
|
allow_clusters (bool, optional) : If set to False, no clusters will be fetched. Mainly used for the option to fetch landmarks nearby.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
tuple[list[Landmark], list[Landmark]]:
|
tuple[list[Landmark], list[Landmark]]:
|
||||||
- A list of all existing landmarks.
|
- A list of all existing landmarks.
|
||||||
- A list of the most important landmarks based on the user's preferences.
|
- A list of the most important landmarks based on the user's preferences.
|
||||||
"""
|
"""
|
||||||
self.logger.debug('Starting to fetch landmarks...')
|
self.logger.info(f'Starting to fetch landmarks around {center_coordinates}...')
|
||||||
max_walk_dist = int((preferences.max_time_minute/2)/60*self.walking_speed*1000/self.detour_factor)
|
max_walk_dist = int((preferences.max_time_minute/2)/60*self.walking_speed*1000/self.detour_factor)
|
||||||
radius = min(max_walk_dist, int(self.max_bbox_side/2))
|
radius = min(max_walk_dist, int(self.max_bbox_side/2))
|
||||||
|
|
||||||
@ -89,10 +95,11 @@ class LandmarkManager:
|
|||||||
all_landmarks.update(current_landmarks)
|
all_landmarks.update(current_landmarks)
|
||||||
self.logger.info(f'Found {len(current_landmarks)} sightseeing landmarks')
|
self.logger.info(f'Found {len(current_landmarks)} sightseeing landmarks')
|
||||||
|
|
||||||
|
if allow_clusters :
|
||||||
# special pipeline for historic neighborhoods
|
# special pipeline for historic neighborhoods
|
||||||
neighborhood_manager = ClusterManager(bbox, 'sightseeing')
|
neighborhood_manager = ClusterManager(bbox, 'sightseeing')
|
||||||
historic_clusters = neighborhood_manager.generate_clusters()
|
historic_clusters = neighborhood_manager.generate_clusters()
|
||||||
all_landmarks.update(historic_clusters)
|
all_landmarks.update(historic_clusters)
|
||||||
|
|
||||||
# list for nature
|
# list for nature
|
||||||
if preferences.nature.score != 0:
|
if preferences.nature.score != 0:
|
||||||
@ -113,14 +120,18 @@ class LandmarkManager:
|
|||||||
landmark.duration = 30
|
landmark.duration = 30
|
||||||
all_landmarks.update(current_landmarks)
|
all_landmarks.update(current_landmarks)
|
||||||
|
|
||||||
# special pipeline for shopping malls
|
if allow_clusters :
|
||||||
shopping_manager = ClusterManager(bbox, 'shopping')
|
# special pipeline for shopping malls
|
||||||
shopping_clusters = shopping_manager.generate_clusters()
|
shopping_manager = ClusterManager(bbox, 'shopping')
|
||||||
all_landmarks.update(shopping_clusters)
|
shopping_clusters = shopping_manager.generate_clusters()
|
||||||
|
all_landmarks.update(shopping_clusters)
|
||||||
|
|
||||||
|
|
||||||
landmarks_constrained = take_most_important(all_landmarks, self.n_important)
|
landmarks_constrained = take_most_important(all_landmarks, self.n_important)
|
||||||
|
|
||||||
|
# DETAILS HERE
|
||||||
# self.logger.info(f'All landmarks generated : {len(all_landmarks)} landmarks around {center_coordinates}, and constrained to {len(landmarks_constrained)} most important ones.')
|
# self.logger.info(f'All landmarks generated : {len(all_landmarks)} landmarks around {center_coordinates}, and constrained to {len(landmarks_constrained)} most important ones.')
|
||||||
|
self.logger.info(f'Found {len(all_landmarks)} landmarks in total.')
|
||||||
|
|
||||||
return all_landmarks, landmarks_constrained
|
return all_landmarks, landmarks_constrained
|
||||||
|
|
||||||
@ -236,6 +247,17 @@ class LandmarkManager:
|
|||||||
continue
|
continue
|
||||||
|
|
||||||
tags = elem.get('tags')
|
tags = elem.get('tags')
|
||||||
|
n_tags=len(tags)
|
||||||
|
|
||||||
|
# Skip this landmark if not suitable
|
||||||
|
if tags.get('building:part') is not None :
|
||||||
|
continue
|
||||||
|
if tags.get('disused') is not None :
|
||||||
|
continue
|
||||||
|
if tags.get('boundary') is not None :
|
||||||
|
continue
|
||||||
|
if tags.get('shop') is not None and landmarktype != 'shopping' :
|
||||||
|
continue
|
||||||
|
|
||||||
# Convert this to Landmark object
|
# Convert this to Landmark object
|
||||||
landmark = Landmark(name=name,
|
landmark = Landmark(name=name,
|
||||||
@ -244,180 +266,36 @@ class LandmarkManager:
|
|||||||
osm_id=id,
|
osm_id=id,
|
||||||
osm_type=osm_type,
|
osm_type=osm_type,
|
||||||
attractiveness=0,
|
attractiveness=0,
|
||||||
n_tags=len(tags))
|
n_tags=n_tags)
|
||||||
|
|
||||||
# Browse through tags to add information to landmark.
|
# Extract useful information for score calculation later down the road.
|
||||||
for key, value in tags.items():
|
landmark.image_url = tags.get('image')
|
||||||
|
landmark.website_url = tags.get('website')
|
||||||
|
landmark.wiki_url = tags.get('wikipedia')
|
||||||
|
landmark.name_en = tags.get('name:en')
|
||||||
|
|
||||||
# Skip this landmark if not suitable.
|
# Check for place of worship
|
||||||
if key == 'building:part' and value == 'yes' :
|
if tags.get('place_of_worship') is not None :
|
||||||
break
|
|
||||||
if 'disused:' in key :
|
|
||||||
break
|
|
||||||
if 'boundary:' in key :
|
|
||||||
break
|
|
||||||
if 'shop' in key and landmarktype != 'shopping' :
|
|
||||||
break
|
|
||||||
# if value == 'apartments' :
|
|
||||||
# break
|
|
||||||
|
|
||||||
# Fill in the other attributes.
|
|
||||||
if key == 'image' :
|
|
||||||
landmark.image_url = value
|
|
||||||
if key == 'website' :
|
|
||||||
landmark.website_url = value
|
|
||||||
if value == 'place_of_worship' :
|
|
||||||
landmark.is_place_of_worship = True
|
landmark.is_place_of_worship = True
|
||||||
if key == 'wikipedia' :
|
landmark.name_en = tags.get('place_of_worship')
|
||||||
landmark.wiki_url = value
|
|
||||||
if key == 'name:en' :
|
|
||||||
landmark.name_en = value
|
|
||||||
if 'building:' in key or 'pay' in key :
|
|
||||||
landmark.n_tags -= 1
|
|
||||||
|
|
||||||
|
# Set the duration. Needed for the optimization.
|
||||||
|
if tags.get('amenity') in ['aquarium', 'planetarium'] or tags.get('tourism') in ['aquarium', 'museum', 'zoo']:
|
||||||
|
landmark.duration = 60
|
||||||
|
elif tags.get('tourism') == 'viewpoint' :
|
||||||
|
landmark.is_viewpoint = True
|
||||||
|
landmark.duration = 10
|
||||||
|
elif tags.get('building') == 'cathedral' :
|
||||||
|
landmark.is_place_of_worship = False
|
||||||
|
landmark.duration = 10
|
||||||
|
|
||||||
# Set the duration.
|
# Compute the score and add landmark to the list.
|
||||||
if value in ['museum', 'aquarium', 'planetarium'] :
|
|
||||||
landmark.duration = 60
|
|
||||||
elif value == 'viewpoint' :
|
|
||||||
landmark.is_viewpoint = True
|
|
||||||
landmark.duration = 10
|
|
||||||
elif value == 'cathedral' :
|
|
||||||
landmark.is_place_of_worship = False
|
|
||||||
landmark.duration = 10
|
|
||||||
|
|
||||||
landmark.description, landmark.keywords = self.description_and_keywords(tags)
|
|
||||||
self.set_landmark_score(landmark, landmarktype, preference_level)
|
self.set_landmark_score(landmark, landmarktype, preference_level)
|
||||||
landmarks.append(landmark)
|
landmarks.append(landmark)
|
||||||
|
|
||||||
continue
|
|
||||||
|
|
||||||
|
|
||||||
return landmarks
|
return landmarks
|
||||||
|
|
||||||
|
|
||||||
def description_and_keywords(self, tags: dict):
|
|
||||||
"""
|
|
||||||
Generates a description and a set of keywords for a given landmark based on its tags.
|
|
||||||
|
|
||||||
Params:
|
|
||||||
tags (dict): A dictionary containing metadata about the landmark, including its name,
|
|
||||||
importance, height, date of construction, and visitor information.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
description (str): A string description of the landmark.
|
|
||||||
keywords (dict): A dictionary of keywords with fields such as 'importance', 'height',
|
|
||||||
'place_type', and 'date'.
|
|
||||||
"""
|
|
||||||
# Extract relevant fields
|
|
||||||
name = tags.get('name')
|
|
||||||
importance = tags.get('importance', None)
|
|
||||||
n_visitors = tags.get('tourism:visitors', None)
|
|
||||||
height = tags.get('height')
|
|
||||||
place_type = self.get_place_type(tags)
|
|
||||||
date = self.get_date(tags)
|
|
||||||
|
|
||||||
if place_type is None :
|
|
||||||
return None, None
|
|
||||||
|
|
||||||
# Start the description.
|
|
||||||
if importance is None :
|
|
||||||
if len(tags.keys()) < 5 :
|
|
||||||
return None, None
|
|
||||||
if len(tags.keys()) < 10 :
|
|
||||||
description = f"{name} is a well known {place_type}."
|
|
||||||
elif len(tags.keys()) < 17 :
|
|
||||||
importance = 'national'
|
|
||||||
description = f"{name} is a {place_type} of national importance."
|
|
||||||
else :
|
|
||||||
importance = 'international'
|
|
||||||
description = f"{name} is an internationally famous {place_type}."
|
|
||||||
else :
|
|
||||||
description = f"{name} is a {place_type} of {importance} importance."
|
|
||||||
|
|
||||||
if height is not None and date is not None :
|
|
||||||
description += f" This {place_type} was constructed in {date} and is ca. {height} meters high."
|
|
||||||
elif height is not None :
|
|
||||||
description += f" This {place_type} stands ca. {height} meters tall."
|
|
||||||
elif date is not None:
|
|
||||||
description += f" It was constructed in {date}."
|
|
||||||
|
|
||||||
# Format the visitor number
|
|
||||||
if n_visitors is not None :
|
|
||||||
n_visitors = int(n_visitors)
|
|
||||||
if n_visitors < 1000000 :
|
|
||||||
description += f" It welcomes {int(n_visitors/1000)} thousand visitors every year."
|
|
||||||
else :
|
|
||||||
description += f" It welcomes {round(n_visitors/1000000, 1)} million visitors every year."
|
|
||||||
|
|
||||||
# Set the keywords.
|
|
||||||
keywords = {"importance": importance,
|
|
||||||
"height": height,
|
|
||||||
"place_type": place_type,
|
|
||||||
"date": date}
|
|
||||||
|
|
||||||
return description, keywords
|
|
||||||
|
|
||||||
|
|
||||||
def get_place_type(self, data):
|
|
||||||
"""
|
|
||||||
Determines the type of the place based on available tags such as 'amenity', 'building',
|
|
||||||
'historic', and 'leisure'. The priority order is: 'historic' > 'building' (if not generic) >
|
|
||||||
'amenity' > 'leisure'.
|
|
||||||
|
|
||||||
Params:
|
|
||||||
data (dict): A dictionary containing metadata about the place.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
place_type (str): The determined type of the place, or None if no relevant type is found.
|
|
||||||
"""
|
|
||||||
amenity = data.get('amenity', None)
|
|
||||||
building = data.get('building', None)
|
|
||||||
historic = data.get('historic', None)
|
|
||||||
leisure = data.get('leisure')
|
|
||||||
|
|
||||||
if historic and historic != "yes":
|
|
||||||
return historic
|
|
||||||
if building and building not in ["yes", "civic", "government", "apartments", "residential", "commericial", "industrial", "retail", "religious", "public", "service"]:
|
|
||||||
return building
|
|
||||||
if amenity:
|
|
||||||
return amenity
|
|
||||||
if leisure:
|
|
||||||
return leisure
|
|
||||||
|
|
||||||
|
|
||||||
return None
|
|
||||||
|
|
||||||
|
|
||||||
def get_date(self, data):
|
|
||||||
"""
|
|
||||||
Extracts the most relevant date from the available tags, prioritizing 'construction_date',
|
|
||||||
'start_date', 'year_of_construction', and 'opening_date' in that order.
|
|
||||||
|
|
||||||
Params:
|
|
||||||
data (dict): A dictionary containing metadata about the place.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
date (str): The most relevant date found, or None if no date is available.
|
|
||||||
"""
|
|
||||||
construction_date = data.get('construction_date', None)
|
|
||||||
opening_date = data.get('opening_date', None)
|
|
||||||
start_date = data.get('start_date', None)
|
|
||||||
year_of_construction = data.get('year_of_construction', None)
|
|
||||||
|
|
||||||
# Prioritize based on availability
|
|
||||||
if construction_date:
|
|
||||||
return construction_date
|
|
||||||
if start_date:
|
|
||||||
return start_date
|
|
||||||
if year_of_construction:
|
|
||||||
return year_of_construction
|
|
||||||
if opening_date:
|
|
||||||
return opening_date
|
|
||||||
|
|
||||||
return None
|
|
||||||
|
|
||||||
|
|
||||||
def dict_to_selector_list(d: dict) -> list:
|
def dict_to_selector_list(d: dict) -> list:
|
||||||
"""
|
"""
|
||||||
Convert a dictionary of key-value pairs to a list of Overpass query strings.
|
Convert a dictionary of key-value pairs to a list of Overpass query strings.
|
||||||
|
117
backend/src/landmarks/landmarks_routes.py
Normal file
117
backend/src/landmarks/landmarks_routes.py
Normal file
@ -0,0 +1,117 @@
|
|||||||
|
"""Main app for backend api"""
|
||||||
|
import logging
|
||||||
|
import time
|
||||||
|
import random
|
||||||
|
from fastapi import HTTPException, APIRouter
|
||||||
|
|
||||||
|
from ..structs.landmark import Landmark
|
||||||
|
from ..structs.preferences import Preferences, Preference
|
||||||
|
from ..landmarks.landmarks_manager import LandmarkManager
|
||||||
|
|
||||||
|
|
||||||
|
# Setup the logger and the Landmarks Manager
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
manager = LandmarkManager()
|
||||||
|
|
||||||
|
|
||||||
|
# Start the router
|
||||||
|
router = APIRouter()
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/get/landmarks")
|
||||||
|
def get_landmarks(
|
||||||
|
preferences: Preferences,
|
||||||
|
start: tuple[float, float],
|
||||||
|
) -> list[Landmark]:
|
||||||
|
"""
|
||||||
|
Function that returns all available landmarks given some preferences and a start position.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
preferences : the preferences specified by the user as the post body
|
||||||
|
start : the coordinates of the starting point
|
||||||
|
Returns:
|
||||||
|
list[Landmark] : The full list of fetched landmarks
|
||||||
|
"""
|
||||||
|
if preferences is None:
|
||||||
|
raise HTTPException(status_code=406, detail="Preferences not provided or incomplete.")
|
||||||
|
if (preferences.shopping.score == 0 and
|
||||||
|
preferences.sightseeing.score == 0 and
|
||||||
|
preferences.nature.score == 0) :
|
||||||
|
raise HTTPException(status_code=406, detail="All preferences are 0.")
|
||||||
|
if start is None:
|
||||||
|
raise HTTPException(status_code=406, detail="Start coordinates not provided")
|
||||||
|
if not (-90 <= start[0] <= 90 or -180 <= start[1] <= 180):
|
||||||
|
raise HTTPException(status_code=422, detail="Start coordinates not in range")
|
||||||
|
|
||||||
|
logger.info(f"Requested new trip generation. Details:\n\tCoordinates: {start}\n\tTime: {preferences.max_time_minute}\n\tSightseeing: {preferences.sightseeing.score}\n\tNature: {preferences.nature.score}\n\tShopping: {preferences.shopping.score}")
|
||||||
|
|
||||||
|
start_time = time.time()
|
||||||
|
|
||||||
|
# Generate the landmarks from the start location
|
||||||
|
landmarks, _ = manager.generate_landmarks_list(
|
||||||
|
center_coordinates = start,
|
||||||
|
preferences = preferences
|
||||||
|
)
|
||||||
|
|
||||||
|
if len(landmarks) == 0 :
|
||||||
|
raise HTTPException(status_code=500, detail="No landmarks were found.")
|
||||||
|
|
||||||
|
t_generate_landmarks = time.time() - start_time
|
||||||
|
logger.info(f'Fetched {len(landmarks)} landmarks in \t: {round(t_generate_landmarks,3)} seconds')
|
||||||
|
|
||||||
|
return landmarks
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/landmarks/get-nearby/{lat}/{lon}")
|
||||||
|
def get_landmarks_nearby(
|
||||||
|
lat: float,
|
||||||
|
lon: float
|
||||||
|
) -> list[Landmark] :
|
||||||
|
"""
|
||||||
|
Suggests nearby landmarks based on a given latitude and longitude.
|
||||||
|
|
||||||
|
This endpoint returns a curated list of up to 5 landmarks around the given geographical coordinates. It uses fixed preferences for
|
||||||
|
sightseeing, shopping, and nature, with a maximum time constraint of 30 minutes to limit the number of landmarks returned.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
lat (float): Latitude of the user's current location.
|
||||||
|
lon (float): Longitude of the user's current location.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
list[Landmark]: A list of selected nearby landmarks.
|
||||||
|
"""
|
||||||
|
logger.info(f'Fetching landmarks nearby ({lat}, {lon}).')
|
||||||
|
# Define fixed preferences:
|
||||||
|
prefs = Preferences(
|
||||||
|
sightseeing = Preference(
|
||||||
|
type='sightseeing',
|
||||||
|
score=5
|
||||||
|
),
|
||||||
|
shopping = Preference(
|
||||||
|
type='shopping',
|
||||||
|
score=2
|
||||||
|
),
|
||||||
|
nature = Preference(
|
||||||
|
type='nature',
|
||||||
|
score=5
|
||||||
|
),
|
||||||
|
max_time_minute=30,
|
||||||
|
detour_tolerance_minute=0,
|
||||||
|
)
|
||||||
|
|
||||||
|
# Find the landmarks around the location
|
||||||
|
_, landmarks_around = manager.generate_landmarks_list(
|
||||||
|
center_coordinates = (lat, lon),
|
||||||
|
preferences = prefs,
|
||||||
|
allow_clusters=False,
|
||||||
|
)
|
||||||
|
if len(landmarks_around) == 0 :
|
||||||
|
raise HTTPException(status_code=500, detail="No landmarks were found.")
|
||||||
|
|
||||||
|
# select 5 landmarks from there
|
||||||
|
if len(landmarks_around) > 6 :
|
||||||
|
landmarks_around = landmarks_around[:2] + random.sample(landmarks_around[3:], 2)
|
||||||
|
|
||||||
|
logger.info(f'Found {len(landmarks_around)} landmarks nearby ({lat}, {lon}).')
|
||||||
|
logger.debug('Suggested landmarks :\n\t' + '\n\t'.join(f'{landmark}' for landmark in landmarks_around))
|
||||||
|
return landmarks_around
|
@ -33,14 +33,14 @@ def configure_logging():
|
|||||||
# silence the chatty logs loki generates itself
|
# silence the chatty logs loki generates itself
|
||||||
logging.getLogger('urllib3.connectionpool').setLevel(logging.WARNING)
|
logging.getLogger('urllib3.connectionpool').setLevel(logging.WARNING)
|
||||||
# no need for time since it's added by loki or can be shown in kube logs
|
# no need for time since it's added by loki or can be shown in kube logs
|
||||||
logging_format = '%(name)s - %(levelname)s - %(message)s'
|
logging_format = '%(name)-55s - %(levelname)-7s - %(message)s'
|
||||||
|
|
||||||
else:
|
else:
|
||||||
# if we are in a debug (local) session, set verbose and rich logging
|
# if we are in a debug (local) session, set verbose and rich logging
|
||||||
from rich.logging import RichHandler
|
from rich.logging import RichHandler
|
||||||
logging_handlers = [RichHandler()]
|
logging_handlers = [RichHandler()]
|
||||||
logging_level = logging.DEBUG if is_debug else logging.INFO
|
logging_level = logging.DEBUG if is_debug else logging.INFO
|
||||||
logging_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
logging_format = '%(asctime)s - %(name)-55s - %(levelname)-7s - %(message)s'
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
@ -1,16 +1,19 @@
|
|||||||
"""Main app for backend api"""
|
"""Main app for backend api"""
|
||||||
import logging
|
import logging
|
||||||
import time
|
import time
|
||||||
|
import random
|
||||||
from contextlib import asynccontextmanager
|
from contextlib import asynccontextmanager
|
||||||
from fastapi import FastAPI, HTTPException, BackgroundTasks
|
from fastapi import FastAPI, HTTPException, BackgroundTasks
|
||||||
|
|
||||||
from .logging_config import configure_logging
|
from .logging_config import configure_logging
|
||||||
from .structs.landmark import Landmark
|
from .structs.landmark import Landmark
|
||||||
from .structs.preferences import Preferences
|
from .structs.preferences import Preferences, Preference
|
||||||
from .structs.linked_landmarks import LinkedLandmarks
|
from .structs.linked_landmarks import LinkedLandmarks
|
||||||
from .structs.trip import Trip
|
from .structs.trip import Trip
|
||||||
from .landmarks.landmarks_manager import LandmarkManager
|
from .landmarks.landmarks_manager import LandmarkManager
|
||||||
from .toilets.toilet_routes import router as toilets_router
|
from .toilets.toilets_route import router as toilets_router
|
||||||
|
from .optimization.optimization_routes import router as optimization_router
|
||||||
|
from .landmarks.landmarks_routes import router as landmarks_router
|
||||||
from .optimization.optimizer import Optimizer
|
from .optimization.optimizer import Optimizer
|
||||||
from .optimization.refiner import Refiner
|
from .optimization.refiner import Refiner
|
||||||
from .overpass.overpass import fill_cache
|
from .overpass.overpass import fill_cache
|
||||||
@ -38,13 +41,17 @@ app = FastAPI(lifespan=lifespan)
|
|||||||
|
|
||||||
|
|
||||||
app.include_router(toilets_router)
|
app.include_router(toilets_router)
|
||||||
|
app.include_router(optimization_router)
|
||||||
|
app.include_router(landmarks_router)
|
||||||
|
|
||||||
|
|
||||||
@app.post("/trip/new")
|
@app.post("/trip/new")
|
||||||
def new_trip(preferences: Preferences,
|
def new_trip(
|
||||||
start: tuple[float, float],
|
preferences: Preferences,
|
||||||
end: tuple[float, float] | None = None,
|
start: tuple[float, float],
|
||||||
background_tasks: BackgroundTasks = None) -> Trip:
|
end: tuple[float, float] | None = None,
|
||||||
|
background_tasks: BackgroundTasks = None
|
||||||
|
) -> Trip:
|
||||||
"""
|
"""
|
||||||
Main function to call the optimizer.
|
Main function to call the optimizer.
|
||||||
|
|
||||||
@ -140,7 +147,7 @@ def new_trip(preferences: Preferences,
|
|||||||
# upon creation of the trip, persistence of both the trip and its landmarks is ensured.
|
# upon creation of the trip, persistence of both the trip and its landmarks is ensured.
|
||||||
trip = Trip.from_linked_landmarks(linked_tour, cache_client)
|
trip = Trip.from_linked_landmarks(linked_tour, cache_client)
|
||||||
logger.info(f'Generated a trip of {trip.total_time} minutes with {len(refined_tour)} landmarks in {round(t_generate_landmarks + t_first_stage + t_second_stage,3)} seconds.')
|
logger.info(f'Generated a trip of {trip.total_time} minutes with {len(refined_tour)} landmarks in {round(t_generate_landmarks + t_first_stage + t_second_stage,3)} seconds.')
|
||||||
logger.debug('Detailed trip :\n\t' + '\n\t'.join(f'{landmark}' for landmark in refined_tour))
|
logger.info('Detailed trip :\n\t' + '\n\t'.join(f'{landmark}' for landmark in refined_tour))
|
||||||
|
|
||||||
background_tasks.add_task(fill_cache)
|
background_tasks.add_task(fill_cache)
|
||||||
|
|
||||||
@ -224,3 +231,4 @@ def update_trip_time(trip_uuid: str, removed_landmark_uuid: str) -> Trip:
|
|||||||
trip = Trip.from_linked_landmarks(linked_tour, cache_client)
|
trip = Trip.from_linked_landmarks(linked_tour, cache_client)
|
||||||
|
|
||||||
return trip
|
return trip
|
||||||
|
|
||||||
|
135
backend/src/optimization/optimization_routes.py
Normal file
135
backend/src/optimization/optimization_routes.py
Normal file
@ -0,0 +1,135 @@
|
|||||||
|
"""Main app for backend api"""
|
||||||
|
import logging
|
||||||
|
import time
|
||||||
|
import yaml
|
||||||
|
from fastapi import HTTPException, APIRouter, BackgroundTasks
|
||||||
|
|
||||||
|
from ..structs.landmark import Landmark
|
||||||
|
from ..structs.preferences import Preferences
|
||||||
|
from ..structs.linked_landmarks import LinkedLandmarks
|
||||||
|
from ..utils.take_most_important import take_most_important
|
||||||
|
from ..structs.trip import Trip
|
||||||
|
from ..optimization.optimizer import Optimizer
|
||||||
|
from ..optimization.refiner import Refiner
|
||||||
|
from ..overpass.overpass import fill_cache
|
||||||
|
from ..cache import client as cache_client
|
||||||
|
from ..constants import LANDMARK_PARAMETERS_PATH
|
||||||
|
|
||||||
|
|
||||||
|
# Setup the Logger, Optimizer and Refiner
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
optimizer = Optimizer()
|
||||||
|
refiner = Refiner(optimizer=optimizer)
|
||||||
|
|
||||||
|
|
||||||
|
# Start the router
|
||||||
|
router = APIRouter()
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/optimize/trip")
|
||||||
|
def optimize_trip(
|
||||||
|
preferences: Preferences,
|
||||||
|
landmarks: list[Landmark],
|
||||||
|
start: tuple[float, float],
|
||||||
|
end: tuple[float, float] | None = None,
|
||||||
|
background_tasks: BackgroundTasks = None
|
||||||
|
) -> Trip:
|
||||||
|
"""
|
||||||
|
Main function to call the optimizer.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
preferences (Preferences) : the preferences specified by the user as the post body.
|
||||||
|
start (tuple[float, float]) : the coordinates of the starting point.
|
||||||
|
end tuple[float, float] : the coordinates of the finishing point.
|
||||||
|
backgroud_tasks (BackgroundTasks) : necessary to fill the cache after the trip has been returned.
|
||||||
|
Returns:
|
||||||
|
(uuid) : The uuid of the first landmark in the optimized route
|
||||||
|
"""
|
||||||
|
if preferences is None:
|
||||||
|
raise HTTPException(status_code=406, detail="Preferences not provided or incomplete.")
|
||||||
|
if len(landmarks) == 0 :
|
||||||
|
raise HTTPException(status_code=406, detail="No landmarks provided for computing the trip.")
|
||||||
|
if (preferences.shopping.score == 0 and
|
||||||
|
preferences.sightseeing.score == 0 and
|
||||||
|
preferences.nature.score == 0) :
|
||||||
|
raise HTTPException(status_code=406, detail="All preferences are 0.")
|
||||||
|
if start is None:
|
||||||
|
raise HTTPException(status_code=406, detail="Start coordinates not provided")
|
||||||
|
if not (-90 <= start[0] <= 90 or -180 <= start[1] <= 180):
|
||||||
|
raise HTTPException(status_code=422, detail="Start coordinates not in range")
|
||||||
|
if end is None:
|
||||||
|
end = start
|
||||||
|
logger.info("No end coordinates provided. Using start=end.")
|
||||||
|
|
||||||
|
# Start the timer
|
||||||
|
start_time = time.time()
|
||||||
|
|
||||||
|
logger.info(f"Requested new trip generation. Details:\n\tCoordinates: {start}\n\tTime: {preferences.max_time_minute}\n\tSightseeing: {preferences.sightseeing.score}\n\tNature: {preferences.nature.score}\n\tShopping: {preferences.shopping.score}")
|
||||||
|
|
||||||
|
start_landmark = Landmark(name='start',
|
||||||
|
type='start',
|
||||||
|
location=(start[0], start[1]),
|
||||||
|
osm_type='start',
|
||||||
|
osm_id=0,
|
||||||
|
attractiveness=0,
|
||||||
|
duration=0,
|
||||||
|
must_do=True,
|
||||||
|
n_tags = 0)
|
||||||
|
|
||||||
|
end_landmark = Landmark(name='finish',
|
||||||
|
type='finish',
|
||||||
|
location=(end[0], end[1]),
|
||||||
|
osm_type='end',
|
||||||
|
osm_id=0,
|
||||||
|
attractiveness=0,
|
||||||
|
duration=0,
|
||||||
|
must_do=True,
|
||||||
|
n_tags=0)
|
||||||
|
|
||||||
|
# From the parameters load the length at which to truncate the landmarks list.
|
||||||
|
with LANDMARK_PARAMETERS_PATH.open('r') as f:
|
||||||
|
parameters = yaml.safe_load(f)
|
||||||
|
n_important = parameters['N_important']
|
||||||
|
|
||||||
|
# Truncate to the most important landmarks for a shorter list
|
||||||
|
landmarks_short = take_most_important(landmarks, n_important)
|
||||||
|
|
||||||
|
# insert start and finish to the shorter landmarks list
|
||||||
|
landmarks_short.insert(0, start_landmark)
|
||||||
|
landmarks_short.append(end_landmark)
|
||||||
|
|
||||||
|
# First stage optimization
|
||||||
|
try:
|
||||||
|
base_tour = optimizer.solve_optimization(preferences.max_time_minute, landmarks_short)
|
||||||
|
except Exception as exc:
|
||||||
|
logger.error(f"Trip generation failed: {str(exc)}")
|
||||||
|
raise HTTPException(status_code=500, detail=f"Optimization failed: {str(exc)}") from exc
|
||||||
|
|
||||||
|
t_first_stage = time.time() - start_time
|
||||||
|
start_time = time.time()
|
||||||
|
|
||||||
|
# Second stage optimization
|
||||||
|
try :
|
||||||
|
refined_tour = refiner.refine_optimization(landmarks, base_tour,
|
||||||
|
preferences.max_time_minute,
|
||||||
|
preferences.detour_tolerance_minute)
|
||||||
|
except Exception as exc :
|
||||||
|
logger.warning(f"Refiner failed. Proceeding with base trip {str(exc)}")
|
||||||
|
refined_tour = base_tour
|
||||||
|
|
||||||
|
t_second_stage = time.time() - start_time
|
||||||
|
|
||||||
|
logger.debug(f'First stage optimization\t: {round(t_first_stage,3)} seconds')
|
||||||
|
logger.debug(f'Second stage optimization\t: {round(t_second_stage,3)} seconds')
|
||||||
|
logger.info(f'Total computation time\t: {round(t_first_stage + t_second_stage,3)} seconds')
|
||||||
|
linked_tour = LinkedLandmarks(refined_tour)
|
||||||
|
|
||||||
|
# upon creation of the trip, persistence of both the trip and its landmarks is ensured.
|
||||||
|
trip = Trip.from_linked_landmarks(linked_tour, cache_client)
|
||||||
|
logger.info(f'Optimized a trip of {trip.total_time} minutes with {len(refined_tour)} landmarks in {round(t_first_stage + t_second_stage,3)} seconds.')
|
||||||
|
logger.info('Detailed trip :\n\t' + '\n\t'.join(f'{landmark}' for landmark in refined_tour))
|
||||||
|
|
||||||
|
background_tasks.add_task(fill_cache)
|
||||||
|
|
||||||
|
return trip
|
||||||
|
|
@ -402,6 +402,8 @@ def fill_cache():
|
|||||||
n_files = 0
|
n_files = 0
|
||||||
total = 0
|
total = 0
|
||||||
|
|
||||||
|
overpass.logger.info('Trip successfully returned, starting to fill cache.')
|
||||||
|
|
||||||
with os.scandir(OSM_CACHE_DIR) as it:
|
with os.scandir(OSM_CACHE_DIR) as it:
|
||||||
for entry in it:
|
for entry in it:
|
||||||
if entry.is_file() and entry.name.startswith('hollow_'):
|
if entry.is_file() and entry.name.startswith('hollow_'):
|
||||||
|
@ -2,6 +2,7 @@
|
|||||||
|
|
||||||
from .landmark import Landmark
|
from .landmark import Landmark
|
||||||
from ..utils.get_time_distance import get_time
|
from ..utils.get_time_distance import get_time
|
||||||
|
from ..utils.description import description_and_keywords
|
||||||
|
|
||||||
class LinkedLandmarks:
|
class LinkedLandmarks:
|
||||||
"""
|
"""
|
||||||
@ -35,18 +36,23 @@ class LinkedLandmarks:
|
|||||||
Create the links between the landmarks in the list by setting their
|
Create the links between the landmarks in the list by setting their
|
||||||
.next_uuid and the .time_to_next attributes.
|
.next_uuid and the .time_to_next attributes.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
# Mark secondary landmarks as such
|
# Mark secondary landmarks as such
|
||||||
self.update_secondary_landmarks()
|
self.update_secondary_landmarks()
|
||||||
|
|
||||||
|
|
||||||
for i, landmark in enumerate(self._landmarks[:-1]):
|
for i, landmark in enumerate(self._landmarks[:-1]):
|
||||||
|
# Set uuid of the next landmark
|
||||||
landmark.next_uuid = self._landmarks[i + 1].uuid
|
landmark.next_uuid = self._landmarks[i + 1].uuid
|
||||||
|
|
||||||
|
# Adjust time to reach and total time
|
||||||
time_to_next = get_time(landmark.location, self._landmarks[i + 1].location)
|
time_to_next = get_time(landmark.location, self._landmarks[i + 1].location)
|
||||||
landmark.time_to_reach_next = time_to_next
|
landmark.time_to_reach_next = time_to_next
|
||||||
self.total_time += time_to_next
|
self.total_time += time_to_next
|
||||||
self.total_time += landmark.duration
|
self.total_time += landmark.duration
|
||||||
|
|
||||||
|
# Fill in the keywords and description. GOOD IDEA, BAD EXECUTION, tags aren't available anymore at this stage
|
||||||
|
# landmark.description, landmark.keywords = description_and_keywords(tags)
|
||||||
|
|
||||||
|
|
||||||
self._landmarks[-1].next_uuid = None
|
self._landmarks[-1].next_uuid = None
|
||||||
self._landmarks[-1].time_to_reach_next = 0
|
self._landmarks[-1].time_to_reach_next = 0
|
||||||
|
|
||||||
|
@ -1,4 +1,4 @@
|
|||||||
"""Collection of tests to ensure correct implementation and track progress. """
|
"""Collection of tests to ensure correct implementation and track progress."""
|
||||||
import time
|
import time
|
||||||
from fastapi.testclient import TestClient
|
from fastapi.testclient import TestClient
|
||||||
import pytest
|
import pytest
|
||||||
@ -6,6 +6,7 @@ import pytest
|
|||||||
from .test_utils import load_trip_landmarks, log_trip_details
|
from .test_utils import load_trip_landmarks, log_trip_details
|
||||||
from ..main import app
|
from ..main import app
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="module")
|
@pytest.fixture(scope="module")
|
||||||
def client():
|
def client():
|
||||||
"""Client used to call the app."""
|
"""Client used to call the app."""
|
||||||
@ -88,15 +89,13 @@ def test_bellecour(client, request) : # pylint: disable=redefined-outer-name
|
|||||||
# Add details to report
|
# Add details to report
|
||||||
log_trip_details(request, landmarks, result['total_time'], duration_minutes)
|
log_trip_details(request, landmarks, result['total_time'], duration_minutes)
|
||||||
|
|
||||||
# for elem in landmarks :
|
|
||||||
# print(elem)
|
|
||||||
|
|
||||||
# checks :
|
# checks :
|
||||||
assert response.status_code == 200 # check for successful planning
|
assert response.status_code == 200 # check for successful planning
|
||||||
assert comp_time < 30, f"Computation time exceeded 30 seconds: {comp_time:.2f} seconds"
|
assert comp_time < 30, f"Computation time exceeded 30 seconds: {comp_time:.2f} seconds"
|
||||||
assert duration_minutes*0.8 < result['total_time'], f"Trip too short: {result['total_time']} instead of {duration_minutes}"
|
assert duration_minutes*0.8 < result['total_time'], f"Trip too short: {result['total_time']} instead of {duration_minutes}"
|
||||||
assert duration_minutes*1.2 > result['total_time'], f"Trip too long: {result['total_time']} instead of {duration_minutes}"
|
assert duration_minutes*1.2 > result['total_time'], f"Trip too long: {result['total_time']} instead of {duration_minutes}"
|
||||||
|
|
||||||
|
|
||||||
def test_cologne(client, request) : # pylint: disable=redefined-outer-name
|
def test_cologne(client, request) : # pylint: disable=redefined-outer-name
|
||||||
"""
|
"""
|
||||||
Test n°3 : Custom test in Cologne to ensure proper decision making in crowded area.
|
Test n°3 : Custom test in Cologne to ensure proper decision making in crowded area.
|
||||||
@ -128,9 +127,6 @@ def test_cologne(client, request) : # pylint: disable=redefined-outer-name
|
|||||||
# Add details to report
|
# Add details to report
|
||||||
log_trip_details(request, landmarks, result['total_time'], duration_minutes)
|
log_trip_details(request, landmarks, result['total_time'], duration_minutes)
|
||||||
|
|
||||||
# for elem in landmarks :
|
|
||||||
# print(elem)
|
|
||||||
|
|
||||||
# checks :
|
# checks :
|
||||||
assert response.status_code == 200 # check for successful planning
|
assert response.status_code == 200 # check for successful planning
|
||||||
assert comp_time < 30, f"Computation time exceeded 30 seconds: {comp_time:.2f} seconds"
|
assert comp_time < 30, f"Computation time exceeded 30 seconds: {comp_time:.2f} seconds"
|
||||||
@ -169,9 +165,6 @@ def test_strasbourg(client, request) : # pylint: disable=redefined-outer-name
|
|||||||
# Add details to report
|
# Add details to report
|
||||||
log_trip_details(request, landmarks, result['total_time'], duration_minutes)
|
log_trip_details(request, landmarks, result['total_time'], duration_minutes)
|
||||||
|
|
||||||
# for elem in landmarks :
|
|
||||||
# print(elem)
|
|
||||||
|
|
||||||
# checks :
|
# checks :
|
||||||
assert response.status_code == 200 # check for successful planning
|
assert response.status_code == 200 # check for successful planning
|
||||||
assert comp_time < 30, f"Computation time exceeded 30 seconds: {comp_time:.2f} seconds"
|
assert comp_time < 30, f"Computation time exceeded 30 seconds: {comp_time:.2f} seconds"
|
||||||
@ -210,9 +203,6 @@ def test_zurich(client, request) : # pylint: disable=redefined-outer-name
|
|||||||
# Add details to report
|
# Add details to report
|
||||||
log_trip_details(request, landmarks, result['total_time'], duration_minutes)
|
log_trip_details(request, landmarks, result['total_time'], duration_minutes)
|
||||||
|
|
||||||
# for elem in landmarks :
|
|
||||||
# print(elem)
|
|
||||||
|
|
||||||
# checks :
|
# checks :
|
||||||
assert response.status_code == 200 # check for successful planning
|
assert response.status_code == 200 # check for successful planning
|
||||||
assert comp_time < 30, f"Computation time exceeded 30 seconds: {comp_time:.2f} seconds"
|
assert comp_time < 30, f"Computation time exceeded 30 seconds: {comp_time:.2f} seconds"
|
||||||
@ -251,9 +241,6 @@ def test_paris(client, request) : # pylint: disable=redefined-outer-name
|
|||||||
# Add details to report
|
# Add details to report
|
||||||
log_trip_details(request, landmarks, result['total_time'], duration_minutes)
|
log_trip_details(request, landmarks, result['total_time'], duration_minutes)
|
||||||
|
|
||||||
# for elem in landmarks :
|
|
||||||
# print(elem)
|
|
||||||
|
|
||||||
# checks :
|
# checks :
|
||||||
assert response.status_code == 200 # check for successful planning
|
assert response.status_code == 200 # check for successful planning
|
||||||
assert comp_time < 30, f"Computation time exceeded 30 seconds: {comp_time:.2f} seconds"
|
assert comp_time < 30, f"Computation time exceeded 30 seconds: {comp_time:.2f} seconds"
|
||||||
@ -292,9 +279,6 @@ def test_new_york(client, request) : # pylint: disable=redefined-outer-name
|
|||||||
# Add details to report
|
# Add details to report
|
||||||
log_trip_details(request, landmarks, result['total_time'], duration_minutes)
|
log_trip_details(request, landmarks, result['total_time'], duration_minutes)
|
||||||
|
|
||||||
# for elem in landmarks :
|
|
||||||
# print(elem)
|
|
||||||
|
|
||||||
# checks :
|
# checks :
|
||||||
assert response.status_code == 200 # check for successful planning
|
assert response.status_code == 200 # check for successful planning
|
||||||
assert comp_time < 30, f"Computation time exceeded 30 seconds: {comp_time:.2f} seconds"
|
assert comp_time < 30, f"Computation time exceeded 30 seconds: {comp_time:.2f} seconds"
|
||||||
@ -333,9 +317,6 @@ def test_shopping(client, request) : # pylint: disable=redefined-outer-name
|
|||||||
# Add details to report
|
# Add details to report
|
||||||
log_trip_details(request, landmarks, result['total_time'], duration_minutes)
|
log_trip_details(request, landmarks, result['total_time'], duration_minutes)
|
||||||
|
|
||||||
# for elem in landmarks :
|
|
||||||
# print(elem)
|
|
||||||
|
|
||||||
# checks :
|
# checks :
|
||||||
assert response.status_code == 200 # check for successful planning
|
assert response.status_code == 200 # check for successful planning
|
||||||
assert comp_time < 30, f"Computation time exceeded 30 seconds: {comp_time:.2f} seconds"
|
assert comp_time < 30, f"Computation time exceeded 30 seconds: {comp_time:.2f} seconds"
|
||||||
|
38
backend/src/tests/test_nearby.py
Normal file
38
backend/src/tests/test_nearby.py
Normal file
@ -0,0 +1,38 @@
|
|||||||
|
"""Collection of tests to ensure correct implementation and track progress of the get_landmarks_nearby feature. """
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from ..main import app
|
||||||
|
|
||||||
|
@pytest.fixture(scope="module")
|
||||||
|
def client():
|
||||||
|
"""Client used to call the app."""
|
||||||
|
return TestClient(app)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"location,status_code",
|
||||||
|
[
|
||||||
|
([45.7576485, 4.8330241], 2000), # Lyon, Bellecour.
|
||||||
|
([41.4020572, 2.1818985], 2000), # Barcelona
|
||||||
|
]
|
||||||
|
)
|
||||||
|
def test_nearby(client, location, status_code): # pylint: disable=redefined-outer-name
|
||||||
|
"""
|
||||||
|
Test n°1 : Verify handling of invalid input.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
client:
|
||||||
|
request:
|
||||||
|
"""
|
||||||
|
response = client.post(f"/landmarks/get-nearby/{location[0]}/{location[1]}")
|
||||||
|
print(response)
|
||||||
|
print(response.json())
|
||||||
|
suggestions = response.json()
|
||||||
|
|
||||||
|
# checks :
|
||||||
|
assert response.status_code == status_code # check for successful planning
|
||||||
|
assert isinstance(suggestions, list) # check that the return type is a list
|
||||||
|
assert len(suggestions) > 0
|
80
backend/src/tests/test_trip_generation.py
Normal file
80
backend/src/tests/test_trip_generation.py
Normal file
@ -0,0 +1,80 @@
|
|||||||
|
"""Collection of tests to ensure correct implementation and track progress."""
|
||||||
|
import time
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from .test_utils import load_trip_landmarks, log_trip_details
|
||||||
|
from ..structs.preferences import Preferences, Preference
|
||||||
|
from ..main import app
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(scope="module")
|
||||||
|
def client():
|
||||||
|
"""Client used to call the app."""
|
||||||
|
return TestClient(app)
|
||||||
|
|
||||||
|
|
||||||
|
def test_bellecour(client, request) : # pylint: disable=redefined-outer-name
|
||||||
|
"""
|
||||||
|
Test n°2 : Custom test in Lyon centre to ensure proper decision making in crowded area.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
client:
|
||||||
|
request:
|
||||||
|
"""
|
||||||
|
start_time = time.time() # Start timer
|
||||||
|
|
||||||
|
# Step 0: Define the trip preferences
|
||||||
|
prefs = Preferences(
|
||||||
|
sightseeing = Preference(
|
||||||
|
type='sightseeing',
|
||||||
|
score=5
|
||||||
|
),
|
||||||
|
shopping = Preference(
|
||||||
|
type='shopping',
|
||||||
|
score=5
|
||||||
|
),
|
||||||
|
nature = Preference(
|
||||||
|
type='nature',
|
||||||
|
score=5
|
||||||
|
),
|
||||||
|
max_time_minute=120,
|
||||||
|
detour_tolerance_minute=0,
|
||||||
|
)
|
||||||
|
|
||||||
|
# Define the starting coordinates
|
||||||
|
start = [45.7576485, 4.8330241]
|
||||||
|
|
||||||
|
# Step 1: request the list of landmarks in the vicinty of the starting point
|
||||||
|
response = client.post(
|
||||||
|
"/get/landmarks",
|
||||||
|
json={
|
||||||
|
"preferences": prefs.model_dump(),
|
||||||
|
"start": start
|
||||||
|
}
|
||||||
|
)
|
||||||
|
landmarks = response.json()
|
||||||
|
|
||||||
|
# Step 2: Feed the landmarks to the optimizer to compute the trip
|
||||||
|
response = client.post(
|
||||||
|
"/optimize/trip",
|
||||||
|
json={
|
||||||
|
"preferences": prefs.model_dump(),
|
||||||
|
"landmarks": landmarks,
|
||||||
|
"start": start
|
||||||
|
}
|
||||||
|
)
|
||||||
|
result = response.json()
|
||||||
|
landmarks = load_trip_landmarks(client, result['first_landmark_uuid'])
|
||||||
|
|
||||||
|
# Get computation time
|
||||||
|
comp_time = time.time() - start_time
|
||||||
|
|
||||||
|
# Add details to report
|
||||||
|
log_trip_details(request, landmarks, result['total_time'], prefs.max_time_minute)
|
||||||
|
|
||||||
|
# checks :
|
||||||
|
assert response.status_code == 200 # check for successful planning
|
||||||
|
assert comp_time < 30, f"Computation time exceeded 30 seconds: {comp_time:.2f} seconds"
|
||||||
|
assert prefs.max_time_minute*0.8 < result['total_time'], f"Trip too short: {result['total_time']} instead of {prefs.max_time_minute}"
|
||||||
|
assert prefs.max_time_minute*1.2 > result['total_time'], f"Trip too long: {result['total_time']} instead of {prefs.max_time_minute}"
|
123
backend/src/utils/description.py
Normal file
123
backend/src/utils/description.py
Normal file
@ -0,0 +1,123 @@
|
|||||||
|
"""Add more information about the landmarks by writing a short description and keywords. """
|
||||||
|
|
||||||
|
|
||||||
|
def description_and_keywords(tags: dict):
|
||||||
|
"""
|
||||||
|
Generates a description and a set of keywords for a given landmark based on its tags.
|
||||||
|
|
||||||
|
Params:
|
||||||
|
tags (dict): A dictionary containing metadata about the landmark, including its name,
|
||||||
|
importance, height, date of construction, and visitor information.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
description (str): A string description of the landmark.
|
||||||
|
keywords (dict): A dictionary of keywords with fields such as 'importance', 'height',
|
||||||
|
'place_type', and 'date'.
|
||||||
|
"""
|
||||||
|
# Extract relevant fields
|
||||||
|
name = tags.get('name')
|
||||||
|
importance = tags.get('importance', None)
|
||||||
|
n_visitors = tags.get('tourism:visitors', None)
|
||||||
|
height = tags.get('height')
|
||||||
|
place_type = get_place_type(tags)
|
||||||
|
date = get_date(tags)
|
||||||
|
|
||||||
|
if place_type is None :
|
||||||
|
return None, None
|
||||||
|
|
||||||
|
# Start the description.
|
||||||
|
if importance is None :
|
||||||
|
if len(tags.keys()) < 5 :
|
||||||
|
return None, None
|
||||||
|
if len(tags.keys()) < 10 :
|
||||||
|
description = f"{name} is a well known {place_type}."
|
||||||
|
elif len(tags.keys()) < 17 :
|
||||||
|
importance = 'national'
|
||||||
|
description = f"{name} is a {place_type} of national importance."
|
||||||
|
else :
|
||||||
|
importance = 'international'
|
||||||
|
description = f"{name} is an internationally famous {place_type}."
|
||||||
|
else :
|
||||||
|
description = f"{name} is a {place_type} of {importance} importance."
|
||||||
|
|
||||||
|
if height is not None and date is not None :
|
||||||
|
description += f" This {place_type} was constructed in {date} and is ca. {height} meters high."
|
||||||
|
elif height is not None :
|
||||||
|
description += f" This {place_type} stands ca. {height} meters tall."
|
||||||
|
elif date is not None:
|
||||||
|
description += f" It was constructed in {date}."
|
||||||
|
|
||||||
|
# Format the visitor number
|
||||||
|
if n_visitors is not None :
|
||||||
|
n_visitors = int(n_visitors)
|
||||||
|
if n_visitors < 1000000 :
|
||||||
|
description += f" It welcomes {int(n_visitors/1000)} thousand visitors every year."
|
||||||
|
else :
|
||||||
|
description += f" It welcomes {round(n_visitors/1000000, 1)} million visitors every year."
|
||||||
|
|
||||||
|
# Set the keywords.
|
||||||
|
keywords = {"importance": importance,
|
||||||
|
"height": height,
|
||||||
|
"place_type": place_type,
|
||||||
|
"date": date}
|
||||||
|
|
||||||
|
return description, keywords
|
||||||
|
|
||||||
|
|
||||||
|
def get_place_type(tags):
|
||||||
|
"""
|
||||||
|
Determines the type of the place based on available tags such as 'amenity', 'building',
|
||||||
|
'historic', and 'leisure'. The priority order is: 'historic' > 'building' (if not generic) >
|
||||||
|
'amenity' > 'leisure'.
|
||||||
|
|
||||||
|
Params:
|
||||||
|
tags (dict): A dictionary containing metadata about the place.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
place_type (str): The determined type of the place, or None if no relevant type is found.
|
||||||
|
"""
|
||||||
|
amenity = tags.get('amenity', None)
|
||||||
|
building = tags.get('building', None)
|
||||||
|
historic = tags.get('historic', None)
|
||||||
|
leisure = tags.get('leisure')
|
||||||
|
|
||||||
|
if historic and historic != "yes":
|
||||||
|
return historic
|
||||||
|
if building and building not in ["yes", "civic", "government", "apartments", "residential", "commericial", "industrial", "retail", "religious", "public", "service"]:
|
||||||
|
return building
|
||||||
|
if amenity:
|
||||||
|
return amenity
|
||||||
|
if leisure:
|
||||||
|
return leisure
|
||||||
|
|
||||||
|
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def get_date(tags):
|
||||||
|
"""
|
||||||
|
Extracts the most relevant date from the available tags, prioritizing 'construction_date',
|
||||||
|
'start_date', 'year_of_construction', and 'opening_date' in that order.
|
||||||
|
|
||||||
|
Params:
|
||||||
|
tags (dict): A dictionary containing metadata about the place.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
date (str): The most relevant date found, or None if no date is available.
|
||||||
|
"""
|
||||||
|
construction_date = tags.get('construction_date', None)
|
||||||
|
opening_date = tags.get('opening_date', None)
|
||||||
|
start_date = tags.get('start_date', None)
|
||||||
|
year_of_construction = tags.get('year_of_construction', None)
|
||||||
|
|
||||||
|
# Prioritize based on availability
|
||||||
|
if construction_date:
|
||||||
|
return construction_date
|
||||||
|
if start_date:
|
||||||
|
return start_date
|
||||||
|
if year_of_construction:
|
||||||
|
return year_of_construction
|
||||||
|
if opening_date:
|
||||||
|
return opening_date
|
||||||
|
|
||||||
|
return None
|
@ -1,15 +1,22 @@
|
|||||||
"""Helper function to return only the major landmarks from a large list."""
|
"""Helper function to return only the major landmarks from a large list."""
|
||||||
from ..structs.landmark import Landmark
|
from ..structs.landmark import Landmark
|
||||||
|
|
||||||
def take_most_important(landmarks: list[Landmark], n_important) -> list[Landmark]:
|
def take_most_important(
|
||||||
|
landmarks: list[Landmark],
|
||||||
|
n_important: int
|
||||||
|
) -> list[Landmark]:
|
||||||
"""
|
"""
|
||||||
Given a list of landmarks, return the n_important most important landmarks
|
Given a list of landmarks, return the most important landmarks based on their attractiveness.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
landmarks: list[Landmark] - list of landmarks
|
landmarks (list[Landmark]): List of landmarks that needs to be truncated
|
||||||
n_important: int - number of most important landmarks to return
|
n_important (int): Number of most important landmarks to return
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
list[Landmark] - list of the n_important most important landmarks
|
list[Landmark]: List of the n_important most important landmarks
|
||||||
"""
|
"""
|
||||||
|
if n_important == 0 :
|
||||||
|
raise ValueError('Number of landmarks to keep cannot be zero.')
|
||||||
|
|
||||||
# Sort landmarks by attractiveness (descending)
|
# Sort landmarks by attractiveness (descending)
|
||||||
sorted_landmarks = sorted(landmarks, key=lambda x: x.attractiveness, reverse=True)
|
sorted_landmarks = sorted(landmarks, key=lambda x: x.attractiveness, reverse=True)
|
||||||
|
Loading…
x
Reference in New Issue
Block a user