Skip to content

feat(equiment): add support for equipment API #88

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jonperron merged 1 commit into from
Nov 10, 2025
Merged

feat(equiment): add support for equipment API #88

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
153 changes: 153 additions & 0 deletions navitia_client/client/apis/equipment_report_apis.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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)
10 changes: 10 additions & 0 deletions navitia_client/client/navitia_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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."""
Expand Down
73 changes: 66 additions & 7 deletions navitia_client/entities/equipment.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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)
26 changes: 23 additions & 3 deletions navitia_client/entities/equipment_reports.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,30 @@
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


@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)
71 changes: 71 additions & 0 deletions tests/client/apis/test_equipment_report_apis.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
Loading
Loading