Madrid
Module to store the city data of Madrid.
MadridCityData (CityData)
pydantic-model
Madrid city data.
Source code in now8_api/service/city_data/madrid.py
class MadridCityData(CityData):
"""Madrid city data."""
@overrides
async def get_estimations(
self,
stop: Stop,
) -> List[VehicleEstimation]:
try:
response = await get_json(
f"https://www.crtm.es/" # type: ignore
f"widgets/api/GetStopsTimes.php"
f"?codStop="
f"{stop.id.removeprefix('par_').removeprefix('est_')}&"
f"type=1&orderBy=2&stopTimesByIti=3"
)
except JSONDecodeError as error:
raise UpstreamError(
"Upstream error. Check the stop id or try later."
) from error
result: List[VehicleEstimation] = []
_estimations = response["stopTimes"]["times"].get("Time", [])
estimations: List[dict] = (
_estimations if isinstance(_estimations, list) else [_estimations]
)
for e in estimations:
vehicle = Vehicle(
id=e["codIssue"] if e["codIssue"] != "" else None,
route_id=e["line"]["codLine"],
route_way=Way(e["direction"])
if e["direction"] in [0, 1]
else None,
destination_stop=Stop(
id=e["destinationStop"]["codStop"],
code=e["destinationStop"]["shortCodStop"],
name=e["destinationStop"]["name"],
)
if "codStop" in e["destinationStop"]
and e["destinationStop"]["codStop"]
else None,
)
estimation = Estimation(
estimation=e["time"],
time=response["stopTimes"]["actualDate"],
)
result.append(
VehicleEstimation(vehicle=vehicle, estimation=estimation)
)
return result
@overrides
async def get_stops_city(
self,
transport_types: List[TransportType] = None,
) -> List[Stop]:
raise NotImplementedError
@overrides
async def get_stops_route(
self,
route: Route,
) -> Tuple[List[Stop], List[Stop]]:
raise NotImplementedError
@overrides
async def get_routes_stop(
self,
stop: Stop,
) -> List[Route]:
raise NotImplementedError
get_estimations(self, stop)
async
Return ETA for the next vehicles to the stop.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
stop |
Stop |
Stop to get the next vehicle arrival estimations for. |
required |
Returns:
| Type | Description |
|---|---|
List[now8_api.domain.VehicleEstimation] |
ETA for the next vehicles to the stop. |
Exceptions:
| Type | Description |
|---|---|
NotImplementedError |
If the method is not implemented for the stop transport type in this city. |
UpstreamError |
If the upstream city API fails. |
Source code in now8_api/service/city_data/madrid.py
@overrides
async def get_estimations(
self,
stop: Stop,
) -> List[VehicleEstimation]:
try:
response = await get_json(
f"https://www.crtm.es/" # type: ignore
f"widgets/api/GetStopsTimes.php"
f"?codStop="
f"{stop.id.removeprefix('par_').removeprefix('est_')}&"
f"type=1&orderBy=2&stopTimesByIti=3"
)
except JSONDecodeError as error:
raise UpstreamError(
"Upstream error. Check the stop id or try later."
) from error
result: List[VehicleEstimation] = []
_estimations = response["stopTimes"]["times"].get("Time", [])
estimations: List[dict] = (
_estimations if isinstance(_estimations, list) else [_estimations]
)
for e in estimations:
vehicle = Vehicle(
id=e["codIssue"] if e["codIssue"] != "" else None,
route_id=e["line"]["codLine"],
route_way=Way(e["direction"])
if e["direction"] in [0, 1]
else None,
destination_stop=Stop(
id=e["destinationStop"]["codStop"],
code=e["destinationStop"]["shortCodStop"],
name=e["destinationStop"]["name"],
)
if "codStop" in e["destinationStop"]
and e["destinationStop"]["codStop"]
else None,
)
estimation = Estimation(
estimation=e["time"],
time=response["stopTimes"]["actualDate"],
)
result.append(
VehicleEstimation(vehicle=vehicle, estimation=estimation)
)
return result
get_routes_stop(self, stop)
async
Return all the routes that pass through the stop.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
stop |
Stop |
Transport stop to get the routes for. |
required |
Returns:
| Type | Description |
|---|---|
List[now8_api.domain.Route] |
Routes that pass through the stop. |
Exceptions:
| Type | Description |
|---|---|
NotImplementedError |
If the method is not implemented for the stop transport type in this city. |
Source code in now8_api/service/city_data/madrid.py
@overrides
async def get_routes_stop(
self,
stop: Stop,
) -> List[Route]:
raise NotImplementedError
get_stops_city(self, transport_types=None)
async
Return all the stops of the selected transport types.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
transport_types |
List[now8_api.domain.TransportType] |
Transport types to get the stops for. If none are passed, all of the available ones for the city will be included. |
None |
Returns:
| Type | Description |
|---|---|
List[now8_api.domain.Stop] |
All the stops of the selected transport types. |
Exceptions:
| Type | Description |
|---|---|
NotImplementedError |
If the method is not implemented for the stop transport type in this city. |
Source code in now8_api/service/city_data/madrid.py
@overrides
async def get_stops_city(
self,
transport_types: List[TransportType] = None,
) -> List[Stop]:
raise NotImplementedError
get_stops_route(self, route)
async
Return all the stops of the selected route.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
route |
Route |
Transport route to get the stops for. |
required |
Returns:
| Type | Description |
|---|---|
Tuple[List[now8_api.domain.Stop], List[now8_api.domain.Stop]] |
Stops of the selected route for both ways in order. |
Exceptions:
| Type | Description |
|---|---|
NotImplementedError |
If the method is not implemented for the route transport type in this city. |
Source code in now8_api/service/city_data/madrid.py
@overrides
async def get_stops_route(
self,
route: Route,
) -> Tuple[List[Stop], List[Stop]]:
raise NotImplementedError