first pylint correction
Some checks failed
Build and deploy the backend to staging / Build and push image (pull_request) Successful in 2m26s
Run linting on the backend code / Build (pull_request) Failing after 30s
Run testing on the backend code / Build (pull_request) Successful in 2m12s
Build and deploy the backend to staging / Deploy to staging (pull_request) Successful in 15s

This commit is contained in:
Helldragon67 2024-11-22 11:55:25 +01:00
parent 1b955f249e
commit 4f169c483e
12 changed files with 237 additions and 118 deletions

View File

@ -1,6 +1,9 @@
import logging.config
from pathlib import Path
"""Module allowing to access the parameters of route generation"""
import logging
import os
from pathlib import Path
LOCATION_PREFIX = Path('src')
PARAMETERS_DIR = LOCATION_PREFIX / 'parameters'
@ -9,12 +12,10 @@ LANDMARK_PARAMETERS_PATH = PARAMETERS_DIR / 'landmark_parameters.yaml'
OPTIMIZER_PARAMETERS_PATH = PARAMETERS_DIR / 'optimizer_parameters.yaml'
cache_dir_string = os.getenv('OSM_CACHE_DIR', './cache')
OSM_CACHE_DIR = Path(cache_dir_string)
import logging
# if we are in a debug session, set verbose and rich logging
if os.getenv('DEBUG', "false") == "true":
from rich.logging import RichHandler

View File

@ -1,5 +1,7 @@
"""Main app for backend api"""
import logging
from fastapi import FastAPI, Query, Body, HTTPException
from fastapi import FastAPI, HTTPException
from .structs.landmark import Landmark
from .structs.preferences import Preferences
@ -21,13 +23,16 @@ refiner = Refiner(optimizer=optimizer)
@app.post("/trip/new")
def new_trip(preferences: Preferences, start: tuple[float, float], end: tuple[float, float] | None = None) -> Trip:
'''
"""
Main function to call the optimizer.
:param preferences: the preferences specified by the user as the post body
:param start: the coordinates of the starting point as a tuple of floats (as url query parameters)
:param end: the coordinates of the finishing point as a tuple of floats (as url query parameters)
:return: the uuid of the first landmark in the optimized route
'''
Args:
preferences (Preferences) : the preferences specified by the user as the post body
start (tuple) : the coordinates of the starting point as a tuple of floats (as url query parameters)
end (tuple) : the coordinates of the finishing point as a tuple of floats (as url query parameters)
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")
if preferences.shopping.score == 0 and preferences.sightseeing.score == 0 and preferences.nature.score == 0:
@ -50,7 +55,7 @@ def new_trip(preferences: Preferences, start: tuple[float, float], end: tuple[fl
# insert start and finish to the 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)
@ -58,7 +63,7 @@ def new_trip(preferences: Preferences, start: tuple[float, float], end: tuple[fl
raise HTTPException(status_code=500, detail="No solution found")
except TimeoutError:
raise HTTPException(status_code=500, detail="Optimzation took too long")
# Second stage optimization
refined_tour = refiner.refine_optimization(landmarks, base_tour, preferences.max_time_minute, preferences.detour_tolerance_minute)
@ -71,6 +76,15 @@ def new_trip(preferences: Preferences, start: tuple[float, float], end: tuple[fl
#### For already existing trips/landmarks
@app.get("/trip/{trip_uuid}")
def get_trip(trip_uuid: str) -> Trip:
"""
Look-up the cache for a trip that has been previously generated using its identifier.
Args:
trip_uuid (str) : unique identifier for a trip.
Returns:
(Trip) : the corresponding trip.
"""
try:
trip = cache_client.get(f"trip_{trip_uuid}")
return trip
@ -80,6 +94,15 @@ def get_trip(trip_uuid: str) -> Trip:
@app.get("/landmark/{landmark_uuid}")
def get_landmark(landmark_uuid: str) -> Landmark:
"""
Returns a Landmark from its unique identifier.
Args:
landmark_uuid (str) : unique identifier for a Landmark.
Returns:
(Landmark) : the corresponding Landmark.
"""
try:
landmark = cache_client.get(f"landmark_{landmark_uuid}")
return landmark

View File

@ -1,17 +1,65 @@
"""Module used for handling cache"""
from pymemcache.client.base import Client
from .constants import MEMCACHED_HOST_PATH
class DummyClient:
"""
A dummy in-memory client that mimics the behavior of a memcached client.
This class is designed to simulate the behavior of the `pymemcache.Client`
for testing or development purposes. It stores data in a Python dictionary
and provides methods to set, get, and update key-value pairs.
Attributes:
_data (dict): A dictionary that holds the key-value pairs.
Methods:
set(key, value, **kwargs):
Stores the given key-value pair in the internal dictionary.
set_many(data, **kwargs):
Updates the internal dictionary with multiple key-value pairs.
get(key, **kwargs):
Retrieves the value associated with the given key from the internal
dictionary.
"""
_data = {}
def set(self, key, value, **kwargs):
"""
Store a key-value pair in the internal dictionary.
Args:
key: The key for the item to be stored.
value: The value to be stored under the given key.
**kwargs: Additional keyword arguments (unused).
"""
self._data[key] = value
def set_many(self, data, **kwargs):
"""
Update the internal dictionary with multiple key-value pairs.
Args:
data: A dictionary containing key-value pairs to be added.
**kwargs: Additional keyword arguments (unused).
"""
self._data.update(data)
def get(self, key, **kwargs):
"""
Retrieve the value associated with the given key.
Args:
key: The key for the item to be retrieved.
**kwargs: Additional keyword arguments (unused).
Returns:
The value associated with the given key if it exists.
"""
return self._data[key]

View File

@ -1,10 +1,41 @@
"""Definition of the Landmark class to handle visitable objects across the world."""
from typing import Optional, Literal
from uuid import uuid4
from pydantic import BaseModel, Field
from uuid import uuid4
# Output to frontend
class Landmark(BaseModel) :
"""
A class representing a landmark or point of interest (POI) in the context of a trip.
The Landmark class is used to model visitable locations, such as tourist attractions,
natural sites, shopping locations, and start/end points in travel itineraries. It
holds information about the landmark's attributes and supports comparisons and
calculations, such as distance between landmarks.
Attributes:
name (str): The name of the landmark.
type (Literal): The type of the landmark, which can be one of ['sightseeing', 'nature',
'shopping', 'start', 'finish'].
location (tuple): A tuple representing the (latitude, longitude) of the landmark.
osm_type (str): The OpenStreetMap (OSM) type of the landmark.
osm_id (int): The OpenStreetMap (OSM) ID of the landmark.
attractiveness (int): A score representing the attractiveness of the landmark.
n_tags (int): The number of tags associated with the landmark.
image_url (Optional[str]): A URL to an image of the landmark.
website_url (Optional[str]): A URL to the landmark's official website.
description (Optional[str]): A text description of the landmark.
duration (Optional[int]): The estimated time to visit the landmark (in minutes).
name_en (Optional[str]): The English name of the landmark.
uuid (str): A unique identifier for the landmark, generated by default using uuid4.
must_do (Optional[bool]): Whether the landmark is a "must-do" attraction.
must_avoid (Optional[bool]): Whether the landmark should be avoided.
is_secondary (Optional[bool]): Whether the landmark is secondary or less important.
time_to_reach_next (Optional[int]): Estimated time (in minutes) to reach the next landmark.
next_uuid (Optional[str]): UUID of the next landmark in sequence (if applicable).
"""
# Properties of the landmark
name : str
@ -26,12 +57,19 @@ class Landmark(BaseModel) :
# Additional properties depending on specific tour
must_do : Optional[bool] = False
must_avoid : Optional[bool] = False
is_secondary : Optional[bool] = False # TODO future
is_secondary : Optional[bool] = False
time_to_reach_next : Optional[int] = 0
next_uuid : Optional[str] = None
def __str__(self) -> str:
"""
String representation of the Landmark object.
Returns:
str: A formatted string with the landmark's type, name, location, attractiveness score,
time to the next landmark (if available), and whether the landmark is secondary.
"""
time_to_next_str = f", time_to_next={self.time_to_reach_next}" if self.time_to_reach_next else ""
is_secondary_str = f", secondary" if self.is_secondary else ""
type_str = '(' + self.type + ')'
@ -39,12 +77,36 @@ class Landmark(BaseModel) :
return f'Landmark{type_str}: [{self.name} @{self.location}, score={self.attractiveness}{time_to_next_str}{is_secondary_str}]'
def distance(self, value: 'Landmark') -> float:
"""
Calculates the squared distance between this landmark and another.
Args:
value (Landmark): Another Landmark object to calculate the distance to.
Returns:
float: The squared Euclidean distance between the two landmarks.
"""
return (self.location[0] - value.location[0])**2 + (self.location[1] - value.location[1])**2
def __hash__(self) -> int:
"""
Generates a hash for the Landmark based on its name.
Returns:
int: The hash of the landmark.
"""
return hash(self.name)
def __eq__(self, value: 'Landmark') -> bool:
"""
Checks equality between two Landmark objects based on UUID, OSM ID, and name.
Args:
value (Landmark): Another Landmark object to compare.
Returns:
bool: True if the landmarks are equal, False otherwise.
"""
# eq and hash must be consistent
# in particular, if two objects are equal, their hash must be equal
# uuid and osm_id are just shortcuts to avoid comparing all the properties

View File

@ -1,10 +1,15 @@
"""Linked and ordered list of Landmarks that represents the visiting order."""
from .landmark import Landmark
from ..utils.get_time_separation import get_time
class LinkedLandmarks:
"""
A list of landmarks that are linked together, e.g. in a route.
Each landmark serves as a node in the linked list, but since we expect these to be consumed through the rest API, a pythonic reference to the next landmark is not well suited. Instead we use the uuid of the next landmark to reference the next landmark in the list. This is not very efficient, but appropriate for the expected use case ("short" trips with onyl few landmarks).
Each landmark serves as a node in the linked list, but since we expect these to be consumed through the rest API,
a pythonic reference to the next landmark is not well suited. Instead we use the uuid of the next landmark
to reference the next landmark in the list. This is not very efficient, but appropriate for the expected use case
("short" trips with onyl few landmarks).
"""
_landmarks = list[Landmark]
@ -12,8 +17,9 @@ class LinkedLandmarks:
def __init__(self, data: list[Landmark] = None) -> None:
"""
Initialize a new LinkedLandmarks object. This expects an ORDERED list of landmarks, where the first landmark is the starting point and the last landmark is the end point.
Initialize a new LinkedLandmarks object. This expects an ORDERED list of landmarks,
where the first landmark is the starting point and the last landmark is the end point.
Args:
data (list[Landmark], optional): The list of landmarks that are linked together. Defaults to None.
"""
@ -41,16 +47,19 @@ class LinkedLandmarks:
self._landmarks[-1].time_to_reach_next = 0
def update_secondary_landmarks(self) -> None:
"""
Mark landmarks with lower importance as secondary.
"""
# Extract the attractiveness scores and sort them in descending order
scores = sorted([landmark.attractiveness for landmark in self._landmarks], reverse=True)
# Determine the 10th highest score
if len(scores) >= 10:
threshold_score = scores[9]
else:
# If there are fewer than 10 landmarks, use the lowest score in the list as the threshold
threshold_score = min(scores) if scores else 0
# Update 'is_secondary' for landmarks with attractiveness below the threshold score
for landmark in self._landmarks:
if landmark.attractiveness < threshold_score and landmark.type not in ["start", "finish"]:
@ -59,7 +68,7 @@ class LinkedLandmarks:
def __getitem__(self, index: int) -> Landmark:
return self._landmarks[index]
def __str__(self) -> str:
return f"LinkedLandmarks [{' ->'.join([str(landmark) for landmark in self._landmarks])}]"

View File

@ -1,12 +1,26 @@
from pydantic import BaseModel
"""Defines the Preferences used as input for trip generation."""
from typing import Optional, Literal
from pydantic import BaseModel
class Preference(BaseModel) :
"""
Type of preference.
Attributes:
type: what kind of landmark type.
score: how important that type is.
"""
type: Literal['sightseeing', 'nature', 'shopping', 'start', 'finish']
score: int # score could be from 1 to 5
# Input for optimization
class Preferences(BaseModel) :
""""
Full collection of preferences needed to generate a personalized trip.
"""
# Sightseeing / History & Culture (Musées, bâtiments historiques, opéras, églises)
sightseeing : Preference

View File

@ -1,10 +1,24 @@
"""Definition of the Trip class."""
import uuid
from pydantic import BaseModel, Field
from pymemcache.client.base import Client
from .linked_landmarks import LinkedLandmarks
import uuid
class Trip(BaseModel):
""""
A Trip represents the final guided tour that can be passed to frontend.
Attributes:
uuid: unique identifier for this particular trip.
total_time: duration of the trip (in minutes).
first_landmark_uuid: unique identifier of the first Landmark to visit.
Methods:
from_linked_landmarks: create a Trip from LinkedLandmarks object.
"""
uuid: str = Field(default_factory=uuid.uuid4)
total_time: int
first_landmark_uuid: str

View File

@ -1,80 +0,0 @@
import logging
import yaml
from .utils.landmarks_manager import LandmarkManager
from .utils.optimizer import Optimizer
from .utils.refiner import Refiner
from .structs.landmark import Landmark
from .structs.linked_landmarks import LinkedLandmarks
from .structs.preferences import Preferences, Preference
logger = logging.getLogger(__name__)
def test(start_coords: tuple[float, float], finish_coords: tuple[float, float] = None) -> list[Landmark]:
manager = LandmarkManager()
optimizer = Optimizer()
refiner = Refiner(optimizer=optimizer)
preferences = Preferences(
sightseeing=Preference(type='sightseeing', score = 5),
nature=Preference(type='nature', score = 5),
shopping=Preference(type='shopping', score = 0),
max_time_minute=30,
detour_tolerance_minute=0
)
# Create start and finish
if finish_coords is None :
finish_coords = start_coords
start = Landmark(name='start', type='start', location=start_coords, osm_type='', osm_id=0, attractiveness=0, n_tags = 0)
finish = Landmark(name='finish', type='finish', location=finish_coords, osm_type='', osm_id=0, attractiveness=0, n_tags = 0)
#finish = Landmark(name='finish', type=LandmarkType(landmark_type='finish'), location=(48.8777055, 2.3640967), osm_type='finish', osm_id=0, attractiveness=0, must_do=True, n_tags = 0)
#start = Landmark(name='start', type=LandmarkType(landmark_type='start'), location=(48.847132, 2.312359), osm_type='start', osm_id=0, attractiveness=0, must_do=True, n_tags = 0)
#finish = Landmark(name='finish', type=LandmarkType(landmark_type='finish'), location=(48.843185, 2.344533), osm_type='finish', osm_id=0, attractiveness=0, must_do=True, n_tags = 0)
#finish = Landmark(name='finish', type=LandmarkType(landmark_type='finish'), location=(48.847132, 2.312359), osm_type='finish', osm_id=0, attractiveness=0, must_do=True, n_tags = 0)
# Generate the landmarks from the start location
landmarks, landmarks_short = manager.generate_landmarks_list(
center_coordinates = start_coords,
preferences = preferences
)
# Store data to file for debug purposes
# write_data(landmarks, "landmarks_Strasbourg.txt")
# Insert start and finish to the landmarks list
landmarks_short.insert(0, start)
landmarks_short.append(finish)
# First stage optimization
base_tour = optimizer.solve_optimization(max_time=preferences.max_time_minute, landmarks=landmarks_short)
# Second stage using linear optimization
refined_tour = refiner.refine_optimization(all_landmarks=landmarks, base_tour=base_tour, max_time = preferences.max_time_minute, detour = preferences.detour_tolerance_minute)
linked_tour = LinkedLandmarks(refined_tour)
total_time = 0
logger.info("Optimized route : ")
for l in linked_tour :
logger.info(f"{l}")
logger.info(f"Estimated length of tour : {linked_tour.total_time} mintutes and visiting {len(linked_tour._landmarks)} landmarks.")
# with open('linked_tour.yaml', 'w') as f:
# yaml.dump(linked_tour.asdict(), f)
return linked_tour
# test(tuple((48.8344400, 2.3220540))) # Café Chez César
# test(tuple((48.8375946, 2.2949904))) # Point random
# test(tuple((47.377859, 8.540585))) # Zurich HB
test(tuple((45.758217, 4.831814))) # Lyon Bellecour
# test(tuple((48.5848435, 7.7332974))) # Strasbourg Gare
# test(tuple((48.2067858, 16.3692340))) # Vienne
# test(tuple((48.2432090, 7.3892691))) # Orschwiller

View File

@ -1,6 +1,9 @@
from fastapi.testclient import TestClient
"""Collection of tests to ensure correct implementation and track progress. """
from typing import List
from fastapi.testclient import TestClient
import pytest
from ..main import app
from ..structs.landmark import Landmark
@ -10,8 +13,13 @@ def client():
return TestClient(app)
# Base test for checking if the API returns correct error code when no preferences are specified.
def test_new_trip_invalid_prefs(client):
"""
Test n°1 : base test for checking if the API returns correct error code when no preferences are specified.
Args:
client:
"""
response = client.post(
"/trip/new",
json={
@ -21,9 +29,15 @@ def test_new_trip_invalid_prefs(client):
)
assert response.status_code == 422
# Test no. 2
def test_turckheim(client, request):
"""
Test n°2 : Custom test in Turckheim to ensure small villages are also supported.
Args:
client:
request:
"""
duration_minutes = 15
response = client.post(
"/trip/new",
@ -47,6 +61,13 @@ def test_turckheim(client, request):
# Test no. 3
def test_bellecour(client, request) :
"""
Test n°3 : Custom test in Lyon centre to ensure proper decision making in crowded area.
Args:
client:
request:
"""
duration_minutes = 60
response = client.post(
"/trip/new",
@ -99,12 +120,12 @@ def fetch_landmark(client, landmark_uuid: str):
if response.status_code != 200:
raise Exception(f"Failed to fetch landmark with UUID {landmark_uuid}: {response.status_code}")
json_data = response.json()
if "detail" in json_data:
raise Exception(json_data["detail"])
return json_data
@ -135,10 +156,17 @@ def load_trip_landmarks(client, first_uuid: str) -> List[Landmark]:
def log_trip_details(request, landmarks: List[Landmark], duration: int, target_duration: int) :
# Create the trip string
trip_string = [f"{landmark.name} ({landmark.attractiveness} | {landmark.duration}) - {landmark.time_to_reach_next}" for landmark in landmarks]
"""
Allows to show the detailed trip in the html test report.
Args:
request:
landmarks (list): the ordered list of visited landmarks
duration (int): the total duration of this trip
target_duration(int): the target duration of this trip
"""
trip_string = [f"{landmark.name} ({landmark.attractiveness} | {landmark.duration}) - {landmark.time_to_reach_next}" for landmark in landmarks]
# Pass additional info to pytest for reporting
request.node.trip_details = trip_string
request.node.trip_duration = str(duration) # result['total_time']

View File

@ -63,7 +63,7 @@ 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.
Parameters:
Args:
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.

View File

@ -38,11 +38,11 @@ class Refiner :
Create a corridor around the path connecting the landmarks.
Args:
landmarks (list[Landmark]): the landmark path around which to create the corridor
width (float): Width of the corridor in meters.
landmarks (list[Landmark]) : the landmark path around which to create the corridor
width (float) : width of the corridor in meters.
Returns:
Geometry: A buffered geometry object representing the corridor around the path.
Geometry: a buffered geometry object representing the corridor around the path.
"""
corrected_width = (180*width)/(6371000*pi)

View File

@ -3,7 +3,7 @@ from ..structs.landmark import Landmark
def take_most_important(landmarks: list[Landmark], n_important) -> list[Landmark]:
"""
Given a list of landmarks, return the n_important most important landmarks
Parameters:
Args:
landmarks: list[Landmark] - list of landmarks
n_important: int - number of most important landmarks to return
Returns: