feat(apis): add support for public transportation apis (#14)

Changelog
* Add support for commercial modes endpoint
* Add support for companies endpoint
* Add support for disruption endpoint
* Add support for lines endpoint
* Add support for networks endpoint
* Add support for physical modes endpoint
* Add support for routes endpoint
* Add support for stop area endpoint
* Add support for stop point endpoint
* Add support for vehicle journey endpoint

README.md

@@ -27,15 +27,15 @@ A base URL for Navitia IO is hardcoded and provided to NavitiaClient by default.
 
 ##  API support
 
-As of April 7th 2024, the library supports the following [APIs](https://doc.navitia.io/#api-catalog):
+The library supports the following [APIs](https://doc.navitia.io/#api-catalog):
 
 | API                                       | Supported ? |
 | ----------------------------------------- | ----------- |
 | Coverage                                  | ✅           |
 | Datasets                                  | ✅           |
 | Contributors                              | ✅           |
 | Inverted geocoding                        | ❌           |
-| Public transportation Objects exploration | ❌           |
+| Public transportation Objects exploration | ✅           |
 | Autocomplete on Public Transport objects  | ❌           |
 | Autocomplete on geographical objects      | ❌           |
 | Places nearby                             | ❌           |

navitia_client/client/apis/public_transportation_apis/__init__.py

@@ -0,0 +1,10 @@
+from .commercial_mode_apis import CommercialModeApiClient  # noqa: F401
+from .company_apis import CompanyApiClient  # noqa: F401
+from .disruption_apis import DisruptionApiClient  # noqa: F401
+from .line_apis import LineApiClient  # noqa: F401
+from .network_apis import NetworkApiClient  # noqa: F401
+from .physical_mode_apis import PhysicalModeApiClient  # noqa: F401
+from .route_apis import RouteApiClient  # noqa: F401
+from .stop_area_apis import StopAreaApiClient  # noqa: F401
+from .stop_point_apis import StopPointApiClient  # noqa: F401
+from .vehicle_journey_apis import VehicleJourneyApiClient  # noqa: F401

navitia_client/client/apis/public_transportation_apis/commercial_mode_apis.py

@@ -0,0 +1,49 @@
+from typing import Any, Sequence
+from navitia_client.client.apis.api_base_client import ApiBaseClient
+from navitia_client.client.apis.public_transportation_apis.entity_apis import EntityApi
+from navitia_client.entities.physical_mode import CommercialMode
+
+
+class CommercialModeApiClient(ApiBaseClient, EntityApi[CommercialMode]):
+    @staticmethod
+    def _get_entity_from_response(raw_entity_response: Any) -> Sequence[CommercialMode]:
+        commercial_modes = []
+        for commercial_mode in raw_entity_response:
+            commercial_modes.append(CommercialMode.from_json(commercial_mode))
+        return commercial_modes
+
+    def list_entity_collection_from_region(
+        self, region_id: str
+    ) -> Sequence[CommercialMode]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{region_id}/commercial_modes"
+        )
+        raw_results = results.json()["commercial_modes"]
+        return self._get_entity_from_response(raw_results)
+
+    def get_entity_by_id(
+        self, region_id: str, object_id: str
+    ) -> Sequence[CommercialMode]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{region_id}/commercial_modes/{object_id}"
+        )
+        raw_results = results.json()["commercial_modes"]
+        return self._get_entity_from_response(raw_results)
+
+    def list_entity_collection_from_coordinates(
+        self, lon: float, lat: float
+    ) -> Sequence[CommercialMode]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{lon};{lat}/commercial_modes/"
+        )
+        raw_results = results.json()["commercial_modes"]
+        return self._get_entity_from_response(raw_results)
+
+    def get_entity_by_id_and_coordinates(
+        self, lon: float, lat: float, object_id: str
+    ) -> Sequence[CommercialMode]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{lon};{lat}/commercial_modes/{object_id}"
+        )
+        raw_results = results.json()["commercial_modes"]
+        return self._get_entity_from_response(raw_results)

navitia_client/client/apis/public_transportation_apis/company_apis.py

@@ -0,0 +1,45 @@
+from typing import Any, Sequence
+from navitia_client.client.apis.api_base_client import ApiBaseClient
+from navitia_client.client.apis.public_transportation_apis.entity_apis import EntityApi
+from navitia_client.entities.company import Company
+
+
+class CompanyApiClient(ApiBaseClient, EntityApi[Company]):
+    @staticmethod
+    def _get_entity_from_response(raw_entity_response: Any) -> Sequence[Company]:
+        companies = []
+        for company in raw_entity_response:
+            companies.append(Company.from_json(company))
+        return companies
+
+    def list_entity_collection_from_region(self, region_id: str) -> Sequence[Company]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{region_id}/companies"
+        )
+        raw_results = results.json()["companies"]
+        return self._get_entity_from_response(raw_results)
+
+    def get_entity_by_id(self, region_id: str, object_id: str) -> Sequence[Company]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{region_id}/companies/{object_id}"
+        )
+        raw_results = results.json()["companies"]
+        return self._get_entity_from_response(raw_results)
+
+    def list_entity_collection_from_coordinates(
+        self, lon: float, lat: float
+    ) -> Sequence[Company]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{lon};{lat}/companies/"
+        )
+        raw_results = results.json()["companies"]
+        return self._get_entity_from_response(raw_results)
+
+    def get_entity_by_id_and_coordinates(
+        self, lon: float, lat: float, object_id: str
+    ) -> Sequence[Company]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{lon};{lat}/companies/{object_id}"
+        )
+        raw_results = results.json()["companies"]
+        return self._get_entity_from_response(raw_results)

navitia_client/client/apis/public_transportation_apis/disruption_apis.py

@@ -0,0 +1,52 @@
+from typing import Any, Sequence
+from navitia_client.client.apis.api_base_client import ApiBaseClient
+from navitia_client.client.apis.public_transportation_apis.entity_apis import EntityApi
+from navitia_client.entities.disruption import (
+    Disruption,
+)
+
+
+class DisruptionApiClient(ApiBaseClient, EntityApi[Disruption]):
+    @staticmethod
+    def _get_entity_from_response(raw_entity_response: Any) -> Sequence[Disruption]:
+        disruptions = []
+        for disruption_data in raw_entity_response:
+            disruption = Disruption.from_json(
+                payload=disruption_data,
+            )
+            disruptions.append(disruption)
+        return disruptions
+
+    def list_entity_collection_from_region(
+        self, region_id: str
+    ) -> Sequence[Disruption]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{region_id}/disruptions"
+        )
+        raw_results = results.json()["disruptions"]
+        return self._get_entity_from_response(raw_results)
+
+    def get_entity_by_id(self, region_id: str, object_id: str) -> Sequence[Disruption]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{region_id}/disruptions/{object_id}"
+        )
+        raw_results = results.json()["disruptions"]
+        return self._get_entity_from_response(raw_results)
+
+    def list_entity_collection_from_coordinates(
+        self, lon: float, lat: float
+    ) -> Sequence[Disruption]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{lon};{lat}/disruptions/"
+        )
+        raw_results = results.json()["disruptions"]
+        return self._get_entity_from_response(raw_results)
+
+    def get_entity_by_id_and_coordinates(
+        self, lon: float, lat: float, object_id: str
+    ) -> Sequence[Disruption]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{lon};{lat}/disruptions/{object_id}"
+        )
+        raw_results = results.json()["disruptions"]
+        return self._get_entity_from_response(raw_results)

navitia_client/client/apis/public_transportation_apis/entity_apis.py

@@ -0,0 +1,60 @@
+from abc import ABC, abstractmethod
+from typing import Any, Generic, Sequence, TypeVar
+
+from navitia_client.entities.company import Company
+from navitia_client.entities.disruption import (
+    Disruption,
+)
+from navitia_client.entities.line_and_route import Line, Route
+from navitia_client.entities.network import Network
+from navitia_client.entities.physical_mode import (
+    CommercialMode,
+    PhysicalMode,
+)
+from navitia_client.entities.stop_area import StopArea, StopPoint
+from navitia_client.entities.vehicle_journey import VehicleJourney
+
+TEntity = TypeVar(
+    "TEntity",
+    Network,
+    Line,
+    Route,
+    StopPoint,
+    StopArea,
+    CommercialMode,
+    PhysicalMode,
+    Disruption,
+    Company,
+    VehicleJourney,
+)
+
+
+class EntityApi(Generic[TEntity], ABC):
+    @staticmethod
+    @abstractmethod
+    def _get_entity_from_response(raw_entity_response: Any) -> Sequence[TEntity]:
+        """Collection of objects of a region"""
+        raise NotImplementedError
+
+    @abstractmethod
+    def list_entity_collection_from_region(self, region_id: str) -> Sequence[TEntity]:
+        """Information about a specific object"""
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_entity_by_id(self, region_id: str, object_id: str) -> Sequence[TEntity]:
+        raise NotImplementedError
+
+    @abstractmethod
+    def list_entity_collection_from_coordinates(
+        self, lon: float, lat: float
+    ) -> Sequence[TEntity]:
+        """Collection of objects of a region, navitia guesses the region from coordinates"""
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_entity_by_id_and_coordinates(
+        self, lon: float, lat: float, object_id: str
+    ) -> Sequence[TEntity]:
+        """Information about a specific object, navitia guesses the region from coordinates"""
+        raise NotImplementedError

navitia_client/client/apis/public_transportation_apis/line_apis.py

@@ -0,0 +1,50 @@
+from typing import Any, Sequence
+from navitia_client.client.apis.api_base_client import ApiBaseClient
+from navitia_client.client.apis.public_transportation_apis.entity_apis import EntityApi
+from navitia_client.entities.line_and_route import Line
+
+
+class LineApiClient(ApiBaseClient, EntityApi[Line]):
+    @staticmethod
+    def _get_entity_from_response(raw_entity_response: Any) -> Sequence[Line]:
+        lines = []
+        for line_data in raw_entity_response:
+            line = Line.from_json(
+                payload=line_data,
+            )
+
+            lines.append(line)
+
+        return lines
+
+    def list_entity_collection_from_region(self, region_id: str) -> Sequence[Line]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{region_id}/lines"
+        )
+        raw_results = results.json()["lines"]
+        return self._get_entity_from_response(raw_results)
+
+    def get_entity_by_id(self, region_id: str, object_id: str) -> Sequence[Line]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{region_id}/lines/{object_id}"
+        )
+        raw_results = results.json()["lines"]
+        return self._get_entity_from_response(raw_results)
+
+    def list_entity_collection_from_coordinates(
+        self, lon: float, lat: float
+    ) -> Sequence[Line]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{lon};{lat}/lines/"
+        )
+        raw_results = results.json()["lines"]
+        return self._get_entity_from_response(raw_results)
+
+    def get_entity_by_id_and_coordinates(
+        self, lon: float, lat: float, object_id: str
+    ) -> Sequence[Line]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{lon};{lat}/lines/{object_id}"
+        )
+        raw_results = results.json()["lines"]
+        return self._get_entity_from_response(raw_results)

navitia_client/client/apis/public_transportation_apis/network_apis.py

@@ -0,0 +1,45 @@
+from typing import Any, Sequence
+from navitia_client.client.apis.api_base_client import ApiBaseClient
+from navitia_client.client.apis.public_transportation_apis.entity_apis import EntityApi
+from navitia_client.entities.network import Network
+
+
+class NetworkApiClient(ApiBaseClient, EntityApi[Network]):
+    @staticmethod
+    def _get_entity_from_response(raw_entity_response: Any) -> Sequence[Network]:
+        networks = []
+        for network in raw_entity_response:
+            networks.append(Network.from_json(network))
+        return networks
+
+    def list_entity_collection_from_region(self, region_id: str) -> Sequence[Network]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{region_id}/networks"
+        )
+        raw_results = results.json()["networks"]
+        return self._get_entity_from_response(raw_results)
+
+    def get_entity_by_id(self, region_id: str, object_id: str) -> Sequence[Network]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{region_id}/networks/{object_id}"
+        )
+        raw_results = results.json()["networks"]
+        return self._get_entity_from_response(raw_results)
+
+    def list_entity_collection_from_coordinates(
+        self, lon: float, lat: float
+    ) -> Sequence[Network]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{lon};{lat}/networks/"
+        )
+        raw_results = results.json()["networks"]
+        return self._get_entity_from_response(raw_results)
+
+    def get_entity_by_id_and_coordinates(
+        self, lon: float, lat: float, object_id: str
+    ) -> Sequence[Network]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{lon};{lat}/networks/{object_id}"
+        )
+        raw_results = results.json()["networks"]
+        return self._get_entity_from_response(raw_results)

navitia_client/client/apis/public_transportation_apis/physical_mode_apis.py

@@ -0,0 +1,53 @@
+from typing import Any, Sequence
+from navitia_client.client.apis.api_base_client import ApiBaseClient
+from navitia_client.client.apis.public_transportation_apis.entity_apis import EntityApi
+from navitia_client.entities.physical_mode import PhysicalMode
+
+
+class PhysicalModeApiClient(ApiBaseClient, EntityApi[PhysicalMode]):
+    @staticmethod
+    def _get_entity_from_response(raw_entity_response: Any) -> Sequence[PhysicalMode]:
+        physical_modes = []
+        for physical_mode in raw_entity_response:
+            physical_modes.append(
+                PhysicalMode.from_json(
+                    physical_mode,
+                )
+            )
+        return physical_modes
+
+    def list_entity_collection_from_region(
+        self, region_id: str
+    ) -> Sequence[PhysicalMode]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{region_id}/physical_modes"
+        )
+        raw_results = results.json()["physical_modes"]
+        return self._get_entity_from_response(raw_results)
+
+    def get_entity_by_id(
+        self, region_id: str, object_id: str
+    ) -> Sequence[PhysicalMode]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{region_id}/physical_modes/{object_id}"
+        )
+        raw_results = results.json()["physical_modes"]
+        return self._get_entity_from_response(raw_results)
+
+    def list_entity_collection_from_coordinates(
+        self, lon: float, lat: float
+    ) -> Sequence[PhysicalMode]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{lon};{lat}/physical_modes/"
+        )
+        raw_results = results.json()["physical_modes"]
+        return self._get_entity_from_response(raw_results)
+
+    def get_entity_by_id_and_coordinates(
+        self, lon: float, lat: float, object_id: str
+    ) -> Sequence[PhysicalMode]:
+        results = self.get_navitia_api(
+            f"{self.base_navitia_url}/coverage/{lon};{lat}/physical_modes/{object_id}"
+        )
+        raw_results = results.json()["physical_modes"]
+        return self._get_entity_from_response(raw_results)