Add public transports ft. Hove's API (#324)

Add public transport support through Hove's API.

This is built by building a translation layer that will convert Hove's API
result into a format mostly compatible with Mapbox (a few structures are
extended with new fields), which can be interpreted by Erdapfel seamlessly and
in the same fashion as with former Combigo API.

I also took my change to move every directions-related stuff in
`idunn.datasources.directions`, which means a bit of refactoring in the mapbox
client.

Author: remi-dupre

idunn/api/directions.py

@@ -4,7 +4,7 @@
 from idunn import settings
 from idunn.places import Latlon
 from idunn.places.exceptions import IdunnPlaceError
-from ..directions.client import directions_client
+from ..datasources.directions import directions_client
 from ..utils.place import place_from_id
 
 
@@ -17,7 +17,7 @@ def directions_request(request: Request, response: Response):
     return request
 
 
-def get_directions_with_coordinates(
+async def get_directions_with_coordinates(
     # URL values
     f_lon: confloat(ge=-180, le=180) = Path(..., title="Origin point longitude"),
     f_lat: confloat(ge=-90, le=90) = Path(..., title="Origin point latitude"),
@@ -34,12 +34,12 @@ def get_directions_with_coordinates(
     to_place = Latlon(t_lat, t_lon)
     if not type:
         raise HTTPException(status_code=400, detail='"type" query param is required')
-    return directions_client.get_directions(
-        from_place, to_place, type, language, params=request.query_params
+    return await directions_client.get_directions(
+        from_place, to_place, type, language, extra=request.query_params
     )
 
 
-def get_directions(
+async def get_directions(
     # Query parameters
     origin: str = Query(..., description="Origin place id."),
     destination: str = Query(..., description="Destination place id."),
@@ -55,6 +55,6 @@ def get_directions(
     except IdunnPlaceError as exc:
         raise HTTPException(status_code=404, detail=exc.message) from exc
 
-    return directions_client.get_directions(
-        from_place, to_place, type, language, params=request.query_params
+    return await directions_client.get_directions(
+        from_place, to_place, type, language, extra=request.query_params
     )

idunn/api/geocoder.py

@@ -7,6 +7,7 @@
 from ..geocoder.bragi_client import bragi_client
 from ..geocoder.models.geocodejson import Intention
 from ..geocoder.nlu_client import nlu_client, NluClientException
+
 from ..geocoder.models import QueryParams, ExtraParams, IdunnAutocomplete
 
 from idunn import settings

idunn/api/urls.py

@@ -6,7 +6,7 @@
 from .places_list import get_places_bbox, PlacesBboxResponse
 from .categories import AllCategoriesResponse, get_all_categories
 from .closest import closest_address
-from ..directions.models import DirectionsResponse
+from ..datasources.directions.mapbox.models import DirectionsResponse
 from .geocoder import get_autocomplete_response
 from ..geocoder.models import IdunnAutocomplete
 from .directions import get_directions_with_coordinates, get_directions

idunn/datasources/directions/__init__.py

@@ -0,0 +1,86 @@
+import logging
+from abc import ABC, abstractmethod, abstractproperty
+from fastapi import HTTPException
+from pydantic import BaseModel
+from typing import Callable, Optional
+
+
+from idunn import settings
+from idunn.datasources.directions.abs_client import AbsDirectionsClient
+from idunn.geocoder.models import QueryParams
+from idunn.places.base import BasePlace
+from .hove.client import HoveClient
+from .mapbox.client import MapboxClient
+from .mapbox.models import DirectionsResponse, IdunnTransportMode
+
+logger = logging.getLogger(__name__)
+
+
+class MapboxAPIExtraParams(BaseModel):
+    steps: str = "true"
+    alternatives: str = "true"
+    overview: str = "full"
+    geometries: str = "geojson"
+    exclude: Optional[str]
+
+
+class DirectionsClient(AbsDirectionsClient):
+    def __init__(self):
+        self.mapbox = MapboxClient()
+        self.hove = HoveClient()
+
+    @staticmethod
+    def client_name() -> str:
+        return "generic"
+
+    def get_method_for_mode(self, mode: IdunnTransportMode) -> AbsDirectionsClient:
+        methods = {"mapbox": self.mapbox, "hove": self.hove}
+
+        match mode:
+            case IdunnTransportMode.CAR:
+                return methods[settings["DIRECTIONS_PROVIDER_DRIVE"]]
+            case IdunnTransportMode.BIKE:
+                return methods[settings["DIRECTIONS_PROVIDER_CYCLE"]]
+            case IdunnTransportMode.WALKING:
+                return methods[settings["DIRECTIONS_PROVIDER_WALK"]]
+            case IdunnTransportMode.PUBLICTRANSPORT:
+                return methods[settings["DIRECTIONS_PROVIDER_PUBLICTRANSPORT"]]
+
+    @staticmethod
+    def place_to_url_coords(place):
+        coord = place.get_coord()
+        lat, lon = coord["lat"], coord["lon"]
+        return (f"{lon:.5f}", f"{lat:.5f}")
+
+    async def get_directions(
+        self,
+        from_place: BasePlace,
+        to_place: BasePlace,
+        mode: IdunnTransportMode,
+        lang: str,
+        extra: Optional[QueryParams] = None,
+    ) -> DirectionsResponse:
+        idunn_mode = IdunnTransportMode.parse(mode)
+
+        if idunn_mode is None:
+            raise HTTPException(status_code=400, detail=f"unknown mode {mode}")
+
+        method = self.get_method_for_mode(idunn_mode)
+
+        logger.info(
+            "Calling directions API '%s'",
+            method.client_name(),
+            extra={
+                "method": method.client_name(),
+                "mode": idunn_mode,
+                "lang": lang,
+                "from_place": from_place.get_id(),
+                "to_place": to_place.get_id(),
+            },
+        )
+
+        # pylint: disable = not-callable
+        return await method.get_directions(from_place, to_place, idunn_mode, lang, extra)
+
+
+directions_client = DirectionsClient()

idunn/datasources/directions/abs_client.py

@@ -0,0 +1,24 @@
+from abc import ABC, abstractmethod
+from typing import Optional
+
+from idunn.geocoder.models.params import QueryParams
+from idunn.places.base import BasePlace
+from .mapbox.models import DirectionsResponse, IdunnTransportMode
+
+
+class AbsDirectionsClient(ABC):
+    @staticmethod
+    @abstractmethod
+    def client_name() -> str:
+        ...
+
+    @abstractmethod
+    async def get_directions(
+        self,
+        from_place: BasePlace,
+        to_place: BasePlace,
+        mode: IdunnTransportMode,
+        lang: str,
+        extra: Optional[QueryParams] = None,
+    ) -> DirectionsResponse:
+        ...

idunn/directions/__init__.py …/datasources/directions/hove/__init__.pyidunn/directions/__init__.py renamed to idunn/datasources/directions/hove/__init__.py idunn/directions/__init__.py renamed to idunn/datasources/directions/hove/__init__.py

@@ -0,0 +1,77 @@
+import httpx
+from fastapi import HTTPException
+from typing import Optional
+
+from idunn import settings
+from idunn.geocoder.models.params import QueryParams
+from idunn.places.base import BasePlace
+from .models import HoveResponse
+from ..abs_client import AbsDirectionsClient
+from ..mapbox.models import IdunnTransportMode
+
+
+DIRECT_PATH_MAX_DURATION = 86400  # 24h
+MIN_NB_JOURNEYS = 2
+MAX_NB_JOURNEYS = 5
+FREE_RADIUS = 50  # meters
+
+
+class HoveClient(AbsDirectionsClient):
+    def __init__(self):
+        self.api_url = settings["HOVE_API_BASE_URL"]
+        self.session = httpx.AsyncClient(verify=settings["VERIFY_HTTPS"])
+        self.session.headers["User-Agent"] = settings["USER_AGENT"]
+
+    @staticmethod
+    def client_name() -> str:
+        return "hove"
+
+    @property
+    def API_ENABLED(self):  # pylint: disable = invalid-name
+        return bool(settings["HOVE_API_TOKEN"])
+
+    async def get_directions(
+        self,
+        from_place: BasePlace,
+        to_place: BasePlace,
+        mode: IdunnTransportMode,
+        _lang: str,
+        _extra: Optional[QueryParams] = None,
+    ) -> HoveResponse:
+        if not self.API_ENABLED:
+            raise HTTPException(
+                status_code=501,
+                detail=f"Directions API is currently unavailable for mode {mode}",
+            )
+
+        from_place = from_place.get_coord()
+        to_place = to_place.get_coord()
+
+        params = {
+            "from": f"{from_place['lon']};{from_place['lat']}",
+            "to": f"{to_place['lon']};{to_place['lat']}",
+            "free_radius_from": FREE_RADIUS,
+            "free_radius_to": FREE_RADIUS,
+            "max_walking_direct_path_duration": DIRECT_PATH_MAX_DURATION,
+            "max_bike_direct_path_duration": DIRECT_PATH_MAX_DURATION,
+            "max_car_no_park_direct_path_duration": DIRECT_PATH_MAX_DURATION,
+            "min_nb_journeys": MIN_NB_JOURNEYS,
+            "max_nb_journeys": MAX_NB_JOURNEYS,
+            **(
+                {"direct_path": "none"}
+                if mode == IdunnTransportMode.PUBLICTRANSPORT
+                else {
+                    "direct_path_mode[]": mode.to_hove(),
+                    "direct_path": "only",
+                }
+            ),
+        }
+
+        response = await self.session.get(
+            self.api_url,
+            params=params,
+            headers={"Authorization": settings["HOVE_API_TOKEN"]},
+        )
+
+        response.raise_for_status()
+        return HoveResponse(**response.json()).as_api_response()