diff --git a/navitia_client/client/apis/equipment_report_apis.py b/navitia_client/client/apis/equipment_report_apis.py new file mode 100644 index 0000000..de83687 --- /dev/null +++ b/navitia_client/client/apis/equipment_report_apis.py @@ -0,0 +1,153 @@ +from typing import Optional, Sequence, Tuple +from navitia_client.client.apis.api_base_client import ApiBaseClient +from navitia_client.entities.equipment_reports import EquipmentReports +from navitia_client.entities.pagination import Pagination + + +class EquipmentReportsApiClient(ApiBaseClient): + """ + A client class to interact with the Navitia API for fetching equipment reports. + + See https://doc.navitia.io/#equipment-reports + + Methods + ------- + _get_equipment_reports( + url: str, filters: dict + ) -> Tuple[Sequence[EquipmentReports], Pagination]: + Retrieves equipment reports from the Navitia API based on provided URL and filters. + + list_equipment_reports( + region_id: str, + count: int = 10, + depth: int = 1, + filter: Optional[str] = None, + forbidden_uris: Optional[Sequence[str]] = None, + start_page: int = 0, + ) -> Tuple[Sequence[EquipmentReports], Pagination]: + Retrieves equipment reports for a specified region from the Navitia API. + + list_equipment_reports_with_resource_path( + region_id: str, + resource_path: str, + count: int = 10, + depth: int = 1, + filter: Optional[str] = None, + forbidden_uris: Optional[Sequence[str]] = None, + start_page: int = 0, + ) -> Tuple[Sequence[EquipmentReports], Pagination]: + Retrieves equipment reports for a specific resource path in a region from the Navitia API. + """ + + def _get_equipment_reports( + self, url: str, filters: dict + ) -> Tuple[Sequence[EquipmentReports], Pagination]: + """ + Retrieves equipment reports from the Navitia API based on provided URL and filters. + + Parameters: + url (str): The URL for the API request. + filters (dict): Filters to apply to the API request. + + Returns: + Tuple[Sequence[EquipmentReports], Pagination]: A tuple containing sequences of EquipmentReports objects and Pagination object. + """ + results = self.get_navitia_api(url + self._generate_filter_query(filters)) + equipment_reports = [ + EquipmentReports.from_payload(data) + for data in results.json()["equipment_reports"] + ] + pagination = Pagination.from_payload(results.json()["pagination"]) + return equipment_reports, pagination + + def list_equipment_reports( + self, + region_id: str, + count: int = 10, + depth: int = 1, + filter: Optional[str] = None, + forbidden_uris: Optional[Sequence[str]] = None, + start_page: int = 0, + ) -> Tuple[Sequence[EquipmentReports], Pagination]: + """ + Retrieves equipment reports for a specified region from the Navitia API. + + This service provides the state of equipments such as lifts or elevators that + are giving better accessibility to public transport facilities. + The endpoint will report accessible equipment per stop area and per line. + + Parameters: + region_id (str): The region ID (coverage identifier). + count (int): Elements per page. Defaults to 10. + depth (int): Json response depth. Defaults to 1. + filter (Optional[str]): A filter to refine your request (e.g., 'line.code=A'). + forbidden_uris (Optional[Sequence[str]]): If you want to avoid lines, modes, networks, etc. + start_page (int): The page number. Defaults to 0. + + Returns: + Tuple[Sequence[EquipmentReports], Pagination]: A tuple containing sequences of EquipmentReports objects and Pagination object. + + Note: + This feature requires a specific configuration from an equipment service provider. + Therefore, this service is not available by default. + """ + request_url = f"{self.base_navitia_url}/coverage/{region_id}/equipment_reports" + + filters = { + "count": count, + "depth": depth, + "start_page": start_page, + "forbidden_uris[]": forbidden_uris, + } + + if filter: + filters["filter"] = filter + + return self._get_equipment_reports(request_url, filters) + + def list_equipment_reports_with_resource_path( + self, + region_id: str, + resource_path: str, + count: int = 10, + depth: int = 1, + filter: Optional[str] = None, + forbidden_uris: Optional[Sequence[str]] = None, + start_page: int = 0, + ) -> Tuple[Sequence[EquipmentReports], Pagination]: + """ + Retrieves equipment reports for a specific resource path in a region from the Navitia API. + + This service provides the state of equipments such as lifts or elevators that + are giving better accessibility to public transport facilities. + The endpoint will report accessible equipment per stop area and per line. + + Parameters: + region_id (str): The region ID (coverage identifier). + resource_path (str): The resource path (e.g., 'lines/line:A'). + count (int): Elements per page. Defaults to 10. + depth (int): Json response depth. Defaults to 1. + filter (Optional[str]): A filter to refine your request (e.g., 'line.code=A'). + forbidden_uris (Optional[Sequence[str]]): If you want to avoid lines, modes, networks, etc. + start_page (int): The page number. Defaults to 0. + + Returns: + Tuple[Sequence[EquipmentReports], Pagination]: A tuple containing sequences of EquipmentReports objects and Pagination object. + + Note: + This feature requires a specific configuration from an equipment service provider. + Therefore, this service is not available by default. + """ + request_url = f"{self.base_navitia_url}/coverage/{region_id}/{resource_path}/equipment_reports" + + filters = { + "count": count, + "depth": depth, + "start_page": start_page, + "forbidden_uris[]": forbidden_uris, + } + + if filter: + filters["filter"] = filter + + return self._get_equipment_reports(request_url, filters) diff --git a/navitia_client/client/navitia_client.py b/navitia_client/client/navitia_client.py index 057b8d5..26eb0c6 100644 --- a/navitia_client/client/navitia_client.py +++ b/navitia_client/client/navitia_client.py @@ -5,6 +5,7 @@ from navitia_client.client.apis.coverage_apis import CoverageApiClient from navitia_client.client.apis.datasets_apis import DatasetsApiClient from navitia_client.client.apis.departure_apis import DepartureApiClient +from navitia_client.client.apis.equipment_report_apis import EquipmentReportsApiClient from navitia_client.client.apis.inverted_geocoding_apis import ( InvertedGeocodingApiClient, ) @@ -103,6 +104,8 @@ class NavitiaClient: Get an instance of LineReportsApiClient for accessing line reports-related endpoints. traffic_reports -> TrafficReportsApiClient: Get an instance of TrafficReportsApiClient for accessing traffic reports-related endpoints. + equipment_reports -> EquipmentReportsApiClient: + Get an instance of EquipmentReportsApiClient for accessing equipment reports-related endpoints. journeys -> JourneyApiClient: Get an instance of JourneyApiClient for accessing journey-related endpoints. isochrones -> IsochronesApiClient: @@ -280,6 +283,13 @@ def traffic_reports(self) -> TrafficReportsApiClient: auth_token=self.auth_token, base_navitia_url=self.base_navitia_url ) + @property + def equipment_reports(self) -> EquipmentReportsApiClient: + """Get an instance of EquipmentReportsApiClient for accessing equipment-reports-related endpoints.""" + return EquipmentReportsApiClient( + auth_token=self.auth_token, base_navitia_url=self.base_navitia_url + ) + @property def journeys(self) -> JourneyApiClient: """Get an instance of JourneyApiClient for accessing journey-related endpoints.""" diff --git a/navitia_client/entities/equipment.py b/navitia_client/entities/equipment.py index e7919fd..6d58d9f 100644 --- a/navitia_client/entities/equipment.py +++ b/navitia_client/entities/equipment.py @@ -1,6 +1,6 @@ from dataclasses import dataclass from enum import Enum -from typing import Optional +from typing import Optional, List, Dict, Any from .base_entity import BaseEntity from .stop_area import StopArea @@ -22,21 +22,80 @@ class Equipment(Enum): BIKE_DEPOT = "has_bike_depot" +@dataclass +class Label: + label: str + + @classmethod + def from_payload(cls, data: Dict[str, Any]) -> "Label": + return cls(label=data.get("label", "")) + + +@dataclass +class Period: + begin: str + end: str + + @classmethod + def from_payload(cls, data: Dict[str, Any]) -> "Period": + return cls(begin=data.get("begin", ""), end=data.get("end", "")) + + @dataclass class EquipmentAvailability: status: str - cause: Optional[str] - effect: Optional[str] - periods: Optional[dict[str, str]] + cause: Optional[Label] = None + effect: Optional[Label] = None + periods: Optional[List[Period]] = None + updated_at: Optional[str] = None + + @classmethod + def from_payload(cls, data: Dict[str, Any]) -> "EquipmentAvailability": + cause = Label.from_payload(data["cause"]) if "cause" in data else None + effect = Label.from_payload(data["effect"]) if "effect" in data else None + periods = [Period.from_payload(p) for p in data.get("periods", [])] + + return cls( + status=data.get("status", "unknown"), + cause=cause, + effect=effect, + periods=periods if periods else None, + updated_at=data.get("updated_at"), + ) @dataclass class EquipmentDetails(BaseEntity): - current_availability: EquipmentAvailability embedded_type: str + current_availability: Optional[EquipmentAvailability] = None + + @classmethod + def from_payload(cls, data: Dict[str, Any]) -> "EquipmentDetails": + return cls( + id=data.get("id", ""), + name=data.get("name", ""), + embedded_type=data.get("embedded_type", ""), + current_availability=EquipmentAvailability.from_payload( + data["current_availability"] + ) + if "current_availability" in data + else None, + ) @dataclass class StopAreaEquipments: - equiment_details: EquipmentDetails - stop_area: StopArea + equipment_details: List[EquipmentDetails] + stop_area: Optional[StopArea] = None + + @classmethod + def from_payload(cls, data: Dict[str, Any]) -> "StopAreaEquipments": + equipment_details = [ + EquipmentDetails.from_payload(item) + for item in data.get("equipment_details", []) + ] + stop_area = ( + StopArea.from_payload(data["stop_area"]) if "stop_area" in data else None + ) + + return cls(equipment_details=equipment_details, stop_area=stop_area) diff --git a/navitia_client/entities/equipment_reports.py b/navitia_client/entities/equipment_reports.py index 2bc540e..702b773 100644 --- a/navitia_client/entities/equipment_reports.py +++ b/navitia_client/entities/equipment_reports.py @@ -1,4 +1,5 @@ -from dataclasses import dataclass +from dataclasses import dataclass, field +from typing import List, Dict, Any, Optional from .line_and_route import Line from .equipment import StopAreaEquipments @@ -6,5 +7,24 @@ @dataclass class EquipmentReports: - line: Line - stop_area_equipments: StopAreaEquipments + line: Optional[Line] = None + stop_area_equipments: List[StopAreaEquipments] = field(default_factory=list) + + @classmethod + def from_payload(cls, data: Dict[str, Any]) -> "EquipmentReports": + """ + Create an EquipmentReports instance from API payload data. + + Parameters: + data: Dictionary containing equipment report data from the API + + Returns: + EquipmentReports: An instance of EquipmentReports + """ + line = Line.from_payload(data["line"]) if "line" in data else None + stop_area_equipments = [ + StopAreaEquipments.from_payload(item) + for item in data.get("stop_area_equipments", []) + ] + + return cls(line=line, stop_area_equipments=stop_area_equipments) diff --git a/tests/client/apis/test_equipment_report_apis.py b/tests/client/apis/test_equipment_report_apis.py new file mode 100644 index 0000000..b8ceb0f --- /dev/null +++ b/tests/client/apis/test_equipment_report_apis.py @@ -0,0 +1,71 @@ +import json +from unittest.mock import MagicMock, patch + +import pytest + +from navitia_client.client.apis.equipment_report_apis import EquipmentReportsApiClient + + +@pytest.fixture +def equipment_report_apis(): + return EquipmentReportsApiClient( + auth_token="foobar", base_navitia_url="https://api.navitia.io/v1/" + ) + + +@patch.object(EquipmentReportsApiClient, "get_navitia_api") +def test_list_equipment_reports( + mock_get_navitia_api: MagicMock, equipment_report_apis: EquipmentReportsApiClient +) -> None: + """ + Test that list_equipment_reports returns equipment reports and pagination. + """ + # Given + mock_response = MagicMock() + with open("tests/test_data/equipment_reports.json", encoding="utf-8") as file: + mock_response.json.return_value = json.load(file) + + mock_get_navitia_api.return_value = mock_response + + # When + equipment_reports, pagination = equipment_report_apis.list_equipment_reports( + region_id="fr-idf" + ) + + # Then + assert len(equipment_reports) == 2 + assert equipment_reports[0].line is not None + assert len(equipment_reports[0].stop_area_equipments) > 0 + assert pagination.total_result == 2 + assert pagination.items_on_page == 2 + + +@patch.object(EquipmentReportsApiClient, "get_navitia_api") +def test_list_equipment_reports_with_resource_path( + mock_get_navitia_api: MagicMock, equipment_report_apis: EquipmentReportsApiClient +) -> None: + """ + Test that list_equipment_reports_with_resource_path returns equipment reports for a specific resource path. + """ + # Given + mock_response = MagicMock() + with open("tests/test_data/equipment_reports.json", encoding="utf-8") as file: + mock_response.json.return_value = json.load(file) + + mock_get_navitia_api.return_value = mock_response + + # When + equipment_reports, pagination = ( + equipment_report_apis.list_equipment_reports_with_resource_path( + region_id="fr-idf", resource_path="lines/line:IDFM:C01742" + ) + ) + + # Then + called_url = mock_get_navitia_api.call_args[0][0] + assert "lines/line:IDFM:C01742/equipment_reports" in called_url + assert len(equipment_reports) == 2 + assert equipment_reports[0].line is not None + assert len(equipment_reports[0].stop_area_equipments) > 0 + assert pagination.total_result == 2 + assert pagination.items_on_page == 2 diff --git a/tests/test_data/equipment_reports.json b/tests/test_data/equipment_reports.json new file mode 100644 index 0000000..2998938 --- /dev/null +++ b/tests/test_data/equipment_reports.json @@ -0,0 +1,117 @@ +{ + "equipment_reports": [ + { + "line": { + "id": "line:IDFM:C01742", + "name": "Metro 1", + "code": "1", + "color": "FFCD00", + "text_color": "000000", + "opening_time": "053000", + "closing_time": "013600", + "physical_modes": [], + "commercial_mode": { + "id": "commercial_mode:Metro", + "name": "Metro" + }, + "network": { + "id": "network:IDFM", + "name": "RATP" + } + }, + "stop_area_equipments": [ + { + "stop_area": { + "id": "stop_area:IDFM:71261", + "name": "La Defense", + "label": "La Defense (Paris)", + "coord": { + "lat": "48.892038", + "lon": "2.237883" + } + }, + "equipment_details": [ + { + "id": "2702", + "name": "Escalator to platform 1", + "embedded_type": "escalator", + "current_availability": { + "status": "unavailable", + "cause": { + "label": "engineering work in progress" + }, + "effect": { + "label": "platform 1 available via stairs only" + }, + "periods": [ + { + "begin": "20190216T000000", + "end": "20190601T220000" + } + ], + "updated_at": "2019-05-17T15:54:53+02:00" + } + }, + { + "id": "2703", + "name": "Elevator to platform 2", + "embedded_type": "elevator", + "current_availability": { + "status": "unknown" + } + } + ] + } + ] + }, + { + "line": { + "id": "line:IDFM:C01743", + "name": "Metro 2", + "code": "2", + "color": "0064B0", + "text_color": "FFFFFF", + "opening_time": "053000", + "closing_time": "013600", + "physical_modes": [], + "commercial_mode": { + "id": "commercial_mode:Metro", + "name": "Metro" + }, + "network": { + "id": "network:IDFM", + "name": "RATP" + } + }, + "stop_area_equipments": [ + { + "stop_area": { + "id": "stop_area:IDFM:71262", + "name": "Nation", + "label": "Nation (Paris)", + "coord": { + "lat": "48.848571", + "lon": "2.396611" + } + }, + "equipment_details": [ + { + "id": "3001", + "name": "Main elevator", + "embedded_type": "elevator", + "current_availability": { + "status": "unknown" + } + } + ] + } + ] + } + ], + "pagination": { + "start_page": 0, + "items_on_page": 2, + "items_per_page": 10, + "total_result": 2 + } +} diff --git a/tests/test_navitia_client.py b/tests/test_navitia_client.py index c0942a5..4520aa7 100644 --- a/tests/test_navitia_client.py +++ b/tests/test_navitia_client.py @@ -4,6 +4,7 @@ from navitia_client.client.apis.coverage_apis import CoverageApiClient from navitia_client.client.apis.datasets_apis import DatasetsApiClient from navitia_client.client.apis.departure_apis import DepartureApiClient +from navitia_client.client.apis.equipment_report_apis import EquipmentReportsApiClient from navitia_client.client.apis.inverted_geocoding_apis import ( InvertedGeocodingApiClient, ) @@ -142,6 +143,10 @@ def test_traffic_reports_client(navitia_client): assert isinstance(navitia_client.traffic_reports, TrafficReportsApiClient) +def test_equipment_reports_client(navitia_client): + assert isinstance(navitia_client.equipment_reports, EquipmentReportsApiClient) + + def test_journeys_client(navitia_client): assert isinstance(navitia_client.journeys, JourneyApiClient)