feat(apis): add support for isochrone apis (#48)

Author: jonperron

README.md

@@ -40,7 +40,7 @@ The library supports the following [APIs](https://doc.navitia.io/#api-catalog):
 | Autocomplete on geographical objects      | ✅           |                                              |
 | Places nearby                             | ✅           |                                              |
 | Journeys                                  | ✅           |                                              |
-| Isochrones                                | ❌           |                                              |
+| Isochrones                                | ✅           | Beta endpoint according to API response      |
 | Route Schedules                           | ✅           |                                              |
 | Stop Schedules                            | ✅           |                                              |
 | Terminus Schedules                        | ✅           |                                              |

navitia_client/client/apis/isochrone_apis.py

@@ -0,0 +1,91 @@
+from datetime import datetime
+from typing import Any, Optional, Sequence
+from navitia_client.client.apis.api_base_client import ApiBaseClient
+from navitia_client.entities.isochrones import Isochrone
+
+
+class IsochronesApiClient(ApiBaseClient):
+    def _get_traffic_reports(self, url: str, filters: dict) -> Sequence[Isochrone]:
+        results = self.get_navitia_api(url + self._generate_filter_query(filters))
+        isochrones = [
+            Isochrone.from_payload(data) for data in results.json()["isochrones"]
+        ]
+        return isochrones
+
+    def list_isochrones_with_region_id(
+        self,
+        from_: str,
+        region_id: str,
+        start_datetime: datetime = datetime.now(),
+        boundary_duration: Sequence[int] = [],
+        to: Optional[str] = None,
+        first_section_mode: Optional[Sequence[str]] = None,
+        last_section_mode: Optional[Sequence[str]] = None,
+        min_duration: Optional[int] = None,
+        max_duration: Optional[int] = None,
+    ) -> Sequence[Isochrone]:
+        request_url = f"{self.base_navitia_url}/coverage/{region_id}/isochrones"
+
+        filters: dict[str, Any] = {
+            "datetime": start_datetime.isoformat(),
+            "boundary_duration[]": boundary_duration,
+            "from": from_,
+        }
+
+        if to:
+            filters["to"] = to
+
+        if min_duration:
+            filters["min_duration"] = min_duration
+
+        if first_section_mode:
+            filters["first_section_mode[]"] = first_section_mode
+
+        if last_section_mode:
+            filters["last_section_mode[]"] = last_section_mode
+
+        if max_duration:
+            filters["max_duration"] = max_duration
+            if len(boundary_duration) == 0:
+                # From API: you should provide a 'boundary_duration[]' or a 'max_duration'
+                filters.pop("min_duration")
+
+        return self._get_traffic_reports(request_url, filters)
+
+    def list_isochrones(
+        self,
+        from_: str,
+        start_datetime: datetime = datetime.now(),
+        boundary_duration: Sequence[int] = [],
+        to: Optional[str] = None,
+        first_section_mode: Optional[Sequence[str]] = None,
+        last_section_mode: Optional[Sequence[str]] = None,
+        min_duration: Optional[int] = None,
+        max_duration: Optional[int] = None,
+    ) -> Sequence[Isochrone]:
+        request_url = f"{self.base_navitia_url}/isochrones"
+
+        filters: dict[str, Any] = {
+            "datetime": start_datetime.isoformat(),
+            "boundary_duration[]": boundary_duration,
+            "from": from_,
+        }
+        if to:
+            filters["to"] = to
+
+        if first_section_mode:
+            filters["first_section_mode[]"] = first_section_mode
+
+        if last_section_mode:
+            filters["last_section_mode[]"] = last_section_mode
+
+        if min_duration:
+            filters["min_duration"] = min_duration
+
+        if max_duration:
+            filters["max_duration"] = max_duration
+            if len(boundary_duration) == 0:
+                # From API: you should provide a 'boundary_duration[]' or a 'max_duration'
+                filters.pop("min_duration")
+
+        return self._get_traffic_reports(request_url, filters)

navitia_client/client/navitia_client.py

@@ -8,6 +8,7 @@
 from navitia_client.client.apis.inverted_geocoding_apis import (
     InvertedGeocodingApiClient,
 )
+from navitia_client.client.apis.isochrone_apis import IsochronesApiClient
 from navitia_client.client.apis.journeys_apis import JourneyApiClient
 from navitia_client.client.apis.line_report_apis import LineReportsApiClient
 from navitia_client.client.apis.place_apis import PlacesApiClient
@@ -193,3 +194,9 @@ def journeys(self) -> JourneyApiClient:
         return JourneyApiClient(
             auth_token=self.auth_token, base_navitia_url=self.base_navitia_url
         )
+
+    @property
+    def isochrones(self) -> IsochronesApiClient:
+        return IsochronesApiClient(
+            auth_token=self.auth_token, base_navitia_url=self.base_navitia_url
+        )

navitia_client/entities/isochrones.py

@@ -0,0 +1,30 @@
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any
+
+from navitia_client.entities.place import Place
+
+
+@dataclass
+class Isochrone:
+    from_: Place
+    geojson: Any
+    max_date_time: datetime
+    max_duration: int
+    min_date_time: datetime
+    min_duration: int
+    requested_date_time: datetime
+
+    @classmethod
+    def from_payload(cls, payload: dict[str, Any]) -> "Isochrone":
+        return cls(
+            from_=Place.from_payload(payload["from"]),
+            geojson=payload["geojson"],
+            max_date_time=datetime.strptime(payload["max_date_time"], "%Y%m%dT%H%M%S"),
+            max_duration=payload["max_duration"],
+            min_date_time=datetime.strptime(payload["min_date_time"], "%Y%m%dT%H%M%S"),
+            min_duration=payload["min_duration"],
+            requested_date_time=datetime.strptime(
+                payload["requested_date_time"], "%Y%m%dT%H%M%S"
+            ),
+        )

tests/client/apis/test_isochrones_apis.py

@@ -0,0 +1,54 @@
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from navitia_client.client.apis.isochrone_apis import IsochronesApiClient
+from navitia_client.entities.isochrones import Isochrone
+
+
+@pytest.fixture
+def isochrones_apis():
+    return IsochronesApiClient(
+        auth_token="foobar", base_navitia_url="https://api.navitia.io/v1/"
+    )
+
+
+@patch.object(IsochronesApiClient, "get_navitia_api")
+def test_list_covered_areas_with_region_id(
+    mock_get_navitia_api: MagicMock, isochrones_apis: IsochronesApiClient
+) -> None:
+    # Given
+    mock_response = MagicMock()
+    with open("tests/test_data/isochrones.json", encoding="utf-8") as file:
+        mock_response.json.return_value = json.load(file)
+
+    mock_get_navitia_api.return_value = mock_response
+
+    # When
+    isocrhones = isochrones_apis.list_isochrones_with_region_id(
+        region_id="bar", from_="foo"
+    )
+
+    # Then
+    assert len(isocrhones) == 1
+    assert isinstance(isocrhones[0], Isochrone)
+
+
+@patch.object(IsochronesApiClient, "get_navitia_api")
+def test_list_covered_areas(
+    mock_get_navitia_api: MagicMock, isochrones_apis: IsochronesApiClient
+) -> None:
+    # Given
+    mock_response = MagicMock()
+    with open("tests/test_data/isochrones.json", encoding="utf-8") as file:
+        mock_response.json.return_value = json.load(file)
+
+    mock_get_navitia_api.return_value = mock_response
+
+    # When
+    isocrhones = isochrones_apis.list_isochrones(from_="foo")
+
+    # Then
+    assert len(isocrhones) == 1
+    assert isinstance(isocrhones[0], Isochrone)