profitroll/PythonRMV
profitroll/PythonRMV
1
1
Fork 0
Code Issues Pull Requests Actions Packages Releases 3 Wiki Activity

board_arrival and board_departure implemented

Browse Source
This commit is contained in:
Profitroll 2022-09-30 11:39:23 +02:00
parent 1c68772ed4
commit 67699352da
2 changed files with 114 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

71
pyrmv/raw/board_arrival.py
View File

@ -1,9 +1,8 @@
from requests import get
from datetime import datetime
from typing import List, Union
from xmltodict import parse as xmlparse
from pyrmv.utility.find_exception import find_exception
try:
from typing import Literal
except ImportError:
@ -11,8 +10,53 @@ except ImportError:
# 2.25. Arrival Board (arrivalBoard)
def board_arrival(accessId: str,
json: bool = True
json: bool = True,
id: str = None, # type: ignore
extId: str = None, # type: ignore
direction: str = None, # type: ignore
date: Union[str, datetime] = None, # type: ignore
time: Union[str, datetime] = None, # type: ignore
duration: int = 60,
maxJourneys: int = -1,
products: int = None, # type: ignore
operators: Union[str, list] = None, # type: ignore
lines: Union[str, list] = None, # type: ignore
filterEquiv: bool = True,
attributes: Union[str, list] = None, # type: ignore
platforms: Union[str, list] = None, # type: ignore
passlist: bool = False,
boardType: Literal["ARR", "ARR_EQUIVS", "ARR_MAST", "ARR_STATION"] = "ARR"
) -> dict:
"""The arrival board can be retrieved by a call to the arrivalBoard service. This method will return the next
arrivals from a given point in time within a duration covered time span. The default duration size is 60 minutes.
Note: The result list always contains all arrivals running the the last minute found even if the requested
maximum was overrun.
Read more about this in section 2.25. "Arrival Board (arrivalBoard)" of HAFAS ReST Documentation.
### Args:
* accessId (str): Access ID for identifying the requesting client. Get your key on [RMV website](https://opendata.rmv.de/site/start.html).
* json (bool, optional): Whether response should be retrieved as JSON. XML is returned if False. Only matters if raw_response is True. Defaults to True.
* id (str, optional): Access ID for identifying the requesting client. Defaults to None.
* extId (str, optional): Deprecated. Please use id as it supports external IDs. Specifies the external station/stop ID for which the arrivals shall be retrieved. Required if id is not present. Such ID can be retrieved from the `stop_by_name` or `stop_by_coords`. Defaults to None.
* direction (str, optional): If only vehicles departing or arriving from a certain direction shall be returned, specify the direction by giving the station/stop ID of the last stop on the journey. Defaults to None.
* date (Union[str, datetime], optional): Sets the start date for which the departures shall be retrieved. Represented in the format YYYY-MM-DD. By default the current server date is used. Defaults to None.
* time (Union[str, datetime], optional): Sets the start time for which the departures shall be retrieved. Represented in the format hh:mm[:ss] in 24h nomenclature. Seconds will be ignored for requests. By default the current server time is used. Defaults to None.
* duration (int, optional): Set the interval size in minutes. Defaults to 60.
* maxJourneys (int, optional): Maximum number of journeys to be returned. If no value is defined or -1, all departing/arriving services within the duration sized period are returned. Defaults to -1.
* products (int, optional): Decimal value defining the product classes to be included in the search. It represents a bitmask combining bit number of a product as defined in the HAFAS raw data. Defaults to None.
* operators (Union[str, list], optional): Only journeys provided by the given operators are part of the result. To filter multiple operators, separate the codes by comma. If the operator should not be part of the be trip, negate it by putting ! in front of it. Example: Filter for operator A and B: `operators=[A,B]`. Defaults to None.
* lines (Union[str, list], optional): Only journeys running the given line are part of the result. To filter multiple lines, provide a list or separate the codes by comma. If the line should not be part of the be trip, negate it by putting ! in front of it. Defaults to None.
* filterEquiv (bool, optional): Use `boardType` instead. Enables/disables the filtering of equivalent marked stops. Defaults to True.
* attributes (Union[str, list], optional): Filter boards by one or more attribute codes of a journey. Multiple attribute as a list or as a string separated by comma. If the attribute should not be part of the result, negate it by putting ! in front of it. Defaults to None.
* platforms (Union[str, list], optional): Filter boards by platform. Multiple platforms provided as a list or as a string separated by comma. Defaults to None.
* passlist (bool, optional): Include a list of all passed waystops. Defaults to False.
* boardType (Literal["ARR", "ARR_EQUIVS", "ARR_MAST", "ARR_STATION"], optional): Set the station arrival board type to be used. ARR: Arrival board as configured in HAFAS; ARR_EQUIVS: Arrival board with all journeys at any masts and equivalent stops; ARR_MAST: Arrival board at mast; ARR_STATION: Arrival board with all journeys at any masts of the requested station. Defaults to "ARR".
### Returns:
* dict: Output from RMV. Dict will contain "errorCode" and "errorText" if exception occurs.
"""
if json:
headers = {"Accept": "application/json"}
@ -22,13 +66,30 @@ def board_arrival(accessId: str,
payload = {}
for var, val in locals().items():
if str(var) not in ["json", "headers", "payload"]:
if str(var) == "date":
if val != None:
if isinstance(val, datetime):
payload[str(var)] = val.strftime("%Y-%m-%d")
else:
payload[str(var)] = val
elif str(var) == "time":
if val != None:
if isinstance(val, datetime):
payload[str(var)] = val.strftime("%H:%M")
else:
payload[str(var)] = val
elif str(var) == "boardType":
if val != None:
payload["type"] = val
elif str(var) not in ["json", "headers", "payload"]:
if val != None:
payload[str(var)] = val
output = get("https://www.rmv.de/hapi/arrivalBoard", params=payload, headers=headers)
find_exception(output.json())
# Exceptions checker moved to normal methods
# and exceptions will no longer raise in raw ones.
# find_exception(output.json())
if json:
return output.json()

53
pyrmv/raw/board_departure.py
View File

@ -3,8 +3,6 @@ from requests import get
from typing import List, Union
from xmltodict import parse as xmlparse
from pyrmv.utility.find_exception import find_exception
try:
from typing import Literal
except ImportError:
@ -25,10 +23,41 @@ def board_departure(accessId: str,
lines: Union[str, list] = None, # type: ignore
filterEquiv: bool = True,
attributes: Union[str, list] = None, # type: ignore
platforms: int = None, # type: ignore
platforms: Union[str, list] = None, # type: ignore
passlist: bool = False,
boardType: Literal["DEP", "DEP_EQUIVS", "DEP_MAST", "DEP_STATION"] = "DEP"
) -> dict:
"""The separture board can be retrieved by a call to the departureBoard service. This method will return the
next departures (or less if not existing) from a given point in time within a duration covered time span. The
default duration size is 60 minutes.
Note: The result list always contains all departures running the the last minute found even if the requested
maximum was overrun.
Read more about this in section 2.24. "Departure Board (departureBoard)" of HAFAS ReST Documentation.
### Args:
* accessId (str): Access ID for identifying the requesting client. Get your key on [RMV website](https://opendata.rmv.de/site/start.html).
* json (bool, optional): Whether response should be retrieved as JSON. XML is returned if False. Only matters if raw_response is True. Defaults to True.
* id (str, optional): Access ID for identifying the requesting client. Defaults to None.
* extId (str, optional): Deprecated. Please use id as it supports external IDs. Specifies the external station/stop ID for which the arrivals shall be retrieved. Required if id is not present. Such ID can be retrieved from the `stop_by_name` or `stop_by_coords`. Defaults to None.
* direction (str, optional): If only vehicles departing or arriving from a certain direction shall be returned, specify the direction by giving the station/stop ID of the last stop on the journey. Defaults to None.
* date (Union[str, datetime], optional): Sets the start date for which the departures shall be retrieved. Represented in the format YYYY-MM-DD. By default the current server date is used. Defaults to None.
* time (Union[str, datetime], optional): Sets the start time for which the departures shall be retrieved. Represented in the format hh:mm[:ss] in 24h nomenclature. Seconds will be ignored for requests. By default the current server time is used. Defaults to None.
* duration (int, optional): Set the interval size in minutes. Defaults to 60.
* maxJourneys (int, optional): Maximum number of journeys to be returned. If no value is defined or -1, all departing/arriving services within the duration sized period are returned. Defaults to -1.
* products (int, optional): Decimal value defining the product classes to be included in the search. It represents a bitmask combining bit number of a product as defined in the HAFAS raw data. Defaults to None.
* operators (Union[str, list], optional): Only journeys provided by the given operators are part of the result. To filter multiple operators, separate the codes by comma. If the operator should not be part of the be trip, negate it by putting ! in front of it. Example: Filter for operator A and B: `operators=[A,B]`. Defaults to None.
* lines (Union[str, list], optional): Only journeys running the given line are part of the result. To filter multiple lines, provide a list or separate the codes by comma. If the line should not be part of the be trip, negate it by putting ! in front of it. Defaults to None.
* filterEquiv (bool, optional): Use `boardType` instead. Enables/disables the filtering of equivalent marked stops. Defaults to True.
* attributes (Union[str, list], optional): Filter boards by one or more attribute codes of a journey. Multiple attribute as a list or as a string separated by comma. If the attribute should not be part of the result, negate it by putting ! in front of it. Defaults to None.
* platforms (Union[str, list], optional): Filter boards by platform. Multiple platforms provided as a list or as a string separated by comma. Defaults to None.
* passlist (bool, optional): Include a list of all passed waystops. Defaults to False.
* boardType (Literal["DEP", "DEP_EQUIVS", "DEP_MAST", "DEP_STATION"], optional): Set the station departure board type to be used. DEP: Departure board as configured in HAFAS; DEP_EQUIVS: Departure board with all journeys at any masts and equivalent stops; DEP_MAST: Departure board at mast; DEP_STATION: Departure board with all journeys at any masts of the requested station. Defaults to "DEP".
### Returns:
* dict: Output from RMV. Dict will contain "errorCode" and "errorText" if exception occurs.
"""
if json:
headers = {"Accept": "application/json"}
@ -38,7 +67,19 @@ def board_departure(accessId: str,
payload = {}
for var, val in locals().items():
if str(var) == "boardType":
if str(var) == "date":
if val != None:
if isinstance(val, datetime):
payload[str(var)] = val.strftime("%Y-%m-%d")
else:
payload[str(var)] = val
elif str(var) == "time":
if val != None:
if isinstance(val, datetime):
payload[str(var)] = val.strftime("%H:%M")
else:
payload[str(var)] = val
elif str(var) == "boardType":
if val != None:
payload["type"] = val
elif str(var) not in ["json", "headers", "payload"]:
@ -47,7 +88,9 @@ def board_departure(accessId: str,
output = get("https://www.rmv.de/hapi/departureBoard", params=payload, headers=headers)
find_exception(output.json())
# Exceptions checker moved to normal methods
# and exceptions will no longer raise in raw ones.
# find_exception(output.json())
if json:
return output.json()