Integrated classes, added "raw_response" argument

This commit is contained in:
Profitroll 2022-09-23 23:00:30 +02:00
parent 74107ceb1b
commit f8e7fae9e3
4 changed files with 154 additions and 95 deletions

View File

@ -1,3 +1,4 @@
from datetime import datetime
from requests import get
from typing import List, Union
from xmltodict import parse as xmlparse
@ -11,7 +12,22 @@ except ImportError:
# 2.24. Departure Board (departureBoard)
def board_departure(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: int = None, # type: ignore
passlist: bool = False,
boardType: Literal["DEP", "DEP_EQUIVS", "DEP_MAST", "DEP_STATION"] = "DEP"
) -> dict:
if json:
@ -22,7 +38,10 @@ def board_departure(accessId: str,
payload = {}
for var, val in locals().items():
if str(var) not in ["json", "headers", "payload"]:
if str(var) == "boardType":
if val != None:
payload["type"] = val
elif str(var) not in ["json", "headers", "payload"]:
if val != None:
payload[str(var)] = val

View File

@ -1,6 +1,7 @@
from requests import get
from typing import Union
from typing import List, Union
from xmltodict import parse as xmlparse
from pyrmv.classes.Stop import Stop
from pyrmv.utility.find_exception import find_exception
@ -15,15 +16,16 @@ def stop_by_coords(accessId: str,
originCoordLong: Union[str, float],
lang: Literal["de", "da", "en", "es", "fr", "hu", "it", "nl", "no", "pl", "sv", "tr"] = "en",
json: bool = True,
raw_response: bool = False,
radius: Union[int, float] = 1000,
maxNo: int = 10,
stopType: Literal["S", "P", "SP", "SE", "SPE"] = "S",
locationSelectionMode: Literal["SLCT_N", "SLCT_A"] = None,
products: int = None,
meta: str = None,
sattributes: Union[str, list] = None,
sinfotexts: Union[str, list] = None
) -> dict:
locationSelectionMode: Literal["SLCT_N", "SLCT_A"] = None, # type: ignore
products: int = None, # type: ignore
meta: str = None, # type: ignore
sattributes: Union[str, list] = None, # type: ignore
sinfotexts: Union[str, list] = None # type: ignore
) -> Union[dict, List[Stop]]:
"""The location.nearbystops service returns a list of stops around a given center coordinate (within a
radius of 1000m). The returned results are ordered by their distance to the center coordinate.
@ -34,7 +36,8 @@ def stop_by_coords(accessId: str,
* originCoordLat (Union[str, float]): Latitude of centre coordinate.
* originCoordLong (Union[str, float]): Longitude of centre coordinate.
* lang (Literal["de","da","en","es","fr","hu","it","nl","no","pl","sv","tr"], optional): The language of response. Defaults to "en".
* json (bool, optional): Whether response should be retrieved as JSON. XML is returned if False. Defaults to True.
* 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.
* raw_response (bool, optional): Whether response should be returned as `dict` instead of `List[Stop]`. Defaults to False.
* radius (Union[int, float], optional): Search radius in meter around the given coordinate if any. Defaults to 1000.
* maxNo (int, optional): Maximum number of returned stops. Defaults to 10.
* stopType (Literal["S", "P", "SP", "SE", "SPE"], optional): Type filter for location types. Defaults to "S".
@ -56,7 +59,7 @@ def stop_by_coords(accessId: str,
payload = {}
for var, val in locals().items():
if str(var) not in ["json", "headers", "payload"]:
if str(var) not in ["json", "headers", "payload", "raw_response"]:
if val != None:
payload[str(var)] = val
@ -64,7 +67,13 @@ def stop_by_coords(accessId: str,
find_exception(output.json())
if json:
return output.json()
if raw_response:
if json:
return output.json()
else:
return xmlparse(output.content)
else:
return xmlparse(output.content)
stops = []
for stop in output.json()["stopLocationOrCoordLocation"]:
stops.append(Stop(stop["StopLocation"]))
return stops

View File

@ -1,6 +1,7 @@
from requests import get
from typing import Union
from typing import List, Union
from xmltodict import parse as xmlparse
from pyrmv.classes.Stop import Stop
from pyrmv.utility.find_exception import find_exception
@ -14,19 +15,20 @@ def stop_by_name(accessId: str,
inputString: str,
lang: Literal["de", "da", "en", "es", "fr", "hu", "it", "nl", "no", "pl", "sv", "tr"] = "en",
json: bool = True,
raw_response: bool = False,
maxNo: int = 10,
stopType: Literal["A", "ALL", "AP", "P", "S", "SA", "SP"] = "ALL",
locationSelectionMode: Literal["SLCT_N", "SLCT_A"] = None,
products: int = None,
coordLat: Union[str, float] = None,
coordLong: Union[str, float] = None,
locationSelectionMode: Literal["SLCT_N", "SLCT_A"] = None, # type: ignore
products: int = None, # type: ignore
coordLat: Union[str, float] = None, # type: ignore
coordLong: Union[str, float] = None, # type: ignore
radius: Union[int, float] = 1000,
refineId: str = None,
meta: str = None,
stations: Union[str, list] = None,
sattributes: Union[str, list] = None,
refineId: str = None, # type: ignore
meta: str = None, # type: ignore
stations: Union[str, list] = None, # type: ignore
sattributes: Union[str, list] = None, # type: ignore
filterMode: Literal["DIST_PERI", "EXCL_PERI", "SLCT_PERI"] = "DIST_PERI"
) -> dict:
) -> Union[dict, List[Stop]]:
"""The location.name service can be used to perform a pattern matching of a user input and to retrieve a list
of possible matches in the journey planner database. Possible matches might be stops/stations, points of
interest and addresses.
@ -41,7 +43,8 @@ def stop_by_name(accessId: str,
* accessId (str): Access ID for identifying the requesting client. Get your key on [RMV website](https://opendata.rmv.de/site/start.html).
* inputString (str): Search for that token.
* lang (Literal["de","da","en","es","fr","hu","it","nl","no","pl","sv","tr"], optional): The language of response. Defaults to "en".
* json (bool, optional): Whether response should be retrieved as JSON. XML is returned if False. Defaults to True.
* 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.
* raw_response (bool, optional): Whether response should be returned as `dict` instead of `List[Stop]`. Defaults to False.
* maxNo (int, optional): Maximum number of returned stops. In range 1-1000. Defaults to 10.
* stopType (Literal["A", "ALL", "AP", "P", "S", "SA", "SP"], optional): Type filter for location types. Defaults to "ALL".
* locationSelectionMode (str, optional): Selection mode for locations. "SLCT_N": Not selectable, "SLCT_A": Selectable. Defaults to None.
@ -70,7 +73,7 @@ def stop_by_name(accessId: str,
if str(var) == "inputString":
if val != None:
payload["input"] = val
if str(var) not in ["json", "headers", "payload"]:
elif str(var) not in ["json", "headers", "payload", "raw_response"]:
if val != None:
payload[str(var)] = val
@ -78,7 +81,13 @@ def stop_by_name(accessId: str,
find_exception(output.json())
if json:
return output.json()
if raw_response:
if json:
return output.json()
else:
return xmlparse(output.content)
else:
return xmlparse(output.content)
stops = []
for stop in output.json()["stopLocationOrCoordLocation"]:
stops.append(Stop(stop["StopLocation"]))
return stops

View File

@ -1,6 +1,8 @@
from datetime import datetime
from requests import get
from typing import List, Union
from xmltodict import parse as xmlparse
from pyrmv.classes.Trip import Trip
from pyrmv.utility.find_exception import find_exception
@ -13,88 +15,89 @@ except ImportError:
def trip_find(accessId: str,
lang: Literal["de", "da", "en", "es", "fr", "hu", "it", "nl", "no", "pl", "sv", "tr"] = "en",
json: bool = True,
raw_response: bool = False,
originId: str = None,
originExtId: str = None,
originCoordLat: Union[str, float] = None,
originCoordLong: Union[str, float] = None,
originCoordName: str = None,
originId: str = None, # type: ignore
originExtId: str = None, # type: ignore
originCoordLat: Union[str, float] = None, # type: ignore
originCoordLong: Union[str, float] = None, # type: ignore
originCoordName: str = None, # type: ignore
destId: str = None,
destExtId: str = None,
destCoordLat: Union[str, float] = None,
destCoordLong: Union[str, float] = None,
destCoordName: str = None,
destId: str = None, # type: ignore
destExtId: str = None, # type: ignore
destCoordLat: Union[str, float] = None, # type: ignore
destCoordLong: Union[str, float] = None, # type: ignore
destCoordName: str = None, # type: ignore
via: str = None,
viaId: str = None,
via: str = None, # type: ignore
viaId: str = None, # type: ignore
viaWaitTime: int = 0,
avoid: str = None,
avoidId: str = None,
avoid: str = None, # type: ignore
avoidId: str = None, # type: ignore
viaGis: str = None,
viaGis: str = None, # type: ignore
changeTimePercent: int = 100,
minChangeTime: int = None,
maxChangeTime: int = None,
addChangeTime: int = None,
maxChange: int = None,
minChangeTime: int = None, # type: ignore
maxChangeTime: int = None, # type: ignore
addChangeTime: int = None, # type: ignore
maxChange: int = None, # type: ignore
date: str = None,
time: str = None,
date: Union[str, datetime] = None, # type: ignore
time: Union[str, datetime] = None, # type: ignore
searchForArrival: bool = False,
numF: int = None,
numB: int = None,
numF: int = None, # type: ignore
numB: int = None, # type: ignore
context: str = None,
context: str = None, # type: ignore
poly: bool = False,
polyEnc: Literal["DLT", "GPA", "N"] = "N",
passlist: bool = False,
products: str = None,
operators: Union[str, list] = None,
products: int = None, # type: ignore
operators: Union[str, list] = None, # type: ignore
attributes: Union[str, list] = None,
sattributes: Union[str, list] = None,
fattributes: Union[str, list] = None,
lines: Union[str, list] = None,
lineids: Union[str, list] = None,
attributes: Union[str, list] = None, # type: ignore
sattributes: Union[str, list] = None, # type: ignore
fattributes: Union[str, list] = None, # type: ignore
lines: Union[str, list] = None, # type: ignore
lineids: Union[str, list] = None, # type: ignore
avoidPaths: List[Literal["SW", "EA", "ES", "RA", "CB"]] = None,
avoidPaths: List[Literal["SW", "EA", "ES", "RA", "CB"]] = None, # type: ignore
originWalk: Union[str, list] = None,
originBike: Union[str, list] = None,
originCar: Union[str, list] = None,
originTaxi: Union[str, list] = None,
originPark: Union[str, list] = None,
originMeta: Union[str, list] = None,
originWalk: Union[str, list] = None, # type: ignore
originBike: Union[str, list] = None, # type: ignore
originCar: Union[str, list] = None, # type: ignore
originTaxi: Union[str, list] = None, # type: ignore
originPark: Union[str, list] = None, # type: ignore
originMeta: Union[str, list] = None, # type: ignore
destWalk: Union[str, list] = None,
destBike: Union[str, list] = None,
destCar: Union[str, list] = None,
destTaxi: Union[str, list] = None,
destPark: Union[str, list] = None,
destMeta: Union[str, list] = None,
destWalk: Union[str, list] = None, # type: ignore
destBike: Union[str, list] = None, # type: ignore
destCar: Union[str, list] = None, # type: ignore
destTaxi: Union[str, list] = None, # type: ignore
destPark: Union[str, list] = None, # type: ignore
destMeta: Union[str, list] = None, # type: ignore
totalWalk: Union[str, list] = None,
totalBike: Union[str, list] = None,
totalCar: Union[str, list] = None,
totalTaxi: Union[str, list] = None,
totalMeta: Union[str, list] = None,
totalWalk: Union[str, list] = None, # type: ignore
totalBike: Union[str, list] = None, # type: ignore
totalCar: Union[str, list] = None, # type: ignore
totalTaxi: Union[str, list] = None, # type: ignore
totalMeta: Union[str, list] = None, # type: ignore
gisProducts: str = None,
gisProducts: str = None, # type: ignore
includeIv: bool = False,
ivOnly: bool = False,
mobilityProfile: str = None,
mobilityProfile: str = None, # type: ignore
bikeCarriage: bool = False,
bikeCarriageType: Literal["SINGLEBIKES", "SMALLGROUPS", "LARGEGROUPS"] = None,
bikeCarriageType: Literal["SINGLEBIKES", "SMALLGROUPS", "LARGEGROUPS"] = None, # type: ignore
sleepingCar: bool = False,
couchetteCoach: bool = False,
@ -103,26 +106,26 @@ def trip_find(accessId: str,
eco: bool = False,
ecoCmp: bool = False,
ecoParams: str = None,
ecoParams: str = None, # type: ignore
rtMode: Literal["FULL", "INFOS", "OFF", "REALTIME", "SERVER_DEFAULT"] = None,
rtMode: Literal["FULL", "INFOS", "OFF", "REALTIME", "SERVER_DEFAULT"] = None, # type: ignore
unsharp: bool = False,
trainFilter: str = None,
trainFilter: str = None, # type: ignore
economic: bool = False,
groupFilter: str = None,
groupFilter: str = None, # type: ignore
blockingList: str = None,
blockedEdges: str = None,
blockingList: str = None, # type: ignore
blockedEdges: str = None, # type: ignore
trainComposition: bool = False,
includeEarlier: bool = False,
withICTAlternatives: bool = False,
tariff: bool = None,
tariff: bool = None, # type: ignore
trafficMessages: bool = False,
travellerProfileData: str = None,
travellerProfileData: str = None, # type: ignore
withFreq: bool = True
) -> dict:
) -> Union[dict, List[Trip]]:
"""The trip service calculates a trip from a specified origin to a specified destination. These might be
stop/station IDs or coordinates based on addresses and points of interest validated by the location service or
coordinates freely defined by the client.
@ -132,7 +135,8 @@ def trip_find(accessId: str,
### Args:
* accessId (str): Access ID for identifying the requesting client. Get your key on [RMV website](https://opendata.rmv.de/site/start.html).
* lang (Literal["de","da","en","es","fr","hu","it","nl","no","pl","sv","tr"], optional): The language of response. Defaults to "en".
* json (bool, optional): Whether response should be retrieved as JSON. XML is returned if False. Defaults to True.
* 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.
* raw_response (bool, optional): Whether response should be returned as `dict` instead of `List[Trip]`. Defaults to False.
* originId (str, optional): Specifies the station/stop ID of the origin for the trip. Such ID can be retrieved from stopByName() or stopByCoords(). Defaults to None.
* originExtId (str, optional): Deprecated. Please use originId as it supports external IDs. Specifies the external station/stop ID of the origin for the trip. Such ID can be retrieved from stopByName() or stopByCoords(). Defaults to None.
* originCoordLat (Union[str, float], optional): Latitude of station/stop coordinate of the trip's origin. The coordinate can be retrieved from stopByName() or stopByCoords(). Defaults to None.
@ -228,7 +232,19 @@ def trip_find(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) not in ["json", "headers", "payload", "raw_response"]:
if val != None:
payload[str(var)] = val
@ -236,7 +252,13 @@ def trip_find(accessId: str,
find_exception(output.json())
if json:
return output.json()
if raw_response:
if json:
return output.json()
else:
return xmlparse(output.content)
else:
return xmlparse(output.content)
trips = []
for trip in output.json()["Trip"]:
trips.append(Trip(trip))
return trips