From f8e7fae9e35b6b72a800540d99f097dc62f72b10 Mon Sep 17 00:00:00 2001 From: Profitroll <47523801+profitrollgame@users.noreply.github.com> Date: Fri, 23 Sep 2022 23:00:30 +0200 Subject: [PATCH] Integrated classes, added "raw_response" argument --- pyrmv/raw/board_departure.py | 23 +++++- pyrmv/raw/stop_by_coords.py | 33 +++++--- pyrmv/raw/stop_by_name.py | 39 +++++---- pyrmv/raw/trip_find.py | 154 ++++++++++++++++++++--------------- 4 files changed, 154 insertions(+), 95 deletions(-) diff --git a/pyrmv/raw/board_departure.py b/pyrmv/raw/board_departure.py index 4e486d3..d7e2ade 100644 --- a/pyrmv/raw/board_departure.py +++ b/pyrmv/raw/board_departure.py @@ -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 diff --git a/pyrmv/raw/stop_by_coords.py b/pyrmv/raw/stop_by_coords.py index a4c9dd2..35af6ca 100644 --- a/pyrmv/raw/stop_by_coords.py +++ b/pyrmv/raw/stop_by_coords.py @@ -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) \ No newline at end of file + stops = [] + for stop in output.json()["stopLocationOrCoordLocation"]: + stops.append(Stop(stop["StopLocation"])) + return stops \ No newline at end of file diff --git a/pyrmv/raw/stop_by_name.py b/pyrmv/raw/stop_by_name.py index c603d82..8a668b3 100644 --- a/pyrmv/raw/stop_by_name.py +++ b/pyrmv/raw/stop_by_name.py @@ -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) \ No newline at end of file + stops = [] + for stop in output.json()["stopLocationOrCoordLocation"]: + stops.append(Stop(stop["StopLocation"])) + return stops \ No newline at end of file diff --git a/pyrmv/raw/trip_find.py b/pyrmv/raw/trip_find.py index e930218..709e6f7 100644 --- a/pyrmv/raw/trip_find.py +++ b/pyrmv/raw/trip_find.py @@ -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) \ No newline at end of file + trips = [] + for trip in output.json()["Trip"]: + trips.append(Trip(trip)) + return trips \ No newline at end of file