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

Methods overhaul

Browse Source
This commit is contained in:
Profitroll 2022-10-07 11:35:02 +02:00
parent 201fcba766
commit 036a3174f8
18 changed files with 670 additions and 782 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
README.md
View File

@ -25,15 +25,15 @@ If you have everything listed in [requirements](#requirements), then let's begin
```py ```py
import pyrmv import pyrmv
# Set API key # Define a Client with API key
access_id = "Something" client = pyrmv.Client("AcessId")
# Get origin's and destination's location # Get origin's and destination's location
origin = pyrmv.stop_by_name(access_id, "Frankfurt Hauptbahnhof", max_number=3)[0] origin = client.stop_by_name("Frankfurt Hauptbahnhof", max_number=3)[0]
destination = pyrmv.stop_by_coords(access_id, 50.099613, 8.685449, max_number=3)[0] destination = client.stop_by_coords(50.099613, 8.685449, max_number=3)[0]
# Find a trip by locations got # Find a trip by locations got
trip = pyrmv.trip_find(access_id, origin_id=origin.id, dest_id=destination.id) trip = client.trip_find(origin_id=origin.id, dest_id=destination.id)
``` ```
# Frequently Asked Questions # Frequently Asked Questions

5
pyrmv/__init__.py
View File

@ -30,5 +30,6 @@ from . import const
from . import enums from . import enums
from . import errors from . import errors
from . import utility from . import utility
from . import classes from .classes import *
from .methods import * from .classes.Client import Client
# from .methods import *

22
pyrmv/classes/Board.py
View File

@ -1,12 +1,10 @@
from datetime import datetime from datetime import datetime
from pyrmv.methods.journey_detail import journey_detail
from pyrmv.methods.stop_by_id import stop_by_id
from pyrmv.classes.Message import Message from pyrmv.classes.Message import Message
class LineArrival(): class LineArrival():
def __init__(self, data, access_id: str): def __init__(self, data, client):
self.journey = journey_detail(access_id, data["JourneyDetailRef"]["ref"]) self.journey = client.journey_detail(data["JourneyDetailRef"]["ref"])
self.status = data["JourneyStatus"] self.status = data["JourneyStatus"]
self.messages = [] self.messages = []
for message in data["Messages"]["Message"]: for message in data["Messages"]["Message"]:
@ -16,7 +14,7 @@ class LineArrival():
self.stop_name = data["stop"] self.stop_name = data["stop"]
self.stop_id = data["stopid"] self.stop_id = data["stopid"]
self.stop_id_ext = data["stopExtId"] self.stop_id_ext = data["stopExtId"]
self.stop = stop_by_id(access_id, self.stop_id) self.stop = client.stop_by_id(self.stop_id)
self.time = datetime.strptime(data["time"], "%H:%M:%S") self.time = datetime.strptime(data["time"], "%H:%M:%S")
self.date = datetime.strptime(data["date"], "%Y-%m-%d") self.date = datetime.strptime(data["date"], "%Y-%m-%d")
if ("rtTime" in data) and ("rtDate" in data): if ("rtTime" in data) and ("rtDate" in data):
@ -30,8 +28,8 @@ class LineArrival():
class LineDeparture(): class LineDeparture():
def __init__(self, data, access_id: str): def __init__(self, data, client):
self.journey = journey_detail(access_id, data["JourneyDetailRef"]["ref"]) self.journey = client.journey_detail(data["JourneyDetailRef"]["ref"])
self.status = data["JourneyStatus"] self.status = data["JourneyStatus"]
self.messages = [] self.messages = []
for message in data["Messages"]["Message"]: for message in data["Messages"]["Message"]:
@ -41,7 +39,7 @@ class LineDeparture():
self.stop_name = data["stop"] self.stop_name = data["stop"]
self.stop_id = data["stopid"] self.stop_id = data["stopid"]
self.stop_id_ext = data["stopExtId"] self.stop_id_ext = data["stopExtId"]
self.stop = stop_by_id(access_id, self.stop_id) self.stop = client.stop_by_id(self.stop_id)
self.time = datetime.strptime(data["time"], "%H:%M:%S") self.time = datetime.strptime(data["time"], "%H:%M:%S")
self.date = datetime.strptime(data["date"], "%Y-%m-%d") self.date = datetime.strptime(data["date"], "%Y-%m-%d")
if ("rtTime" in data) and ("rtDate" in data): if ("rtTime" in data) and ("rtDate" in data):
@ -56,10 +54,10 @@ class LineDeparture():
class BoardArrival(list): class BoardArrival(list):
def __init__(self, data: dict, access_id: str): def __init__(self, data: dict, client):
super().__init__([]) super().__init__([])
for line in data["Arrival"]: for line in data["Arrival"]:
self.append(LineArrival(line, access_id)) self.append(LineArrival(line, client))
def __str__(self) -> str: def __str__(self) -> str:
lines = [] lines = []
@ -69,10 +67,10 @@ class BoardArrival(list):
class BoardDeparture(list): class BoardDeparture(list):
def __init__(self, data: dict, access_id: str): def __init__(self, data: dict, client):
super().__init__([]) super().__init__([])
for line in data["Departure"]: for line in data["Departure"]:
self.append(LineDeparture(line, access_id)) self.append(LineDeparture(line, client))
def __str__(self) -> str: def __str__(self) -> str:
lines = [] lines = []

631
pyrmv/classes/Client.py Normal file
View File

@ -0,0 +1,631 @@
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.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.trip_find import trip_find as raw_trip_find
from pyrmv.raw.trip_recon import trip_recon as raw_trip_recon
from pyrmv.utility.find_exception import find_exception
from pyrmv.errors.not_ready import NotReadyYetError
try:
from typing import Literal
except ImportError:
from typing_extensions import Literal
class Client():
def __init__(self, access_id: str) -> None:
self.access_id = access_id
def board_arrival(self,
id: str = None, # type: ignore
id_ext: str = None, # type: ignore
direction: Union[str, Stop, StopTrip] = None, # type: ignore
date: Union[str, datetime] = None, # type: ignore
time: Union[str, datetime] = None, # type: ignore
duration: Union[int, timedelta] = 60,
journeys_max: int = -1,
operators: Union[str, list] = None, # type: ignore
lines: Union[str, list] = None, # type: ignore
passlist: bool = False,
board_type: Literal[BoardArrivalType.ARR, BoardArrivalType.ARR_EQUIVS, BoardArrivalType.ARR_MAST, BoardArrivalType.ARR_STATION] = BoardArrivalType.ARR,
) -> BoardArrival:
"""Method returns a board with arriving transport.
More detailed request is available as `raw.board_arrival()`, however returns `dict` instead of `Board`.
### Args:
* id (`str`, **optional**): Specifies the station/stop ID. Such ID can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* id_ext (`str`, **optional**): Deprecated. Please use `id` as it supports external IDs. Specifies the external station/stop ID. Such ID can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* direction (`Union[str, Stop, StopTrip]`, **optional**): If only vehicles departing or arriving from a certain direction shall be returned, specify the direction by giving the station/stop ID of the last stop on the journey or the Stop object itself. Defaults to `None`.
* date (`Union[str, datetime]`, **optional**): Sets the start date for which the departures shall be retrieved. Represented in the format YYYY-MM-DD or as a datetime object. By default the current server date is used. Defaults to `None`.
* time (`Union[str, datetime]`, **optional**): Sets the start time for which the departures shall be retrieved. Represented in the format hh:mm[:ss] in 24h nomenclature or as a datetime object. Seconds will be ignored for requests. By default the current server time is used. Defaults to `None`.
* duration (`Union[int, timedelta]`, **optional**): Set the interval size in minutes. Can also be provided as a timedelta. Defaults to `60`.
* journeys_max (`int`, **optional**): Maximum number of journeys to be returned. If no value is defined or -1, all departing/arriving services within the duration sized period are returned. Defaults to `-1`.
* operators (`Union[str, list]`, **optional**): Only journeys provided by the given operators are part of the result. To filter multiple operators, separate the codes by comma or provide a list. If the operator should not be part of the be trip, negate it by putting ! in front of it. Example: Filter for operator A and B: `operators=[A,B]`. Defaults to `None`.
* lines (`Union[str, list]`, **optional**): Only journeys running the given line are part of the result. To filter multiple lines, provide a list or separate the codes by comma or provide a list. If the line should not be part of the be trip, negate it by putting ! in front of it. Defaults to `None`.
* passlist (`bool`, **optional**): Include a list of all passed waystops. Defaults to `False`.
* board_type (`Union[BoardArrivalType.ARR, BoardArrivalType.ARR_EQUIVS, BoardArrivalType.ARR_MAST, BoardArrivalType.ARR_STATION]`, optional): Set the station arrival board type to be used. Defaults to `BoardArrivalType.ARR`.
### Returns:
* BoardArrival: Instance of `BoardArrival` object.
"""
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,
date=date,
time=time,
duration=duration,
maxJourneys=journeys_max,
operators=operators,
lines=lines,
passlist=passlist,
boardType=board_type.code
)
find_exception(board_raw)
return BoardArrival(board_raw, self)
def board_departure(self,
id: str = None, # type: ignore
id_ext: str = None, # type: ignore
direction: Union[str, Stop, StopTrip] = None, # type: ignore
date: Union[str, datetime] = None, # type: ignore
time: Union[str, datetime] = None, # type: ignore
duration: Union[int, timedelta] = 60,
journeys_max: int = -1,
operators: Union[str, list] = None, # type: ignore
lines: Union[str, list] = None, # type: ignore
passlist: bool = False,
board_type: Literal[BoardDepartureType.DEP, BoardDepartureType.DEP_EQUIVS, BoardDepartureType.DEP_MAST, BoardDepartureType.DEP_STATION] = BoardDepartureType.DEP,
) -> BoardDeparture:
"""Method returns a board with departing transport.
More detailed request is available as `raw.board_departure()`, however returns `dict` instead of `Board`.
### Args:
* id (`str`, **optional**): Specifies the station/stop ID. Such ID can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* id_ext (`str`, **optional**): Deprecated. Please use `id` as it supports external IDs. Specifies the external station/stop ID. Such ID can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* direction (`Union[str, Stop, StopTrip]`, **optional**): If only vehicles departing or arriving from a certain direction shall be returned, specify the direction by giving the station/stop ID of the last stop on the journey or the Stop object itself. Defaults to `None`.
* date (`Union[str, datetime]`, **optional**): Sets the start date for which the departures shall be retrieved. Represented in the format YYYY-MM-DD or as a datetime object. By default the current server date is used. Defaults to `None`.
* time (`Union[str, datetime]`, **optional**): Sets the start time for which the departures shall be retrieved. Represented in the format hh:mm[:ss] in 24h nomenclature or as a datetime object. Seconds will be ignored for requests. By default the current server time is used. Defaults to `None`.
* duration (`Union[int, timedelta]`, **optional**): Set the interval size in minutes. Can also be provided as a timedelta. Defaults to `60`.
* journeys_max (`int`, **optional**): Maximum number of journeys to be returned. If no value is defined or -1, all departing/arriving services within the duration sized period are returned. Defaults to `-1`.
* operators (`Union[str, list]`, **optional**): Only journeys provided by the given operators are part of the result. To filter multiple operators, separate the codes by comma or provide a list. If the operator should not be part of the be trip, negate it by putting ! in front of it. Example: Filter for operator A and B: `operators=[A,B]`. Defaults to `None`.
* lines (`Union[str, list]`, **optional**): Only journeys running the given line are part of the result. To filter multiple lines, provide a list or separate the codes by comma or provide a list. If the line should not be part of the be trip, negate it by putting ! in front of it. Defaults to `None`.
* passlist (`bool`, **optional**): Include a list of all passed waystops. Defaults to `False`.
* board_type (`Union[BoardDepartureType.DEP, BoardDepartureType.DEP_EQUIVS, BoardDepartureType.DEP_MAST, BoardDepartureType.DEP_STATION]`, optional): Set the station departure board type to be used. Defaults to `BoardDepartureType.DEP`.
### Returns:
* BoardDeparture: Instance of `BoardDeparture` object.
"""
if (isinstance(direction, Stop) or isinstance(direction, StopTrip)):
direction = direction.id
board_raw = raw_board_departure(
accessId=self.access_id,
id=id,
extId=id_ext,
direction=direction,
date=date,
time=time,
duration=duration,
maxJourneys=journeys_max,
operators=operators,
lines=lines,
passlist=passlist,
boardType=board_type.code
)
find_exception(board_raw)
return BoardDeparture(board_raw, self)
def him_search(self,
date_begin: Union[str, datetime] = None,
date_end: Union[str, datetime] = None,
time_begin: Union[str, datetime] = None,
time_end: Union[str, datetime] = None,
weekdays: Union[str, OrderedDict[str, bool]] = None,
ids: list = None,
operators: list = None,
categories: list = None,
channels: list = None,
companies: list = None,
lines: list = None,
lineids: list = None,
stations: list = None,
station_from: Union[str, Stop] = None,
station_to: Union[str, Stop] = None,
bothways: bool = None,
trainnames: list = None,
search_mode: Literal[SearchMode.MATCH, SearchMode.NOMATCH, SearchMode.TFMATCH] = None,
affected_journey_mode: Literal[AffectedJourneyMode.ALL, AffectedJourneyMode.OFF] = None,
affected_journey_stop_mode: Literal[AffectedJourneyStopMode.ALL, AffectedJourneyStopMode.IMP, AffectedJourneyStopMode.OFF] = None,
priority_min: int = None,
priority_max: int = None
) -> None: #List[Message]:
raise NotReadyYetError()
def journey_detail(self,
id: str,
date: Union[str, datetime] = None, # type: ignore
real_time_mode: Literal[RealTimeMode.FULL, RealTimeMode.INFOS, RealTimeMode.OFF, RealTimeMode.REALTIME, RealTimeMode.SERVER_DEFAULT] = None, # type: ignore
from_id: str = None, # type: ignore
from_index: int = None, # type: ignore
to_id: str = None, # type: ignore
to_index: int = None # type: ignore
) -> 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.
More detailed request is available as `raw.journey_detail()`, however returns `dict` instead of `Journey`.
### Args:
* id (`str`): Specifies the internal journey id of the journey shall be retrieved. Maximum length 512.
* date (`Union[str, datetime]`, **optional**): Day of operation. Represented in the format `YYYY-MM-DD` or as a datetime object. By default the current server date is used. Defaults to `None`.
* real_time_mode (`Literal[RealTimeMode.FULL, RealTimeMode.INFOS, RealTimeMode.OFF, RealTimeMode.REALTIME, RealTimeMode.SERVER_DEFAULT]`, **optional**): Set the realtime mode to be used. Read more about this in HAFAS ReST Documentation. Defaults to `None`.
* from_id (`str`, **optional**): Specifies the station/stop ID the partial itinerary shall start from. Defaults to `None`.
* from_index (`int`, **optional**): Specifies the station/stop index the partial itinerary shall start from. Defaults to `None`.
* to_id (`str`, **optional**): Specifies the station/stop ID the partial itinerary shall end at. Defaults to `None`.
* to_index (`int`, **optional**): Specifies the station/stop index the partial itinerary shall end at. Defaults to `None`.
### Returns:
* Journey: Instance of `Journey` object.
"""
if real_time_mode == None:
real_time_mode = None
else:
real_time_mode = real_time_mode.code
journey_raw = raw_journey_detail(
accessId=self.access_id,
id=id,
date=date,
rtMode=real_time_mode,
fromId=from_id,
fromIdx=from_index,
toId=to_id,
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: Literal[SelectionMode.SLCT_A, SelectionMode.SLCT_N] = None, # type: ignore
) -> 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.
More detailed request is available as `raw.stop_by_coords()`, however returns `dict` instead of `List[Stop]`.
### Args:
* coords_lat (`Union[str, float]`): Latitude of centre coordinate.
* coords_lon (`Union[str, float]`): Longitude of centre coordinate.
* 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`.
* radius (`Union[int, float]`, **optional**): Search radius in meter around the given coordinate if any. Defaults to `1000`.
* max_number (`int`, **optional**): Maximum number of returned stops. Defaults to `10`.
* stop_type (`Literal[LocationType.S, LocationType.P, LocationType.SP, LocationType.SE, LocationType.SPE]`, **optional**): Type filter for location types. Defaults to `LocationType.S`.
* selection_mode (`Literal[SelectionMode.SLCT_A, SelectionMode.SLCT_N]`, **optional**): Selection mode for locations. `SelectionMode.SLCT_N`: Not selectable, `SelectionMode.SLCT_A`: Selectable. Defaults to `None`.
### 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
stops = []
stops_raw = raw_stop_by_coords(
accessId=self.access_id,
originCoordLat=coords_lat,
originCoordLong=coords_lon,
lang=lang.code,
radius=radius,
maxNo=max_number,
stopType=stop_type.code,
locationSelectionMode=selection_mode
)
find_exception(stops_raw)
for stop in stops_raw["stopLocationOrCoordLocation"]:
if "StopLocation" in stop:
stops.append(Stop(stop["StopLocation"]))
elif "CoordLocation" in stop:
stops.append(Stop(stop["CoordLocation"]))
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]:
"""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:
* Stop: 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
)
find_exception(stops_raw)
if len(stops_raw["stopLocationOrCoordLocation"]) > 0:
stop = stops_raw["stopLocationOrCoordLocation"][0]
if "StopLocation" in stop:
return Stop(stop["StopLocation"])
elif "CoordLocation" in stop:
return Stop(stop["CoordLocation"])
else:
return None
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: Literal[SelectionMode.SLCT_A, SelectionMode.SLCT_N] = None, # type: ignore
coord_lat: Union[str, float] = None, # type: ignore
coord_lon: Union[str, float] = None, # type: ignore
radius: Union[int, float] = 1000,
refine_id: str = None, # type: ignore
stations: Union[str, list] = None, # type: ignore
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.
More detailed request is available as `raw.stop_by_name()`, however returns `dict` instead of `List[Stop]`.
### 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`.
* max_number (`int`, **optional**): Maximum number of returned stops. In range 1-1000. Defaults to `10`.
* stop_type (`Literal[LocationType.A, LocationType.ALL, LocationType.AP, LocationType.P, LocationType.S, LocationType.SA, LocationType.SP]`, **optional**): Type filter for location types. Defaults to `LocationType.ALL`.
* selection_mode (`Literal[SelectionMode.SLCT_A, SelectionMode.SLCT_N]`, **optional**): Selection mode for locations. `SelectionMode.SLCT_N`: Not selectable, `SelectionMode.SLCT_A`: Selectable. Defaults to `None`.
* coord_lat (`Union[str, float]`, **optional**): Latitude of centre coordinate. Defaults to `None`.
* coord_lon (`Union[str, float]`, **optional**): Longitude of centre coordinate. Defaults to `None`.
* radius (`Union[int, float]`, **optional**): Search radius in meter around the given coordinate if any. Defaults to `1000`.
* refine_id (`str`, **optional**): In case of an refinable location, this value takes the ID of the refinable one of a previous result. Defaults to `None`.
* stations (`Union[str, list]`, **optional**): Filter for stations. Matches if the given value is prefix of any station in the itinerary. As a list or as a string separated by comma. Defaults to `None`.
* filter_mode (`Literal[FilterMode.DIST_PERI, FilterMode.EXCL_PERI, FilterMode.SLCT_PERI]`, **optional**): Filter modes for nearby searches. Defaults to `FilterMode.DIST_PERI`.
### 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
stops = []
stops_raw = raw_stop_by_name(
accessId=self.access_id,
inputString=query,
lang=lang.code,
maxNo=max_number,
stopType=stop_type.code,
locationSelectionMode=selection_mode,
coordLat=coord_lat,
coordLong=coord_lon,
radius=radius,
refineId=refine_id,
stations=stations,
filterMode=filter_mode.code
)
find_exception(stops_raw)
for stop in stops_raw["stopLocationOrCoordLocation"]:
if "StopLocation" in stop:
stops.append(Stop(stop["StopLocation"]))
elif "CoordLocation" in stop:
stops.append(Stop(stop["CoordLocation"]))
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: str = None, # type: ignore
origin_id_ext: str = None, # type: ignore
origin_coord_lat: Union[str, float] = None, # type: ignore
origin_coord_lon: Union[str, float] = None, # type: ignore
origin_coord_name: str = None, # type: ignore
destination_id: str = None, # type: ignore
destination_id_ext: str = None, # type: ignore
destination_coord_lat: Union[str, float] = None, # type: ignore
destination_coord_lon: Union[str, float] = None, # type: ignore
destination_coord_name: str = None, # type: ignore
via: str = None, # type: ignore
via_id: str = None, # type: ignore
via_gis: str = None, # type: ignore
via_wait_time: int = 0,
avoid: str = None, # type: ignore
avoid_id: str = None, # type: ignore
change_time_percent: int = 100,
change_time_min: int = None, # type: ignore
change_time_max: int = None, # type: ignore
change_time_add: int = None, # type: ignore
change_max: int = None, # type: ignore
date: Union[str, datetime] = None, # type: ignore
time: Union[str, datetime] = None, # type: ignore
search_arrival: bool = False,
trips_after_time: int = None, # type: ignore
trips_before_time: int = None, # type: ignore
context: str = None, # type: ignore
passlist: bool = False,
operators: Union[str, list] = None, # type: ignore
lines: Union[str, list] = None, # type: ignore
lineids: Union[str, list] = None, # type: ignore
iv_include: bool = False,
iv_only: bool = False,
bike_carriage: bool = False,
passing_points: bool = False,
real_time_mode: Literal[RealTimeMode.FULL, RealTimeMode.INFOS, RealTimeMode.OFF, RealTimeMode.REALTIME, RealTimeMode.SERVER_DEFAULT] = None, # type: ignore
include_earlier: bool = False,
ict_alternatives: bool = False,
tariff: bool = None, # type: ignore
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]`.
### 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`.
* origin_id (`str`, **optional**): Specifies the station/stop ID of the origin for the trip. Such ID can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* origin_id_ext (`str`, **optional**): Deprecated. Please use originId as it supports external IDs. Specifies the external station/stop ID of the origin for the trip. Such ID can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* origin_coord_lat (`Union[str, float]`, **optional**): Latitude of station/stop coordinate of the trip's origin. The coordinate can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* origin_coord_lon (`Union[str, float]`, **optional**): Longitude of station/stop coordinate of the trip's origin. The coordinate can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* origin_coord_name (`str`, **optional**): Name of the trip's origin if coordinate cannot be resolved to an address or poi. Defaults to `None`.
* destination_id (`str`, **optional**): Specifies the station/stop ID of the destination for the trip. Such ID can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* destination_id_ext (`str`, **optional**): Deprecated. Please use destId as it supports external IDs. Specifies the external station/stop ID of the destination for the trip. Such ID can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* destination_coord_lat (`Union[str, float]`, **optional**): Latitude of station/stop coordinate of the trip's destination. The coordinate can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* destination_coord_lon (`Union[str, float]`, **optional**): Longitude of station/stop coordinate of the trip's destination. The coordinate can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* destination_coord_name (`str`, **optional**): Name of the trip's destination if coordinate cannot be resolved to an address or poi. Defaults to `None`.
* via (`str`, **optional**): Complex structure to provide multiple via points separated by semicolon. This structure is build like this: `viaId|waittime|viastatus|products|direct|sleepingCar|couchetteCoach|attributes`. Read more about this in HAFAS ReST Documentation. Defaults to `None`.
* via_id (`str`, **optional**): ID of a station/stop used as a via for the trip. Specifying a via station forces the trip search to look for trips which must pass through this station. Such ID can be retrieved from stop_by_name() or stop_by_coords(). If `via` is used, `via_id` and `via_wait_time ` are having no effect. Defaults to `None`.
* via_gis (`str`, **optional**): Complex structure to provide multiple GIS via locations separated by semicolon. This structure is build like this: `locationId|locationMode|transportMode|placeType|usageType|mode|durationOfStay`. Read more about this in HAFAS ReST Documentation. Defaults to `None`.
* via_wait_time (`int`, **optional**): Defines the waiting time spent at via station in minutes. If `via` is used, `via_id` and `via_wait_time` are having no effect. Defaults to 0.
* avoid (`str`, **optional**): Complex structure to provide multiple points to be avoided separated by semicolon. This structure is build like this: `avoidId|avoidstatus` avoidId: id, extId or altId of the avoid, mandatory avoidstatus: one of NPAVM (do not run through if this is a meta station), NPAVO (do not run through), NCAVM (do not change if this is a meta station), NCAVO (do not change), optional but defaults to NCAVM Example: Just define three avoids by extId: `avoid="801234;801235;801236"`. Defaults to `None`.
* avoid_id (`str`, **optional**): ID of a station/stop to be avoided as transfer stop for the trip. Such ID can be retrieved from stop_by_name() or stop_by_coords(). If `avoid` is used, `avoid_id` has no effect. Defaults to `None`.
* change_time_percent (`int`, **optional**): Configures the walking speed when changing from one leg of the journey to the next one. It extends the time required for changes by a specified percentage. A value of 200 doubles the change time as initially calculated by the system. In the response, change time is presented in full minutes. If the calculation based on changeTime-Percent does not result in a full minute, it is rounded using "round half up" method. Defaults to `100`.
* change_time_min (`int`, **optional**): Minimum change time at stop in minutes. Defaults to `None`.
* change_time_max (`int`, **optional**): Maximum change time at stop in minutes. Defaults to `None`.
* change_time_add (`int`, **optional**): This amount of minutes is added to the change time at each stop. Defaults to `None`.
* change_max (`int`, **optional**): Maximum number of changes. In range 0-11. Defaults to `None`.
* date (`Union[str, datetime]`, **optional**): Sets the start date for which the departures shall be retrieved. Represented in the format `YYYY-MM-DD` or as a datetime object. By default the current server date is used. Defaults to `None`.
* time (`Union[str, datetime]`, **optional**): Sets the start time for which the departures shall be retrieved. Represented in the format `hh:mm[:ss]` in 24h nomenclature or as a datetime object. Seconds will be ignored for requests. By default the current server time is used. Defaults to `None`.
* search_arrival (`bool`, **optional**): If set, the date and time parameters specify the arrival time for the trip search instead of the departure time. Defaults to `False`.
* trips_after_time (`int`, **optional**): Minimum number of trips after the search time. Sum of `trips_after_time` and `trips_before_time` has to be less or equal 6. Read more about this in HAFAS ReST Documentation. In range 1-6. Defaults to `None`.
* trips_before_time (`int`, **optional**): Minimum number of trips before the search time. Sum of `trips_after_time` and `trips_before_time` has to be less or equal 6. Read more about this in HAFAS ReST Documentation. In range 0-6. Defaults to `None`.
* context (`str`, **optional**): Defines the starting point for the scroll back or forth operation. Use the scrB value from a previous result to scroll backwards in time and use the scrF value to scroll forth. Defaults to `None`.
* passlist (`bool`, **optional**): Enables/disables the return of the passlist for each leg of the trip. Defaults to `False`.
* operators (`Union[str, list]`, **optional**): Only trips provided by the given operators are part of the result. If the operator should not be part of the be trip, negate it by putting ! in front of it. Example: Filter for operator A and B: `operators=["A","B"]`. Defaults to `None`.
* lines (`Union[str, list]`, **optional**): Only journeys running the given line are part of the result. If the line should not be part of the be trip, negate it by putting ! in front of it. Defaults to `None`.
* lineids (`Union[str, list]`, **optional**): Only journeys running the given line (identified by its line ID) are part of the result. If the line should not be part of the be trip, negate it by putting ! in front of it. Defaults to `None`.
* iv_include (`bool`, **optional**): Enables/disables search for individual transport routes. Defaults to `False`.
* iv_only (`bool`, **optional**): Enables/disables search for individual transport routes only. Defaults to `False`.
* bike_carriage (`bool`, **optional**): Enables/disables search for trips explicit allowing bike carriage. This will only work in combination with `change_max=0` as those trips are always meant to be direct connections. Defaults to `False`.
* passing_points (`bool`, **optional**): Enables/disables the return of stops having no alighting and boarding in its passlist for each leg of the trip. Needs passlist enabled. Defaults to `False`.
* real_time_mode (`Literal[RealTimeMode.FULL, RealTimeMode.INFOS, RealTimeMode.OFF, RealTimeMode.REALTIME, RealTimeMode.SERVER_DEFAULT]`, **optional**): Set the realtime mode to be used. Read more about this in HAFAS ReST Documentation. Defaults to `None`.
* include_earlier (`bool`, **optional**): Disables search optimization in relation of duration. Defaults to `False`.
* ict_alternatives (`bool`, **optional**): Enables/disables the search for alternatives with individualized change times (ICT). Defaults to `False`.
* tariff (`bool`, **optional**): Enables/disables the output of tariff data. The default is configurable via provisioning. Defaults to `None`.
* messages (`bool`, **optional**): Enables/disables the output of traffic messages. The default is configurable via provisioning. Defaults to `False`.
* frequency (`bool`, **optional**): Enables/disables the calculation of frequency information. Defaults to `True`.
### 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
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,
includeEarlier=include_earlier,
withICTAlternatives=ict_alternatives,
tariff=tariff,
trafficMessages=messages,
withFreq=frequency
)
find_exception(trips_raw)
for trip in trips_raw["Trip"]:
trips.append(Trip(trip))
return trips
def trip_recon(self,
context: Union[str, Journey],
date: Union[str, datetime] = None,
match_real_time: bool = None,
enable_replacements: bool = None,
arrival_dev_lower: int = None,
arrival_dev_upper: int = None,
departure_dev_lower: int = None,
departure_dev_upper: int = None,
passlist: bool = None,
passing_points: bool = False,
real_time_mode: Literal[RealTimeMode.FULL, RealTimeMode.INFOS, RealTimeMode.OFF, RealTimeMode.REALTIME, RealTimeMode.SERVER_DEFAULT] = None,
tariff: bool = 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.
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.
* date (`Union[str, datetime]`, **optional**): Sets the start date for which the departures shall be retrieved. Represented in the format `YYYY-MM-DD` or as a datetime object. This parameter will force the service to reconstruct the trip on that specific date. If the trip is not available on that date, because it does not operate, exception SvcNoResultError will be raised. Defaults to `None`.
* match_real_time (`bool`, **optional**): Whether the realtime type that journeys are based on be considered. Defaults to `None`.
* enable_replacements (`bool`, **optional**): If set to `True` replaces cancelled journeys with their replacement journeys if possible. Defaults to `None`.
* arrival_dev_lower (`int`, **optional**): Lower deviation in minutes within interval `[0, 720]` indicating "how much earlier than original arrival". Defaults to `None`.
* arrival_dev_upper (`int`, **optional**): Upper deviation in minutes within interval `[0, 720]` indicating "how much later than original arrival". Defaults to `None`.
* departure_dev_lower (`int`, **optional**): Lower deviation in minutes within interval `[0, 720]` indicating "how much earlier than original departure". Defaults to `None`.
* departure_dev_upper (`int`, **optional**): Upper deviation in minutes within interval `[0, 720]` indicating "how much later than original departure". Defaults to `None`.
* passlist (`bool`, **optional**): Enables/disables the return of the passlist for each leg of the trip. Defaults to `None`.
* passing_points (`bool`, **optional**): Enables/disables the return of stops having no alighting and boarding in its passlist for each leg of the trip. Needs passlist parameter enabled. Defaults to `False`.
* real_time_mode (`Literal[RealTimeMode.FULL, RealTimeMode.INFOS, RealTimeMode.OFF, RealTimeMode.REALTIME, RealTimeMode.SERVER_DEFAULT]`, **optional**): Set the realtime mode to be used. Read more about this in HAFAS ReST Documentation. Defaults to `None`.
* tariff (`bool`, **optional**): Enables/disables the output of tariff data. The default is configurable via provisioning. Defaults to `None`.
* messages (`bool`, **optional**): Enables/disables the output of traffic messages. The default is configurable via provisioning. Defaults to `False`.
### 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,
date=date,
matchRtType=match_real_time,
enableReplacements=enable_replacements,
arrL=arrival_dev_lower,
arrU=arrival_dev_upper,
depL=departure_dev_lower,
depU=departure_dev_upper,
passlist=passlist,
showPassingPoints=passing_points,
rtMode=real_time_mode,
tariff=tariff,
trafficMessages=messages,
)
find_exception(trips_raw)
for trip in trips_raw["Trip"]:
trips.append(Trip(trip))
return trips

1
pyrmv/classes/Journey.py
View File

@ -6,6 +6,7 @@ class Journey():
def __init__(self, data: dict): def __init__(self, data: dict):
self.stops = [] self.stops = []
self.ref = data["ref"]
self.direction = data["Directions"]["Direction"][0]["value"] self.direction = data["Directions"]["Direction"][0]["value"]
self.direction_flag = data["Directions"]["Direction"][0]["flag"] self.direction_flag = data["Directions"]["Direction"][0]["flag"]
self.messages = [] self.messages = []

9
pyrmv/classes/__init__.py
View File

@ -0,0 +1,9 @@
from .Board import BoardArrival, BoardDeparture, LineArrival, LineDeparture
from .Gis import Gis
from .Journey import Journey
from .Leg import Leg
from .Message import Message, Channel, Url
from .Stop import Stop, StopTrip
from .Ticket import Ticket
from .Trip import Trip
from .Client import Client

9
pyrmv/methods/__init__.py
View File

@ -1,9 +0,0 @@
from .board_arrival import board_arrival
from .board_departure import board_departure
from .him_search import him_search
from .journey_detail import journey_detail
from .stop_by_coords import stop_by_coords
from .stop_by_id import stop_by_id
from .stop_by_name import stop_by_name
from .trip_find import trip_find
from .trip_recon import trip_recon

71
pyrmv/methods/board_arrival.py
View File

@ -1,71 +0,0 @@
from datetime import datetime, timedelta
from typing import Union
from pyrmv.classes.Board import BoardArrival
from pyrmv.classes.Stop import Stop, StopTrip
from pyrmv.enums.board_type import BoardArrivalType
from pyrmv.raw import board_arrival as raw_board_arrival
from pyrmv.utility.find_exception import find_exception
try:
from typing import Literal
except ImportError:
from typing_extensions import Literal
def board_arrival(
access_id: str,
id: str = None, # type: ignore
id_ext: str = None, # type: ignore
direction: Union[str, Stop, StopTrip] = None, # type: ignore
date: Union[str, datetime] = None, # type: ignore
time: Union[str, datetime] = None, # type: ignore
duration: Union[int, timedelta] = 60,
journeys_max: int = -1,
operators: Union[str, list] = None, # type: ignore
lines: Union[str, list] = None, # type: ignore
passlist: bool = False,
board_type: Literal[BoardArrivalType.ARR, BoardArrivalType.ARR_EQUIVS, BoardArrivalType.ARR_MAST, BoardArrivalType.ARR_STATION] = BoardArrivalType.ARR,
) -> BoardArrival:
"""Method returns a board with arriving transport.
More detailed request is available as `raw.board_arrival()`, however returns `dict` instead of `Board`.
### Args:
* access_id (`str`): Access ID for identifying the requesting client. Get your key on [RMV website](https://opendata.rmv.de/site/start.html).
* id (`str`, **optional**): Specifies the station/stop ID. Such ID can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* id_ext (`str`, **optional**): Deprecated. Please use `id` as it supports external IDs. Specifies the external station/stop ID. Such ID can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* direction (`Union[str, Stop, StopTrip]`, **optional**): If only vehicles departing or arriving from a certain direction shall be returned, specify the direction by giving the station/stop ID of the last stop on the journey or the Stop object itself. Defaults to `None`.
* date (`Union[str, datetime]`, **optional**): Sets the start date for which the departures shall be retrieved. Represented in the format YYYY-MM-DD or as a datetime object. By default the current server date is used. Defaults to `None`.
* time (`Union[str, datetime]`, **optional**): Sets the start time for which the departures shall be retrieved. Represented in the format hh:mm[:ss] in 24h nomenclature or as a datetime object. Seconds will be ignored for requests. By default the current server time is used. Defaults to `None`.
* duration (`Union[int, timedelta]`, **optional**): Set the interval size in minutes. Can also be provided as a timedelta. Defaults to `60`.
* journeys_max (`int`, **optional**): Maximum number of journeys to be returned. If no value is defined or -1, all departing/arriving services within the duration sized period are returned. Defaults to `-1`.
* operators (`Union[str, list]`, **optional**): Only journeys provided by the given operators are part of the result. To filter multiple operators, separate the codes by comma or provide a list. If the operator should not be part of the be trip, negate it by putting ! in front of it. Example: Filter for operator A and B: `operators=[A,B]`. Defaults to `None`.
* lines (`Union[str, list]`, **optional**): Only journeys running the given line are part of the result. To filter multiple lines, provide a list or separate the codes by comma or provide a list. If the line should not be part of the be trip, negate it by putting ! in front of it. Defaults to `None`.
* passlist (`bool`, **optional**): Include a list of all passed waystops. Defaults to `False`.
* board_type (`Union[BoardArrivalType.ARR, BoardArrivalType.ARR_EQUIVS, BoardArrivalType.ARR_MAST, BoardArrivalType.ARR_STATION]`, optional): Set the station arrival board type to be used. Defaults to `BoardArrivalType.ARR`.
### Returns:
* BoardArrival: Instance of `BoardArrival` object.
"""
if (isinstance(direction, Stop) or isinstance(direction, StopTrip)):
direction = direction.id
board_raw = raw_board_arrival(
accessId=access_id,
id=id,
extId=id_ext,
direction=direction,
date=date,
time=time,
duration=duration,
maxJourneys=journeys_max,
operators=operators,
lines=lines,
passlist=passlist,
boardType=board_type.code
)
find_exception(board_raw)
return BoardArrival(board_raw, access_id)

71
pyrmv/methods/board_departure.py
View File

@ -1,71 +0,0 @@
from datetime import datetime, timedelta
from typing import Union
from pyrmv.classes.Board import BoardDeparture
from pyrmv.classes.Stop import Stop, StopTrip
from pyrmv.enums.board_type import BoardDepartureType
from pyrmv.raw import board_departure as raw_board_departure
from pyrmv.utility.find_exception import find_exception
try:
from typing import Literal
except ImportError:
from typing_extensions import Literal
def board_departure(
access_id: str,
id: str = None, # type: ignore
id_ext: str = None, # type: ignore
direction: Union[str, Stop, StopTrip] = None, # type: ignore
date: Union[str, datetime] = None, # type: ignore
time: Union[str, datetime] = None, # type: ignore
duration: Union[int, timedelta] = 60,
journeys_max: int = -1,
operators: Union[str, list] = None, # type: ignore
lines: Union[str, list] = None, # type: ignore
passlist: bool = False,
board_type: Literal[BoardDepartureType.DEP, BoardDepartureType.DEP_EQUIVS, BoardDepartureType.DEP_MAST, BoardDepartureType.DEP_STATION] = BoardDepartureType.DEP,
) -> BoardDeparture:
"""Method returns a board with departing transport.
More detailed request is available as `raw.board_departure()`, however returns `dict` instead of `Board`.
### Args:
* access_id (`str`): Access ID for identifying the requesting client. Get your key on [RMV website](https://opendata.rmv.de/site/start.html).
* id (`str`, **optional**): Specifies the station/stop ID. Such ID can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* id_ext (`str`, **optional**): Deprecated. Please use `id` as it supports external IDs. Specifies the external station/stop ID. Such ID can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* direction (`Union[str, Stop, StopTrip]`, **optional**): If only vehicles departing or arriving from a certain direction shall be returned, specify the direction by giving the station/stop ID of the last stop on the journey or the Stop object itself. Defaults to `None`.
* date (`Union[str, datetime]`, **optional**): Sets the start date for which the departures shall be retrieved. Represented in the format YYYY-MM-DD or as a datetime object. By default the current server date is used. Defaults to `None`.
* time (`Union[str, datetime]`, **optional**): Sets the start time for which the departures shall be retrieved. Represented in the format hh:mm[:ss] in 24h nomenclature or as a datetime object. Seconds will be ignored for requests. By default the current server time is used. Defaults to `None`.
* duration (`Union[int, timedelta]`, **optional**): Set the interval size in minutes. Can also be provided as a timedelta. Defaults to `60`.
* journeys_max (`int`, **optional**): Maximum number of journeys to be returned. If no value is defined or -1, all departing/arriving services within the duration sized period are returned. Defaults to `-1`.
* operators (`Union[str, list]`, **optional**): Only journeys provided by the given operators are part of the result. To filter multiple operators, separate the codes by comma or provide a list. If the operator should not be part of the be trip, negate it by putting ! in front of it. Example: Filter for operator A and B: `operators=[A,B]`. Defaults to `None`.
* lines (`Union[str, list]`, **optional**): Only journeys running the given line are part of the result. To filter multiple lines, provide a list or separate the codes by comma or provide a list. If the line should not be part of the be trip, negate it by putting ! in front of it. Defaults to `None`.
* passlist (`bool`, **optional**): Include a list of all passed waystops. Defaults to `False`.
* board_type (`Union[BoardDepartureType.DEP, BoardDepartureType.DEP_EQUIVS, BoardDepartureType.DEP_MAST, BoardDepartureType.DEP_STATION]`, optional): Set the station departure board type to be used. Defaults to `BoardDepartureType.DEP`.
### Returns:
* BoardDeparture: Instance of `BoardDeparture` object.
"""
if (isinstance(direction, Stop) or isinstance(direction, StopTrip)):
direction = direction.id
board_raw = raw_board_departure(
accessId=access_id,
id=id,
extId=id_ext,
direction=direction,
date=date,
time=time,
duration=duration,
maxJourneys=journeys_max,
operators=operators,
lines=lines,
passlist=passlist,
boardType=board_type.code
)
find_exception(board_raw)
return BoardDeparture(board_raw, access_id)

42
pyrmv/methods/him_search.py
View File

@ -1,42 +0,0 @@
from datetime import datetime
from typing import OrderedDict, Union
from pyrmv.classes.Stop import Stop
from pyrmv.errors.not_ready import NotReadyYetError
from pyrmv.enums.search_mode import SearchMode
from pyrmv.enums.aff_journey_mode import AffectedJourneyMode
from pyrmv.enums.aff_journey_stop_mode import AffectedJourneyStopMode
try:
from typing import Literal
except ImportError:
from typing_extensions import Literal
def him_search(
access_id: str,
date_begin: Union[str, datetime] = None,
date_end: Union[str, datetime] = None,
time_begin: Union[str, datetime] = None,
time_end: Union[str, datetime] = None,
weekdays: Union[str, OrderedDict[str, bool]] = None,
ids: list = None,
operators: list = None,
categories: list = None,
channels: list = None,
companies: list = None,
lines: list = None,
lineids: list = None,
stations: list = None,
station_from: Union[str, Stop] = None,
station_to: Union[str, Stop] = None,
bothways: bool = None,
trainnames: list = None,
search_mode: Literal[SearchMode.MATCH, SearchMode.NOMATCH, SearchMode.TFMATCH] = None,
affected_journey_mode: Literal[AffectedJourneyMode.ALL, AffectedJourneyMode.OFF] = None,
affected_journey_stop_mode: Literal[AffectedJourneyStopMode.ALL, AffectedJourneyStopMode.IMP, AffectedJourneyStopMode.OFF] = None,
priority_min: int = None,
priority_max: int = None
) -> None: #List[Message]:
raise NotReadyYetError()

63
pyrmv/methods/journey_detail.py
View File

@ -1,63 +0,0 @@
from datetime import datetime
from typing import Union
from pyrmv.classes.Journey import Journey
from pyrmv.enums.rt_mode import RealTimeMode
from pyrmv.raw.journey_detail import journey_detail as raw_journey_detail
from pyrmv.utility.find_exception import find_exception
try:
from typing import Literal
except ImportError:
from typing_extensions import Literal
def journey_detail(
access_id: str,
id: str,
date: Union[str, datetime] = None, # type: ignore
real_time_mode: Literal[RealTimeMode.FULL, RealTimeMode.INFOS, RealTimeMode.OFF, RealTimeMode.REALTIME, RealTimeMode.SERVER_DEFAULT] = None, # type: ignore
from_id: str = None, # type: ignore
from_index: int = None, # type: ignore
to_id: str = None, # type: ignore
to_index: int = None # type: ignore
) -> 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.
More detailed request is available as `raw.journey_detail()`, however returns `dict` instead of `Journey`.
### Args:
* access_id (`str`): Access ID for identifying the requesting client. Get your key on [RMV website](https://opendata.rmv.de/site/start.html).
* id (`str`): Specifies the internal journey id of the journey shall be retrieved. Maximum length 512.
* date (`Union[str, datetime]`, **optional**): Day of operation. Represented in the format `YYYY-MM-DD` or as a datetime object. By default the current server date is used. Defaults to `None`.
* real_time_mode (`Literal[RealTimeMode.FULL, RealTimeMode.INFOS, RealTimeMode.OFF, RealTimeMode.REALTIME, RealTimeMode.SERVER_DEFAULT]`, **optional**): Set the realtime mode to be used. Read more about this in HAFAS ReST Documentation. Defaults to `None`.
* from_id (`str`, **optional**): Specifies the station/stop ID the partial itinerary shall start from. Defaults to `None`.
* from_index (`int`, **optional**): Specifies the station/stop index the partial itinerary shall start from. Defaults to `None`.
* to_id (`str`, **optional**): Specifies the station/stop ID the partial itinerary shall end at. Defaults to `None`.
* to_index (`int`, **optional**): Specifies the station/stop index the partial itinerary shall end at. Defaults to `None`.
### Returns:
* Journey: Instance of `Journey` object.
"""
if real_time_mode == None:
real_time_mode = None
else:
real_time_mode = real_time_mode.code
journey_raw = raw_journey_detail(
accessId=access_id,
id=id,
date=date,
rtMode=real_time_mode,
fromId=from_id,
fromIdx=from_index,
toId=to_id,
toIdx=to_index
)
find_exception(journey_raw)
return Journey(journey_raw)

70
pyrmv/methods/stop_by_coords.py
View File

@ -1,70 +0,0 @@
from typing import List, Union
from pyrmv.classes.Stop import Stop
from pyrmv.enums.location_type import LocationType
from pyrmv.enums.lang import Language
from pyrmv.enums.selection_mode import SelectionMode
from pyrmv.raw.stop_by_coords import stop_by_coords as raw_stop_by_coords
from pyrmv.utility.find_exception import find_exception
try:
from typing import Literal
except ImportError:
from typing_extensions import Literal
def stop_by_coords(
access_id: str,
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: Literal[SelectionMode.SLCT_A, SelectionMode.SLCT_N] = None, # type: ignore
) -> 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.
More detailed request is available as `raw.stop_by_coords()`, however returns `dict` instead of `List[Stop]`.
### Args:
* access_id (`str`): Access ID for identifying the requesting client. Get your key on [RMV website](https://opendata.rmv.de/site/start.html).
* coords_lat (`Union[str, float]`): Latitude of centre coordinate.
* coords_lon (`Union[str, float]`): Longitude of centre coordinate.
* 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`.
* radius (`Union[int, float]`, **optional**): Search radius in meter around the given coordinate if any. Defaults to `1000`.
* max_number (`int`, **optional**): Maximum number of returned stops. Defaults to `10`.
* stop_type (`Literal[LocationType.S, LocationType.P, LocationType.SP, LocationType.SE, LocationType.SPE]`, **optional**): Type filter for location types. Defaults to `LocationType.S`.
* selection_mode (`Literal[SelectionMode.SLCT_A, SelectionMode.SLCT_N]`, **optional**): Selection mode for locations. `SelectionMode.SLCT_N`: Not selectable, `SelectionMode.SLCT_A`: Selectable. Defaults to `None`.
### 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
stops = []
stops_raw = raw_stop_by_coords(
accessId=access_id,
originCoordLat=coords_lat,
originCoordLong=coords_lon,
lang=lang.code,
radius=radius,
maxNo=max_number,
stopType=stop_type.code,
locationSelectionMode=selection_mode
)
find_exception(stops_raw)
for stop in stops_raw["stopLocationOrCoordLocation"]:
if "StopLocation" in stop:
stops.append(Stop(stop["StopLocation"]))
elif "CoordLocation" in stop:
stops.append(Stop(stop["CoordLocation"]))
return stops

48
pyrmv/methods/stop_by_id.py
View File

@ -1,48 +0,0 @@
from typing import List, Union
from pyrmv.classes.Stop import Stop
from pyrmv.enums.lang import Language
from pyrmv.raw.stop_by_name import stop_by_name as raw_stop_by_name
from pyrmv.utility.find_exception import find_exception
try:
from typing import Literal
except ImportError:
from typing_extensions import Literal
def stop_by_id(
access_id: str,
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:
* access_id (`str`): Access ID for identifying the requesting client. Get your key on [RMV website](https://opendata.rmv.de/site/start.html).
* 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:
* Stop: Instance of `Stop` object or `None` if not found.
"""
stops_raw = raw_stop_by_name(
accessId=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 "StopLocation" in stop:
return Stop(stop["StopLocation"])
elif "CoordLocation" in stop:
return Stop(stop["CoordLocation"])
else:
return None
else:
return 0

84
pyrmv/methods/stop_by_name.py
View File

@ -1,84 +0,0 @@
from typing import List, Union
from pyrmv.classes.Stop import Stop
from pyrmv.enums.location_type import LocationType
from pyrmv.enums.lang import Language
from pyrmv.enums.selection_mode import SelectionMode
from pyrmv.enums.filter_mode import FilterMode
from pyrmv.raw.stop_by_name import stop_by_name as raw_stop_by_name
from pyrmv.utility.find_exception import find_exception
try:
from typing import Literal
except ImportError:
from typing_extensions import Literal
def stop_by_name(
access_id: str,
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: Literal[SelectionMode.SLCT_A, SelectionMode.SLCT_N] = None, # type: ignore
coord_lat: Union[str, float] = None, # type: ignore
coord_lon: Union[str, float] = None, # type: ignore
radius: Union[int, float] = 1000,
refine_id: str = None, # type: ignore
stations: Union[str, list] = None, # type: ignore
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.
More detailed request is available as `raw.stop_by_name()`, however returns `dict` instead of `List[Stop]`.
### Args:
* access_id (`str`): Access ID for identifying the requesting client. Get your key on [RMV website](https://opendata.rmv.de/site/start.html).
* 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`.
* max_number (`int`, **optional**): Maximum number of returned stops. In range 1-1000. Defaults to `10`.
* stop_type (`Literal[LocationType.A, LocationType.ALL, LocationType.AP, LocationType.P, LocationType.S, LocationType.SA, LocationType.SP]`, **optional**): Type filter for location types. Defaults to `LocationType.ALL`.
* selection_mode (`Literal[SelectionMode.SLCT_A, SelectionMode.SLCT_N]`, **optional**): Selection mode for locations. `SelectionMode.SLCT_N`: Not selectable, `SelectionMode.SLCT_A`: Selectable. Defaults to `None`.
* coord_lat (`Union[str, float]`, **optional**): Latitude of centre coordinate. Defaults to `None`.
* coord_lon (`Union[str, float]`, **optional**): Longitude of centre coordinate. Defaults to `None`.
* radius (`Union[int, float]`, **optional**): Search radius in meter around the given coordinate if any. Defaults to `1000`.
* refine_id (`str`, **optional**): In case of an refinable location, this value takes the ID of the refinable one of a previous result. Defaults to `None`.
* stations (`Union[str, list]`, **optional**): Filter for stations. Matches if the given value is prefix of any station in the itinerary. As a list or as a string separated by comma. Defaults to `None`.
* filter_mode (`Literal[FilterMode.DIST_PERI, FilterMode.EXCL_PERI, FilterMode.SLCT_PERI]`, **optional**): Filter modes for nearby searches. Defaults to `FilterMode.DIST_PERI`.
### 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
stops = []
stops_raw = raw_stop_by_name(
accessId=access_id,
inputString=query,
lang=lang.code,
maxNo=max_number,
stopType=stop_type.code,
locationSelectionMode=selection_mode,
coordLat=coord_lat,
coordLong=coord_lon,
radius=radius,
refineId=refine_id,
stations=stations,
filterMode=filter_mode.code
)
find_exception(stops_raw)
for stop in stops_raw["stopLocationOrCoordLocation"]:
if "StopLocation" in stop:
stops.append(Stop(stop["StopLocation"]))
elif "CoordLocation" in stop:
stops.append(Stop(stop["CoordLocation"]))
return stops

205
pyrmv/methods/trip_find.py
View File

@ -1,205 +0,0 @@
from datetime import datetime
from typing import List, Union
from pyrmv.classes.Trip import Trip
from pyrmv.raw.trip_find import trip_find as raw_trip_find
from pyrmv.enums.rt_mode import RealTimeMode
from pyrmv.enums.lang import Language
from pyrmv.utility.find_exception import find_exception
try:
from typing import Literal
except ImportError:
from typing_extensions import Literal
def trip_find(
access_id: 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,
origin_id: str = None, # type: ignore
origin_id_ext: str = None, # type: ignore
origin_coord_lat: Union[str, float] = None, # type: ignore
origin_coord_lon: Union[str, float] = None, # type: ignore
origin_coord_name: str = None, # type: ignore
destination_id: str = None, # type: ignore
destination_id_ext: str = None, # type: ignore
destination_coord_lat: Union[str, float] = None, # type: ignore
destination_coord_lon: Union[str, float] = None, # type: ignore
destination_coord_name: str = None, # type: ignore
via: str = None, # type: ignore
via_id: str = None, # type: ignore
via_gis: str = None, # type: ignore
via_wait_time: int = 0,
avoid: str = None, # type: ignore
avoid_id: str = None, # type: ignore
change_time_percent: int = 100,
change_time_min: int = None, # type: ignore
change_time_max: int = None, # type: ignore
change_time_add: int = None, # type: ignore
change_max: int = None, # type: ignore
date: Union[str, datetime] = None, # type: ignore
time: Union[str, datetime] = None, # type: ignore
search_arrival: bool = False,
trips_after_time: int = None, # type: ignore
trips_before_time: int = None, # type: ignore
context: str = None, # type: ignore
passlist: bool = False,
operators: Union[str, list] = None, # type: ignore
lines: Union[str, list] = None, # type: ignore
lineids: Union[str, list] = None, # type: ignore
iv_include: bool = False,
iv_only: bool = False,
bike_carriage: bool = False,
passing_points: bool = False,
real_time_mode: Literal[RealTimeMode.FULL, RealTimeMode.INFOS, RealTimeMode.OFF, RealTimeMode.REALTIME, RealTimeMode.SERVER_DEFAULT] = None, # type: ignore
include_earlier: bool = False,
ict_alternatives: bool = False,
tariff: bool = None, # type: ignore
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]`.
### Args:
* access_id (`str`): Access ID for identifying the requesting client. Get your key on [RMV website](https://opendata.rmv.de/site/start.html).
* 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`.
* origin_id (`str`, **optional**): Specifies the station/stop ID of the origin for the trip. Such ID can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* origin_id_ext (`str`, **optional**): Deprecated. Please use originId as it supports external IDs. Specifies the external station/stop ID of the origin for the trip. Such ID can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* origin_coord_lat (`Union[str, float]`, **optional**): Latitude of station/stop coordinate of the trip's origin. The coordinate can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* origin_coord_lon (`Union[str, float]`, **optional**): Longitude of station/stop coordinate of the trip's origin. The coordinate can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* origin_coord_name (`str`, **optional**): Name of the trip's origin if coordinate cannot be resolved to an address or poi. Defaults to `None`.
* destination_id (`str`, **optional**): Specifies the station/stop ID of the destination for the trip. Such ID can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* destination_id_ext (`str`, **optional**): Deprecated. Please use destId as it supports external IDs. Specifies the external station/stop ID of the destination for the trip. Such ID can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* destination_coord_lat (`Union[str, float]`, **optional**): Latitude of station/stop coordinate of the trip's destination. The coordinate can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* destination_coord_lon (`Union[str, float]`, **optional**): Longitude of station/stop coordinate of the trip's destination. The coordinate can be retrieved from stop_by_name() or stop_by_coords(). Defaults to `None`.
* destination_coord_name (`str`, **optional**): Name of the trip's destination if coordinate cannot be resolved to an address or poi. Defaults to `None`.
* via (`str`, **optional**): Complex structure to provide multiple via points separated by semicolon. This structure is build like this: `viaId|waittime|viastatus|products|direct|sleepingCar|couchetteCoach|attributes`. Read more about this in HAFAS ReST Documentation. Defaults to `None`.
* via_id (`str`, **optional**): ID of a station/stop used as a via for the trip. Specifying a via station forces the trip search to look for trips which must pass through this station. Such ID can be retrieved from stop_by_name() or stop_by_coords(). If `via` is used, `via_id` and `via_wait_time ` are having no effect. Defaults to `None`.
* via_gis (`str`, **optional**): Complex structure to provide multiple GIS via locations separated by semicolon. This structure is build like this: `locationId|locationMode|transportMode|placeType|usageType|mode|durationOfStay`. Read more about this in HAFAS ReST Documentation. Defaults to `None`.
* via_wait_time (`int`, **optional**): Defines the waiting time spent at via station in minutes. If `via` is used, `via_id` and `via_wait_time` are having no effect. Defaults to 0.
* avoid (`str`, **optional**): Complex structure to provide multiple points to be avoided separated by semicolon. This structure is build like this: `avoidId|avoidstatus` avoidId: id, extId or altId of the avoid, mandatory avoidstatus: one of NPAVM (do not run through if this is a meta station), NPAVO (do not run through), NCAVM (do not change if this is a meta station), NCAVO (do not change), optional but defaults to NCAVM Example: Just define three avoids by extId: `avoid="801234;801235;801236"`. Defaults to `None`.
* avoid_id (`str`, **optional**): ID of a station/stop to be avoided as transfer stop for the trip. Such ID can be retrieved from stop_by_name() or stop_by_coords(). If `avoid` is used, `avoid_id` has no effect. Defaults to `None`.
* change_time_percent (`int`, **optional**): Configures the walking speed when changing from one leg of the journey to the next one. It extends the time required for changes by a specified percentage. A value of 200 doubles the change time as initially calculated by the system. In the response, change time is presented in full minutes. If the calculation based on changeTime-Percent does not result in a full minute, it is rounded using "round half up" method. Defaults to `100`.
* change_time_min (`int`, **optional**): Minimum change time at stop in minutes. Defaults to `None`.
* change_time_max (`int`, **optional**): Maximum change time at stop in minutes. Defaults to `None`.
* change_time_add (`int`, **optional**): This amount of minutes is added to the change time at each stop. Defaults to `None`.
* change_max (`int`, **optional**): Maximum number of changes. In range 0-11. Defaults to `None`.
* date (`Union[str, datetime]`, **optional**): Sets the start date for which the departures shall be retrieved. Represented in the format `YYYY-MM-DD` or as a datetime object. By default the current server date is used. Defaults to `None`.
* time (`Union[str, datetime]`, **optional**): Sets the start time for which the departures shall be retrieved. Represented in the format `hh:mm[:ss]` in 24h nomenclature or as a datetime object. Seconds will be ignored for requests. By default the current server time is used. Defaults to `None`.
* search_arrival (`bool`, **optional**): If set, the date and time parameters specify the arrival time for the trip search instead of the departure time. Defaults to `False`.
* trips_after_time (`int`, **optional**): Minimum number of trips after the search time. Sum of `trips_after_time` and `trips_before_time` has to be less or equal 6. Read more about this in HAFAS ReST Documentation. In range 1-6. Defaults to `None`.
* trips_before_time (`int`, **optional**): Minimum number of trips before the search time. Sum of `trips_after_time` and `trips_before_time` has to be less or equal 6. Read more about this in HAFAS ReST Documentation. In range 0-6. Defaults to `None`.
* context (`str`, **optional**): Defines the starting point for the scroll back or forth operation. Use the scrB value from a previous result to scroll backwards in time and use the scrF value to scroll forth. Defaults to `None`.
* passlist (`bool`, **optional**): Enables/disables the return of the passlist for each leg of the trip. Defaults to `False`.
* operators (`Union[str, list]`, **optional**): Only trips provided by the given operators are part of the result. If the operator should not be part of the be trip, negate it by putting ! in front of it. Example: Filter for operator A and B: `operators=["A","B"]`. Defaults to `None`.
* lines (`Union[str, list]`, **optional**): Only journeys running the given line are part of the result. If the line should not be part of the be trip, negate it by putting ! in front of it. Defaults to `None`.
* lineids (`Union[str, list]`, **optional**): Only journeys running the given line (identified by its line ID) are part of the result. If the line should not be part of the be trip, negate it by putting ! in front of it. Defaults to `None`.
* iv_include (`bool`, **optional**): Enables/disables search for individual transport routes. Defaults to `False`.
* iv_only (`bool`, **optional**): Enables/disables search for individual transport routes only. Defaults to `False`.
* bike_carriage (`bool`, **optional**): Enables/disables search for trips explicit allowing bike carriage. This will only work in combination with `change_max=0` as those trips are always meant to be direct connections. Defaults to `False`.
* passing_points (`bool`, **optional**): Enables/disables the return of stops having no alighting and boarding in its passlist for each leg of the trip. Needs passlist enabled. Defaults to `False`.
* real_time_mode (`Literal[RealTimeMode.FULL, RealTimeMode.INFOS, RealTimeMode.OFF, RealTimeMode.REALTIME, RealTimeMode.SERVER_DEFAULT]`, **optional**): Set the realtime mode to be used. Read more about this in HAFAS ReST Documentation. Defaults to `None`.
* include_earlier (`bool`, **optional**): Disables search optimization in relation of duration. Defaults to `False`.
* ict_alternatives (`bool`, **optional**): Enables/disables the search for alternatives with individualized change times (ICT). Defaults to `False`.
* tariff (`bool`, **optional**): Enables/disables the output of tariff data. The default is configurable via provisioning. Defaults to `None`.
* messages (`bool`, **optional**): Enables/disables the output of traffic messages. The default is configurable via provisioning. Defaults to `False`.
* frequency (`bool`, **optional**): Enables/disables the calculation of frequency information. Defaults to `True`.
### 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
trips = []
trips_raw = raw_trip_find(
accessId=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,
includeEarlier=include_earlier,
withICTAlternatives=ict_alternatives,
tariff=tariff,
trafficMessages=messages,
withFreq=frequency
)
find_exception(trips_raw)
for trip in trips_raw["Trip"]:
trips.append(Trip(trip))
return trips

88
pyrmv/methods/trip_recon.py
View File

@ -1,88 +0,0 @@
from datetime import datetime
from typing import List, Union
from pyrmv.classes.Journey import Journey
from pyrmv.classes.Trip import Trip
from pyrmv.enums.rt_mode import RealTimeMode
from pyrmv.raw.trip_recon import trip_recon as raw_trip_recon
from pyrmv.utility.find_exception import find_exception
try:
from typing import Literal
except ImportError:
from typing_extensions import Literal
def trip_recon(
access_id: str,
context: Union[str, Journey],
date: Union[str, datetime] = None,
match_real_time: bool = None,
enable_replacements: bool = None,
arrival_dev_lower: int = None,
arrival_dev_upper: int = None,
departure_dev_lower: int = None,
departure_dev_upper: int = None,
passlist: bool = None,
passing_points: bool = False,
real_time_mode: Literal[RealTimeMode.FULL, RealTimeMode.INFOS, RealTimeMode.OFF, RealTimeMode.REALTIME, RealTimeMode.SERVER_DEFAULT] = None,
tariff: bool = 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.
More detailed request is available as `raw.trip_recon()`, however returns `dict` instead of `List[Trip]`.
### Args:
* access_id (`str`): Access ID for identifying the requesting client. Get your key on [RMV website](https://opendata.rmv.de/site/start.html).
* context (`Union[str, Journey]`): Specifies the reconstruction context.
* date (`Union[str, datetime]`, **optional**): Sets the start date for which the departures shall be retrieved. Represented in the format `YYYY-MM-DD` or as a datetime object. This parameter will force the service to reconstruct the trip on that specific date. If the trip is not available on that date, because it does not operate, exception SvcNoResultError will be raised. Defaults to `None`.
* match_real_time (`bool`, **optional**): Whether the realtime type that journeys are based on be considered. Defaults to `None`.
* enable_replacements (`bool`, **optional**): If set to `True` replaces cancelled journeys with their replacement journeys if possible. Defaults to `None`.
* arrival_dev_lower (`int`, **optional**): Lower deviation in minutes within interval `[0, 720]` indicating "how much earlier than original arrival". Defaults to `None`.
* arrival_dev_upper (`int`, **optional**): Upper deviation in minutes within interval `[0, 720]` indicating "how much later than original arrival". Defaults to `None`.
* departure_dev_lower (`int`, **optional**): Lower deviation in minutes within interval `[0, 720]` indicating "how much earlier than original departure". Defaults to `None`.
* departure_dev_upper (`int`, **optional**): Upper deviation in minutes within interval `[0, 720]` indicating "how much later than original departure". Defaults to `None`.
* passlist (`bool`, **optional**): Enables/disables the return of the passlist for each leg of the trip. Defaults to `None`.
* passing_points (`bool`, **optional**): Enables/disables the return of stops having no alighting and boarding in its passlist for each leg of the trip. Needs passlist parameter enabled. Defaults to `False`.
* real_time_mode (`Literal[RealTimeMode.FULL, RealTimeMode.INFOS, RealTimeMode.OFF, RealTimeMode.REALTIME, RealTimeMode.SERVER_DEFAULT]`, **optional**): Set the realtime mode to be used. Read more about this in HAFAS ReST Documentation. Defaults to `None`.
* tariff (`bool`, **optional**): Enables/disables the output of tariff data. The default is configurable via provisioning. Defaults to `None`.
* messages (`bool`, **optional**): Enables/disables the output of traffic messages. The default is configurable via provisioning. Defaults to `False`.
### 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, Journey):
context = context.ctx_recon
trips = []
trips_raw = raw_trip_recon(
accessId=access_id,
ctx=context,
date=date,
matchRtType=match_real_time,
enableReplacements=enable_replacements,
arrL=arrival_dev_lower,
arrU=arrival_dev_upper,
depL=departure_dev_lower,
depU=departure_dev_upper,
passlist=passlist,
showPassingPoints=passing_points,
rtMode=real_time_mode,
tariff=tariff,
trafficMessages=messages,
)
find_exception(trips_raw)
for trip in trips_raw["Trip"]:
trips.append(Trip(trip))
return trips

3
setup.py
View File

@ -5,7 +5,7 @@ setup(
version="0.3.0", version="0.3.0",
author="Profitroll", author="Profitroll",
description="Small module that makes your journey with RMV REST API somehow easier.", description="Small module that makes your journey with RMV REST API somehow easier.",
long_description="## PythonRMV\n\nSmall module that makes your journey with RMV REST API somehow easier. Based fully on official RMV API reference and HAFAS documentation.\n\n# Usage\n\n```py\nimport pyrmv\n\n# Set API key\naccess_id = \"Something\"\n\n# Get origin's and destination's location\norigin = pyrmv.stop_by_name(access_id, \"Frankfurt Hauptbahnhof\", max_number=3)[0]\ndestination = pyrmv.stop_by_coords(access_id, 50.099613, 8.685449, max_number=3)[0]\n\n# Find a trip by locations got\ntrip = pyrmv.trip_find(access_id, origin_id=origin.id, dest_id=destination.id)\n```\n\n# Frequently Asked Questions\n\n- [Why are there raw versions and formatted ones?](#why-are-there-raw-versions-and-formatted-ones)\n- [Some methods work slightly different](#some-methods-work-slightly-different)\n- [Documentation is not perfectly clear](#documentation-is-not-perfectly-clear)\n\n## Why are there raw versions and formatted ones?\n\nFor the purposes of my projects I don't really need all the stuff RMV gives (even though it's not much).\nI only need some specific things. However I do understand that in some cases other users may find\nthose methods quite useful so I implemented them as well.\n\n\n## Some methods work slightly different\n\nCan be. Not all function arguments written may work perfectly because I simply did not test each and\nevery request. Some of arguments may be irrelevant in my use-case and the others are used quite rare at all.\nJust [make an issue](https://git.end-play.xyz/profitroll/PythonRMV/issues/new) and I'll implement it correct when I'll have some free time.\n\n## Documentation is not perfectly clear\n\nOf course docs cannot be perfect as a python docstring, especially if sometimes I don't\nknow how things should correctly work. That's why you get HAFAS API docs in addition to your\nRMV API key. Just use my functions together with those docs, if you want to build something\nreally sophisticated. However I'm not sure whether RMV supports that many HAFAS features publicly.\n\n# To-Do\n## General\n- [ ] Docs in Wiki\n\n## Raw methods\n- [x] arrivalBoard (board_arrival) \n- [x] departureBoard (board_departure) \n- [x] himsearch (him_search) \n- [x] journeyDetail (journey_detail)\n- [x] location.nearbystops (stop_by_coords) \n- [x] location.name (stop_by_name) \n- [x] trip (trip_find) \n- [x] recon (trip_recon)\n\n## Normal methods\n- [x] arrivalBoard (board_arrival) \n- [x] departureBoard (board_departure) \n- [ ] himsearch (him_search) \n- [x] journeyDetail (journey_detail)\n- [x] location.nearbystops (stop_by_coords) \n- [x] location.name (stop_by_name) \n- [x] trip (trip_find) \n- [x] recon (trip_recon)", long_description="## PythonRMV\n\nSmall module that makes your journey with RMV REST API somehow easier. Based fully on official RMV API reference and HAFAS documentation.\n\n# Usage\n\n```py\nimport pyrmv\n\n# Define a Client with API key\nclient = pyrmv.Client(\"AccessId\")\n\n# Get origin's and destination's location\norigin = client.stop_by_name(\"Frankfurt Hauptbahnhof\", max_number=3)[0]\ndestination = client.stop_by_coords(50.099613, 8.685449, max_number=3)[0]\n\n# Find a trip by locations got\ntrip = client.trip_find(origin_id=origin.id, dest_id=destination.id)\n```\n\n# Frequently Asked Questions\n\n- [Why are there raw versions and formatted ones?](#why-are-there-raw-versions-and-formatted-ones)\n- [Some methods work slightly different](#some-methods-work-slightly-different)\n- [Documentation is not perfectly clear](#documentation-is-not-perfectly-clear)\n\n## Why are there raw versions and formatted ones?\n\nFor the purposes of my projects I don't really need all the stuff RMV gives (even though it's not much).\nI only need some specific things. However I do understand that in some cases other users may find\nthose methods quite useful so I implemented them as well.\n\n\n## Some methods work slightly different\n\nCan be. Not all function arguments written may work perfectly because I simply did not test each and\nevery request. Some of arguments may be irrelevant in my use-case and the others are used quite rare at all.\nJust [make an issue](https://git.end-play.xyz/profitroll/PythonRMV/issues/new) and I'll implement it correct when I'll have some free time.\n\n## Documentation is not perfectly clear\n\nOf course docs cannot be perfect as a python docstring, especially if sometimes I don't\nknow how things should correctly work. That's why you get HAFAS API docs in addition to your\nRMV API key. Just use my functions together with those docs, if you want to build something\nreally sophisticated. However I'm not sure whether RMV supports that many HAFAS features publicly.\n\n# To-Do\n## General\n- [ ] Docs in Wiki\n\n## Raw methods\n- [x] arrivalBoard (board_arrival) \n- [x] departureBoard (board_departure) \n- [x] himsearch (him_search) \n- [x] journeyDetail (journey_detail)\n- [x] location.nearbystops (stop_by_coords) \n- [x] location.name (stop_by_name) \n- [x] trip (trip_find) \n- [x] recon (trip_recon)\n\n## Normal methods\n- [x] arrivalBoard (board_arrival) \n- [x] departureBoard (board_departure) \n- [ ] himsearch (him_search) \n- [x] journeyDetail (journey_detail)\n- [x] location.nearbystops (stop_by_coords) \n- [x] location.name (stop_by_name) \n- [x] trip (trip_find) \n- [x] recon (trip_recon)",
long_description_content_type="text/markdown", long_description_content_type="text/markdown",
author_email="profitroll@end-play.xyz", author_email="profitroll@end-play.xyz",
url="https://git.end-play.xyz/profitroll/PythonRMV", url="https://git.end-play.xyz/profitroll/PythonRMV",
@ -20,7 +20,6 @@ setup(
"pyrmv.const", "pyrmv.const",
"pyrmv.enums", "pyrmv.enums",
"pyrmv.errors", "pyrmv.errors",
"pyrmv.methods",
"pyrmv.utility", "pyrmv.utility",
"pyrmv.classes" "pyrmv.classes"
], ],

20
test.py
View File

@ -1,13 +1,12 @@
from datetime import datetime, timedelta from datetime import datetime, timedelta
from os import makedirs from os import makedirs
from pyrmv.enums.rt_mode import RealTimeMode from pyrmv.enums import RealTimeMode
from typing import Union from typing import Union
from ujson import loads, dumps, JSONDecodeError from ujson import loads, dumps, JSONDecodeError
from test_colors import * from test_colors import *
import pyrmv import pyrmv
try: try:
with open("test_key.json", mode="r", encoding="utf-8") as file: with open("test_key.json", mode="r", encoding="utf-8") as file:
key = loads(file.read())["key"] key = loads(file.read())["key"]
@ -62,6 +61,7 @@ def test(name: str, data: Union[str, dict], raw: bool = False) -> None:
else: else:
print(f"{BBLACK}[{BGREEN}OK{BBLACK}] {RESET}Test of {YELLOW}{name} {RESET}is not empty, so might be fine. Check {CYAN}tests/{name}.txt {RESET}for more information.", flush=True) print(f"{BBLACK}[{BGREEN}OK{BBLACK}] {RESET}Test of {YELLOW}{name} {RESET}is not empty, so might be fine. Check {CYAN}tests/{name}.txt {RESET}for more information.", flush=True)
client = pyrmv.Client(key)
test("raw_board_arrival", pyrmv.raw.board_arrival(key, id="A=1@O=Offenbach (Main)-Tempelsee Wilhelm-Schramm-Straße@X=8783648@Y=50083822@U=80@L=3008012@", maxJourneys=5), raw=True) test("raw_board_arrival", pyrmv.raw.board_arrival(key, id="A=1@O=Offenbach (Main)-Tempelsee Wilhelm-Schramm-Straße@X=8783648@Y=50083822@U=80@L=3008012@", maxJourneys=5), raw=True)
test("raw_board_departure", pyrmv.raw.board_departure(key, id="A=1@O=Offenbach (Main)-Tempelsee Wilhelm-Schramm-Straße@X=8783648@Y=50083822@U=80@L=3008012@", maxJourneys=5), raw=True) test("raw_board_departure", pyrmv.raw.board_departure(key, id="A=1@O=Offenbach (Main)-Tempelsee Wilhelm-Schramm-Straße@X=8783648@Y=50083822@U=80@L=3008012@", maxJourneys=5), raw=True)
@ -72,11 +72,11 @@ test("raw_stop_by_coords", pyrmv.raw.stop_by_coords(key, 50.131140, 8.733362, ra
test("raw_trip_find", pyrmv.raw.trip_find(key, originCoordLat="50.084659", originCoordLong="8.785948", destCoordLat=50.1233048, destCoordLong=8.6129742, trafficMessages=True, numF=3), raw=True) test("raw_trip_find", pyrmv.raw.trip_find(key, originCoordLat="50.084659", originCoordLong="8.785948", destCoordLat=50.1233048, destCoordLong=8.6129742, trafficMessages=True, numF=3), raw=True)
test("raw_trip_recon", pyrmv.raw.trip_recon(key, ctx="¶HKI¶G@F$A=2@O=50.084659, 8.785948@X=8785948@Y=50084659@u=0@a=128@$A=1@O=Offenbach (Main)-Tempelsee Wilhelm-Schramm-Straße@L=3008012@a=128@$202210061243$202210061247$$$1$$$$$$§T$A=1@O=Offenbach (Main)-Tempelsee Wilhelm-Schramm-Straße@L=3008012@a=128@$A=1@O=Offenbach (Main)-Zentrum Marktplatz\/Frankf. Straße@L=3002510@a=128@$202210061247$202210061300$Bus 101 $$1$$$$$$§W$A=1@O=Offenbach (Main)-Zentrum Marktplatz\/Frankf. Straße@L=3002510@a=128@$A=1@O=Offenbach (Main)-Zentrum Marktplatz@L=3011265@a=128@$202210061300$202210061304$$$1$$$$$$§T$A=1@O=Offenbach (Main)-Zentrum Marktplatz@L=3011265@a=128@$A=1@O=Frankfurt (Main) Taunusanlage@L=3000011@a=128@$202210061306$202210061319$ S2$$1$$$$$$§T$A=1@O=Frankfurt (Main) Taunusanlage@L=3000011@a=128@$A=1@O=Frankfurt (Main) Rödelheim Bahnhof@L=3001217@a=128@$202210061322$202210061333$ S5$$1$$$$$$§G@F$A=1@O=Frankfurt (Main) Rödelheim Bahnhof@L=3001217@a=128@$A=2@O=50.123304, 8.612974@X=8612974@Y=50123304@u=0@a=128@$202210061333$202210061344$$$1$$$$$$¶GP¶ft@0@2000@120@1@100@1@@0@@@@@false@0@-1@0@-1@-1@$f@$f@$f@$f@$f@$§bt@0@2000@120@1@100@1@@0@@@@@false@0@-1@0@-1@-1@$f@$f@$f@$f@$f@$§tt@0@2000@120@1@100@1@@0@@@@@false@0@-1@0@-1@-1@$f@$f@$f@$f@$f@$§¶KRCC¶#VE#1#"), raw=True) test("raw_trip_recon", pyrmv.raw.trip_recon(key, ctx="¶HKI¶G@F$A=2@O=50.084659, 8.785948@X=8785948@Y=50084659@u=0@a=128@$A=1@O=Offenbach (Main)-Tempelsee Wilhelm-Schramm-Straße@L=3008012@a=128@$202210061243$202210061247$$$1$$$$$$§T$A=1@O=Offenbach (Main)-Tempelsee Wilhelm-Schramm-Straße@L=3008012@a=128@$A=1@O=Offenbach (Main)-Zentrum Marktplatz\/Frankf. Straße@L=3002510@a=128@$202210061247$202210061300$Bus 101 $$1$$$$$$§W$A=1@O=Offenbach (Main)-Zentrum Marktplatz\/Frankf. Straße@L=3002510@a=128@$A=1@O=Offenbach (Main)-Zentrum Marktplatz@L=3011265@a=128@$202210061300$202210061304$$$1$$$$$$§T$A=1@O=Offenbach (Main)-Zentrum Marktplatz@L=3011265@a=128@$A=1@O=Frankfurt (Main) Taunusanlage@L=3000011@a=128@$202210061306$202210061319$ S2$$1$$$$$$§T$A=1@O=Frankfurt (Main) Taunusanlage@L=3000011@a=128@$A=1@O=Frankfurt (Main) Rödelheim Bahnhof@L=3001217@a=128@$202210061322$202210061333$ S5$$1$$$$$$§G@F$A=1@O=Frankfurt (Main) Rödelheim Bahnhof@L=3001217@a=128@$A=2@O=50.123304, 8.612974@X=8612974@Y=50123304@u=0@a=128@$202210061333$202210061344$$$1$$$$$$¶GP¶ft@0@2000@120@1@100@1@@0@@@@@false@0@-1@0@-1@-1@$f@$f@$f@$f@$f@$§bt@0@2000@120@1@100@1@@0@@@@@false@0@-1@0@-1@-1@$f@$f@$f@$f@$f@$§tt@0@2000@120@1@100@1@@0@@@@@false@0@-1@0@-1@-1@$f@$f@$f@$f@$f@$§¶KRCC¶#VE#1#"), raw=True)
test("board_arrival", pyrmv.board_arrival(key, "A=1@O=Frankfurt (Main) Taunusanlage@X=8668765@Y=50113478@U=80@L=3000011@", journeys_max=5)) test("board_arrival", client.board_arrival("A=1@O=Frankfurt (Main) Taunusanlage@X=8668765@Y=50113478@U=80@L=3000011@", journeys_max=5))
test("board_departure", pyrmv.board_departure(key, "A=1@O=Frankfurt (Main) Taunusanlage@X=8668765@Y=50113478@U=80@L=3000011@", journeys_max=5)) test("board_departure", client.board_departure("A=1@O=Frankfurt (Main) Taunusanlage@X=8668765@Y=50113478@U=80@L=3000011@", journeys_max=5))
test("journey_detail", pyrmv.journey_detail(key, "2|#VN#1#ST#1664906549#PI#0#ZI#12709#TA#0#DA#61022#1S#3008007#1T#1248#LS#3008043#LT#1323#PU#80#RT#1#CA#1aE#ZE#101#ZB#Bus 101 #PC#6#FR#3008007#FT#1248#TO#3008043#TT#1323#", real_time_mode=RealTimeMode.FULL)) test("journey_detail", client.journey_detail("2|#VN#1#ST#1664906549#PI#0#ZI#12709#TA#0#DA#61022#1S#3008007#1T#1248#LS#3008043#LT#1323#PU#80#RT#1#CA#1aE#ZE#101#ZB#Bus 101 #PC#6#FR#3008007#FT#1248#TO#3008043#TT#1323#", real_time_mode=RealTimeMode.FULL))
test("stop_by_coords", pyrmv.stop_by_coords(key, 50.131140, 8.733362, radius=300, max_number=3)) test("stop_by_coords", client.stop_by_coords(50.131140, 8.733362, radius=300, max_number=3))
test("stop_by_id", pyrmv.stop_by_id(key, "A=1@O=Offenbach (Main)-Zentrum Marktplatz\/Frankf. Straße@X=8764456@Y=50105181@U=80@L=3002510@")) test("stop_by_id", client.stop_by_id("A=1@O=Offenbach (Main)-Zentrum Marktplatz\/Frankf. Straße@X=8764456@Y=50105181@U=80@L=3002510@"))
test("stop_by_name", pyrmv.stop_by_name(key, "Groß Karben", max_number=3)) test("stop_by_name", client.stop_by_name("Groß Karben", max_number=3))
test("trip_find", pyrmv.trip_find(key, origin_coord_lat="50.084659", origin_coord_lon="8.785948", destination_coord_lat=50.1233048, destination_coord_lon=8.6129742, messages=True)) test("trip_find", 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))
test("trip_recon", pyrmv.trip_recon(key, pyrmv.trip_find(key, origin_coord_lat="50.084659", origin_coord_lon="8.785948", destination_coord_lat=50.1233048, destination_coord_lon=8.6129742, messages=True, trips_after_time=1)[0] )) test("trip_recon", client.trip_recon( 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, trips_after_time=1)[0] ))