anydev/anyway
anydev/anyway
0
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

Merge pull request 'Adding features to find public toilets and shopping streets' (#41) from feature/backend/toilets-and-shopping-streets into main
Some checks failed
Build and deploy the backend to production / Build and push image (push) Failing after 43s
Details
Build and deploy the backend to production / Deploy to production (push) Has been skipped
Details
/ push-to-remote (push) Failing after 13s
Details

Browse Source 
Reviewed-on: #41
This commit is contained in:
kscheidecker 2024-12-14 15:57:08 +00:00
commit ddd2e91328
29 changed files with 198819 additions and 2023 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.gitea/workflows/backend_run_lint.yaml
View File

@ -30,5 +30,5 @@ jobs:
working-directory: backend
- name: Run linter
run: pipenv run pylint src
run: pipenv run pylint src --fail-under=9
working-directory: backend

3
backend/.pylintrc
View File

@ -439,7 +439,8 @@ disable=raw-checker-failed,
use-symbolic-message-instead,
use-implicit-booleaness-not-comparison-to-string,
use-implicit-booleaness-not-comparison-to-zero,
import-error
import-error,
line-too-long
# Enable the message, report, category or checker with the given id(s). You can
# either give multiple identifier separated by comma (,) or put this option

2
backend/Pipfile
View File

@ -23,3 +23,5 @@ osmpythontools = "*"
pywikibot = "*"
pymemcache = "*"
fastapi-cli = "*"
scikit-learn = "*"
pyqt6 = "*"

3975
backend/Pipfile.lock generated
View File

File diff suppressed because it is too large Load Diff

10
backend/report.html
View File

File diff suppressed because one or more lines are too long

52
backend/src/main.py
View File

@ -1,13 +1,14 @@
"""Main app for backend api"""
import logging
from fastapi import FastAPI, HTTPException
from fastapi import FastAPI, HTTPException, Query
from .structs.landmark import Landmark
from .structs.landmark import Landmark, Toilets
from .structs.preferences import Preferences
from .structs.linked_landmarks import LinkedLandmarks
from .structs.trip import Trip
from .utils.landmarks_manager import LandmarkManager
from .utils.toilets_manager import ToiletsManager
from .utils.optimizer import Optimizer
from .utils.refiner import Refiner
from .persistence import client as cache_client
@ -36,19 +37,15 @@ def new_trip(preferences: Preferences,
(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.")
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.")
raise HTTPException(status_code=406, detail="All preferences are 0.")
if start is None:
raise HTTPException(status_code=406,
detail="Start coordinates not provided")
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=423,
detail="Start coordinates not in range")
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.")
@ -61,7 +58,7 @@ def new_trip(preferences: Preferences,
attractiveness=0,
must_do=True,
n_tags = 0)
end_landmark = Landmark(name='finish',
type='finish',
location=(end[0], end[1]),
@ -69,7 +66,7 @@ def new_trip(preferences: Preferences,
osm_id=0,
attractiveness=0,
must_do=True,
n_tags = 0)
n_tags=0)
# Generate the landmarks from the start location
landmarks, landmarks_short = manager.generate_landmarks_list(
@ -134,5 +131,32 @@ def get_landmark(landmark_uuid: str) -> Landmark:
landmark = cache_client.get(f"landmark_{landmark_uuid}")
return landmark
except KeyError as exc:
raise HTTPException(status_code=404,
detail="Landmark not found") from exc
raise HTTPException(status_code=404, detail="Landmark not found") from exc
@app.post("/toilets/new")
def get_toilets(location: tuple[float, float] = Query(...), radius: int = 500) -> list[Toilets] :
"""
Endpoint to find toilets within a specified radius from a given location.
This endpoint expects the `location` and `radius` as **query parameters**, not in the request body.
Args:
location (tuple[float, float]): The latitude and longitude of the location to search from.
radius (int, optional): The radius (in meters) within which to search for toilets. Defaults to 500 meters.
Returns:
list[Toilets]: A list of Toilets objects that meet the criteria.
"""
if location is None:
raise HTTPException(status_code=406, detail="Coordinates not provided or invalid")
if not (-90 <= location[0] <= 90 or -180 <= location[1] <= 180):
raise HTTPException(status_code=422, detail="Start coordinates not in range")
toilets_manager = ToiletsManager(location, radius)
try :
toilets_list = toilets_manager.generate_toilet_list()
return toilets_list
except KeyError as exc:
raise HTTPException(status_code=404, detail="No toilets found") from exc

5
backend/src/persistence.py
View File

@ -1,5 +1,5 @@
"""Module used for handling cache"""
from pymemcache import serde
from pymemcache.client.base import Client
from .constants import MEMCACHED_HOST_PATH
@ -70,5 +70,6 @@ else:
MEMCACHED_HOST_PATH,
timeout=1,
allow_unicode_keys=True,
encoding='utf-8'
encoding='utf-8',
serde=serde.pickle_serde
)

4442
backend/src/sandbox/bandung_data.json Normal file
View File

File diff suppressed because it is too large Load Diff

698
backend/src/sandbox/colmar_data.json Normal file
View File

@ -0,0 +1,698 @@
{
"type": "FeatureCollection",
"generator": "overpass-turbo",
"copyright": "The data included in this document is from www.openstreetmap.org. The data is made available under ODbL.",
"timestamp": "2024-12-02T21:14:59Z",
"features": [
{
"type": "Feature",
"properties": {
"@id": "node/1345741798",
"name": "Cordonnerie Saint-Joseph",
"shop": "shoes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3481705,
48.0816462
]
},
"id": "node/1345741798"
},
{
"type": "Feature",
"properties": {
"@id": "node/2659184738",
"brand": "Armand Thiery",
"brand:wikidata": "Q2861975",
"brand:wikipedia": "fr:Armand Thiery",
"name": "Armand Thiery",
"opening_hours": "Mo-Sa 09:30-19:00",
"shop": "clothes",
"wheelchair": "limited"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3594454,
48.0785574
]
},
"id": "node/2659184738"
},
{
"type": "Feature",
"properties": {
"@id": "node/3618136290",
"name": "Chez Dominique",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3362362,
48.0712174
]
},
"id": "node/3618136290"
},
{
"type": "Feature",
"properties": {
"@id": "node/3618136605",
"name": "Divamod",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3304253,
48.0782989
]
},
"id": "node/3618136605"
},
{
"type": "Feature",
"properties": {
"@id": "node/3618284507",
"name": "Star tendances et voyages",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3474029,
48.0830993
]
},
"id": "node/3618284507"
},
{
"type": "Feature",
"properties": {
"@id": "node/3619696125",
"brand": "Zeeman",
"brand:wikidata": "Q184399",
"name": "Zeeman",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3413834,
48.0638444
]
},
"id": "node/3619696125"
},
{
"type": "Feature",
"properties": {
"@id": "node/4594398129",
"name": "Miss et Mister",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3308309,
48.0779118
]
},
"id": "node/4594398129"
},
{
"type": "Feature",
"properties": {
"@id": "node/4907320441",
"brand": "Sergent Major",
"brand:wikidata": "Q62521738",
"clothes": "babies;children",
"name": "Sergent Major",
"opening_hours": "Mo-Sa 09:30-19:00",
"shop": "clothes",
"wheelchair": "no"
},
"geometry": {
"type": "Point",
"coordinates": [
7.359116,
48.0787229
]
},
"id": "node/4907320441"
},
{
"type": "Feature",
"properties": {
"@id": "node/4907364791",
"brand": "Armand Thiery",
"brand:wikidata": "Q2861975",
"brand:wikipedia": "fr:Armand Thiery",
"clothes": "women",
"name": "Armand Thiery",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3601857,
48.0783373
]
},
"id": "node/4907364791"
},
{
"type": "Feature",
"properties": {
"@id": "node/4907385675",
"check_date": "2024-05-19",
"clothes": "children",
"name": "Du Pareil...au même",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3604521,
48.0779726
]
},
"id": "node/4907385675"
},
{
"type": "Feature",
"properties": {
"@id": "node/4922191645",
"name": "Abilos",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3566167,
48.0794136
]
},
"id": "node/4922191645"
},
{
"type": "Feature",
"properties": {
"@id": "node/4922191648",
"brand": "Esprit",
"brand:wikidata": "Q532746",
"brand:wikipedia": "en:Esprit Holdings",
"name": "Esprit",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3554004,
48.0787549
]
},
"id": "node/4922191648"
},
{
"type": "Feature",
"properties": {
"@id": "node/4922191972",
"brand": "Guess",
"brand:wikidata": "Q2470307",
"brand:wikipedia": "en:Guess (clothing)",
"name": "Guess",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.355273,
48.0788003
]
},
"id": "node/4922191972"
},
{
"type": "Feature",
"properties": {
"@id": "node/4922192001",
"name": "Lingerie",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3575453,
48.0779317
]
},
"id": "node/4922192001"
},
{
"type": "Feature",
"properties": {
"@id": "node/5359915869",
"name": "Al Assil",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3305665,
48.0780902
]
},
"id": "node/5359915869"
},
{
"type": "Feature",
"properties": {
"@id": "node/9089360040",
"brand": "Grain de Malice",
"brand:wikidata": "Q66757157",
"clothes": "women",
"name": "Grain de Malice",
"shop": "clothes",
"short_name": "GDM"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3593125,
48.0786234
]
},
"id": "node/9089360040"
},
{
"type": "Feature",
"properties": {
"@id": "node/9095193153",
"brand": "Undiz",
"brand:wikidata": "Q105306275",
"clothes": "underwear",
"name": "Undiz",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3599579,
48.0782846
]
},
"id": "node/9095193153"
},
{
"type": "Feature",
"properties": {
"@id": "node/9095193154",
"branch": "Lingerie",
"brand": "RougeGorge",
"brand:wikidata": "Q104600739",
"clothes": "underwear",
"name": "RougeGorge",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3604883,
48.0781607
]
},
"id": "node/9095193154"
},
{
"type": "Feature",
"properties": {
"@id": "node/9095212690",
"alt_name": "North Face",
"brand": "The North Face",
"brand:wikidata": "Q152784",
"brand:wikipedia": "en:The North Face",
"check_date": "2024-05-19",
"name": "The North Face",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3603923,
48.0773727
]
},
"id": "node/9095212690"
},
{
"type": "Feature",
"properties": {
"@id": "node/9095270059",
"air_conditioning": "no",
"clothes": "men",
"level": "0",
"name": "Maison Aume",
"second_hand": "no",
"shop": "clothes",
"wheelchair": "no"
},
"geometry": {
"type": "Point",
"coordinates": [
7.361364,
48.0799999
]
},
"id": "node/9095270059"
},
{
"type": "Feature",
"properties": {
"@id": "node/9098624272",
"name": "Destock Place",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3575161,
48.0793009
]
},
"id": "node/9098624272"
},
{
"type": "Feature",
"properties": {
"@id": "node/9123861652",
"name": "Weackers",
"shop": "shoes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.361329,
48.0785972
]
},
"id": "node/9123861652"
},
{
"type": "Feature",
"properties": {
"@id": "node/9162179887",
"brand": "Calzedonia",
"brand:wikidata": "Q1027874",
"brand:wikipedia": "en:Calzedonia",
"name": "Calzedonia",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3606374,
48.0780809
]
},
"id": "node/9162179887"
},
{
"type": "Feature",
"properties": {
"@id": "node/9162206449",
"clothes": "women",
"name": "Cop. Copine",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3600947,
48.078399
]
},
"id": "node/9162206449"
},
{
"type": "Feature",
"properties": {
"@id": "node/9162226360",
"brand": "Okaïdi",
"brand:wikidata": "Q3350027",
"brand:wikipedia": "fr:Okaïdi",
"name": "Okaïdi",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3596986,
48.078428
]
},
"id": "node/9162226360"
},
{
"type": "Feature",
"properties": {
"@id": "node/9162227010",
"brand": "Jules",
"brand:wikidata": "Q3188386",
"brand:wikipedia": "fr:Jules (enseigne)",
"clothes": "men",
"name": "Jules",
"opening_hours": "Mo-Sa 09:30-19:00",
"phone": "+33 3 89 41 03 62",
"shop": "clothes",
"website": "https://www.jules.com/fr-fr/magasins/1600133/"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3600323,
48.0782229
]
},
"id": "node/9162227010"
},
{
"type": "Feature",
"properties": {
"@id": "node/10151865029",
"name": "Atelier Cinq",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3571756,
48.0772657
]
},
"id": "node/10151865029"
},
{
"type": "Feature",
"properties": {
"@id": "node/10862176110",
"name": "L'hexagone",
"shop": "bag"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3808571,
48.0814138
]
},
"id": "node/10862176110"
},
{
"type": "Feature",
"properties": {
"@id": "node/11150877331",
"brand": "Punt Roma",
"brand:wikidata": "Q101423290",
"clothes": "women",
"name": "Punt Roma",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3571859,
48.0779406
]
},
"id": "node/11150877331"
},
{
"type": "Feature",
"properties": {
"@id": "node/11150959880",
"name": "Caroll",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3579354,
48.0779291
]
},
"id": "node/11150959880"
},
{
"type": "Feature",
"properties": {
"@id": "node/11302242094",
"branch": "Wintzenheim",
"name": "Label Fripe",
"opening_hours": "Mo-Sa 09:00-18:45",
"phone": "+33 3 89 27 39 25",
"second_hand": "only",
"shop": "clothes",
"website": "https://labelfripe.fr/label-fripe-wintzenheim/"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3109899,
48.0850362
]
},
"id": "node/11302242094"
},
{
"type": "Feature",
"properties": {
"@id": "node/11392247003",
"name": "Lingerie Sipp",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3111507,
48.0841835
]
},
"id": "node/11392247003"
},
{
"type": "Feature",
"properties": {
"@id": "node/11778819781",
"addr:city": "Colmar",
"addr:housenumber": "10",
"addr:postcode": "68000",
"addr:street": "Rue des Têtes",
"clothes": "suits;hats;men",
"name": "Phillipe",
"phone": "0389411983",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3559389,
48.0789064
]
},
"id": "node/11778819781"
},
{
"type": "Feature",
"properties": {
"@id": "node/11799215969",
"brand": "Petit Bateau",
"brand:wikidata": "Q3377090",
"name": "Petit Bateau",
"opening_hours": "Mo-Sa 10:00-19:00; Su 10:00-18:00",
"phone": "+33 3 89 24 97 85",
"shop": "clothes",
"website": "https://stores.petit-bateau.com/france/colmar/9-rue-des-boulangers"
},
"geometry": {
"type": "Point",
"coordinates": [
7.355149,
48.0780213
]
},
"id": "node/11799215969"
},
{
"type": "Feature",
"properties": {
"@id": "node/11816704669",
"addr:housenumber": "10",
"addr:street": "Rue des Boulangers",
"name": "des petits hauts",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3555001,
48.0780768
]
},
"id": "node/11816704669"
},
{
"type": "Feature",
"properties": {
"@id": "node/12320343534",
"addr:city": "Colmar",
"addr:housenumber": "44",
"addr:postcode": "68000",
"addr:street": "Rue des Clefs",
"brand": "Un Jour Ailleurs",
"brand:wikidata": "Q105106211",
"clothes": "women",
"name": "Un Jour Ailleurs",
"opening_hours": "Mo-Fr 10:00-19:00; Sa 10:00-18:30",
"phone": "+33368318572",
"shop": "clothes",
"website": "https://boutique.unjourailleurs.com/fr/mode-femme/boutique-colmar-76"
},
"geometry": {
"type": "Point",
"coordinates": [
7.35897,
48.0789807
]
},
"id": "node/12320343534"
},
{
"type": "Feature",
"properties": {
"@id": "node/12320343536",
"addr:city": "Colmar",
"addr:housenumber": "38",
"addr:postcode": "68000",
"addr:street": "Rue des Clefs",
"brand": "Timberland",
"brand:wikidata": "Q1539185",
"name": "Timberland",
"opening_hours": "Mo-Sa 10:00-19:00",
"phone": "+33389298650",
"shop": "clothes"
},
"geometry": {
"type": "Point",
"coordinates": [
7.3592409,
48.0788785
]
},
"id": "node/12320343536"
}
]
}

350
backend/src/sandbox/get_streets.py Normal file
View File

@ -0,0 +1,350 @@
# pylint: skip-file
import numpy as np
import json
import os
from typing import Optional, Literal
from sklearn.cluster import DBSCAN
from sklearn.decomposition import PCA
import matplotlib.pyplot as plt
from pydantic import BaseModel
from OSMPythonTools.overpass import Overpass, overpassQueryBuilder
from OSMPythonTools.cachingStrategy import CachingStrategy, JSON
from math import sin, cos, sqrt, atan2, radians
EARTH_RADIUS_KM = 6373
class ShoppingLocation(BaseModel):
type: Literal['street', 'area']
importance: int
centroid: tuple
start: Optional[list] = None
end: Optional[list] = None
# Output to frontend
class Landmark(BaseModel) :
# Properties of the landmark
name : str
type: Literal['sightseeing', 'nature', 'shopping', 'start', 'finish']
location : tuple
osm_type : str
osm_id : int
attractiveness : int
n_tags : int
image_url : Optional[str] = None
website_url : Optional[str] = None
description : Optional[str] = None # TODO future
duration : Optional[int] = 0
name_en : Optional[str] = None
# Additional properties depending on specific tour
must_do : Optional[bool] = False
must_avoid : Optional[bool] = False
is_secondary : Optional[bool] = False
time_to_reach_next : Optional[int] = 0
next_uuid : Optional[str] = None
def extract_points(filestr: str) :
"""
Extract points from geojson file.
Returns :
np.array containing the points
"""
points = []
with open(os.path.dirname(__file__) + '/' + filestr, 'r') as f:
geojson = json.load(f)
for feature in geojson['features']:
if feature['geometry']['type'] == 'Point':
centroid = feature['geometry']['coordinates']
points.append(centroid)
elif feature['geometry']['type'] == 'Polygon':
centroid = np.array(feature['geometry']['coordinates'][0][0])
points.append(centroid)
# Convert the list of points to a NumPy array
return np.array(points)
def get_distance(p1: tuple[float, float], p2: tuple[float, float]) -> int:
"""
Calculate the time in minutes to travel from one location to another.
Args:
p1 (tuple[float, float]): Coordinates of the starting location.
p2 (tuple[float, float]): Coordinates of the destination.
Returns:
int: Time to travel from p1 to p2 in minutes.
"""
if p1 == p2:
return 0
else:
# Compute the distance in km along the surface of the Earth
# (assume spherical Earth)
# this is the haversine formula, stolen from stackoverflow
# in order to not use any external libraries
lat1, lon1 = radians(p1[0]), radians(p1[1])
lat2, lon2 = radians(p2[0]), radians(p2[1])
dlon = lon2 - lon1
dlat = lat2 - lat1
a = sin(dlat / 2)**2 + cos(lat1) * cos(lat2) * sin(dlon / 2)**2
c = 2 * atan2(sqrt(a), sqrt(1 - a))
return EARTH_RADIUS_KM * c
def filter_clusters(cluster_points, cluster_labels):
"""
Remove clusters of less importance.
"""
label_counts = np.bincount(cluster_labels)
# Step 3: Get the indices (labels) of the 5 largest clusters
top_5_labels = np.argsort(label_counts)[-5:] # Get the largest 5 clusters
# Step 4: Filter points to keep only the points in the top 5 clusters
filtered_cluster_points = []
filtered_cluster_labels = []
for label in top_5_labels:
filtered_cluster_points.append(cluster_points[cluster_labels == label])
filtered_cluster_labels.append(np.full((label_counts[label],), label)) # Replicate the label
# Concatenate filtered clusters into a single array
return np.vstack(filtered_cluster_points), np.concatenate(filtered_cluster_labels)
def fit_lines(points, labels):
"""
Fit lines to identified clusters.
"""
all_x = []
all_y = []
lines = []
locations = []
for label in set(labels):
cluster_points = points[labels == label]
# If there's not enough points, skip
if len(cluster_points) < 2:
continue
# Apply PCA to find the principal component (i.e., the line of best fit)
pca = PCA(n_components=1)
pca.fit(cluster_points)
direction = pca.components_[0]
centroid = pca.mean_
# Project the cluster points onto the principal direction (line direction)
projections = np.dot(cluster_points - centroid, direction)
# Get the range of the projections to find the approximate length of the cluster
cluster_length = projections.max() - projections.min()
# Now adjust `t` so that it scales with the cluster length
t = np.linspace(-cluster_length / 2.75, cluster_length / 2.75, 10)
# Calculate the start and end of the line based on min/max projections
start_point = centroid[0] + t*direction[0]
end_point = centroid[1] + t*direction[1]
# Store the line
lines.append((start_point, end_point))
# For visualization, store the points
all_x.append(min(start_point))
all_x.append(max(start_point))
all_y.append(min(end_point))
all_y.append(max(end_point))
if np.linalg.norm(t) <= 0.0045 :
loc = ShoppingLocation(
type='area',
centroid=tuple((centroid[1], centroid[0])),
importance = len(cluster_points),
)
else :
loc = ShoppingLocation(
type='street',
centroid=tuple((centroid[1], centroid[0])),
importance = len(cluster_points),
start=start_point,
end=end_point
)
locations.append(loc)
xmin = min(all_x)
xmax = max(all_x)
ymin = min(all_y)
ymax = max(all_y)
corners = (xmin, xmax, ymin, ymax)
return corners, locations
def create_landmark(shopping_location: ShoppingLocation):
# Define the bounding box for a given radius around the coordinates
lat, lon = shopping_location.centroid
bbox = ("around:1000", str(lat), str(lon))
overpass = Overpass()
# CachingStrategy.use(JSON, cacheDir=OSM_CACHE_DIR)
# Query neighborhoods and shopping malls
selectors = ['"place"~"^(suburb|neighborhood|neighbourhood|quarter|city_block)$"', '"shop"="mall"']
min_dist = float('inf')
new_name = 'Shopping Area'
new_name_en = None
osm_id = 0
osm_type = 'node'
for sel in selectors :
query = overpassQueryBuilder(
bbox = bbox,
elementType = ['node', 'way', 'relation'],
selector = sel,
includeCenter = True,
out = 'center'
)
try:
result = overpass.query(query)
except Exception as e:
raise Exception("query unsuccessful")
for elem in result.elements():
location = (elem.centerLat(), elem.centerLon())
if location[0] is None :
location = (elem.lat(), elem.lon())
if location[0] is None :
# print(f"Fetching coordinates failed with {elem.type()}/{elem.id()}")
continue
# print(f"Distance : {get_distance(shopping_location.centroid, location)}")
d = get_distance(shopping_location.centroid, location)
if d < min_dist :
min_dist = d
new_name = elem.tag('name')
osm_type = elem.type() # Add type: 'way' or 'relation'
osm_id = elem.id() # Add OSM id
# add english name if it exists
try :
new_name_en = elem.tag('name:en')
except:
pass
return Landmark(
name=new_name,
type='shopping',
location=shopping_location.centroid, # TODO: use the fact the we can also recognize streets.
attractiveness=shopping_location.importance,
n_tags=0,
osm_id=osm_id,
osm_type=osm_type,
name_en=new_name_en
)
# Extract points
points = extract_points('vienna_data.json')
# print(len(points))
######## Create a figure with 1 row and 3 columns for side-by-side plots
fig, axes = plt.subplots(1, 3, figsize=(15, 5))
# Plot Raw data points
axes[0].set_title('Raw Data')
axes[0].scatter(points[:, 0], points[:, 1], color='blue', s=20)
# Apply DBSCAN to find clusters. Choose different settings for different cities.
if len(points) > 400 :
dbscan = DBSCAN(eps=0.00118, min_samples=15, algorithm='kd_tree') # for large cities
else :
dbscan = DBSCAN(eps=0.00075, min_samples=10, algorithm='kd_tree') # for small cities
labels = dbscan.fit_predict(points)
# Separate clustered points and noise points
clustered_points = points[labels != -1]
clustered_labels = labels[labels != -1]
noise_points = points[labels == -1]
######## Plot n°1: DBSCAN Clustering Results
axes[1].set_title('DBSCAN Clusters')
axes[1].scatter(clustered_points[:, 0], clustered_points[:, 1], c=clustered_labels, cmap='rainbow', s=20)
axes[1].scatter(noise_points[:, 0], noise_points[:, 1], c='blue', s=7, label='Noise')
# Keep the 5 biggest clusters
clustered_points, clustered_labels = filter_clusters(clustered_points, clustered_labels)
# Fit lines
corners, locations = fit_lines(clustered_points, clustered_labels)
(xmin, xmax, ymin, ymax) = corners
######## Plot clustered points in normal size and noise points separately
axes[2].scatter(clustered_points[:, 0], clustered_points[:, 1], c=clustered_labels, cmap='rainbow', s=30)
axes[2].set_title('PCA Fitted Lines on Clusters')
# Create a list of Landmarks for the shopping things
shopping_landmarks = []
for loc in locations :
axes[2].scatter(loc.centroid[1], loc.centroid[0], color='red', marker='x', s=200, linewidth=3)
landmark = create_landmark(loc)
shopping_landmarks.append(landmark)
axes[2].text(loc.centroid[1], loc.centroid[0], landmark.name,
ha='center', va='top', fontsize=6,
bbox=dict(facecolor='white', edgecolor='black', boxstyle='round,pad=0.2'),
zorder=3)
####### Plot the detected lines in the final plot #######
# for loc in locations:
# if loc.type == 'street' :
# line_x = loc.start
# line_y = loc.end
# axes[2].plot(line_x, line_y, color='lime', linewidth=3)
# else :
axes[0].set_xlim(xmin-0.01, xmax+0.01)
axes[0].set_ylim(ymin-0.01, ymax+0.01)
axes[1].set_xlim(xmin-0.01, xmax+0.01)
axes[1].set_ylim(ymin-0.01, ymax+0.01)
axes[2].set_xlim(xmin-0.01, xmax+0.01)
axes[2].set_ylim(ymin-0.01, ymax+0.01)
print("\n\n\n")
for landmark in shopping_landmarks :
print(f"{landmark.name} is a shopping area with a score of {landmark.attractiveness}")
plt.tight_layout()
plt.show()

17824
backend/src/sandbox/lyon_data.json Normal file
View File

File diff suppressed because it is too large Load Diff

42085
backend/src/sandbox/newyork_data.json Normal file
View File

File diff suppressed because it is too large Load Diff

83615
backend/src/sandbox/paris_data.json Normal file
View File

File diff suppressed because it is too large Load Diff

4947
backend/src/sandbox/strasbourg_data.json Normal file
View File

File diff suppressed because it is too large Load Diff

23140
backend/src/sandbox/vienna_data.json Normal file
View File

File diff suppressed because it is too large Load Diff

2844
backend/src/sandbox/winterthur_data.json Normal file
View File

File diff suppressed because it is too large Load Diff

16070
backend/src/sandbox/zurich_data.json Normal file
View File

File diff suppressed because it is too large Load Diff

27
backend/src/structs/landmark.py
View File

@ -73,8 +73,6 @@ class Landmark(BaseModel) :
t_to_next_str = f", time_to_next={self.time_to_reach_next}" if self.time_to_reach_next else ""
is_secondary_str = ", secondary" if self.is_secondary else ""
type_str = '(' + self.type + ')'
if self.type in ["start", "finish", "nature", "shopping"] :
type_str += '\t '
return (f'Landmark{type_str}: [{self.name} @{self.location}, '
f'score={self.attractiveness}{t_to_next_str}{is_secondary_str}]')
@ -117,3 +115,28 @@ class Landmark(BaseModel) :
return (self.uuid == value.uuid or
self.osm_id == value.osm_id or
(self.name == value.name and self.distance(value) < 0.001))
class Toilets(BaseModel) :
"""
Model for toilets. When false/empty the information is either false either not known.
"""
location : tuple
wheelchair : Optional[bool] = False
changing_table : Optional[bool] = False
fee : Optional[bool] = False
opening_hours : Optional[str] = ""
def __str__(self) -> str:
"""
String representation of the Toilets object.
Returns:
str: A formatted string with the toilets location.
"""
return f'Toilets @{self.location}'
class Config:
# This allows us to easily convert the model to and from dictionaries
orm_mode = True

42
backend/src/tests/test_cache.py Normal file
View File

@ -0,0 +1,42 @@
"""Collection of tests to ensure correct handling of invalid input."""
from fastapi.testclient import TestClient
import pytest
from .test_utils import load_trip_landmarks
from ..main import app
@pytest.fixture(scope="module")
def client():
"""Client used to call the app."""
return TestClient(app)
def test_cache(client, request): # pylint: disable=redefined-outer-name
"""
Test n°1 : Custom test in Turckheim to ensure small villages are also supported.
Args:
client:
request:
"""
duration_minutes = 15
response = client.post(
"/trip/new",
json={
"preferences": {"sightseeing": {"type": "sightseeing", "score": 5},
"nature": {"type": "nature", "score": 5},
"shopping": {"type": "shopping", "score": 5},
"max_time_minute": duration_minutes,
"detour_tolerance_minute": 0},
"start": [48.084588, 7.280405]
}
)
result = response.json()
landmarks = load_trip_landmarks(client, result['first_landmark_uuid'])
landmarks_cached = load_trip_landmarks(client, result['first_landmark_uuid'], True)
# checks :
assert response.status_code == 200 # check for successful planning
assert landmarks_cached == landmarks

8
backend/src/tests/test_invalid_input.py
View File

@ -33,19 +33,19 @@ def invalid_client():
([91, 181], {"sightseeing": {"type": "nature", "score": 5},
"nature": {"type": "nature", "score": 5},
"shopping": {"type": "shopping", "score": 5},
}, 423),
}, 422),
([-91, 181], {"sightseeing": {"type": "nature", "score": 5},
"nature": {"type": "nature", "score": 5},
"shopping": {"type": "shopping", "score": 5},
}, 423),
}, 422),
([91, -181], {"sightseeing": {"type": "nature", "score": 5},
"nature": {"type": "nature", "score": 5},
"shopping": {"type": "shopping", "score": 5},
}, 423),
}, 422),
([-91, -181], {"sightseeing": {"type": "nature", "score": 5},
"nature": {"type": "nature", "score": 5},
"shopping": {"type": "shopping", "score": 5},
}, 423),
}, 422),
]
)
def test_input(invalid_client, start, preferences, status_code): # pylint: disable=redefined-outer-name

30
backend/src/tests/test_main.py
View File

@ -78,6 +78,36 @@ def test_bellecour(client, request) : # pylint: disable=redefined-outer-name
assert 136200148 in osm_ids # check for Cathédrale St. Jean in trip
def test_shopping(client, request) : # pylint: disable=redefined-outer-name
"""
Test n°3 : Custom test in Lyon centre to ensure shopping clusters are found.
Args:
client:
request:
"""
duration_minutes = 600
response = client.post(
"/trip/new",
json={
"preferences": {"sightseeing": {"type": "sightseeing", "score": 0},
"nature": {"type": "nature", "score": 0},
"shopping": {"type": "shopping", "score": 5},
"max_time_minute": duration_minutes,
"detour_tolerance_minute": 0},
"start": [45.7576485, 4.8330241]
}
)
result = response.json()
landmarks = load_trip_landmarks(client, result['first_landmark_uuid'])
# osm_ids = landmarks_to_osmid(landmarks)
# Add details to report
log_trip_details(request, landmarks, result['total_time'], duration_minutes)
# checks :
assert response.status_code == 200 # check for successful planning
assert duration_minutes*0.8 < int(result['total_time']) < duration_minutes*1.2
# def test_new_trip_single_prefs(client):
# response = client.post(

102
backend/src/tests/test_toilets.py Normal file
View File

@ -0,0 +1,102 @@
"""Collection of tests to ensure correct implementation and track progress. """
from fastapi.testclient import TestClient
import pytest
from ..structs.landmark import Toilets
from ..main import app
@pytest.fixture(scope="module")
def client():
"""Client used to call the app."""
return TestClient(app)
@pytest.mark.parametrize(
"location,radius,status_code",
[
({}, None, 422), # Invalid case: no location at all.
([443], None, 422), # Invalid cases: invalid location.
([443, 433], None, 422), # Invalid cases: invalid location.
]
)
def test_invalid_input(client, location, radius, status_code): # pylint: disable=redefined-outer-name
"""
Test n°1 : Verify handling of invalid input.
Args:
client:
request:
"""
response = client.post(
"/toilets/new",
params={
"location": location,
"radius": radius
}
)
# checks :
assert response.status_code == status_code
@pytest.mark.parametrize(
"location,status_code",
[
([48.2270, 7.4370], 200), # Orschwiller.
([10.2012, 10.123], 200), # Nigerian desert.
([63.989, -19.677], 200), # Hekla volcano, Iceland
]
)
def test_no_toilets(client, location, status_code): # pylint: disable=redefined-outer-name
"""
Test n°3 : Verify the code finds some toilets in big cities.
Args:
client:
request:
"""
response = client.post(
"/toilets/new",
params={
"location": location
}
)
toilets_list = [Toilets.model_validate(toilet) for toilet in response.json()]
# checks :
assert response.status_code == 200 # check for successful planning
assert isinstance(toilets_list, list) # check that the return type is a list
@pytest.mark.parametrize(
"location,status_code",
[
([45.7576485, 4.8330241], 200), # Lyon, Bellecour.
([-6.913795, 107.60278], 200), # Bandung, train station
([-22.970140, -43.18181], 200), # Rio de Janeiro, Copacabana
]
)
def test_toilets(client, location, status_code): # pylint: disable=redefined-outer-name
"""
Test n°3 : Verify the code finds some toilets in big cities.
Args:
client:
request:
"""
response = client.post(
"/toilets/new",
params={
"location": location,
"radius" : 600
}
)
toilets_list = [Toilets.model_validate(toilet) for toilet in response.json()]
# checks :
assert response.status_code == 200 # check for successful planning
assert isinstance(toilets_list, list) # check that the return type is a list
assert len(toilets_list) > 0

79
backend/src/tests/test_utils.py
View File

@ -1,11 +1,13 @@
"""Helper methods for testing."""
from typing import List
import logging
from fastapi import HTTPException
from pydantic import ValidationError
from ..structs.landmark import Landmark
from ..persistence import client as cache_client
def landmarks_to_osmid(landmarks: List[Landmark]) -> List[int] :
def landmarks_to_osmid(landmarks: list[Landmark]) -> list[int] :
"""
Convert the list of landmarks into a list containing their osm ids for quick landmark checking.
@ -31,22 +33,68 @@ def fetch_landmark(client, landmark_uuid: str):
Returns:
dict: Landmark data fetched from the API.
"""
logger = logging.getLogger(__name__)
response = client.get(f"/landmark/{landmark_uuid}")
if response.status_code != 200:
raise HTTPException(status_code=999,
raise HTTPException(status_code=500,
detail=f"Failed to fetch landmark with UUID {landmark_uuid}: {response.status_code}")
json_data = response.json()
try:
json_data = response.json()
logger.info(f"API Response: {json_data}")
except ValueError as e:
logger.error(f"Failed to parse response as JSON: {response.text}")
raise HTTPException(status_code=500, detail="Invalid response format from API")
# Try validating against the Landmark model here to ensure consistency
try:
landmark = Landmark(**json_data)
except ValidationError as ve:
logging.error(f"Validation error: {ve}")
raise HTTPException(status_code=500, detail="Invalid data format received from API")
if "detail" in json_data:
raise HTTPException(status_code=999, detail=json_data["detail"])
raise HTTPException(status_code=500, detail=json_data["detail"])
return Landmark(**json_data)
return json_data
def fetch_landmark_cache(landmark_uuid: str):
"""
Fetch landmark data from the cache based on the landmark UUID.
Args:
landmark_uuid (str): The UUID of the landmark.
Returns:
dict: Landmark data fetched from the cache or raises an HTTP exception.
"""
logger = logging.getLogger(__name__)
# Try to fetch the landmark data from the cache
try:
landmark = cache_client.get(f"landmark_{landmark_uuid}")
if not landmark :
logger.warning(f"Cache miss for landmark UUID: {landmark_uuid}")
raise HTTPException(status_code=404, detail=f"Landmark with UUID {landmark_uuid} not found in cache.")
# Validate that the fetched data is a dictionary
if not isinstance(landmark, Landmark):
logger.error(f"Invalid cache data format for landmark UUID: {landmark_uuid}. Expected dict, got {type(landmark).__name__}.")
raise HTTPException(status_code=500, detail="Invalid cache data format.")
return landmark
except Exception as exc:
logger.error(f"Unexpected error occurred while fetching landmark UUID {landmark_uuid}: {exc}")
raise HTTPException(status_code=500, detail="An unexpected error occurred while fetching the landmark from the cache") from exc
def load_trip_landmarks(client, first_uuid: str) -> List[Landmark]:
def load_trip_landmarks(client, first_uuid: str, from_cache=None) -> list[Landmark]:
"""
Load all landmarks for a trip using the response from the API.
@ -60,19 +108,18 @@ def load_trip_landmarks(client, first_uuid: str) -> List[Landmark]:
next_uuid = first_uuid
while next_uuid is not None:
landmark_data = fetch_landmark(client, next_uuid)
# # Convert UUIDs to strings explicitly
# landmark_data = {
# key: str(value) if isinstance(value, UUID) else value
# for key, value in landmark_data.items()
# }
landmarks.append(Landmark(**landmark_data)) # Create Landmark objects
next_uuid = landmark_data.get('next_uuid') # Prepare for the next iteration
if from_cache :
landmark = fetch_landmark_cache(next_uuid)
else :
landmark = fetch_landmark(client, next_uuid)
landmarks.append(landmark)
next_uuid = landmark.next_uuid # Prepare for the next iteration
return landmarks
def log_trip_details(request, landmarks: List[Landmark], duration: int, target_duration: int) :
def log_trip_details(request, landmarks: list[Landmark], duration: int, target_duration: int) :
"""
Allows to show the detailed trip in the html test report.

283
backend/src/utils/cluster_processing.py Normal file
View File

@ -0,0 +1,283 @@
import logging
from typing import Literal
import numpy as np
from sklearn.cluster import DBSCAN
from pydantic import BaseModel
from OSMPythonTools.overpass import Overpass, overpassQueryBuilder
from OSMPythonTools.cachingStrategy import CachingStrategy, JSON
from ..structs.landmark import Landmark
from ..utils.get_time_separation import get_distance
from ..constants import AMENITY_SELECTORS_PATH, LANDMARK_PARAMETERS_PATH, OPTIMIZER_PARAMETERS_PATH, OSM_CACHE_DIR
class ShoppingLocation(BaseModel):
""""
A classe representing an interesting area for shopping.
It can represent either a general area or a specifc route with start and end point.
The importance represents the number of shops found in this cluster.
Attributes:
type : either a 'street' or 'area' (representing a denser field of shops).
importance : size of the cluster (number of points).
centroid : center of the cluster.
start : if the type is a street it goes from here...
end : ...to here
"""
type: Literal['street', 'area']
importance: int
centroid: tuple
# start: Optional[list] = None # for later use if we want to have streets as well
# end: Optional[list] = None
class ShoppingManager:
logger = logging.getLogger(__name__)
# NOTE: all points are in (lat, lon) format
valid: bool # Ensure the manager is valid (ie there are some clusters to be found)
all_points: list
cluster_points: list
cluster_labels: list
shopping_locations: list[ShoppingLocation]
def __init__(self, bbox: tuple) -> None:
"""
Upon intialization, generate the point cloud used for cluster detection.
The points represent bag/clothes shops and general boutiques.
Args:
bbox: The bounding box coordinates (around:radius, center_lat, center_lon).
"""
# Initialize overpass and cache
self.overpass = Overpass()
CachingStrategy.use(JSON, cacheDir=OSM_CACHE_DIR)
# Initialize the points for cluster detection
query = overpassQueryBuilder(
bbox = bbox,
elementType = ['node'],
selector = ['"shop"~"^(bag|boutique|clothes)$"'],
includeCenter = True,
out = 'skel'
)
try:
result = self.overpass.query(query)
except Exception as e:
self.logger.error(f"Error fetching landmarks: {e}")
if len(result.elements()) == 0 :
self.valid = False
else :
points = []
for elem in result.elements() :
points.append(tuple((elem.lat(), elem.lon())))
self.all_points = np.array(points)
self.valid = True
def generate_shopping_landmarks(self) -> list[Landmark]:
"""
Generate shopping landmarks based on clustered locations.
This method first generates clusters of locations and then extracts shopping-related
locations from these clusters. It transforms each shopping location into a `Landmark` object.
Returns:
list[Landmark]: A list of `Landmark` objects representing shopping locations.
Returns an empty list if no clusters are found.
"""
self.generate_clusters()
if len(set(self.cluster_labels)) == 0 :
return [] # Return empty list if no clusters were found
# Then generate the shopping locations
self.generate_shopping_locations()
# Transform the locations in landmarks and return the list
shopping_landmarks = []
for location in self.shopping_locations :
shopping_landmarks.append(self.create_landmark(location))
return shopping_landmarks
def generate_clusters(self) :
"""
Generate clusters of points using DBSCAN.
This method applies the DBSCAN clustering algorithm with different
parameters depending on the size of the city (number of points).
It filters out noise points and keeps only the largest clusters.
The method updates:
- `self.cluster_points`: The points belonging to clusters.
- `self.cluster_labels`: The labels for the points in clusters.
The method also calls `filter_clusters()` to retain only the largest clusters.
"""
# Apply DBSCAN to find clusters. Choose different settings for different cities.
if len(self.all_points) > 200 :
dbscan = DBSCAN(eps=0.00118, min_samples=15, algorithm='kd_tree') # for large cities
else :
dbscan = DBSCAN(eps=0.00075, min_samples=10, algorithm='kd_tree') # for small cities
labels = dbscan.fit_predict(self.all_points)
# Separate clustered points and noise points
self.cluster_points = self.all_points[labels != -1]
self.cluster_labels = labels[labels != -1]
# filter the clusters to keep only the largest ones
self.filter_clusters()
def generate_shopping_locations(self) :
"""
Generate shopping locations based on clustered points.
This method iterates over the different clusters, calculates the centroid
(as the mean of the points within each cluster), and assigns an importance
based on the size of the cluster.
The generated shopping locations are stored in `self.shopping_locations`
as a list of `ShoppingLocation` objects, each with:
- `type`: Set to 'area'.
- `centroid`: The calculated centroid of the cluster.
- `importance`: The number of points in the cluster.
"""
locations = []
# loop through the different clusters
for label in set(self.cluster_labels):
# Extract points belonging to the current cluster
current_cluster = self.cluster_points[self.cluster_labels == label]
# Calculate the centroid as the mean of the points
centroid = np.mean(current_cluster, axis=0)
locations.append(ShoppingLocation(
type='area',
centroid=centroid,
importance = len(current_cluster)
))
self.shopping_locations = locations
def create_landmark(self, shopping_location: ShoppingLocation) -> Landmark:
"""
Create a Landmark object based on the given shopping location.
This method queries the Overpass API for nearby neighborhoods and shopping malls
within a 1000m radius around the shopping location centroid. It selects the closest
result and creates a landmark with the associated details such as name, type, and OSM ID.
Parameters:
shopping_location (ShoppingLocation): A ShoppingLocation object containing
the centroid and importance of the area.
Returns:
Landmark: A Landmark object containing details such as the name, type,
location, attractiveness, and OSM details.
"""
# Define the bounding box for a given radius around the coordinates
lat, lon = shopping_location.centroid
bbox = ("around:1000", str(lat), str(lon))
# Query neighborhoods and shopping malls
selectors = ['"place"~"^(suburb|neighborhood|neighbourhood|quarter|city_block)$"', '"shop"="mall"']
min_dist = float('inf')
new_name = 'Shopping Area'
new_name_en = None
osm_id = 0
osm_type = 'node'
for sel in selectors :
query = overpassQueryBuilder(
bbox = bbox,
elementType = ['node', 'way', 'relation'],
selector = sel,
includeCenter = True,
out = 'center'
)
try:
result = self.overpass.query(query)
except Exception as e:
self.logger.error(f"Error fetching landmarks: {e}")
continue
for elem in result.elements():
location = (elem.centerLat(), elem.centerLon())
if location[0] is None :
location = (elem.lat(), elem.lon())
if location[0] is None :
continue
d = get_distance(shopping_location.centroid, location)
if d < min_dist :
min_dist = d
new_name = elem.tag('name')
osm_type = elem.type() # Add type: 'way' or 'relation'
osm_id = elem.id() # Add OSM id
# Add english name if it exists
try :
new_name_en = elem.tag('name:en')
except:
pass
return Landmark(
name=new_name,
type='shopping',
location=shopping_location.centroid, # TODO: use the fact the we can also recognize streets.
attractiveness=shopping_location.importance,
n_tags=0,
osm_id=osm_id,
osm_type=osm_type,
name_en=new_name_en
)
def filter_clusters(self):
"""
Filter clusters to retain only the 5 largest clusters by point count.
This method calculates the size of each cluster and filters out all but the
5 largest clusters. It then updates the cluster points and labels to reflect
only those from the top 5 clusters.
"""
label_counts = np.bincount(self.cluster_labels)
# Step 3: Get the indices (labels) of the 5 largest clusters
top_5_labels = np.argsort(label_counts)[-5:] # Get the largest 5 clusters
# Step 4: Filter points to keep only the points in the top 5 clusters
filtered_cluster_points = []
filtered_cluster_labels = []
for label in top_5_labels:
filtered_cluster_points.append(self.cluster_points[self.cluster_labels == label])
filtered_cluster_labels.append(np.full((label_counts[label],), label)) # Replicate the label
# update the cluster points and labels with the filtered data
self.cluster_points = np.vstack(filtered_cluster_points)
self.cluster_labels = np.concatenate(filtered_cluster_labels)

36
backend/src/utils/get_time_separation.py
View File

@ -15,8 +15,8 @@ def get_time(p1: tuple[float, float], p2: tuple[float, float]) -> int:
Calculate the time in minutes to travel from one location to another.
Args:
p1 (Tuple[float, float]): Coordinates of the starting location.
p2 (Tuple[float, float]): Coordinates of the destination.
p1 (tuple[float, float]): Coordinates of the starting location.
p2 (tuple[float, float]): Coordinates of the destination.
Returns:
int: Time to travel from p1 to p2 in minutes.
@ -48,3 +48,35 @@ def get_time(p1: tuple[float, float], p2: tuple[float, float]) -> int:
walk_time = walk_distance / AVERAGE_WALKING_SPEED * 60
return round(walk_time)
def get_distance(p1: tuple[float, float], p2: tuple[float, float]) -> int:
"""
Calculate the time in minutes to travel from one location to another.
Args:
p1 (tuple[float, float]): Coordinates of the starting location.
p2 (tuple[float, float]): Coordinates of the destination.
Returns:
int: Time to travel from p1 to p2 in minutes.
"""
if p1 == p2:
return 0
else:
# Compute the distance in km along the surface of the Earth
# (assume spherical Earth)
# this is the haversine formula, stolen from stackoverflow
# in order to not use any external libraries
lat1, lon1 = radians(p1[0]), radians(p1[1])
lat2, lon2 = radians(p2[0]), radians(p2[1])
dlon = lon2 - lon1
dlat = lat2 - lat1
a = sin(dlat / 2)**2 + cos(lat1) * cos(lat2) * sin(dlon / 2)**2
c = 2 * atan2(sqrt(a), sqrt(1 - a))
return EARTH_RADIUS_KM * c

68
backend/src/utils/landmarks_manager.py
View File

@ -1,13 +1,11 @@
import math
import yaml
import logging
import math, yaml, logging
from OSMPythonTools.overpass import Overpass, overpassQueryBuilder
from OSMPythonTools.cachingStrategy import CachingStrategy, JSON
from ..structs.preferences import Preferences
from ..structs.landmark import Landmark
from .take_most_important import take_most_important
from .cluster_processing import ShoppingManager
from ..constants import AMENITY_SELECTORS_PATH, LANDMARK_PARAMETERS_PATH, OPTIMIZER_PARAMETERS_PATH, OSM_CACHE_DIR
@ -79,7 +77,9 @@ class LandmarkManager:
# use set to avoid duplicates, this requires some __methods__ to be set in Landmark
all_landmarks = set()
bbox = self.create_bbox(center_coordinates, reachable_bbox_side)
# Create a bbox using the around technique
bbox = tuple((f"around:{reachable_bbox_side/2}", str(center_coordinates[0]), str(center_coordinates[1])))
# list for sightseeing
if preferences.sightseeing.score != 0:
score_function = lambda score: score * 10 * preferences.sightseeing.score / 5
@ -96,10 +96,19 @@ class LandmarkManager:
if preferences.shopping.score != 0:
score_function = lambda score: score * 10 * preferences.shopping.score / 5
current_landmarks = self.fetch_landmarks(bbox, self.amenity_selectors['shopping'], preferences.shopping.type, score_function)
# set time for all shopping activites :
for landmark in current_landmarks : landmark.duration = 45
for landmark in current_landmarks : landmark.duration = 30
all_landmarks.update(current_landmarks)
# special pipeline for shopping malls
shopping_manager = ShoppingManager(bbox)
if shopping_manager.valid :
shopping_clusters = shopping_manager.generate_shopping_landmarks()
for landmark in shopping_clusters : landmark.duration = 45
all_landmarks.update(shopping_clusters)
landmarks_constrained = take_most_important(all_landmarks, self.N_important)
self.logger.info(f'Generated {len(all_landmarks)} landmarks around {center_coordinates}, and constrained to {len(landmarks_constrained)} most important ones.')
@ -151,36 +160,24 @@ class LandmarkManager:
return 0
def create_bbox(self, coordinates: tuple[float, float], reachable_bbox_side: int) -> tuple[float, float, float, float]:
"""
Create a bounding box around the given coordinates.
# def create_bbox(self, coordinates: tuple[float, float], reachable_bbox_side: int) -> tuple[float, float, float, float]:
# """
# Create a bounding box around the given coordinates.
Args:
coordinates (tuple[float, float]): The latitude and longitude of the center of the bounding box.
reachable_bbox_side (int): The side length of the bounding box in meters.
# Args:
# coordinates (tuple[float, float]): The latitude and longitude of the center of the bounding box.
# reachable_bbox_side (int): The side length of the bounding box in meters.
Returns:
tuple[float, float, float, float]: The minimum latitude, minimum longitude, maximum latitude, and maximum longitude
defining the bounding box.
"""
lat = coordinates[0]
lon = coordinates[1]
# Returns:
# tuple[float, float, float, float]: The minimum latitude, minimum longitude, maximum latitude, and maximum longitude
# defining the bounding box.
# """
# Half the side length in km (since it's a square bbox)
half_side_length_km = reachable_bbox_side / 2 / 1000
# # Half the side length in m (since it's a square bbox)
# half_side_length_m = reachable_bbox_side / 2
# Convert distance to degrees
lat_diff = half_side_length_km / 111 # 1 degree latitude is approximately 111 km
lon_diff = half_side_length_km / (111 * math.cos(math.radians(lat))) # Adjust for longitude based on latitude
# return tuple((f"around:{half_side_length_m}", str(coordinates[0]), str(coordinates[1])))
# Calculate bbox
min_lat = lat - lat_diff
max_lat = lat + lat_diff
min_lon = lon - lon_diff
max_lon = lon + lon_diff
return min_lat, min_lon, max_lat, max_lon
def fetch_landmarks(self, bbox: tuple, amenity_selector: dict, landmarktype: str, score_function: callable) -> list[Landmark]:
@ -188,7 +185,7 @@ class LandmarkManager:
Fetches landmarks of a specified type from OpenStreetMap (OSM) within a bounding box centered on given coordinates.
Args:
bbox (tuple[float, float, float, float]): The bounding box coordinates (min_lat, min_lon, max_lat, max_lon).
bbox (tuple[float, float, float, float]): The bounding box coordinates (around:radius, center_lat, center_lon).
amenity_selector (dict): The Overpass API query selector for the desired landmark type.
landmarktype (str): The type of the landmark (e.g., 'sightseeing', 'nature', 'shopping').
score_function (callable): The function to compute the score of the landmark based on its attributes.
@ -212,7 +209,9 @@ class LandmarkManager:
for sel in dict_to_selector_list(amenity_selector):
self.logger.debug(f"Current selector: {sel}")
query_conditions = ['count_tags()>5']
# query_conditions = ['count_tags()>5']
# if landmarktype == 'shopping' : # use this later for shopping clusters
# element_types = ['node']
element_types = ['way', 'relation']
if 'viewpoint' in sel :
@ -228,7 +227,7 @@ class LandmarkManager:
selector = sel,
conditions = query_conditions, # except for nature....
includeCenter = True,
out = 'body'
out = 'center'
)
self.logger.debug(f"Query: {query}")
@ -365,7 +364,6 @@ class LandmarkManager:
return return_list
def dict_to_selector_list(d: dict) -> list:
"""
Convert a dictionary of key-value pairs to a list of Overpass query strings.

22
backend/src/utils/optimizer.py
View File

@ -44,7 +44,7 @@ class Optimizer:
resx (list[float]): List of edge weights.
Returns:
Tuple[list[int], list[int]]: A tuple containing a new row for constraint matrix and new value for upper bound vector.
tuple[list[int], list[int]]: A tuple containing a new row for constraint matrix and new value for upper bound vector.
"""
for i, elem in enumerate(resx):
@ -79,7 +79,7 @@ class Optimizer:
L (int): Number of landmarks.
Returns:
Tuple[np.ndarray, list[int]]: A tuple containing a new row for constraint matrix and new value for upper bound vector.
tuple[np.ndarray, list[int]]: A tuple containing a new row for constraint matrix and new value for upper bound vector.
"""
l1 = [0]*L*L
@ -107,7 +107,7 @@ class Optimizer:
resx (list): List of edge weights.
Returns:
Tuple[list[int], Optional[list[list[int]]]]: A tuple containing the visit order and a list of any detected circles.
tuple[list[int], Optional[list[list[int]]]]: A tuple containing the visit order and a list of any detected circles.
"""
# first round the results to have only 0-1 values
@ -180,7 +180,7 @@ class Optimizer:
max_time (int): Maximum time of visit allowed.
Returns:
Tuple[list[float], list[float], list[int]]: Objective function coefficients, inequality constraint coefficients, and the right-hand side of the inequality constraint.
tuple[list[float], list[float], list[int]]: Objective function coefficients, inequality constraint coefficients, and the right-hand side of the inequality constraint.
"""
# Objective function coefficients. a*x1 + b*x2 + c*x3 + ...
@ -212,7 +212,7 @@ class Optimizer:
L (int): Number of landmarks.
Returns:
Tuple[np.ndarray, list[int]]: Inequality constraint coefficients and the right-hand side of the inequality constraints.
tuple[np.ndarray, list[int]]: Inequality constraint coefficients and the right-hand side of the inequality constraints.
"""
ones = [1]*L
@ -239,7 +239,7 @@ class Optimizer:
L (int): Number of landmarks.
Returns:
Tuple[np.ndarray, list[int]]: Inequality constraint coefficients and the right-hand side of the inequality constraints.
tuple[np.ndarray, list[int]]: Inequality constraint coefficients and the right-hand side of the inequality constraints.
"""
upper_ind = np.triu_indices(L,0,L)
@ -270,7 +270,7 @@ class Optimizer:
L (int): Number of landmarks.
Returns:
Tuple[list[np.ndarray], list[int]]: Equality constraint coefficients and the right-hand side of the equality constraints.
tuple[list[np.ndarray], list[int]]: Equality constraint coefficients and the right-hand side of the equality constraints.
"""
l = [0]*L*L
@ -293,7 +293,7 @@ class Optimizer:
landmarks (list[Landmark]): List of landmarks, where some are marked as 'must_do'.
Returns:
Tuple[np.ndarray, list[int]]: Inequality constraint coefficients and the right-hand side of the inequality constraints.
tuple[np.ndarray, list[int]]: Inequality constraint coefficients and the right-hand side of the inequality constraints.
"""
L = len(landmarks)
@ -319,7 +319,7 @@ class Optimizer:
landmarks (list[Landmark]): List of landmarks, where some are marked as 'must_avoid'.
Returns:
Tuple[np.ndarray, list[int]]: Inequality constraint coefficients and the right-hand side of the inequality constraints.
tuple[np.ndarray, list[int]]: Inequality constraint coefficients and the right-hand side of the inequality constraints.
"""
L = len(landmarks)
@ -346,7 +346,7 @@ class Optimizer:
L (int): Number of landmarks.
Returns:
Tuple[np.ndarray, list[int]]: Inequality constraint coefficients and the right-hand side of the inequality constraints.
tuple[np.ndarray, list[int]]: Inequality constraint coefficients and the right-hand side of the inequality constraints.
"""
l_start = [1]*L + [0]*L*(L-1) # sets departures only for start (horizontal ones)
@ -374,7 +374,7 @@ class Optimizer:
L (int): Number of landmarks.
Returns:
Tuple[np.ndarray, list[int]]: Inequality constraint coefficients and the right-hand side of the inequality constraints.
tuple[np.ndarray, list[int]]: Inequality constraint coefficients and the right-hand side of the inequality constraints.
"""
A = [0]*L*L

3
backend/src/utils/refiner.py
View File

@ -2,7 +2,6 @@ import yaml, logging
from shapely import buffer, LineString, Point, Polygon, MultiPoint, concave_hull
from math import pi
from typing import List
from ..structs.landmark import Landmark
from . import take_most_important, get_time_separation
@ -135,7 +134,7 @@ class Refiner :
return tour
def integrate_landmarks(self, sub_list: List[Landmark], main_list: List[Landmark]) :
def integrate_landmarks(self, sub_list: list[Landmark], main_list: list[Landmark]) :
"""
Inserts 'sub_list' of Landmarks inside the 'main_list' by leaving the ends untouched.

78
backend/src/utils/toilets_manager.py Normal file
View File

@ -0,0 +1,78 @@
import logging, yaml
from OSMPythonTools.overpass import Overpass, overpassQueryBuilder
from OSMPythonTools.cachingStrategy import CachingStrategy, JSON
from ..structs.landmark import Toilets
from ..constants import LANDMARK_PARAMETERS_PATH, OSM_CACHE_DIR
# silence the overpass logger
logging.getLogger('OSMPythonTools').setLevel(level=logging.CRITICAL)
class ToiletsManager:
logger = logging.getLogger(__name__)
location: tuple[float, float]
radius: int # radius in meters
def __init__(self, location: tuple[float, float], radius : int) -> None:
self.radius = radius
self.location = location
self.overpass = Overpass()
CachingStrategy.use(JSON, cacheDir=OSM_CACHE_DIR)
def generate_toilet_list(self) -> list[Toilets] :
# Create a bbox using the around technique
bbox = tuple((f"around:{self.radius}", str(self.location[0]), str(self.location[1])))
toilets_list = []
query = overpassQueryBuilder(
bbox = bbox,
elementType = ['node', 'way', 'relation'],
# selector can in principle be a list already,
# but it generates the intersection of the queries
# we want the union
selector = ['"amenity"="toilets"'],
includeCenter = True,
out = 'center'
)
self.logger.debug(f"Query: {query}")
try:
result = self.overpass.query(query)
except Exception as e:
self.logger.error(f"Error fetching landmarks: {e}")
return None
for elem in result.elements():
location = (elem.centerLat(), elem.centerLon())
# handle unprecise and no-name locations
if location[0] is None:
location = (elem.lat(), elem.lon())
else :
continue
toilets = Toilets(location=location)
if 'wheelchair' in elem.tags().keys() and elem.tag('wheelchair') == 'yes':
toilets.wheelchair = True
if 'changing_table' in elem.tags().keys() and elem.tag('changing_table') == 'yes':
toilets.changing_table = True
if 'fee' in elem.tags().keys() and elem.tag('fee') == 'yes':
toilets.fee = True
if 'opening_hours' in elem.tags().keys() :
toilets.opening_hours = elem.tag('opening_hours')
toilets_list.append(toilets)
return toilets_list