Merge modifications for more separate backend functions #69

Open
kscheidecker wants to merge 2 commits from backend/micro-services-restructuring into main
16 changed files with 597 additions and 219 deletions

3
backend/.gitignore vendored
View File

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

View File

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

View File

@ -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,6 +95,7 @@ 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()
@ -113,6 +120,7 @@ class LandmarkManager:
landmark.duration = 30 landmark.duration = 30
all_landmarks.update(current_landmarks) all_landmarks.update(current_landmarks)
if allow_clusters :
# special pipeline for shopping malls # special pipeline for shopping malls
shopping_manager = ClusterManager(bbox, 'shopping') shopping_manager = ClusterManager(bbox, 'shopping')
shopping_clusters = shopping_manager.generate_clusters() shopping_clusters = shopping_manager.generate_clusters()
@ -120,7 +128,10 @@ class LandmarkManager:
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.
# Set the duration. if tags.get('amenity') in ['aquarium', 'planetarium'] or tags.get('tourism') in ['aquarium', 'museum', 'zoo']:
if value in ['museum', 'aquarium', 'planetarium'] :
landmark.duration = 60 landmark.duration = 60
elif value == 'viewpoint' : elif tags.get('tourism') == 'viewpoint' :
landmark.is_viewpoint = True landmark.is_viewpoint = True
landmark.duration = 10 landmark.duration = 10
elif value == 'cathedral' : elif tags.get('building') == 'cathedral' :
landmark.is_place_of_worship = False landmark.is_place_of_worship = False
landmark.duration = 10 landmark.duration = 10
landmark.description, landmark.keywords = self.description_and_keywords(tags) # Compute the score and add landmark to the list.
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.

View 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

View File

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

View File

@ -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(
preferences: Preferences,
start: tuple[float, float], start: tuple[float, float],
end: tuple[float, float] | None = None, end: tuple[float, float] | None = None,
background_tasks: BackgroundTasks = None) -> Trip: 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

View 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

View File

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

View File

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

View File

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

View 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

View 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}"

View 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

View File

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