diff --git a/README.md b/README.md index 979f8e2..6345086 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,9 @@ -# PythonRMV +

PythonRMV

+ +

+License: MIT +Code style: black +

Small module that makes your journey with RMV REST API somehow easier. Based fully on official RMV API reference and HAFAS documentation. diff --git a/pyproject.toml b/pyproject.toml index 40287a7..499db6e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -44,6 +44,7 @@ where = ["src"] [tool.black] target-version = ['py38', 'py39', 'py310', 'py311'] +line-length = 94 [tool.isort] profile = "black" diff --git a/src/pyrmv/classes/Board.py b/src/pyrmv/classes/Board.py index 70d3429..a52a1d5 100644 --- a/src/pyrmv/classes/Board.py +++ b/src/pyrmv/classes/Board.py @@ -1,94 +1,99 @@ from datetime import datetime +from typing import Any, Mapping + from pyrmv.classes.Message import Message -class LineArrival(): - def __init__(self, data, client, retrieve_stops: bool = True): +class LineArrival: + def __init__(self, data: Mapping[str, Any], client, retrieve_stops: bool = True): self.journey = client.journey_detail(data["JourneyDetailRef"]["ref"]) self.status = data["JourneyStatus"] self.messages = [] - if "Messages" in data: - self.messages.extend( - Message(message) for message in data["Messages"]["Message"] - ) self.name = data["name"] self.type = data["type"] self.stop_name = data["stop"] self.stop_id = data["stopid"] self.stop_id_ext = data["stopExtId"] - if retrieve_stops: - self.stop = client.stop_by_id(self.stop_id) - else: - self.stop = None + self.stop = client.stop_by_id(self.stop_id) if retrieve_stops else None self.stop = client.stop_by_id(self.stop_id) self.time = datetime.strptime(data["time"], "%H:%M:%S") self.date = datetime.strptime(data["date"], "%Y-%m-%d") + self.reachable = data["reachable"] + self.origin = data["origin"] + self.origin = data["origin"] + + if "Messages" in data: + self.messages.extend(Message(message) for message in data["Messages"]["Message"]) + if ("rtTime" in data) and ("rtDate" in data): self.time_real_time = datetime.strptime(data["rtTime"], "%H:%M:%S") self.date_real_time = datetime.strptime(data["rtDate"], "%Y-%m-%d") - self.reachable = data["reachable"] - self.origin = data["origin"] + else: + self.time_real_time = None + self.date_real_time = None def __str__(self) -> str: - return f"{self.name} coming from {self.origin} at {self.time.time()} {self.date.date()}" + return ( + f"{self.name} coming from {self.origin} at {self.time.time()} {self.date.date()}" + ) -class LineDeparture(): - def __init__(self, data, client, retrieve_stops: bool = True): +class LineDeparture: + def __init__(self, data: Mapping[str, Any], client, retrieve_stops: bool = True): self.journey = client.journey_detail(data["JourneyDetailRef"]["ref"]) self.status = data["JourneyStatus"] self.messages = [] - if "Messages" in data: - self.messages.extend( - Message(message) for message in data["Messages"]["Message"] - ) self.name = data["name"] self.type = data["type"] self.stop_name = data["stop"] self.stop_id = data["stopid"] self.stop_id_ext = data["stopExtId"] - if retrieve_stops: - self.stop = client.stop_by_id(self.stop_id) - else: - self.stop = None + self.stop = client.stop_by_id(self.stop_id) if retrieve_stops else None self.time = datetime.strptime(data["time"], "%H:%M:%S") self.date = datetime.strptime(data["date"], "%Y-%m-%d") - if ("rtTime" in data) and ("rtDate" in data): - self.time_real_time = datetime.strptime(data["rtTime"], "%H:%M:%S") - self.date_real_time = datetime.strptime(data["rtDate"], "%Y-%m-%d") self.reachable = data["reachable"] self.direction = data["direction"] self.direction_flag = data["directionFlag"] - - def __str__(self) -> str: - return f"{self.name} heading {self.direction} at {self.time.time()} {self.date.date()}" - -class BoardArrival(list): - def __init__(self, data: dict, client, retrieve_stops: bool = True): + if "Messages" in data: + self.messages.extend(Message(message) for message in data["Messages"]["Message"]) + + if ("rtTime" in data) and ("rtDate" in data): + self.time_real_time = datetime.strptime(data["rtTime"], "%H:%M:%S") + self.date_real_time = datetime.strptime(data["rtDate"], "%Y-%m-%d") + else: + self.time_real_time = None + self.date_real_time = None + + def __str__(self) -> str: + return ( + f"{self.name} heading {self.direction} at {self.time.time()} {self.date.date()}" + ) + + +class BoardArrival(list): + def __init__(self, data: Mapping[str, Any], client, retrieve_stops: bool = True): super().__init__([]) + if "Arrival" not in data: return + for line in data["Arrival"]: self.append(LineArrival(line, client, retrieve_stops=retrieve_stops)) def __str__(self) -> str: - lines = [] - for line in self: - lines.append(str(line)) - return "Arrival board\n" + "\n".join(lines) + return "Arrival board\n" + "\n".join([str(line) for line in self]) + class BoardDeparture(list): - - def __init__(self, data: dict, client, retrieve_stops: bool = True): + def __init__(self, data: Mapping[str, Any], client, retrieve_stops: bool = True): super().__init__([]) + if "Departure" not in data: return + for line in data["Departure"]: self.append(LineDeparture(line, client, retrieve_stops=retrieve_stops)) def __str__(self) -> str: - lines = [] - for line in self: - lines.append(str(line)) - return "Departure board\n" + "\n".join(lines) \ No newline at end of file + return "Departure board\n" + "\n".join([str(line) for line in self]) diff --git a/src/pyrmv/classes/Client.py b/src/pyrmv/classes/Client.py index a625026..8676f76 100644 --- a/src/pyrmv/classes/Client.py +++ b/src/pyrmv/classes/Client.py @@ -1,24 +1,26 @@ from datetime import datetime, timedelta from typing import List, OrderedDict, Union + from pyrmv.classes import * from pyrmv.enums import * from pyrmv.raw import board_arrival as raw_board_arrival from pyrmv.raw.board_departure import board_departure as raw_board_departure +from pyrmv.raw.him_search import him_search as raw_him_search from pyrmv.raw.journey_detail import journey_detail as raw_journey_detail -from pyrmv.raw.stop_by_name import stop_by_name as raw_stop_by_name from pyrmv.raw.stop_by_coords import stop_by_coords as raw_stop_by_coords +from pyrmv.raw.stop_by_name import stop_by_name as raw_stop_by_name from pyrmv.raw.trip_find import trip_find as raw_trip_find from pyrmv.raw.trip_recon import trip_recon as raw_trip_recon -from pyrmv.raw.him_search import him_search as raw_him_search -from pyrmv.utility.find_exception import find_exception +from pyrmv.utility import find_exception try: from typing import Literal except ImportError: from typing_extensions import Literal -class Client(): - """The main class in the whole module. Is used to use all non-raw methods. + +class Client: + """The main class in the whole module. Is used to use all non-raw methods. More detailed docs for each method can be found by using IDE's docstring highlighting system or in project's wiki ([can be found here](https://git.end-play.xyz/profitroll/PythonRMV/wiki)) @@ -63,27 +65,33 @@ class Client(): # Find a trip and reconstruct it trip = client.trip_find(origin_coord_lat="50.084659", origin_coord_lon="8.785948", destination_coord_lat=50.1233048, destination_coord_lon=8.6129742, messages=True)[0] trip_recon = client.trip_recon(trip)[0] - ``` - """ + ``` + """ def __init__(self, access_id: str) -> None: self.access_id = access_id - def board_arrival(self, - id: Union[str, None] = None, - id_ext: Union[str, None] = None, - direction: Union[str, Stop, StopTrip, None] = None, - date: Union[str, datetime, None] = None, - time: Union[str, datetime, None] = None, - duration: Union[int, timedelta] = 60, - journeys_max: int = -1, - operators: Union[str, list, None] = None, - lines: Union[str, list, None] = None, - passlist: bool = False, - board_type: Literal[BoardArrivalType.ARR, BoardArrivalType.ARR_EQUIVS, BoardArrivalType.ARR_MAST, BoardArrivalType.ARR_STATION] = BoardArrivalType.ARR, - retrieve_stops: bool = True - ) -> BoardArrival: - """Method returns a board with arriving transport. + def board_arrival( + self, + id: Union[str, None] = None, + id_ext: Union[str, None] = None, + direction: Union[str, Stop, StopTrip, None] = None, + date: Union[str, datetime, None] = None, + time: Union[str, datetime, None] = None, + duration: Union[int, timedelta] = 60, + journeys_max: int = -1, + operators: Union[str, list, None] = None, + lines: Union[str, list, None] = None, + passlist: bool = False, + board_type: Literal[ + BoardArrivalType.ARR, + BoardArrivalType.ARR_EQUIVS, + BoardArrivalType.ARR_MAST, + BoardArrivalType.ARR_STATION, + ] = BoardArrivalType.ARR, + retrieve_stops: bool = True, + ) -> BoardArrival: + """Method returns a board with arriving transport. More detailed request is available as `raw.board_arrival()`, however returns `dict` instead of `Board`. @@ -105,14 +113,14 @@ class Client(): * BoardArrival: Instance of `BoardArrival` object. """ - if (isinstance(direction, Stop) or isinstance(direction, StopTrip)): + if isinstance(direction, Stop) or isinstance(direction, StopTrip): direction = direction.id board_raw = raw_board_arrival( accessId=self.access_id, id=id, extId=id_ext, - direction=direction, # type: ignore + direction=direction, # type: ignore date=date, time=time, duration=duration, @@ -120,28 +128,34 @@ class Client(): operators=operators, lines=lines, passlist=passlist, - boardType=board_type.code + boardType=board_type.code, ) find_exception(board_raw) return BoardArrival(board_raw, self, retrieve_stops=retrieve_stops) - def board_departure(self, - id: Union[str, None] = None, - id_ext: Union[str, None] = None, - direction: Union[str, Stop, StopTrip, None] = None, - date: Union[str, datetime, None] = None, - time: Union[str, datetime, None] = None, - duration: Union[int, timedelta] = 60, - journeys_max: int = -1, - operators: Union[str, list, None] = None, - lines: Union[str, list, None] = None, - passlist: bool = False, - board_type: Literal[BoardDepartureType.DEP, BoardDepartureType.DEP_EQUIVS, BoardDepartureType.DEP_MAST, BoardDepartureType.DEP_STATION] = BoardDepartureType.DEP, - retrieve_stops: bool = True - ) -> BoardDeparture: - """Method returns a board with departing transport. + def board_departure( + self, + id: Union[str, None] = None, + id_ext: Union[str, None] = None, + direction: Union[str, Stop, StopTrip, None] = None, + date: Union[str, datetime, None] = None, + time: Union[str, datetime, None] = None, + duration: Union[int, timedelta] = 60, + journeys_max: int = -1, + operators: Union[str, list, None] = None, + lines: Union[str, list, None] = None, + passlist: bool = False, + board_type: Literal[ + BoardDepartureType.DEP, + BoardDepartureType.DEP_EQUIVS, + BoardDepartureType.DEP_MAST, + BoardDepartureType.DEP_STATION, + ] = BoardDepartureType.DEP, + retrieve_stops: bool = True, + ) -> BoardDeparture: + """Method returns a board with departing transport. More detailed request is available as `raw.board_departure()`, however returns `dict` instead of `Board`. @@ -163,14 +177,14 @@ class Client(): * BoardDeparture: Instance of `BoardDeparture` object. """ - if (isinstance(direction, Stop) or isinstance(direction, StopTrip)): + if isinstance(direction, (Stop, StopTrip)): direction = direction.id board_raw = raw_board_departure( accessId=self.access_id, id=id, extId=id_ext, - direction=direction, # type: ignore + direction=direction, # type: ignore date=date, time=time, duration=duration, @@ -178,41 +192,53 @@ class Client(): operators=operators, lines=lines, passlist=passlist, - boardType=board_type.code + boardType=board_type.code, ) find_exception(board_raw) return BoardDeparture(board_raw, self, retrieve_stops=retrieve_stops) - def him_search(self, - date_begin: Union[str, datetime, None] = None, - date_end: Union[str, datetime, None] = None, - time_begin: Union[str, datetime, None] = None, - time_end: Union[str, datetime, None] = None, - weekdays: Union[str, OrderedDict[str, bool], None] = None, - ids: Union[list, None] = None, - operators: Union[list, None] = None, - categories: Union[list, None] = None, - channels: Union[list, None] = None, - companies: Union[list, None] = None, - lines: Union[list, None] = None, - line_ids: Union[list, None] = None, - stations: Union[list, List[Stop], None] = None, - station_from: Union[str, Stop, None] = None, - station_to: Union[str, Stop, None] = None, - both_ways: Union[bool, None] = None, - train_names: Union[list, None] = None, - search_mode: Union[Literal[SearchMode.MATCH, SearchMode.NOMATCH, SearchMode.TFMATCH], None] = None, - affected_journey_mode: Union[Literal[AffectedJourneyMode.ALL, AffectedJourneyMode.OFF], None] = None, - affected_journey_stop_mode: Union[Literal[AffectedJourneyStopMode.ALL, AffectedJourneyStopMode.IMP, AffectedJourneyStopMode.OFF], None] = None, - priority_min: Union[int, None] = None, - priority_max: Union[int, None] = None - ) -> List[Message]: + def him_search( + self, + date_begin: Union[str, datetime, None] = None, + date_end: Union[str, datetime, None] = None, + time_begin: Union[str, datetime, None] = None, + time_end: Union[str, datetime, None] = None, + weekdays: Union[str, OrderedDict[str, bool], None] = None, + ids: Union[list, None] = None, + operators: Union[list, None] = None, + categories: Union[list, None] = None, + channels: Union[list, None] = None, + companies: Union[list, None] = None, + lines: Union[list, None] = None, + line_ids: Union[list, None] = None, + stations: Union[list, List[Stop], None] = None, + station_from: Union[str, Stop, None] = None, + station_to: Union[str, Stop, None] = None, + both_ways: Union[bool, None] = None, + train_names: Union[list, None] = None, + search_mode: Union[ + Literal[SearchMode.MATCH, SearchMode.NOMATCH, SearchMode.TFMATCH], None + ] = None, + affected_journey_mode: Union[ + Literal[AffectedJourneyMode.ALL, AffectedJourneyMode.OFF], None + ] = None, + affected_journey_stop_mode: Union[ + Literal[ + AffectedJourneyStopMode.ALL, + AffectedJourneyStopMode.IMP, + AffectedJourneyStopMode.OFF, + ], + None, + ] = None, + priority_min: Union[int, None] = None, + priority_max: Union[int, None] = None, + ) -> List[Message]: """The him_search method will deliver a list of HIM messages if matched by the given criteria as - well as affected products if any. + well as affected products if any. - More detailed request is available as `raw.him_search()`, however returns `dict` instead of `List[Message]`. + More detailed request is available as `raw.him_search()`, however returns `dict` instead of `List[Message]`. ### Args: * date_begin (`Union[str, datetime]`, optional): Sets the event period start date. Defaults to `None`. @@ -240,7 +266,7 @@ class Client(): ### Returns: * List[Message]: List of `Message` objects. Empty list if none found. - """ + """ if isinstance(station_from, Stop): station_from = station_from.ext_id @@ -257,20 +283,13 @@ class Client(): new_stations.append(stop) stations = new_stations - if search_mode == None: - search_mode = None - else: - search_mode = search_mode.code - - if affected_journey_mode == None: - affected_journey_mode = None - else: - affected_journey_mode = affected_journey_mode.code - - if affected_journey_stop_mode == None: - affected_journey_stop_mode = None - else: - affected_journey_stop_mode = affected_journey_stop_mode.code + search_mode = None if search_mode is None else search_mode.code + affected_journey_mode = ( + None if affected_journey_mode is None else affected_journey_mode.code + ) + affected_journey_stop_mode = ( + None if affected_journey_stop_mode is None else affected_journey_stop_mode.code + ) messages = [] messages_raw = raw_him_search( @@ -288,38 +307,47 @@ class Client(): lines=lines, lineids=line_ids, stations=stations, - fromstation=station_from, # type: ignore - tostation=station_to, # type: ignore + fromstation=station_from, # type: ignore + tostation=station_to, # type: ignore bothways=both_ways, trainnames=train_names, - searchmode=search_mode, # type: ignore - affectedJourneyMode=affected_journey_mode, # type: ignore - affectedJourneyStopMode=affected_journey_stop_mode, # type: ignore + searchmode=search_mode, # type: ignore + affectedJourneyMode=affected_journey_mode, # type: ignore + affectedJourneyStopMode=affected_journey_stop_mode, # type: ignore maxprio=priority_max, - minprio=priority_min + minprio=priority_min, ) find_exception(messages_raw) if "Message" in messages_raw: - for message in messages_raw["Message"]: - messages.append(Message(message)) + messages.extend(Message(message) for message in messages_raw["Message"]) return messages - def journey_detail(self, - id: str, - date: Union[str, datetime, None] = None, - real_time_mode: Union[Literal[RealTimeMode.FULL, RealTimeMode.INFOS, RealTimeMode.OFF, RealTimeMode.REALTIME, RealTimeMode.SERVER_DEFAULT], None] = None, - from_id: Union[str, None] = None, - from_index: Union[int, None] = None, - to_id: Union[str, None] = None, - to_index: Union[int, None] = None - ) -> Journey: + def journey_detail( + self, + id: str, + date: Union[str, datetime, None] = None, + real_time_mode: Union[ + Literal[ + RealTimeMode.FULL, + RealTimeMode.INFOS, + RealTimeMode.OFF, + RealTimeMode.REALTIME, + RealTimeMode.SERVER_DEFAULT, + ], + None, + ] = None, + from_id: Union[str, None] = None, + from_index: Union[int, None] = None, + to_id: Union[str, None] = None, + to_index: Union[int, None] = None, + ) -> Journey: """The journey_detail method will deliver information about the complete route of a vehicle. The journey identifier is part of a trip or `board_departure()` response. It contains a list of all stops/stations of this journey including all departure and arrival times (with real-time data if available) and additional information like - specific attributes about facilities and other texts. + specific attributes about facilities and other texts. More detailed request is available as `raw.journey_detail()`, however returns `dict` instead of `Journey`. @@ -334,40 +362,58 @@ class Client(): ### Returns: * Journey: Instance of `Journey` object. - """ + """ - if real_time_mode == None: - real_time_mode = None - else: - real_time_mode = real_time_mode.code + real_time_mode = None if real_time_mode is None else real_time_mode.code journey_raw = raw_journey_detail( accessId=self.access_id, id=id, date=date, - rtMode=real_time_mode, # type: ignore + rtMode=real_time_mode, # type: ignore fromId=from_id, fromIdx=from_index, toId=to_id, - toIdx=to_index + toIdx=to_index, ) find_exception(journey_raw) return Journey(journey_raw) - def stop_by_coords(self, - coords_lat: Union[str, float], - coords_lon: Union[str, float], - - lang: Literal[Language.DE, Language.DA, Language.EN, Language.ES, Language.FR, Language.HU, Language.IT, Language.NL, Language.NO, Language.PL, Language.SV, Language.TR] = Language.EN, - radius: Union[int, float] = 1000, - max_number: int = 10, - stop_type: Literal[LocationType.S, LocationType.P, LocationType.SP, LocationType.SE, LocationType.SPE] = LocationType.S, - selection_mode: Union[Literal[SelectionMode.SLCT_A, SelectionMode.SLCT_N], None] = None, - ) -> List[Stop]: + def stop_by_coords( + self, + coords_lat: Union[str, float], + coords_lon: Union[str, float], + lang: Literal[ + Language.DE, + Language.DA, + Language.EN, + Language.ES, + Language.FR, + Language.HU, + Language.IT, + Language.NL, + Language.NO, + Language.PL, + Language.SV, + Language.TR, + ] = Language.EN, + radius: Union[int, float] = 1000, + max_number: int = 10, + stop_type: Literal[ + LocationType.S, + LocationType.P, + LocationType.SP, + LocationType.SE, + LocationType.SPE, + ] = LocationType.S, + selection_mode: Union[ + Literal[SelectionMode.SLCT_A, SelectionMode.SLCT_N], None + ] = None, + ) -> List[Stop]: """Method returns a list of stops around a given center coordinate. - The returned results are ordered by their distance to the center coordinate. + The returned results are ordered by their distance to the center coordinate. More detailed request is available as `raw.stop_by_coords()`, however returns `dict` instead of `List[Stop]`. @@ -382,12 +428,9 @@ class Client(): ### Returns: * List[Stop]: List of `Stop` objects. Empty list if none found. - """ + """ - if selection_mode == None: - selection_mode = None - else: - selection_mode = selection_mode.code + selection_mode = None if selection_mode is None else selection_mode.code stops = [] stops_raw = raw_stop_by_coords( @@ -398,7 +441,7 @@ class Client(): radius=radius, maxNo=max_number, stopType=stop_type.code, - locationSelectionMode=selection_mode # type: ignore + locationSelectionMode=selection_mode, # type: ignore ) find_exception(stops_raw) @@ -412,59 +455,96 @@ class Client(): return stops - def stop_by_id(self, - query: str, - lang: Literal[Language.DE, Language.DA, Language.EN, Language.ES, Language.FR, Language.HU, Language.IT, Language.NL, Language.NO, Language.PL, Language.SV, Language.TR] = Language.EN, - ) -> Union[Stop, None]: + def stop_by_id( + self, + query: str, + lang: Literal[ + Language.DE, + Language.DA, + Language.EN, + Language.ES, + Language.FR, + Language.HU, + Language.IT, + Language.NL, + Language.NO, + Language.PL, + Language.SV, + Language.TR, + ] = Language.EN, + ) -> Union[Stop, None]: """Method can be used to get Stop object whilst only having id available. ### Args: * query (`str`): Search for that token. * lang (`Literal[Language.DE, Language.DA, Language.EN, Language.ES, Language.FR, Language.HU, Language.IT, Language.NL, Language.NO, Language.PL, Language.SV, Language.TR]`, **optional**): The language of response. Defaults to `Language.EN`. - + ### Returns: * Union[Stop, None]: Instance of `Stop` object or `None` if not found. - """ + """ stops_raw = raw_stop_by_name( - accessId=self.access_id, - inputString=query, - lang=lang.code, - maxNo=1 + accessId=self.access_id, inputString=query, lang=lang.code, maxNo=1 ) find_exception(stops_raw) - if len(stops_raw["stopLocationOrCoordLocation"]) > 0: - stop = stops_raw["stopLocationOrCoordLocation"][0] + if len(stops_raw["stopLocationOrCoordLocation"]) <= 0: + return None - if "StopLocation" in stop: - return Stop(stop["StopLocation"]) - elif "CoordLocation" in stop: - return Stop(stop["CoordLocation"]) - else: - return None + stop = stops_raw["stopLocationOrCoordLocation"][0] + + if "StopLocation" in stop: + return Stop(stop["StopLocation"]) + elif "CoordLocation" in stop: + return Stop(stop["CoordLocation"]) else: return None - def stop_by_name(self, - query: str, - lang: Literal[Language.DE, Language.DA, Language.EN, Language.ES, Language.FR, Language.HU, Language.IT, Language.NL, Language.NO, Language.PL, Language.SV, Language.TR] = Language.EN, - max_number: int = 10, - stop_type: Literal[LocationType.A, LocationType.ALL, LocationType.AP, LocationType.P, LocationType.S, LocationType.SA, LocationType.SP] = LocationType.ALL, - selection_mode: Union[Literal[SelectionMode.SLCT_A, SelectionMode.SLCT_N], None] = None, - coord_lat: Union[str, float, None] = None, - coord_lon: Union[str, float, None] = None, - radius: Union[int, float] = 1000, - refine_id: Union[str, None] = None, - stations: Union[str, list, None] = None, - filter_mode: Literal[FilterMode.DIST_PERI, FilterMode.EXCL_PERI, FilterMode.SLCT_PERI] = FilterMode.DIST_PERI - ) -> List[Stop]: + def stop_by_name( + self, + query: str, + lang: Literal[ + Language.DE, + Language.DA, + Language.EN, + Language.ES, + Language.FR, + Language.HU, + Language.IT, + Language.NL, + Language.NO, + Language.PL, + Language.SV, + Language.TR, + ] = Language.EN, + max_number: int = 10, + stop_type: Literal[ + LocationType.A, + LocationType.ALL, + LocationType.AP, + LocationType.P, + LocationType.S, + LocationType.SA, + LocationType.SP, + ] = LocationType.ALL, + selection_mode: Union[ + Literal[SelectionMode.SLCT_A, SelectionMode.SLCT_N], None + ] = None, + coord_lat: Union[str, float, None] = None, + coord_lon: Union[str, float, None] = None, + radius: Union[int, float] = 1000, + refine_id: Union[str, None] = None, + stations: Union[str, list, None] = None, + filter_mode: Literal[ + FilterMode.DIST_PERI, FilterMode.EXCL_PERI, FilterMode.SLCT_PERI + ] = FilterMode.DIST_PERI, + ) -> List[Stop]: """Method 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. + points of interest and addresses. - More detailed request is available as `raw.stop_by_name()`, however returns `dict` instead of `List[Stop]`. + More detailed request is available as `raw.stop_by_name()`, however returns `dict` instead of `List[Stop]`. ### Args: * query (`str`): Search for that token. @@ -481,12 +561,9 @@ class Client(): ### Returns: * List[Stop]: List of `Stop` objects. Empty list if none found. - """ + """ - if selection_mode == None: - selection_mode = None - else: - selection_mode = selection_mode.code + selection_mode = None if selection_mode is None else selection_mode.code stops = [] stops_raw = raw_stop_by_name( @@ -495,13 +572,13 @@ class Client(): lang=lang.code, maxNo=max_number, stopType=stop_type.code, - locationSelectionMode=selection_mode, # type: ignore + locationSelectionMode=selection_mode, # type: ignore coordLat=coord_lat, coordLong=coord_lon, radius=radius, refineId=refine_id, stations=stations, - filterMode=filter_mode.code + filterMode=filter_mode.code, ) find_exception(stops_raw) @@ -515,71 +592,78 @@ class Client(): return stops - def trip_find(self, - lang: Literal[Language.DE, Language.DA, Language.EN, Language.ES, Language.FR, Language.HU, Language.IT, Language.NL, Language.NO, Language.PL, Language.SV, Language.TR] = Language.EN, - - origin_id: Union[str, None] = None, - origin_id_ext: Union[str, None] = None, - origin_coord_lat: Union[str, float, None] = None, - origin_coord_lon: Union[str, float, None] = None, - origin_coord_name: Union[str, None] = None, - - destination_id: Union[str, None] = None, - destination_id_ext: Union[str, None] = None, - destination_coord_lat: Union[str, float, None] = None, - destination_coord_lon: Union[str, float, None] = None, - destination_coord_name: Union[str, None] = None, - - via: Union[str, None] = None, - via_id: Union[str, None] = None, - via_gis: Union[str, None] = None, - via_wait_time: int = 0, - - avoid: Union[str, None] = None, - avoid_id: Union[str, None] = None, - - change_time_percent: int = 100, - change_time_min: Union[int, None] = None, - change_time_max: Union[int, None] = None, - change_time_add: Union[int, None] = None, - change_max: Union[int, None] = None, - - date: Union[str, datetime, None] = None, - time: Union[str, datetime, None] = None, - - search_arrival: bool = False, - - trips_after_time: Union[int, None] = None, - trips_before_time: Union[int, None] = None, - - context: Union[str, None] = None, - - passlist: bool = False, - operators: Union[str, list, None] = None, - - lines: Union[str, list, None] = None, - lineids: Union[str, list, None] = None, - - iv_include: bool = False, - iv_only: bool = False, - - bike_carriage: bool = False, - - passing_points: bool = False, - - real_time_mode: Union[Literal[RealTimeMode.FULL, RealTimeMode.INFOS, RealTimeMode.OFF, RealTimeMode.REALTIME, RealTimeMode.SERVER_DEFAULT], None] = None, - - include_earlier: bool = False, - ict_alternatives: bool = False, - tariff: Union[bool, None] = None, - messages: bool = False, - frequency: bool = True - ) -> List[Trip]: + def trip_find( + self, + lang: Literal[ + Language.DE, + Language.DA, + Language.EN, + Language.ES, + Language.FR, + Language.HU, + Language.IT, + Language.NL, + Language.NO, + Language.PL, + Language.SV, + Language.TR, + ] = Language.EN, + origin_id: Union[str, None] = None, + origin_id_ext: Union[str, None] = None, + origin_coord_lat: Union[str, float, None] = None, + origin_coord_lon: Union[str, float, None] = None, + origin_coord_name: Union[str, None] = None, + destination_id: Union[str, None] = None, + destination_id_ext: Union[str, None] = None, + destination_coord_lat: Union[str, float, None] = None, + destination_coord_lon: Union[str, float, None] = None, + destination_coord_name: Union[str, None] = None, + via: Union[str, None] = None, + via_id: Union[str, None] = None, + via_gis: Union[str, None] = None, + via_wait_time: int = 0, + avoid: Union[str, None] = None, + avoid_id: Union[str, None] = None, + change_time_percent: int = 100, + change_time_min: Union[int, None] = None, + change_time_max: Union[int, None] = None, + change_time_add: Union[int, None] = None, + change_max: Union[int, None] = None, + date: Union[str, datetime, None] = None, + time: Union[str, datetime, None] = None, + search_arrival: bool = False, + trips_after_time: Union[int, None] = None, + trips_before_time: Union[int, None] = None, + context: Union[str, None] = None, + passlist: bool = False, + operators: Union[str, list, None] = None, + lines: Union[str, list, None] = None, + lineids: Union[str, list, None] = None, + iv_include: bool = False, + iv_only: bool = False, + bike_carriage: bool = False, + passing_points: bool = False, + real_time_mode: Union[ + Literal[ + RealTimeMode.FULL, + RealTimeMode.INFOS, + RealTimeMode.OFF, + RealTimeMode.REALTIME, + RealTimeMode.SERVER_DEFAULT, + ], + None, + ] = None, + include_earlier: bool = False, + ict_alternatives: bool = False, + tariff: Union[bool, None] = None, + messages: bool = False, + frequency: bool = True, + ) -> 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. - More detailed request is available as `raw.trip_find()`, however returns `dict` instead of `List[Trip]`. + More detailed request is available as `raw.trip_find()`, however returns `dict` instead of `List[Trip]`. ### Args: * lang (`Literal[Language.DE, Language.DA, Language.EN, Language.ES, Language.FR, Language.HU, Language.IT, Language.NL, Language.NO, Language.PL, Language.SV, Language.TR]`, **optional**): The language of response. Defaults to `Language.EN`. @@ -627,105 +711,94 @@ class Client(): ### Returns: * List[Trip]: List of `Trip` objects. Empty list if none found. - """ + """ - if real_time_mode == None: - real_time_mode = None - else: - real_time_mode = real_time_mode.code + real_time_mode = None if real_time_mode is None else real_time_mode.code trips = [] trips_raw = raw_trip_find( - accessId=self.access_id, lang=lang.code, - originId=origin_id, originExtId=origin_id_ext, originCoordLat=origin_coord_lat, originCoordLong=origin_coord_lon, originCoordName=origin_coord_name, - destId=destination_id, destExtId=destination_id_ext, destCoordLat=destination_coord_lat, destCoordLong=destination_coord_lon, destCoordName=destination_coord_name, - via=via, viaId=via_id, viaGis=via_gis, viaWaitTime=via_wait_time, - avoid=avoid, avoidId=avoid_id, - changeTimePercent=change_time_percent, minChangeTime=change_time_min, maxChangeTime=change_time_max, addChangeTime=change_time_add, maxChange=change_max, - date=date, time=time, - searchForArrival=search_arrival, - numF=trips_after_time, numB=trips_before_time, - context=context, - passlist=passlist, operators=operators, - lines=lines, lineids=lineids, - includeIv=iv_include, ivOnly=iv_only, - bikeCarriage=bike_carriage, - showPassingPoints=passing_points, - - rtMode=real_time_mode, # type: ignore - + rtMode=real_time_mode, # type: ignore includeEarlier=include_earlier, withICTAlternatives=ict_alternatives, tariff=tariff, trafficMessages=messages, - withFreq=frequency + withFreq=frequency, ) find_exception(trips_raw) if "Trip" in trips_raw: - for trip in trips_raw["Trip"]: - trips.append(Trip(trip)) + trips.extend(Trip(trip) for trip in trips_raw["Trip"]) return trips - def trip_recon(self, - context: Union[str, Trip], - date: Union[str, datetime, None] = None, - match_real_time: Union[bool, None] = None, - enable_replacements: Union[bool, None] = None, - arrival_dev_lower: Union[int, None] = None, - arrival_dev_upper: Union[int, None] = None, - departure_dev_lower: Union[int, None] = None, - departure_dev_upper: Union[int, None] = None, - passlist: bool = False, - passing_points: bool = False, - real_time_mode: Union[Literal[RealTimeMode.FULL, RealTimeMode.INFOS, RealTimeMode.OFF, RealTimeMode.REALTIME, RealTimeMode.SERVER_DEFAULT], None] = None, - tariff: Union[bool, None] = None, - messages: bool = False - ) -> List[Trip]: + def trip_recon( + self, + context: Union[str, Trip], + date: Union[str, datetime, None] = None, + match_real_time: Union[bool, None] = None, + enable_replacements: Union[bool, None] = None, + arrival_dev_lower: Union[int, None] = None, + arrival_dev_upper: Union[int, None] = None, + departure_dev_lower: Union[int, None] = None, + departure_dev_upper: Union[int, None] = None, + passlist: bool = False, + passing_points: bool = False, + real_time_mode: Union[ + Literal[ + RealTimeMode.FULL, + RealTimeMode.INFOS, + RealTimeMode.OFF, + RealTimeMode.REALTIME, + RealTimeMode.SERVER_DEFAULT, + ], + None, + ] = None, + tariff: Union[bool, None] = None, + messages: bool = False, + ) -> List[Trip]: """Reconstructing a trip can be achieved using the reconstruction context provided by any trip result in the `ctx_recon` attribute of `Trip` object. The result will be a true copy of the original trip search result given - that the underlying data did not change. + that the underlying data did not change. - More detailed request is available as `raw.trip_recon()`, however returns `dict` instead of `List[Trip]`. + More detailed request is available as `raw.trip_recon()`, however returns `dict` instead of `List[Trip]`. ### Args: * context (`Union[str, Journey]`): Specifies the reconstruction context. @@ -744,20 +817,20 @@ class Client(): ### Returns: * List[Trip]: List of `Trip` objects. Empty list if none found. - """ - + """ + if real_time_mode == None: real_time_mode = None else: real_time_mode = real_time_mode.code - + if isinstance(context, Trip): context = context.ctx_recon trips = [] trips_raw = raw_trip_recon( accessId=self.access_id, - ctx=context, # type: ignore + ctx=context, # type: ignore date=date, matchRtType=match_real_time, enableReplacements=enable_replacements, @@ -767,7 +840,7 @@ class Client(): depU=departure_dev_upper, passlist=passlist, showPassingPoints=passing_points, - rtMode=real_time_mode, # type: ignore + rtMode=real_time_mode, # type: ignore tariff=tariff, trafficMessages=messages, ) @@ -775,7 +848,6 @@ class Client(): find_exception(trips_raw) if "Trip" in trips_raw: - for trip in trips_raw["Trip"]: - trips.append(Trip(trip)) + trips.extend(Trip(trip) for trip in trips_raw["Trip"]) - return trips \ No newline at end of file + return trips diff --git a/src/pyrmv/classes/Gis.py b/src/pyrmv/classes/Gis.py index b0c3990..ff972b6 100644 --- a/src/pyrmv/classes/Gis.py +++ b/src/pyrmv/classes/Gis.py @@ -1,11 +1,13 @@ +from typing import Any, Mapping + from isodate import parse_duration -class Gis(): - """Gis object.""" - - def __init__(self, ref: str, route: dict): - + +class Gis: + """Gis object.""" + + def __init__(self, ref: str, route: Mapping[str, Any]): self.ref = ref self.dist = route["dist"] self.duration = parse_duration(route["durS"]) - self.geo = route["dirGeo"] \ No newline at end of file + self.geo = route["dirGeo"] diff --git a/src/pyrmv/classes/Journey.py b/src/pyrmv/classes/Journey.py index 2f08d7e..a2deca4 100644 --- a/src/pyrmv/classes/Journey.py +++ b/src/pyrmv/classes/Journey.py @@ -1,22 +1,22 @@ -from pyrmv.classes.Stop import Stop +from typing import Any, Mapping + from pyrmv.classes.Message import Message +from pyrmv.classes.Stop import Stop -class Journey(): - """Journey object.""" - def __init__(self, data: dict): +class Journey: + """Journey object.""" + + def __init__(self, data: Mapping[str, Any]): self.stops = [] self.ref = data["ref"] self.direction = data["Directions"]["Direction"][0]["value"] self.direction_flag = data["Directions"]["Direction"][0]["flag"] + self.stops.extend(Stop(stop) for stop in data["Stops"]["Stop"]) self.messages = [] - self.stops.extend(Stop(stop) for stop in data["Stops"]["Stop"]) - if "Messages" in data: - self.messages.extend( - Message(message) for message in data["Messages"]["Message"] - ) + self.messages.extend(Message(message) for message in data["Messages"]["Message"]) def __str__(self) -> str: - return f"Journey with total of {len(self.stops)} stops and {len(self.messages)} messages heading {self.direction} ({self.direction_flag})" \ No newline at end of file + return f"Journey with total of {len(self.stops)} stops and {len(self.messages)} messages heading {self.direction} ({self.direction_flag})" diff --git a/src/pyrmv/classes/Leg.py b/src/pyrmv/classes/Leg.py index 7de63bc..be3594d 100644 --- a/src/pyrmv/classes/Leg.py +++ b/src/pyrmv/classes/Leg.py @@ -1,43 +1,29 @@ +from typing import Any, Mapping + +from isodate import parse_duration + from pyrmv.classes.Gis import Gis from pyrmv.classes.Message import Message from pyrmv.classes.Stop import StopTrip -from isodate import parse_duration -class Leg(): - """Trip leg object.""" - - def __init__(self, data: dict): +class Leg: + """Trip leg object.""" + + def __init__(self, data: Mapping[str, Any]): self.origin = StopTrip(data["Origin"]) self.destination = StopTrip(data["Destination"]) - - if "GisRef" in data: - self.gis = Gis(data["GisRef"]["ref"], data["GisRoute"]) - else: - self.gis = None - + self.gis = ( + None if "GisRef" not in data else Gis(data["GisRef"]["ref"], data["GisRoute"]) + ) + self.messages = [] self.index = data["idx"] self.name = data["name"] self.type = data["type"] - - if "direction" in data: - self.direction = data["direction"] - else: - self.direction = None - - self.messages = [] - if "Messages" in data: - for message in data["Messages"]["Message"]: - self.messages.append(Message(message)) - - if "number" in data: - self.number = data["number"] - else: - self.number = None - + self.direction = data.get("direction") + self.number = data.get("number") self.duration = parse_duration(data["duration"]) - - if "dist" in data: - self.distance = data["dist"] - else: - self.distance = None \ No newline at end of file + self.distance = data.get("dist") + + if "Messages" in data: + self.messages.extend(Message(message) for message in data["Messages"]["Message"]) diff --git a/src/pyrmv/classes/Message.py b/src/pyrmv/classes/Message.py index f93754d..4b151bc 100644 --- a/src/pyrmv/classes/Message.py +++ b/src/pyrmv/classes/Message.py @@ -1,25 +1,29 @@ -from pyrmv.classes.Stop import Stop from datetime import datetime +from typing import Any, Mapping + from isodate import parse_duration -class Url(): - """Traffic message channel url object.""" +from pyrmv.classes.Stop import Stop - def __init__(self, data: dict) -> None: + +class Url: + """Traffic message channel url object.""" + + def __init__(self, data: Mapping[str, Any]) -> None: self.name = data["name"] self.url = data["url"] def __str__(self) -> str: return f"{self.name}: {self.url}" -class Channel(): - """Traffic message channel object.""" - def __init__(self, data: dict) -> None: +class Channel: + """Traffic message channel object.""" + + def __init__(self, data: Mapping[str, Any]) -> None: self.name = data["name"] url = [] - for link in url: - url.append(Url(link)) + url.extend(Url(link) for link in url) self.url = url self.time_start = datetime.strptime(data["validFromTime"], "%H:%M:%S") self.date_start = datetime.strptime(data["validFromDate"], "%Y-%m-%d") @@ -30,30 +34,17 @@ class Channel(): return f"{self.name}: from {self.time_start} {self.date_start} until {self.time_end} {self.date_end}" -class Message(): - """Traffic message object.""" - - def __init__(self, data: dict) -> None: +class Message: + """Traffic message object.""" + def __init__(self, data: Mapping[str, Any]) -> None: self.affected_stops = [] - if "affectedStops" in data: - for stop in data["affectedStops"]["StopLocation"]: - self.affected_stops.append(Stop(stop)) - - if "validFromStop" in data: - self.valid_from_stop = Stop(data["validFromStop"]) - else: - self.valid_from_stop = None - - if "validToStop" in data: - self.valid_to_stop = Stop(data["validToStop"]) - else: - self.valid_to_stop = None - + self.valid_from_stop = ( + None if "validFromStop" not in data else Stop(data["validFromStop"]) + ) + self.valid_to_stop = None if "validToStop" not in data else Stop(data["validToStop"]) self.channels = [] - for channel in data["channel"]: - self.channels.append(Channel(channel)) - + self.channels.extend(Channel(channel) for channel in data["channel"]) self.id = data["id"] self.active = data["act"] self.head = data["head"] @@ -74,11 +65,12 @@ class Message(): self.date_modified = datetime.strptime(data["modDate"], "%Y-%m-%d") self.daily_start = datetime.strptime(data["dailyStartingAt"], "%H:%M:%S") self.daily_duration = parse_duration(data["dailyDuration"]) + self.base_type = data["baseType"] if "baseType" in data else None - if "baseType" in data: - self.base_type = data["baseType"] - else: - self.base_type = None + if "affectedStops" in data: + self.affected_stops.extend( + Stop(stop) for stop in data["affectedStops"]["StopLocation"] + ) def __str__(self) -> str: - return f"{self.base_type} message with priority {self.products} valid from {self.time_start.time()} {self.date_start.date()} until {self.time_end.time()} {self.date_end.date()}: {self.head} - {self.lead}" \ No newline at end of file + return f"{self.base_type} message with priority {self.products} valid from {self.time_start.time()} {self.date_start.date()} until {self.time_end.time()} {self.date_end.date()}: {self.head} - {self.lead}" diff --git a/src/pyrmv/classes/Stop.py b/src/pyrmv/classes/Stop.py index fbd1d77..e197e1c 100644 --- a/src/pyrmv/classes/Stop.py +++ b/src/pyrmv/classes/Stop.py @@ -1,54 +1,32 @@ from datetime import datetime -class Stop(): - """Stop object.""" - - def __init__(self, data: dict): +class Stop: + """Stop object.""" + def __init__(self, data: dict): self.name = data["name"] self.id = data["id"] - - if "extId" in data: - self.ext_id = data["extId"] - else: - self.ext_id = None - - if "description" in data: - self.description = data["description"] - else: - self.description = None - + self.ext_id = data.get("extId") + self.description = data.get("description") self.lon = data["lon"] self.lat = data["lat"] - - if "routeIdx" in data: - self.route_index = data["routeIdx"] - else: - self.route_index = None - - if "arrTrack" in data: - self.track_arrival = data["arrTrack"] - else: - self.track_arrival = None - - if "depTrack" in data: - self.track_departure = data["depTrack"] - else: - self.track_departure = None + self.route_index = data.get("routeIdx") + self.track_arrival = data.get("arrTrack") + self.track_departure = data.get("depTrack") def __str__(self) -> str: return f"Stop {self.name} at {self.lon}, {self.lat}" + class StopTrip(Stop): - """Trip stop object. It's like a Stop object, but with a date and time.""" + """Trip stop object. It's like a Stop object, but with a date and time.""" def __init__(self, data: dict): - self.type = data["type"] self.date = datetime.strptime(data["date"], "%Y-%m-%d") self.time = datetime.strptime(data["time"], "%H:%M:%S") super().__init__(data) def __str__(self) -> str: - return f"Stop {self.name} at {self.lon}, {self.lat} at {self.time.time()} {self.date.date()}" \ No newline at end of file + return f"Stop {self.name} at {self.lon}, {self.lat} at {self.time.time()} {self.date.date()}" diff --git a/src/pyrmv/classes/Trip.py b/src/pyrmv/classes/Trip.py index 7422d12..809d85e 100644 --- a/src/pyrmv/classes/Trip.py +++ b/src/pyrmv/classes/Trip.py @@ -1,37 +1,28 @@ -from pyrmv.classes.Leg import Leg -from pyrmv.classes.Stop import StopTrip from isodate import parse_duration -class Trip(): - """Trip object.""" - +from pyrmv.classes.Leg import Leg +from pyrmv.classes.Stop import StopTrip + + +class Trip: + """Trip object.""" + def __init__(self, data: dict): - self.raw_data = data self.origin = StopTrip(data["Origin"]) self.destination = StopTrip(data["Destination"]) - self.legs = [] - for leg in data["LegList"]["Leg"]: - self.legs.append(Leg(leg)) - + self.legs.extend(Leg(leg) for leg in data["LegList"]["Leg"]) self.calculation = data["calculation"] self.index = data["idx"] self.id = data["tripId"] self.ctx_recon = data["ctxRecon"] self.duration = parse_duration(data["duration"]) - - if "rtDuration" in data: - self.real_time_duration = parse_duration(data["rtDuration"]) - else: - self.real_time_duration = None - + self.real_time_duration = ( + None if "rtDuration" not in data else parse_duration(data["rtDuration"]) + ) self.checksum = data["checksum"] - - if "transferCount" in data: - self.transfer_count = data["transferCount"] - else: - self.transfer_count = 0 + self.transfer_count = data.get("transferCount", 0) def __str__(self) -> str: - return f"Trip from {self.origin.name} to {self.destination.name} lasting {self.duration} ({self.real_time_duration}) with {len(self.legs)} legs and {self.transfer_count} transfers" \ No newline at end of file + return f"Trip from {self.origin.name} to {self.destination.name} lasting {self.duration} ({self.real_time_duration}) with {len(self.legs)} legs and {self.transfer_count} transfers" diff --git a/src/pyrmv/classes/__init__.py b/src/pyrmv/classes/__init__.py index c0d5e4b..ab6bb04 100644 --- a/src/pyrmv/classes/__init__.py +++ b/src/pyrmv/classes/__init__.py @@ -1,9 +1,9 @@ from .Board import BoardArrival, BoardDeparture, LineArrival, LineDeparture +from .Client import Client from .Gis import Gis from .Journey import Journey from .Leg import Leg -from .Message import Message, Channel, Url +from .Message import Channel, Message, Url from .Stop import Stop, StopTrip from .Ticket import Ticket from .Trip import Trip -from .Client import Client \ No newline at end of file diff --git a/src/pyrmv/const/__init__.py b/src/pyrmv/const/__init__.py index e69de29..ffb3cc7 100644 --- a/src/pyrmv/const/__init__.py +++ b/src/pyrmv/const/__init__.py @@ -0,0 +1 @@ +from .product import PRODUCTS diff --git a/src/pyrmv/const/product.py b/src/pyrmv/const/product.py index fe6087e..84794ba 100644 --- a/src/pyrmv/const/product.py +++ b/src/pyrmv/const/product.py @@ -17,4 +17,4 @@ PRODUCTS: Dict[str, int] = { "ferry": 256, "taxi": 512, "bahn": 1024, -} \ No newline at end of file +} diff --git a/src/pyrmv/enums/__init__.py b/src/pyrmv/enums/__init__.py index 425b262..92cffb6 100644 --- a/src/pyrmv/enums/__init__.py +++ b/src/pyrmv/enums/__init__.py @@ -1,10 +1,10 @@ -from .product import Product -from .rt_mode import RealTimeMode +from .aff_journey_mode import AffectedJourneyMode +from .aff_journey_stop_mode import AffectedJourneyStopMode +from .board_type import BoardArrivalType, BoardDepartureType +from .filter_mode import FilterMode from .lang import Language from .location_type import LocationType -from .selection_mode import SelectionMode +from .product import Product +from .rt_mode import RealTimeMode from .search_mode import SearchMode -from .filter_mode import FilterMode -from .board_type import BoardArrivalType, BoardDepartureType -from .aff_journey_mode import AffectedJourneyMode -from .aff_journey_stop_mode import AffectedJourneyStopMode \ No newline at end of file +from .selection_mode import SelectionMode diff --git a/src/pyrmv/enums/aff_journey_mode.py b/src/pyrmv/enums/aff_journey_mode.py index cf67c93..15d7e96 100644 --- a/src/pyrmv/enums/aff_journey_mode.py +++ b/src/pyrmv/enums/aff_journey_mode.py @@ -1,6 +1,8 @@ from enum import auto + from .auto_name import AutoName + class AffectedJourneyMode(AutoName): """Enumeration used to declare types of HIM search modes.""" @@ -8,4 +10,4 @@ class AffectedJourneyMode(AutoName): "Return affected journeys." OFF = auto() - "Do not return affected journeys." \ No newline at end of file + "Do not return affected journeys." diff --git a/src/pyrmv/enums/aff_journey_stop_mode.py b/src/pyrmv/enums/aff_journey_stop_mode.py index 85066ae..d21034f 100644 --- a/src/pyrmv/enums/aff_journey_stop_mode.py +++ b/src/pyrmv/enums/aff_journey_stop_mode.py @@ -1,6 +1,8 @@ from enum import auto + from .auto_name import AutoName + class AffectedJourneyStopMode(AutoName): """Enumeration used to declare types of affected journey stops return modes.""" @@ -11,4 +13,4 @@ class AffectedJourneyStopMode(AutoName): "Return important stops of affected journeys." OFF = auto() - "Do not return stops of affected journeys." \ No newline at end of file + "Do not return stops of affected journeys." diff --git a/src/pyrmv/enums/auto_name.py b/src/pyrmv/enums/auto_name.py index fa8a5fe..d0c07cf 100644 --- a/src/pyrmv/enums/auto_name.py +++ b/src/pyrmv/enums/auto_name.py @@ -2,7 +2,9 @@ # https://github.com/pyrogram/pyrogram/blob/master/pyrogram/enums/auto_name.py from enum import Enum -from pyrmv.const.product import PRODUCTS + +from pyrmv.const import PRODUCTS + class AutoName(Enum): def __init__(self, code) -> None: @@ -14,6 +16,7 @@ class AutoName(Enum): def __repr__(self): return f"pyrmv.enums.{self}" + class AutoNameProduct(AutoName): def __init__(self, code) -> None: - self.code = PRODUCTS[code] \ No newline at end of file + self.code = PRODUCTS[code] diff --git a/src/pyrmv/enums/board_type.py b/src/pyrmv/enums/board_type.py index 426da65..0d4ccfa 100644 --- a/src/pyrmv/enums/board_type.py +++ b/src/pyrmv/enums/board_type.py @@ -1,9 +1,11 @@ from enum import auto + from .auto_name import AutoName + class BoardArrivalType(AutoName): - """Enumeration used to declare types of arrival board. - + """Enumeration used to declare types of arrival board. + * 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 @@ -22,9 +24,10 @@ class BoardArrivalType(AutoName): ARR_STATION = auto() "Arrival board with all journeys at any masts of the requested station" + class BoardDepartureType(AutoName): - """Enumeration used to declare types of departure board. - + """Enumeration used to declare types of departure board. + * 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 @@ -41,4 +44,4 @@ class BoardDepartureType(AutoName): "Departure board at mast" DEP_STATION = auto() - "Departure board with all journeys at any masts of the requested station" \ No newline at end of file + "Departure board with all journeys at any masts of the requested station" diff --git a/src/pyrmv/enums/filter_mode.py b/src/pyrmv/enums/filter_mode.py index 192c3c0..b7ce25d 100644 --- a/src/pyrmv/enums/filter_mode.py +++ b/src/pyrmv/enums/filter_mode.py @@ -1,9 +1,11 @@ from enum import auto + from .auto_name import AutoName + class FilterMode(AutoName): - """Enumeration used to declare filters for nearby searches. - + """Enumeration used to declare filters for nearby searches. + * DIST_PERI - Accentuate matches. Matches in the radius are first * EXCL_PERI - Returns matches inside the radius only * SLCT_PERI - Matches in the radius are excluded. Returns matches outside the radius only @@ -16,4 +18,4 @@ class FilterMode(AutoName): "Returns matches inside the radius only." SLCT_PERI = auto() - "Matches in the radius are excluded. Returns matches outside the radius only." \ No newline at end of file + "Matches in the radius are excluded. Returns matches outside the radius only." diff --git a/src/pyrmv/enums/lang.py b/src/pyrmv/enums/lang.py index f87c13f..57cf286 100644 --- a/src/pyrmv/enums/lang.py +++ b/src/pyrmv/enums/lang.py @@ -1,24 +1,44 @@ from enum import auto + from .auto_name import AutoName + class Language(AutoName): """Enumeration used to declare locales as ISO-3166 codes (but only available in HAFAS ones).""" - DE = auto() - "German" + AR = auto() + "Arabic" + + CA = auto() + "Catalan, Valencian" DA = auto() "Danish" + DE = auto() + "German" + + EL = auto() + "Greek" + EN = auto() "English" ES = auto() "Spanish" + FI = auto() + "Finnish" + FR = auto() "French" + HI = auto() + "Hindi" + + HR = auto() + "Croatian" + HU = auto() "Hungarian" @@ -34,8 +54,26 @@ class Language(AutoName): PL = auto() "Polish" + RU = auto() + "Russian" + + SK = auto() + "Slovak" + + SL = auto() + "Slovenian" + SV = auto() "Swedish" + TL = auto() + "Tagalog" + TR = auto() - "Turkish" \ No newline at end of file + "Turkish" + + UR = auto() + "Urdu" + + ZH = auto() + "Chinese" diff --git a/src/pyrmv/enums/location_type.py b/src/pyrmv/enums/location_type.py index 9631444..45e4a94 100644 --- a/src/pyrmv/enums/location_type.py +++ b/src/pyrmv/enums/location_type.py @@ -1,9 +1,11 @@ from enum import auto + from .auto_name import AutoName + class LocationType(AutoName): - """Enumeration used to declare types of location filter. - + """Enumeration used to declare types of location filter. + * S - Search for station/stops only * A - Search for addresses only * P - Search for POIs only @@ -15,7 +17,7 @@ class LocationType(AutoName): * SPE - Search for stations/stops, POIs and entrypoints Note that not all location types may be available for a certain methods. - Make sure that selected type is supported by method by looking up + Make sure that selected type is supported by method by looking up its acceptable types in argument description. """ @@ -44,4 +46,4 @@ class LocationType(AutoName): "Search in all existing location pools" SPE = auto() - "Search for stations/stops, POIs and entrypoints" \ No newline at end of file + "Search for stations/stops, POIs and entrypoints" diff --git a/src/pyrmv/enums/product.py b/src/pyrmv/enums/product.py index 45bf631..c0c6cb6 100644 --- a/src/pyrmv/enums/product.py +++ b/src/pyrmv/enums/product.py @@ -1,6 +1,8 @@ from enum import auto + from .auto_name import AutoNameProduct + class Product(AutoNameProduct): """Enumeration used to declare types of transport.""" @@ -11,7 +13,7 @@ class Product(AutoNameProduct): "InterCity (commonly abbreviated IC on timetables and tickets) is the classification applied to certain long-distance passenger train services in Europe. Such trains (in contrast to regional, local, or commuter trains) generally call at major stations only." EC = auto() - "EuroCity, abbreviated as EC, is a cross-border train category within the European inter-city rail network. In contrast to trains allocated to the lower-level \"IC\" (InterCity) category, EC trains are international services that meet 20 criteria covering comfort, speed, food service, and cleanliness." + 'EuroCity, abbreviated as EC, is a cross-border train category within the European inter-city rail network. In contrast to trains allocated to the lower-level "IC" (InterCity) category, EC trains are international services that meet 20 criteria covering comfort, speed, food service, and cleanliness.' R = auto() "Regional rail, also known as local trains and stopping trains, are passenger rail services that operate between towns and cities. These trains operate with more stops over shorter distances than inter-city rail, but fewer stops and faster service than commuter rail." @@ -44,4 +46,4 @@ class Product(AutoNameProduct): "A taxi, also known as a taxicab or simply a cab, is a type of vehicle for hire with a driver, used by a single passenger or small group of passengers, often for a non-shared ride." BAHN = auto() - "Rail transport (also known as train transport) is a means of transport that transfers passengers and goods on wheeled vehicles running on rails, which are located on tracks. In contrast to road transport, where the vehicles run on a prepared flat surface, rail vehicles (rolling stock) are directionally guided by the tracks on which they run." \ No newline at end of file + "Rail transport (also known as train transport) is a means of transport that transfers passengers and goods on wheeled vehicles running on rails, which are located on tracks. In contrast to road transport, where the vehicles run on a prepared flat surface, rail vehicles (rolling stock) are directionally guided by the tracks on which they run." diff --git a/src/pyrmv/enums/rt_mode.py b/src/pyrmv/enums/rt_mode.py index 97c1b4b..c5e5f32 100644 --- a/src/pyrmv/enums/rt_mode.py +++ b/src/pyrmv/enums/rt_mode.py @@ -1,6 +1,8 @@ from enum import auto + from .auto_name import AutoName + class RealTimeMode(AutoName): """Enumeration used to declare types of real-time traffic.""" @@ -17,4 +19,4 @@ class RealTimeMode(AutoName): "Search on real-time data: Connections are computed on the basis of real-time data, using planned schedule only whenever no real-time data is available. All connections computed are feasible with respect to the currently known real-time situation. Additional trains (supplied via real-time feed) will be found if these are part of a fast, comfortable, or direct connection (or economic connection, if economic search is activated)." SERVER_DEFAULT = auto() - "One of the other real-times modes used by default for RMV." \ No newline at end of file + "One of the other real-times modes used by default for RMV." diff --git a/src/pyrmv/enums/search_mode.py b/src/pyrmv/enums/search_mode.py index e6956b9..150aafc 100644 --- a/src/pyrmv/enums/search_mode.py +++ b/src/pyrmv/enums/search_mode.py @@ -1,6 +1,8 @@ from enum import auto + from .auto_name import AutoName + class SearchMode(AutoName): """Enumeration used to declare types of HIM search modes.""" @@ -11,4 +13,4 @@ class SearchMode(AutoName): "Iterate over all HIM messages available." TFMATCH = auto() - "Uses filters defined `metas` parameter." \ No newline at end of file + "Uses filters defined `metas` parameter." diff --git a/src/pyrmv/enums/selection_mode.py b/src/pyrmv/enums/selection_mode.py index 4bbd9fb..7724be6 100644 --- a/src/pyrmv/enums/selection_mode.py +++ b/src/pyrmv/enums/selection_mode.py @@ -1,9 +1,11 @@ from enum import auto + from .auto_name import AutoName + class SelectionMode(AutoName): """Enumeration used to declare location selection modes. - + * SLCT_A - Selectable * SLCT_N - Not selectable """ @@ -12,4 +14,4 @@ class SelectionMode(AutoName): "Selectable" SLCT_N = auto() - "Not selectable" \ No newline at end of file + "Not selectable" diff --git a/src/pyrmv/errors/__init__.py b/src/pyrmv/errors/__init__.py index e69de29..1671d6b 100644 --- a/src/pyrmv/errors/__init__.py +++ b/src/pyrmv/errors/__init__.py @@ -0,0 +1,7 @@ +from .api_errors import * +from .int_errors import * +from .not_ready import * +from .ps_errors import * +from .sot_errors import * +from .svc_errors import * +from .unknown_error import * diff --git a/src/pyrmv/errors/api_errors.py b/src/pyrmv/errors/api_errors.py index 8119d16..1f057ba 100644 --- a/src/pyrmv/errors/api_errors.py +++ b/src/pyrmv/errors/api_errors.py @@ -2,32 +2,38 @@ class ApiAuthError(Exception): """ Access denied for accessId provided. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): return self.__doc__ + class ApiQuotaError(Exception): """ Quota exceeded for accessId provided. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): return self.__doc__ + class ApiTooManyRequests(Exception): """ Too many requests. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): return self.__doc__ + class ApiParamError(Exception): """Exception raised for errors in the input arguments. @@ -42,14 +48,16 @@ class ApiParamError(Exception): super().__init__(self.errorText) def __str__(self): - return f'{self.errorCode} -> {self.errorText}' + return f"{self.errorCode} -> {self.errorText}" + class ApiFormatError(Exception): """ Response format not supported. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): - return self.__doc__ \ No newline at end of file + return self.__doc__ diff --git a/src/pyrmv/errors/int_errors.py b/src/pyrmv/errors/int_errors.py index b92dde8..cc5fc44 100644 --- a/src/pyrmv/errors/int_errors.py +++ b/src/pyrmv/errors/int_errors.py @@ -2,28 +2,33 @@ class IntError(Exception): """ Internal RMV server error. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): return self.__doc__ + class IntGatewayError(Exception): """ Communication error with RMV backend systems. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): return self.__doc__ + class IntTimeoutError(Exception): """ Timeout during service processing. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): - return self.__doc__ \ No newline at end of file + return self.__doc__ diff --git a/src/pyrmv/errors/not_ready.py b/src/pyrmv/errors/not_ready.py index 0ed3c03..b1c428e 100644 --- a/src/pyrmv/errors/not_ready.py +++ b/src/pyrmv/errors/not_ready.py @@ -2,8 +2,9 @@ class NotReadyYetError(Exception): """ This method is not finished yet. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): - return self.__doc__ \ No newline at end of file + return self.__doc__ diff --git a/src/pyrmv/errors/ps_errors.py b/src/pyrmv/errors/ps_errors.py index 83ac32c..adc8392 100644 --- a/src/pyrmv/errors/ps_errors.py +++ b/src/pyrmv/errors/ps_errors.py @@ -6,8 +6,9 @@ class PsIncorrectParamError(Exception): forward/backward from the end/beginning of the connection. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): - return self.__doc__ \ No newline at end of file + return self.__doc__ diff --git a/src/pyrmv/errors/sot_errors.py b/src/pyrmv/errors/sot_errors.py index 6b2459f..cc99f9b 100644 --- a/src/pyrmv/errors/sot_errors.py +++ b/src/pyrmv/errors/sot_errors.py @@ -2,48 +2,57 @@ class SotAlreadyArrivedError(Exception): """ Trip already arrived. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): return self.__doc__ + class SotNotStartedError(Exception): """ Trip not started. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): return self.__doc__ + class SotCancelledError(Exception): """ Trip cancelled. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): return self.__doc__ + class SotAllTrainsFilteredError(Exception): """ All trips filtered. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): return self.__doc__ + class SotStayOnTripError(Exception): """ No change. Stay on trip. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): - return self.__doc__ \ No newline at end of file + return self.__doc__ diff --git a/src/pyrmv/errors/svc_errors.py b/src/pyrmv/errors/svc_errors.py index c2073ad..b28b7e6 100644 --- a/src/pyrmv/errors/svc_errors.py +++ b/src/pyrmv/errors/svc_errors.py @@ -12,134 +12,160 @@ class SvcParamError(Exception): super().__init__(self.errorText) def __str__(self): - return f'{self.errorCode} -> {self.errorText}' + return f"{self.errorCode} -> {self.errorText}" + class SvcLocationError(Exception): """ Location missing or invalid. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): return self.__doc__ + class SvcLocationArrivalError(Exception): """ Arrival location missing or invalid. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): return self.__doc__ + class SvcLocationDepartureError(Exception): """ Departure location missing or invalid. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): return self.__doc__ + class SvcLocationViaError(Exception): """ Unknown change stop. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): return self.__doc__ + class SvcLocationEqualError(Exception): """ Origin/destination or vias are equal. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): return self.__doc__ + class SvcLocationNearError(Exception): """ Origin and destination are too close. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): return self.__doc__ + class SvcDatetimeError(Exception): """ Date/time are missing or invalid. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): return self.__doc__ + class SvcDatetimePeriodError(Exception): """ Date/time are not in timetable or allowed period. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): return self.__doc__ + class SvcProductError(Exception): """ Product field missing or invalid. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): return self.__doc__ + class SvcContextError(Exception): """ Context is invalid. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): return self.__doc__ + class SvcNoResultError(Exception): """ No result found. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): return self.__doc__ + class SvcSearchError(Exception): """ Unsuccessful search. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): return self.__doc__ + class SvcNoMatchError(Exception): """ No match found. """ + def __init__(self): super().__init__(self.__doc__) def __str__(self): - return self.__doc__ \ No newline at end of file + return self.__doc__ diff --git a/src/pyrmv/errors/unknown_error.py b/src/pyrmv/errors/unknown_error.py index 0f56ca6..786ffae 100644 --- a/src/pyrmv/errors/unknown_error.py +++ b/src/pyrmv/errors/unknown_error.py @@ -12,4 +12,4 @@ class UnknownError(Exception): super().__init__(self.errorText) def __str__(self): - return f'{self.errorCode} -> {self.errorText}' \ No newline at end of file + return f"{self.errorCode} -> {self.errorText}" diff --git a/src/pyrmv/raw/__init__.py b/src/pyrmv/raw/__init__.py index 62fb3fd..3c8c095 100644 --- a/src/pyrmv/raw/__init__.py +++ b/src/pyrmv/raw/__init__.py @@ -5,4 +5,4 @@ from .journey_detail import journey_detail from .stop_by_coords import stop_by_coords from .stop_by_name import stop_by_name from .trip_find import trip_find -from .trip_recon import trip_recon \ No newline at end of file +from .trip_recon import trip_recon diff --git a/src/pyrmv/raw/board_arrival.py b/src/pyrmv/raw/board_arrival.py index 9b8f347..fee0b99 100644 --- a/src/pyrmv/raw/board_arrival.py +++ b/src/pyrmv/raw/board_arrival.py @@ -1,6 +1,7 @@ -from requests import get from datetime import datetime, timedelta from typing import Union + +from requests import get from xmltodict import parse as xmlparse try: @@ -8,32 +9,34 @@ try: except ImportError: from typing_extensions import Literal + # 2.25. Arrival Board (arrivalBoard) -def board_arrival(accessId: str, - json: bool = True, - id: Union[str, None] = None, - extId: Union[str, None] = None, - direction: Union[str, None] = None, - date: Union[str, datetime, None] = None, - time: Union[str, datetime, None] = None, - duration: Union[int, timedelta] = 60, - maxJourneys: int = -1, - products: Union[int, None] = None, - operators: Union[str, list, None] = None, - lines: Union[str, list, None] = None, - filterEquiv: bool = True, - attributes: Union[str, list, None] = None, - platforms: Union[str, list, None] = None, - passlist: bool = False, - boardType: Literal["ARR", "ARR_EQUIVS", "ARR_MAST", "ARR_STATION"] = "ARR" - ) -> dict: +def board_arrival( + accessId: str, + json: bool = True, + id: Union[str, None] = None, + extId: Union[str, None] = None, + direction: Union[str, None] = None, + date: Union[str, datetime, None] = None, + time: Union[str, datetime, None] = None, + duration: Union[int, timedelta] = 60, + maxJourneys: int = -1, + products: Union[int, None] = None, + operators: Union[str, list, None] = None, + lines: Union[str, list, None] = None, + filterEquiv: bool = True, + attributes: Union[str, list, None] = None, + platforms: Union[str, list, None] = None, + 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. + 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. + maximum was overrun. - Read more about this in section 2.25. "Arrival Board (arrivalBoard)" of HAFAS ReST Documentation. + 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). @@ -56,14 +59,10 @@ def board_arrival(accessId: str, ### Returns: * dict: Output from RMV. Dict will contain "errorCode" and "errorText" if exception occurs. - """ - - if json: - headers = {"Accept": "application/json"} - else: - headers = {"Accept": "application/xml"} + """ payload = {} + headers = {"Accept": "application/json"} if json else {"Accept": "application/xml"} for var, val in locals().items(): if str(var) == "date": @@ -74,16 +73,12 @@ def board_arrival(accessId: str, 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 + payload[str(var)] = ( + val.strftime("%H:%M") if isinstance(val, datetime) else val + ) elif str(var) == "duration": if val != None: - if isinstance(val, timedelta): - payload[str(var)] = val.minutes # type: ignore - else: - payload[str(var)] = val + payload[str(var)] = val.minutes if isinstance(val, timedelta) else val elif str(var) == "boardType": if val != None: payload["type"] = val.upper() @@ -93,7 +88,4 @@ def board_arrival(accessId: str, output = get("https://www.rmv.de/hapi/arrivalBoard", params=payload, headers=headers) - if json: - return output.json() - else: - return xmlparse(output.content) \ No newline at end of file + return output.json() if json else xmlparse(output.content) diff --git a/src/pyrmv/raw/board_departure.py b/src/pyrmv/raw/board_departure.py index ed30539..57e76b2 100644 --- a/src/pyrmv/raw/board_departure.py +++ b/src/pyrmv/raw/board_departure.py @@ -1,6 +1,7 @@ -from requests import get from datetime import datetime, timedelta from typing import Union + +from requests import get from xmltodict import parse as xmlparse try: @@ -8,33 +9,35 @@ try: except ImportError: from typing_extensions import Literal + # 2.24. Departure Board (departureBoard) -def board_departure(accessId: str, - json: bool = True, - id: Union[str, None] = None, - extId: Union[str, None] = None, - direction: Union[str, None] = None, - date: Union[str, datetime, None] = None, - time: Union[str, datetime, None] = None, - duration: Union[int, timedelta] = 60, - maxJourneys: int = -1, - products: Union[int, None] = None, - operators: Union[str, list, None] = None, - lines: Union[str, list, None] = None, - filterEquiv: bool = True, - attributes: Union[str, list, None] = None, - platforms: Union[str, list, None] = None, - passlist: bool = False, - boardType: Literal["DEP", "DEP_EQUIVS", "DEP_MAST", "DEP_STATION"] = "DEP" - ) -> dict: +def board_departure( + accessId: str, + json: bool = True, + id: Union[str, None] = None, + extId: Union[str, None] = None, + direction: Union[str, None] = None, + date: Union[str, datetime, None] = None, + time: Union[str, datetime, None] = None, + duration: Union[int, timedelta] = 60, + maxJourneys: int = -1, + products: Union[int, None] = None, + operators: Union[str, list, None] = None, + lines: Union[str, list, None] = None, + filterEquiv: bool = True, + attributes: Union[str, list, None] = None, + platforms: Union[str, list, None] = None, + 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. + 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. + maximum was overrun. - Read more about this in section 2.24. "Departure Board (departureBoard)" of HAFAS ReST Documentation. + 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). @@ -57,14 +60,10 @@ def board_departure(accessId: str, ### Returns: * dict: Output from RMV. Dict will contain "errorCode" and "errorText" if exception occurs. - """ - - if json: - headers = {"Accept": "application/json"} - else: - headers = {"Accept": "application/xml"} + """ payload = {} + headers = {"Accept": "application/json"} if json else {"Accept": "application/xml"} for var, val in locals().items(): if str(var) == "date": @@ -75,16 +74,12 @@ def board_departure(accessId: str, 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 + payload[str(var)] = ( + val.strftime("%H:%M") if isinstance(val, datetime) else val + ) elif str(var) == "duration": if val != None: - if isinstance(val, timedelta): - payload[str(var)] = val.minutes # type: ignore - else: - payload[str(var)] = val + payload[str(var)] = val.minutes if isinstance(val, timedelta) else val elif str(var) == "boardType": if val != None: payload["type"] = val.upper() @@ -94,7 +89,4 @@ def board_departure(accessId: str, output = get("https://www.rmv.de/hapi/departureBoard", params=payload, headers=headers) - if json: - return output.json() - else: - return xmlparse(output.content) \ No newline at end of file + return output.json() if json else xmlparse(output.content) diff --git a/src/pyrmv/raw/him_search.py b/src/pyrmv/raw/him_search.py index 0ebf5fe..ae38717 100644 --- a/src/pyrmv/raw/him_search.py +++ b/src/pyrmv/raw/him_search.py @@ -1,55 +1,58 @@ -from requests import get -from typing import OrderedDict, Union -from xmltodict import parse as xmlparse from datetime import datetime +from typing import OrderedDict, Union -from pyrmv.utility.weekdays_bitmask import weekdays_bitmask +from requests import get +from xmltodict import parse as xmlparse + +from pyrmv.utility import weekdays_bitmask try: from typing import Literal except ImportError: from typing_extensions import Literal + # 2.37. HIM Search (himsearch) -def him_search(accessId: str, - json: bool = True, - dateB: Union[str, datetime, None] = None, - dateE: Union[str, datetime, None] = None, - timeB: Union[str, datetime, None] = None, - timeE: Union[str, datetime, None] = None, - weekdays: Union[str, OrderedDict[str, bool], None] = None, - himIds: Union[str, list, None] = None, - hierarchicalView: bool = False, - operators: Union[str, list, None] = None, - categories: Union[str, list, None] = None, - channels: Union[str, list, None] = None, - companies: Union[str, list, None] = None, - lines: Union[str, list, None] = None, - lineids: Union[str, list, None] = None, - stations: Union[str, list, None] = None, - fromstation: Union[str, None] = None, - tostation: Union[str, None] = None, - bothways: Union[bool, None] = None, - trainnames: Union[str, list, None] = None, - metas: Union[str, list, None] = None, - himcategory: Union[str, None] = None, - himtags: Union[str, list, None] = None, - regions: Union[str, list, None] = None, - himtext: Union[str, list, None] = None, - himtexttags: Union[str, list, None] = None, - additionalfields: Union[str, list, dict, None] = None, - poly: bool = False, - searchmode: Union[Literal["MATCH", "NOMATCH", "TFMATCH"], None] = None, - affectedJourneyMode: Union[Literal["ALL", "OFF"], None] = None, - affectedJourneyStopMode: Union[Literal["ALL", "IMP", "OFF"], None] = None, - orderBy: Union[str, list, None] = None, - minprio: Union[str, int, None] = None, - maxprio: Union[str, int, None] = None - ) -> dict: +def him_search( + accessId: str, + json: bool = True, + dateB: Union[str, datetime, None] = None, + dateE: Union[str, datetime, None] = None, + timeB: Union[str, datetime, None] = None, + timeE: Union[str, datetime, None] = None, + weekdays: Union[str, OrderedDict[str, bool], None] = None, + himIds: Union[str, list, None] = None, + hierarchicalView: bool = False, + operators: Union[str, list, None] = None, + categories: Union[str, list, None] = None, + channels: Union[str, list, None] = None, + companies: Union[str, list, None] = None, + lines: Union[str, list, None] = None, + lineids: Union[str, list, None] = None, + stations: Union[str, list, None] = None, + fromstation: Union[str, None] = None, + tostation: Union[str, None] = None, + bothways: Union[bool, None] = None, + trainnames: Union[str, list, None] = None, + metas: Union[str, list, None] = None, + himcategory: Union[str, None] = None, + himtags: Union[str, list, None] = None, + regions: Union[str, list, None] = None, + himtext: Union[str, list, None] = None, + himtexttags: Union[str, list, None] = None, + additionalfields: Union[str, list, dict, None] = None, + poly: bool = False, + searchmode: Union[Literal["MATCH", "NOMATCH", "TFMATCH"], None] = None, + affectedJourneyMode: Union[Literal["ALL", "OFF"], None] = None, + affectedJourneyStopMode: Union[Literal["ALL", "IMP", "OFF"], None] = None, + orderBy: Union[str, list, None] = None, + minprio: Union[str, int, None] = None, + maxprio: Union[str, int, None] = None, +) -> dict: """The himSearch will return a list of HIM messages if matched by the given criteria as well as affected products if any. - Read more about this in section 2.37. "HIM Search (himsearch)" of HAFAS ReST Documentation. + Read more about this in section 2.37. "HIM Search (himsearch)" 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). @@ -89,28 +92,23 @@ def him_search(accessId: str, ### Returns: * dict: Output from RMV. Dict will contain "errorCode" and "errorText" if exception occurs. - """ - - if json: - headers = {"Accept": "application/json"} - else: - headers = {"Accept": "application/xml"} + """ payload = {} + headers = {"Accept": "application/json"} if json else {"Accept": "application/xml"} for var, val in locals().items(): - if str(var) in ["dateB", "dateE"]: + if str(var) in {"dateB", "dateE"}: if val != None: if isinstance(val, datetime): payload[str(var)] = val.strftime("%Y-%m-%d") else: payload[str(var)] = val - elif str(var) in ["timeB", "timeE"]: + elif str(var) in {"timeB", "timeE"}: if val != None: - if isinstance(val, datetime): - payload[str(var)] = val.strftime("%H:%M") - else: - payload[str(var)] = val + payload[str(var)] = ( + val.strftime("%H:%M") if isinstance(val, datetime) else val + ) elif str(var) == "weekdays": if val != None: if isinstance(val, OrderedDict): @@ -132,7 +130,4 @@ def him_search(accessId: str, output = get("https://www.rmv.de/hapi/himsearch", params=payload, headers=headers) - if json: - return output.json() - else: - return xmlparse(output.content) \ No newline at end of file + return output.json() if json else xmlparse(output.content) diff --git a/src/pyrmv/raw/journey_detail.py b/src/pyrmv/raw/journey_detail.py index 56347a0..6aa9be8 100644 --- a/src/pyrmv/raw/journey_detail.py +++ b/src/pyrmv/raw/journey_detail.py @@ -1,6 +1,7 @@ from datetime import datetime -from requests import get from typing import Union + +from requests import get from xmltodict import parse as xmlparse try: @@ -8,27 +9,29 @@ try: except ImportError: from typing_extensions import Literal + # 2.26. Journey Detail (journeyDetail) -def journey_detail(accessId: str, - id: str, - json: bool = True, - date: Union[str, datetime, None] = None, - poly: bool = False, - polyEnc: Literal["DLT", "GPA", "N"] = "N", - showPassingPoints: bool = False, - rtMode: Union[Literal["FULL", "INFOS", "OFF", "REALTIME", "SERVER_DEFAULT"], None] = None, - fromId: Union[str, None] = None, - fromIdx: Union[int, None] = None, - toId: Union[str, None] = None, - toIdx: Union[int, None] = None, - baim: bool = False - ) -> dict: +def journey_detail( + accessId: str, + id: str, + json: bool = True, + date: Union[str, datetime, None] = None, + poly: bool = False, + polyEnc: Literal["DLT", "GPA", "N"] = "N", + showPassingPoints: bool = False, + rtMode: Union[Literal["FULL", "INFOS", "OFF", "REALTIME", "SERVER_DEFAULT"], None] = None, + fromId: Union[str, None] = None, + fromIdx: Union[int, None] = None, + toId: Union[str, None] = None, + toIdx: Union[int, None] = None, + baim: bool = False, +) -> dict: """The journey_detail method will deliver information about the complete route of a vehicle. The journey identifier is part of a trip or departureBoard response. It contains a list of all stops/stations of this journey including all departure and arrival times (with real-time data if available) and additional information like - specific attributes about facilities and other texts. + specific attributes about facilities and other texts. - Read more about this in section 2.26. "Journey Detail (journeyDetail)" of HAFAS ReST Documentation. + Read more about this in section 2.26. "Journey Detail (journeyDetail)" 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). @@ -47,14 +50,10 @@ def journey_detail(accessId: str, ### Returns: * dict: Output from RMV. Dict will contain "errorCode" and "errorText" if exception occurs. - """ - - if json: - headers = {"Accept": "application/json"} - else: - headers = {"Accept": "application/xml"} + """ payload = {} + headers = {"Accept": "application/json"} if json else {"Accept": "application/xml"} for var, val in locals().items(): if str(var) == "rtMode": @@ -66,7 +65,4 @@ def journey_detail(accessId: str, output = get("https://www.rmv.de/hapi/journeyDetail", params=payload, headers=headers) - if json: - return output.json() - else: - return xmlparse(output.content) \ No newline at end of file + return output.json() if json else xmlparse(output.content) diff --git a/src/pyrmv/raw/stop_by_coords.py b/src/pyrmv/raw/stop_by_coords.py index 160b2e4..1122d08 100644 --- a/src/pyrmv/raw/stop_by_coords.py +++ b/src/pyrmv/raw/stop_by_coords.py @@ -1,5 +1,6 @@ -from requests import get from typing import Union + +from requests import get from xmltodict import parse as xmlparse try: @@ -7,25 +8,29 @@ try: except ImportError: from typing_extensions import Literal -# 2.4. Location Search by Coordinate (location.nearbystops) -def stop_by_coords(accessId: str, - originCoordLat: Union[str, float], - originCoordLong: Union[str, float], - lang: Literal["de", "da", "en", "es", "fr", "hu", "it", "nl", "no", "pl", "sv", "tr"] = "en", - json: bool = True, - radius: Union[int, float] = 1000, - maxNo: int = 10, - stopType: Literal["S", "P", "SP", "SE", "SPE"] = "S", - locationSelectionMode: Union[Literal["SLCT_N", "SLCT_A"], None] = None, - products: Union[int, None] = None, - meta: Union[str, None] = None, - sattributes: Union[str, list, None] = None, - sinfotexts: Union[str, list, None] = None - ) -> dict: - """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. - Read more about this in section 2.4. "Location Search by Coordinate (location.nearbystops)" of HAFAS ReST Documentation. +# 2.4. Location Search by Coordinate (location.nearbystops) +def stop_by_coords( + accessId: str, + originCoordLat: Union[str, float], + originCoordLong: Union[str, float], + lang: Literal[ + "de", "da", "en", "es", "fr", "hu", "it", "nl", "no", "pl", "sv", "tr" + ] = "en", + json: bool = True, + radius: Union[int, float] = 1000, + maxNo: int = 10, + stopType: Literal["S", "P", "SP", "SE", "SPE"] = "S", + locationSelectionMode: Union[Literal["SLCT_N", "SLCT_A"], None] = None, + products: Union[int, None] = None, + meta: Union[str, None] = None, + sattributes: Union[str, list, None] = None, + sinfotexts: Union[str, list, None] = None, +) -> dict: + """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. + + Read more about this in section 2.4. "Location Search by Coordinate (location.nearbystops)" 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). @@ -44,14 +49,10 @@ def stop_by_coords(accessId: str, ### Returns: * dict: Output from RMV. Dict will contain "errorCode" and "errorText" if exception occurs. - """ - - if json: - headers = {"Accept": "application/json"} - else: - headers = {"Accept": "application/xml"} + """ payload = {} + headers = {"Accept": "application/json"} if json else {"Accept": "application/xml"} for var, val in locals().items(): if str(var) == "stopType": @@ -64,9 +65,8 @@ def stop_by_coords(accessId: str, if val != None: payload[str(var)] = val - output = get("https://www.rmv.de/hapi/location.nearbystops", params=payload, headers=headers) + output = get( + "https://www.rmv.de/hapi/location.nearbystops", params=payload, headers=headers + ) - if json: - return output.json() - else: - return xmlparse(output.content) \ No newline at end of file + return output.json() if json else xmlparse(output.content) diff --git a/src/pyrmv/raw/stop_by_name.py b/src/pyrmv/raw/stop_by_name.py index ac0adb8..f46f4c8 100644 --- a/src/pyrmv/raw/stop_by_name.py +++ b/src/pyrmv/raw/stop_by_name.py @@ -1,5 +1,6 @@ -from requests import get from typing import Union + +from requests import get from xmltodict import parse as xmlparse try: @@ -7,33 +8,37 @@ try: except ImportError: from typing_extensions import Literal + # 2.3. Location Search by Name (location.name) -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, - maxNo: int = 10, - stopType: Literal["A", "ALL", "AP", "P", "S", "SA", "SP"] = "ALL", - locationSelectionMode: Union[Literal["SLCT_N", "SLCT_A"], None] = None, - products: Union[int, None] = None, - coordLat: Union[str, float, None] = None, - coordLong: Union[str, float, None] = None, - radius: Union[int, float] = 1000, - refineId: Union[str, None] = None, - meta: Union[str, None] = None, - stations: Union[str, list, None] = None, - sattributes: Union[str, list, None] = None, - filterMode: Literal["DIST_PERI", "EXCL_PERI", "SLCT_PERI"] = "DIST_PERI" - ) -> dict: +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, + maxNo: int = 10, + stopType: Literal["A", "ALL", "AP", "P", "S", "SA", "SP"] = "ALL", + locationSelectionMode: Union[Literal["SLCT_N", "SLCT_A"], None] = None, + products: Union[int, None] = None, + coordLat: Union[str, float, None] = None, + coordLong: Union[str, float, None] = None, + radius: Union[int, float] = 1000, + refineId: Union[str, None] = None, + meta: Union[str, None] = None, + stations: Union[str, list, None] = None, + sattributes: Union[str, list, None] = None, + filterMode: Literal["DIST_PERI", "EXCL_PERI", "SLCT_PERI"] = "DIST_PERI", +) -> dict: """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. + interest and addresses. The result is a list of possible matches (locations) where the user might pick one entry to perform a trip request with this location as origin or destination or to ask for a departure board or arrival board of this - location (stops/stations only). + location (stops/stations only). - Read more about this in section 2.3. "Location Search by Name (location.name)" of HAFAS ReST Documentation. + Read more about this in section 2.3. "Location Search by Name (location.name)" 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). @@ -55,14 +60,10 @@ def stop_by_name(accessId: str, ### Returns: * dict: Output from RMV. Dict will contain "errorCode" and "errorText" if exception occurs. - """ - - if json: - headers = {"Accept": "application/json"} - else: - headers = {"Accept": "application/xml"} + """ payload = {} + headers = {"Accept": "application/json"} if json else {"Accept": "application/xml"} for var, val in locals().items(): if str(var) == "inputString": @@ -80,7 +81,4 @@ def stop_by_name(accessId: str, output = get("https://www.rmv.de/hapi/location.name", params=payload, headers=headers) - if json: - return output.json() - else: - return xmlparse(output.content) \ No newline at end of file + return output.json() if json else xmlparse(output.content) diff --git a/src/pyrmv/raw/trip_find.py b/src/pyrmv/raw/trip_find.py index 19336a6..659632f 100644 --- a/src/pyrmv/raw/trip_find.py +++ b/src/pyrmv/raw/trip_find.py @@ -1,6 +1,7 @@ from datetime import datetime -from requests import get from typing import List, Union + +from requests import get from xmltodict import parse as xmlparse try: @@ -8,125 +9,103 @@ try: except ImportError: from typing_extensions import Literal -# 2.12. Trip Search (trip) -def trip_find(accessId: str, - lang: Literal["de", "da", "en", "es", "fr", "hu", "it", "nl", "no", "pl", "sv", "tr"] = "en", - json: bool = True, - originId: Union[str, None] = None, - originExtId: Union[str, None] = None, - originCoordLat: Union[str, float, None] = None, - originCoordLong: Union[str, float, None] = None, - originCoordName: Union[str, None] = None, - - destId: Union[str, None] = None, - destExtId: Union[str, None] = None, - destCoordLat: Union[str, float, None] = None, - destCoordLong: Union[str, float, None] = None, - destCoordName: Union[str, None] = None, - - via: Union[str, None] = None, - viaId: Union[str, None] = None, - viaWaitTime: int = 0, - - avoid: Union[str, None] = None, - avoidId: Union[str, None] = None, - - viaGis: Union[str, None] = None, - - changeTimePercent: int = 100, - minChangeTime: Union[int, None] = None, - maxChangeTime: Union[int, None] = None, - addChangeTime: Union[int, None] = None, - maxChange: Union[int, None] = None, - - date: Union[str, datetime, None] = None, - time: Union[str, datetime, None] = None, - - searchForArrival: bool = False, - - numF: Union[int, None] = None, - numB: Union[int, None] = None, - - context: Union[str, None] = None, - - poly: bool = False, - polyEnc: Literal["DLT", "GPA", "N"] = "N", - - passlist: bool = False, - products: Union[int, None] = None, - operators: Union[str, list, None] = None, - - attributes: Union[str, list, None] = None, - sattributes: Union[str, list, None] = None, - fattributes: Union[str, list, None] = None, - lines: Union[str, list, None] = None, - lineids: Union[str, list, None] = None, - - avoidPaths: Union[List[Literal["SW", "EA", "ES", "RA", "CB"]], None] = None, - - originWalk: Union[str, list, None] = None, - originBike: Union[str, list, None] = None, - originCar: Union[str, list, None] = None, - originTaxi: Union[str, list, None] = None, - originPark: Union[str, list, None] = None, - originMeta: Union[str, list, None] = None, - - destWalk: Union[str, list, None] = None, - destBike: Union[str, list, None] = None, - destCar: Union[str, list, None] = None, - destTaxi: Union[str, list, None] = None, - destPark: Union[str, list, None] = None, - destMeta: Union[str, list, None] = None, - - totalWalk: Union[str, list, None] = None, - totalBike: Union[str, list, None] = None, - totalCar: Union[str, list, None] = None, - totalTaxi: Union[str, list, None] = None, - totalMeta: Union[str, list, None] = None, - - gisProducts: Union[str, None] = None, - - includeIv: bool = False, - ivOnly: bool = False, - - mobilityProfile: Union[str, None] = None, - - bikeCarriage: bool = False, - bikeCarriageType: Union[Literal["SINGLEBIKES", "SMALLGROUPS", "LARGEGROUPS"], None] = None, - - sleepingCar: bool = False, - couchetteCoach: bool = False, - showPassingPoints: bool = False, - baim: bool = False, - - eco: bool = False, - ecoCmp: bool = False, - ecoParams: Union[str, None] = None, - - rtMode: Union[Literal["FULL", "INFOS", "OFF", "REALTIME", "SERVER_DEFAULT"], None] = None, - - unsharp: bool = False, - trainFilter: Union[str, None] = None, - economic: bool = False, - groupFilter: Union[str, None] = None, - - blockingList: Union[str, None] = None, - blockedEdges: Union[str, None] = None, - - trainComposition: bool = False, - includeEarlier: bool = False, - withICTAlternatives: bool = False, - tariff: Union[bool, None] = None, - trafficMessages: bool = False, - travellerProfileData: Union[str, None] = None, - withFreq: bool = True - ) -> dict: +def trip_find( + accessId: str, + lang: Literal[ + "de", "da", "en", "es", "fr", "hu", "it", "nl", "no", "pl", "sv", "tr" + ] = "en", + json: bool = True, + originId: Union[str, None] = None, + originExtId: Union[str, None] = None, + originCoordLat: Union[str, float, None] = None, + originCoordLong: Union[str, float, None] = None, + originCoordName: Union[str, None] = None, + destId: Union[str, None] = None, + destExtId: Union[str, None] = None, + destCoordLat: Union[str, float, None] = None, + destCoordLong: Union[str, float, None] = None, + destCoordName: Union[str, None] = None, + via: Union[str, None] = None, + viaId: Union[str, None] = None, + viaWaitTime: int = 0, + avoid: Union[str, None] = None, + avoidId: Union[str, None] = None, + viaGis: Union[str, None] = None, + changeTimePercent: int = 100, + minChangeTime: Union[int, None] = None, + maxChangeTime: Union[int, None] = None, + addChangeTime: Union[int, None] = None, + maxChange: Union[int, None] = None, + date: Union[str, datetime, None] = None, + time: Union[str, datetime, None] = None, + searchForArrival: bool = False, + numF: Union[int, None] = None, + numB: Union[int, None] = None, + context: Union[str, None] = None, + poly: bool = False, + polyEnc: Literal["DLT", "GPA", "N"] = "N", + passlist: bool = False, + products: Union[int, None] = None, + operators: Union[str, list, None] = None, + attributes: Union[str, list, None] = None, + sattributes: Union[str, list, None] = None, + fattributes: Union[str, list, None] = None, + lines: Union[str, list, None] = None, + lineids: Union[str, list, None] = None, + avoidPaths: Union[List[Literal["SW", "EA", "ES", "RA", "CB"]], None] = None, + originWalk: Union[str, list, None] = None, + originBike: Union[str, list, None] = None, + originCar: Union[str, list, None] = None, + originTaxi: Union[str, list, None] = None, + originPark: Union[str, list, None] = None, + originMeta: Union[str, list, None] = None, + destWalk: Union[str, list, None] = None, + destBike: Union[str, list, None] = None, + destCar: Union[str, list, None] = None, + destTaxi: Union[str, list, None] = None, + destPark: Union[str, list, None] = None, + destMeta: Union[str, list, None] = None, + totalWalk: Union[str, list, None] = None, + totalBike: Union[str, list, None] = None, + totalCar: Union[str, list, None] = None, + totalTaxi: Union[str, list, None] = None, + totalMeta: Union[str, list, None] = None, + gisProducts: Union[str, None] = None, + includeIv: bool = False, + ivOnly: bool = False, + mobilityProfile: Union[str, None] = None, + bikeCarriage: bool = False, + bikeCarriageType: Union[ + Literal["SINGLEBIKES", "SMALLGROUPS", "LARGEGROUPS"], None + ] = None, + sleepingCar: bool = False, + couchetteCoach: bool = False, + showPassingPoints: bool = False, + baim: bool = False, + eco: bool = False, + ecoCmp: bool = False, + ecoParams: Union[str, None] = None, + rtMode: Union[Literal["FULL", "INFOS", "OFF", "REALTIME", "SERVER_DEFAULT"], None] = None, + unsharp: bool = False, + trainFilter: Union[str, None] = None, + economic: bool = False, + groupFilter: Union[str, None] = None, + blockingList: Union[str, None] = None, + blockedEdges: Union[str, None] = None, + trainComposition: bool = False, + includeEarlier: bool = False, + withICTAlternatives: bool = False, + tariff: Union[bool, None] = None, + trafficMessages: bool = False, + travellerProfileData: Union[str, None] = None, + withFreq: bool = True, +) -> dict: """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. + coordinates freely defined by the client. - Read more about this in section 2.12. "Trip Search (trip)" of HAFAS ReST Documentation. + Read more about this in section 2.12. "Trip Search (trip)" 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). @@ -217,14 +196,10 @@ def trip_find(accessId: str, ### Returns: * dict: Output from RMV. Dict will contain "errorCode" and "errorText" if exception occurs. - """ - - if json: - headers = {"Accept": "application/json"} - else: - headers = {"Accept": "application/xml"} + """ payload = {} + headers = {"Accept": "application/json"} if json else {"Accept": "application/xml"} for var, val in locals().items(): if str(var) == "date": @@ -235,10 +210,9 @@ def trip_find(accessId: str, 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 + payload[str(var)] = ( + val.strftime("%H:%M") if isinstance(val, datetime) else val + ) elif str(var) == "rtMode": if val != None: payload["rtMode"] = val.upper() @@ -248,7 +222,4 @@ def trip_find(accessId: str, output = get("https://www.rmv.de/hapi/trip", params=payload, headers=headers) - if json: - return output.json() - else: - return xmlparse(output.content) \ No newline at end of file + return output.json() if json else xmlparse(output.content) diff --git a/src/pyrmv/raw/trip_recon.py b/src/pyrmv/raw/trip_recon.py index 9b4ac45..d0ac727 100644 --- a/src/pyrmv/raw/trip_recon.py +++ b/src/pyrmv/raw/trip_recon.py @@ -1,6 +1,7 @@ from datetime import datetime -from requests import get from typing import Union + +from requests import get from xmltodict import parse as xmlparse try: @@ -8,43 +9,44 @@ try: except ImportError: from typing_extensions import Literal -# 2.17. Reconstruction (recon) -def trip_recon(accessId: str, - ctx: str, - json: bool = True, - poly: bool = False, - polyEnc: Literal["DLT", "GPA", "N"] = "N", - date: Union[str, datetime, None] = None, - useCombinedComparison: Union[bool, None] = None, - acceptGaps: Union[bool, None] = None, - allowDummySections: Union[bool, None] = None, - flagAllNonReachable: Union[bool, None] = None, - matchCatStrict: Union[bool, None] = None, - matchIdNonBlank: Union[bool, None] = None, - matchIdStrict: Union[bool, None] = None, - matchNumStrict: Union[bool, None] = None, - matchRtType: Union[bool, None] = None, - enableRtFullSearch: Union[bool, None] = None, - enableReplacements: Union[bool, None] = None, - arrL: Union[int, None] = None, - arrU: Union[int, None] = None, - depL: Union[int, None] = None, - depU: Union[int, None] = None, - passlist: bool = False, - showPassingPoints: bool = False, - rtMode: Union[Literal["FULL", "INFOS", "OFF", "REALTIME", "SERVER_DEFAULT"], None] = None, - eco: bool = False, - ecoCmp: bool = False, - ecoParams: Union[str, None] = None, - tariff: Union[bool, None] = None, - trafficMessages: Union[bool, None] = None, - travellerProfileData: Union[str, None] = None - ) -> dict: + +def trip_recon( + accessId: str, + ctx: str, + json: bool = True, + poly: bool = False, + polyEnc: Literal["DLT", "GPA", "N"] = "N", + date: Union[str, datetime, None] = None, + useCombinedComparison: Union[bool, None] = None, + acceptGaps: Union[bool, None] = None, + allowDummySections: Union[bool, None] = None, + flagAllNonReachable: Union[bool, None] = None, + matchCatStrict: Union[bool, None] = None, + matchIdNonBlank: Union[bool, None] = None, + matchIdStrict: Union[bool, None] = None, + matchNumStrict: Union[bool, None] = None, + matchRtType: Union[bool, None] = None, + enableRtFullSearch: Union[bool, None] = None, + enableReplacements: Union[bool, None] = None, + arrL: Union[int, None] = None, + arrU: Union[int, None] = None, + depL: Union[int, None] = None, + depU: Union[int, None] = None, + passlist: bool = False, + showPassingPoints: bool = False, + rtMode: Union[Literal["FULL", "INFOS", "OFF", "REALTIME", "SERVER_DEFAULT"], None] = None, + eco: bool = False, + ecoCmp: bool = False, + ecoParams: Union[str, None] = None, + tariff: Union[bool, None] = None, + trafficMessages: Union[bool, None] = None, + travellerProfileData: Union[str, None] = None, +) -> dict: """Reconstructing a trip can be achieved using the reconstruction context provided by any trip result in the ctxRecon attribute of Trip element. The result will be a true copy of the original trip search result given - that the underlying data did not change. + that the underlying data did not change. - Read more about this in section 2.17. "Reconstruction (recon)" of HAFAS ReST Documentation. + Read more about this in section 2.17. "Reconstruction (recon)" 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). @@ -80,14 +82,10 @@ def trip_recon(accessId: str, ### Returns: * dict: Output from RMV. Dict will contain "errorCode" and "errorText" if exception occurs. - """ - - if json: - headers = {"Accept": "application/json"} - else: - headers = {"Accept": "application/xml"} + """ payload = {} + headers = {"Accept": "application/json"} if json else {"Accept": "application/xml"} for var, val in locals().items(): if str(var) == "date": @@ -102,7 +100,4 @@ def trip_recon(accessId: str, output = get("https://www.rmv.de/hapi/recon", params=payload, headers=headers) - if json: - return output.json() - else: - return xmlparse(output.content) \ No newline at end of file + return output.json() if json else xmlparse(output.content) diff --git a/src/pyrmv/utility/__init__.py b/src/pyrmv/utility/__init__.py index e69de29..1a20d53 100644 --- a/src/pyrmv/utility/__init__.py +++ b/src/pyrmv/utility/__init__.py @@ -0,0 +1,2 @@ +from .find_exception import find_exception +from .weekdays_bitmask import weekdays_bitmask diff --git a/src/pyrmv/utility/find_exception.py b/src/pyrmv/utility/find_exception.py index 066ed8d..eb18ae9 100644 --- a/src/pyrmv/utility/find_exception.py +++ b/src/pyrmv/utility/find_exception.py @@ -1,22 +1,46 @@ -from pyrmv.errors.api_errors import ApiAuthError, ApiFormatError, ApiParamError, ApiQuotaError, ApiTooManyRequests -from pyrmv.errors.int_errors import IntError, IntGatewayError, IntTimeoutError -from pyrmv.errors.ps_errors import PsIncorrectParamError -from pyrmv.errors.sot_errors import SotAllTrainsFilteredError, SotAlreadyArrivedError, SotCancelledError, SotNotStartedError, SotStayOnTripError -from pyrmv.errors.svc_errors import SvcContextError, SvcDatetimeError, SvcDatetimePeriodError, SvcLocationArrivalError, SvcLocationDepartureError, SvcLocationEqualError, SvcLocationError, SvcLocationNearError, SvcLocationViaError, SvcNoMatchError, SvcNoResultError, SvcProductError, SvcSearchError -from pyrmv.errors.unknown_error import UnknownError +from pyrmv.errors import ( + ApiAuthError, + ApiFormatError, + ApiParamError, + ApiQuotaError, + ApiTooManyRequests, + IntError, + IntGatewayError, + IntTimeoutError, + PsIncorrectParamError, + SotAllTrainsFilteredError, + SotAlreadyArrivedError, + SotCancelledError, + SotNotStartedError, + SotStayOnTripError, + SvcContextError, + SvcDatetimeError, + SvcDatetimePeriodError, + SvcLocationArrivalError, + SvcLocationDepartureError, + SvcLocationEqualError, + SvcLocationError, + SvcLocationNearError, + SvcLocationViaError, + SvcNoMatchError, + SvcNoResultError, + SvcProductError, + SvcSearchError, + UnknownError, +) + def find_exception(data: dict): """Scan returned dict for errorCode from RMV. - Raises different exceptions if errorCode is not None. + Raises different exceptions if errorCode is not None. ### Args: * data (dict): Response from RMV as a dict. ### Raises: * Any: Formatted as "errorCode -> errorText" if ApiParamError and UnknownError or as a single massage for others. - """ + """ if "errorCode" in data: - if data["errorCode"] == "API_AUTH": raise ApiAuthError() @@ -25,7 +49,7 @@ def find_exception(data: dict): elif data["errorCode"] == "API_TOO_MANY_REQUESTS": raise ApiTooManyRequests() - + elif data["errorCode"] == "API_PARAM": raise ApiParamError(errorCode=data["errorCode"], errorText=data["errorText"]) @@ -99,4 +123,4 @@ def find_exception(data: dict): raise PsIncorrectParamError() else: - raise UnknownError(errorCode=data["errorCode"], errorText=data["errorText"]) \ No newline at end of file + raise UnknownError(errorCode=data["errorCode"], errorText=data["errorText"]) diff --git a/src/pyrmv/utility/weekdays_bitmask.py b/src/pyrmv/utility/weekdays_bitmask.py index c5456d3..d426cbe 100644 --- a/src/pyrmv/utility/weekdays_bitmask.py +++ b/src/pyrmv/utility/weekdays_bitmask.py @@ -9,13 +9,10 @@ def weekdays_bitmask(data: OrderedDict[str, bool]) -> str: ### Returns: * str: _description_ - """ - output = "" + """ if len(data) != 7: - raise ValueError("OrderedDict must be formatted as follows: OrderedDict(Monday=bool, Tuesday=bool, Wednesday=bool, Thursday=bool, Friday=bool, Saturday=bool, Sunday=bool)") - for day in data: - if data[day]: - output += "1" - else: - output += "0" - return output \ No newline at end of file + raise ValueError( + "OrderedDict must be formatted as follows: OrderedDict(Monday=bool, Tuesday=bool, Wednesday=bool, Thursday=bool, Friday=bool, Saturday=bool, Sunday=bool)" + ) + + return "".join("1" if data[day] else "0" for day in data)