88 lines
3.6 KiB
Python
88 lines
3.6 KiB
Python
import logging
|
|
from fastapi import FastAPI, Query, Body, HTTPException
|
|
|
|
from structs.landmark import Landmark
|
|
from structs.preferences import Preferences
|
|
from structs.linked_landmarks import LinkedLandmarks
|
|
from structs.trip import Trip
|
|
from utils.landmarks_manager import LandmarkManager
|
|
from utils.optimizer import Optimizer
|
|
from utils.refiner import Refiner
|
|
from persistence import client as cache_client
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
app = FastAPI()
|
|
manager = LandmarkManager()
|
|
optimizer = Optimizer()
|
|
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
|
|
'''
|
|
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:
|
|
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 end is None:
|
|
end = start
|
|
logger.info("No end coordinates provided. Using start=end.")
|
|
|
|
start_landmark = Landmark(name='start', type='start', location=(start[0], start[1]), osm_type='start', osm_id=0, attractiveness=0, must_do=True, n_tags = 0)
|
|
end_landmark = Landmark(name='finish', type='finish', location=(end[0], end[1]), osm_type='end', 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,
|
|
preferences = preferences
|
|
)
|
|
|
|
# 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)
|
|
except ArithmeticError:
|
|
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)
|
|
|
|
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)
|
|
return trip
|
|
|
|
|
|
#### For already existing trips/landmarks
|
|
@app.get("/trip/{trip_uuid}")
|
|
def get_trip(trip_uuid: str) -> Trip:
|
|
try:
|
|
trip = cache_client.get(f"trip_{trip_uuid}")
|
|
return trip
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="Trip not found")
|
|
|
|
|
|
@app.get("/landmark/{landmark_uuid}")
|
|
def get_landmark(landmark_uuid: str) -> Landmark:
|
|
try:
|
|
landmark = cache_client.get(f"landmark_{landmark_uuid}")
|
|
return landmark
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="Landmark not found")
|