Merge modifications for more separate backend functions #69

Open
kscheidecker wants to merge 2 commits from backend/micro-services-restructuring into main
8 changed files with 298 additions and 186 deletions
Showing only changes of commit 6921ab57f8 - Show all commits

File diff suppressed because one or more lines are too long

View File

@ -146,7 +146,7 @@ class ClusterManager:
self.valid = False
else :
self.logger.debug(f"Detected 0 {cluster_type} clusters.")
self.logger.debug(f"Found 0 {cluster_type} clusters.")
self.valid = False

View File

@ -55,7 +55,12 @@ class LandmarkManager:
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.
@ -63,16 +68,17 @@ class LandmarkManager:
and current location. It scores and corrects these landmarks, removes duplicates, and then selects the most important
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.
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:
tuple[list[Landmark], list[Landmark]]:
- A list of all existing landmarks.
- 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)
radius = min(max_walk_dist, int(self.max_bbox_side/2))
@ -89,6 +95,7 @@ class LandmarkManager:
all_landmarks.update(current_landmarks)
self.logger.info(f'Found {len(current_landmarks)} sightseeing landmarks')
if allow_clusters :
# special pipeline for historic neighborhoods
neighborhood_manager = ClusterManager(bbox, 'sightseeing')
historic_clusters = neighborhood_manager.generate_clusters()
@ -113,6 +120,7 @@ class LandmarkManager:
landmark.duration = 30
all_landmarks.update(current_landmarks)
if allow_clusters :
# special pipeline for shopping malls
shopping_manager = ClusterManager(bbox, 'shopping')
shopping_clusters = shopping_manager.generate_clusters()
@ -120,7 +128,10 @@ class LandmarkManager:
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'Found {len(all_landmarks)} landmarks in total.')
return all_landmarks, landmarks_constrained
@ -236,188 +247,63 @@ class LandmarkManager:
continue
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
# Maybe a bit too inefficient
# for key in tags.keys():
# if 'disused:' in key or 'boundary:' in key :
# break
# if 'building:' in key or 'pay' in key :
# n_tags -= 1
# Convert this to Landmark object
# TODO: convert to proto landmark and store rest to memcache
landmark = Landmark(name=name,
type=landmarktype,
location=coords,
osm_id=id,
osm_type=osm_type,
attractiveness=0,
n_tags=len(tags))
n_tags=n_tags)
# Browse through tags to add information to landmark.
for key, value in tags.items():
# Extract useful information for score calculation later down the road.
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.
if key == 'building:part' and value == 'yes' :
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' :
# Check for place of worship
if tags.get('place_of_worship') is not None :
landmark.is_place_of_worship = True
if key == 'wikipedia' :
landmark.wiki_url = value
if key == 'name:en' :
landmark.name_en = value
if 'building:' in key or 'pay' in key :
landmark.n_tags -= 1
landmark.name_en = tags.get('place_of_worship')
# Set the duration.
if value in ['museum', 'aquarium', 'planetarium'] :
# 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 value == 'viewpoint' :
elif tags.get('tourism') == 'viewpoint' :
landmark.is_viewpoint = True
landmark.duration = 10
elif value == 'cathedral' :
elif tags.get('building') == 'cathedral' :
landmark.is_place_of_worship = False
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)
landmarks.append(landmark)
continue
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:
"""
Convert a dictionary of key-value pairs to a list of Overpass query strings.

View File

@ -1,12 +1,13 @@
"""Main app for backend api"""
import logging
import time
import random
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, BackgroundTasks
from .logging_config import configure_logging
from .structs.landmark import Landmark
from .structs.preferences import Preferences
from .structs.preferences import Preferences, Preference
from .structs.linked_landmarks import LinkedLandmarks
from .structs.trip import Trip
from .landmarks.landmarks_manager import LandmarkManager
@ -140,7 +141,7 @@ def new_trip(preferences: Preferences,
# 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'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)
@ -224,3 +225,45 @@ def update_trip_time(trip_uuid: str, removed_landmark_uuid: str) -> Trip:
trip = Trip.from_linked_landmarks(linked_tour, cache_client)
return trip
# TODO: get stuff to do nearby. The idea is to have maybe thhe 3 best things to do within 500m and 2 hidden gems.
@app.post("/landmarks/get-nearby/{lat}/{lon}")
def get_landmarks_nearby(lat: float, lon: float) -> list[Landmark] :
# preferences = {"sightseeing": {"type": "sightseeing", "score": 0},
# "nature": {"type": "nature", "score": 0},
# "shopping": {"type": "shopping", "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 = 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),
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('Suggested landmarks :\n\t' + '\n\t'.join(f'{landmark}' for landmark in landmarks_around))
return landmarks_around

View File

@ -2,6 +2,7 @@
from .landmark import Landmark
from ..utils.get_time_distance import get_time
from ..utils.description import description_and_keywords
class LinkedLandmarks:
"""
@ -35,18 +36,23 @@ class LinkedLandmarks:
Create the links between the landmarks in the list by setting their
.next_uuid and the .time_to_next attributes.
"""
# Mark secondary landmarks as such
self.update_secondary_landmarks()
for i, landmark in enumerate(self._landmarks[:-1]):
# Set uuid of the next landmark
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)
landmark.time_to_reach_next = time_to_next
self.total_time += time_to_next
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].time_to_reach_next = 0

View File

@ -0,0 +1,16 @@
from typing import Optional
from uuid import uuid4, UUID
from pydantic import BaseModel, Field
# Output to frontend
class ProtoLandmark(BaseModel) :
"""fef"""
uuid: UUID = Field(default_factory=uuid4)
location : tuple
attractiveness : int
duration : Optional[int] = 5
must_do : Optional[bool] = False
must_avoid : Optional[bool] = False

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