working split

backend/.gitignore

@@ -12,6 +12,9 @@ __pycache__/
 # C extensions
 *.so
 
+# Pytest reports
+report.html
+
 # Distribution / packaging
 .Python
 build/

backend/src/landmarks/landmarks_manager.py

@@ -259,15 +259,7 @@ class LandmarkManager:
             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,

backend/src/landmarks/landmarks_routes.py

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

backend/src/logging_config.py

@@ -33,14 +33,14 @@ def configure_logging():
         # silence the chatty logs loki generates itself
         logging.getLogger('urllib3.connectionpool').setLevel(logging.WARNING)
         # 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:
         # if we are in a debug (local) session, set verbose and rich logging
         from rich.logging import RichHandler
         logging_handlers = [RichHandler()]
         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'
 
 
 

backend/src/main.py

@@ -11,7 +11,9 @@ from .structs.preferences import Preferences, Preference
 from .structs.linked_landmarks import LinkedLandmarks
 from .structs.trip import Trip
 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.refiner import Refiner
 from .overpass.overpass import fill_cache
@@ -39,13 +41,17 @@ app = FastAPI(lifespan=lifespan)
 
 
 app.include_router(toilets_router)
+app.include_router(optimization_router)
+app.include_router(landmarks_router)
 
 
 @app.post("/trip/new")
-def new_trip(preferences: Preferences,
-             start: tuple[float, float],
-             end: tuple[float, float] | None = None,
-             background_tasks: BackgroundTasks = None) -> Trip:
+def new_trip(
+    preferences: Preferences,
+    start: tuple[float, float],
+    end: tuple[float, float] | None = None,
+    background_tasks: BackgroundTasks = None
+) -> Trip:
     """
     Main function to call the optimizer.
 
@@ -226,44 +232,3 @@ def update_trip_time(trip_uuid: str, removed_landmark_uuid: str) -> Trip:
 
     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

backend/src/optimization/optimization_routes.py

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

backend/src/overpass/overpass.py

@@ -402,6 +402,8 @@ def fill_cache():
     n_files = 0
     total = 0
 
+    overpass.logger.info('Trip successfully returned, starting to fill cache.')
+
     with os.scandir(OSM_CACHE_DIR) as it:
         for entry in it:
             if entry.is_file() and entry.name.startswith('hollow_'):

backend/src/structs/proto_landmark.py

@@ -1,16 +0,0 @@
-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

backend/src/tests/test_main.py

@@ -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
 from fastapi.testclient import TestClient
 import pytest
@@ -6,6 +6,7 @@ import pytest
 from .test_utils import load_trip_landmarks, log_trip_details
 from ..main import app
 
+
 @pytest.fixture(scope="module")
 def client():
     """Client used to call the app."""
@@ -88,15 +89,13 @@ def test_bellecour(client, request) :   # pylint: disable=redefined-outer-name
     # Add details to report
     log_trip_details(request, landmarks, result['total_time'], duration_minutes)
 
-    # for elem in landmarks :
-    #     print(elem)
-
     # 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 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}"
 
+
 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.
@@ -128,9 +127,6 @@ def test_cologne(client, request) :   # pylint: disable=redefined-outer-name
     # Add details to report
     log_trip_details(request, landmarks, result['total_time'], duration_minutes)
 
-    # for elem in landmarks :
-    #     print(elem)
-
     # checks :
     assert response.status_code == 200  # check for successful planning
     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
     log_trip_details(request, landmarks, result['total_time'], duration_minutes)
 
-    # for elem in landmarks :
-    #     print(elem)
-
     # checks :
     assert response.status_code == 200  # check for successful planning
     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
     log_trip_details(request, landmarks, result['total_time'], duration_minutes)
 
-    # for elem in landmarks :
-    #     print(elem)
-
     # checks :
     assert response.status_code == 200  # check for successful planning
     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
     log_trip_details(request, landmarks, result['total_time'], duration_minutes)
 
-    # for elem in landmarks :
-    #     print(elem)
-
     # checks :
     assert response.status_code == 200  # check for successful planning
     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
     log_trip_details(request, landmarks, result['total_time'], duration_minutes)
 
-    # for elem in landmarks :
-    #     print(elem)
-
     # checks :
     assert response.status_code == 200  # check for successful planning
     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
     log_trip_details(request, landmarks, result['total_time'], duration_minutes)
 
-    # for elem in landmarks :
-    #     print(elem)
-
     # checks :
     assert response.status_code == 200  # check for successful planning
     assert comp_time < 30, f"Computation time exceeded 30 seconds: {comp_time:.2f} seconds"

backend/src/tests/test_trip_generation.py

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

backend/src/utils/take_most_important.py

@@ -1,15 +1,22 @@
 """Helper function to return only the major landmarks from a large list."""
 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:
-    landmarks: list[Landmark] - list of landmarks
-    n_important: int - number of most important landmarks to return
+        landmarks (list[Landmark]): List of landmarks that needs to be truncated
+        n_important (int): Number of most important landmarks to return
+    
     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)
     sorted_landmarks = sorted(landmarks, key=lambda x: x.attractiveness, reverse=True)