Small refactor and isort+black formatting
Some checks reported warnings
Tests / test (3.11) (push) Has been cancelled
Tests / test (3.8) (push) Has been cancelled
Tests / test (3.9) (push) Has been cancelled
Tests / test (3.10) (push) Has been cancelled

This commit is contained in:
2023-11-24 11:21:02 +01:00
parent fa4f7b83ec
commit f31fa65d78
45 changed files with 1035 additions and 923 deletions

View File

@@ -1,94 +1,99 @@
from datetime import datetime
from typing import Any, Mapping
from pyrmv.classes.Message import Message
class LineArrival():
def __init__(self, data, client, retrieve_stops: bool = True):
class LineArrival:
def __init__(self, data: Mapping[str, Any], client, retrieve_stops: bool = True):
self.journey = client.journey_detail(data["JourneyDetailRef"]["ref"])
self.status = data["JourneyStatus"]
self.messages = []
if "Messages" in data:
self.messages.extend(
Message(message) for message in data["Messages"]["Message"]
)
self.name = data["name"]
self.type = data["type"]
self.stop_name = data["stop"]
self.stop_id = data["stopid"]
self.stop_id_ext = data["stopExtId"]
if retrieve_stops:
self.stop = client.stop_by_id(self.stop_id)
else:
self.stop = None
self.stop = client.stop_by_id(self.stop_id) if retrieve_stops else None
self.stop = client.stop_by_id(self.stop_id)
self.time = datetime.strptime(data["time"], "%H:%M:%S")
self.date = datetime.strptime(data["date"], "%Y-%m-%d")
self.reachable = data["reachable"]
self.origin = data["origin"]
self.origin = data["origin"]
if "Messages" in data:
self.messages.extend(Message(message) for message in data["Messages"]["Message"])
if ("rtTime" in data) and ("rtDate" in data):
self.time_real_time = datetime.strptime(data["rtTime"], "%H:%M:%S")
self.date_real_time = datetime.strptime(data["rtDate"], "%Y-%m-%d")
self.reachable = data["reachable"]
self.origin = data["origin"]
else:
self.time_real_time = None
self.date_real_time = None
def __str__(self) -> str:
return f"{self.name} coming from {self.origin} at {self.time.time()} {self.date.date()}"
return (
f"{self.name} coming from {self.origin} at {self.time.time()} {self.date.date()}"
)
class LineDeparture():
def __init__(self, data, client, retrieve_stops: bool = True):
class LineDeparture:
def __init__(self, data: Mapping[str, Any], client, retrieve_stops: bool = True):
self.journey = client.journey_detail(data["JourneyDetailRef"]["ref"])
self.status = data["JourneyStatus"]
self.messages = []
if "Messages" in data:
self.messages.extend(
Message(message) for message in data["Messages"]["Message"]
)
self.name = data["name"]
self.type = data["type"]
self.stop_name = data["stop"]
self.stop_id = data["stopid"]
self.stop_id_ext = data["stopExtId"]
if retrieve_stops:
self.stop = client.stop_by_id(self.stop_id)
else:
self.stop = None
self.stop = client.stop_by_id(self.stop_id) if retrieve_stops else None
self.time = datetime.strptime(data["time"], "%H:%M:%S")
self.date = datetime.strptime(data["date"], "%Y-%m-%d")
if ("rtTime" in data) and ("rtDate" in data):
self.time_real_time = datetime.strptime(data["rtTime"], "%H:%M:%S")
self.date_real_time = datetime.strptime(data["rtDate"], "%Y-%m-%d")
self.reachable = data["reachable"]
self.direction = data["direction"]
self.direction_flag = data["directionFlag"]
def __str__(self) -> str:
return f"{self.name} heading {self.direction} at {self.time.time()} {self.date.date()}"
class BoardArrival(list):
def __init__(self, data: dict, client, retrieve_stops: bool = True):
if "Messages" in data:
self.messages.extend(Message(message) for message in data["Messages"]["Message"])
if ("rtTime" in data) and ("rtDate" in data):
self.time_real_time = datetime.strptime(data["rtTime"], "%H:%M:%S")
self.date_real_time = datetime.strptime(data["rtDate"], "%Y-%m-%d")
else:
self.time_real_time = None
self.date_real_time = None
def __str__(self) -> str:
return (
f"{self.name} heading {self.direction} at {self.time.time()} {self.date.date()}"
)
class BoardArrival(list):
def __init__(self, data: Mapping[str, Any], client, retrieve_stops: bool = True):
super().__init__([])
if "Arrival" not in data:
return
for line in data["Arrival"]:
self.append(LineArrival(line, client, retrieve_stops=retrieve_stops))
def __str__(self) -> str:
lines = []
for line in self:
lines.append(str(line))
return "Arrival board\n" + "\n".join(lines)
return "Arrival board\n" + "\n".join([str(line) for line in self])
class BoardDeparture(list):
def __init__(self, data: dict, client, retrieve_stops: bool = True):
def __init__(self, data: Mapping[str, Any], client, retrieve_stops: bool = True):
super().__init__([])
if "Departure" not in data:
return
for line in data["Departure"]:
self.append(LineDeparture(line, client, retrieve_stops=retrieve_stops))
def __str__(self) -> str:
lines = []
for line in self:
lines.append(str(line))
return "Departure board\n" + "\n".join(lines)
return "Departure board\n" + "\n".join([str(line) for line in self])

View File

@@ -1,24 +1,26 @@
from datetime import datetime, timedelta
from typing import List, OrderedDict, Union
from pyrmv.classes import *
from pyrmv.enums import *
from pyrmv.raw import board_arrival as raw_board_arrival
from pyrmv.raw.board_departure import board_departure as raw_board_departure
from pyrmv.raw.him_search import him_search as raw_him_search
from pyrmv.raw.journey_detail import journey_detail as raw_journey_detail
from pyrmv.raw.stop_by_name import stop_by_name as raw_stop_by_name
from pyrmv.raw.stop_by_coords import stop_by_coords as raw_stop_by_coords
from pyrmv.raw.stop_by_name import stop_by_name as raw_stop_by_name
from pyrmv.raw.trip_find import trip_find as raw_trip_find
from pyrmv.raw.trip_recon import trip_recon as raw_trip_recon
from pyrmv.raw.him_search import him_search as raw_him_search
from pyrmv.utility.find_exception import find_exception
from pyrmv.utility import find_exception
try:
from typing import Literal
except ImportError:
from typing_extensions import Literal
class Client():
"""The main class in the whole module. Is used to use all non-raw methods.
class Client:
"""The main class in the whole module. Is used to use all non-raw methods.
More detailed docs for each method can be found by using IDE's docstring
highlighting system or in project's wiki ([can be found here](https://git.end-play.xyz/profitroll/PythonRMV/wiki))
@@ -63,27 +65,33 @@ class Client():
# Find a trip and reconstruct it
trip = client.trip_find(origin_coord_lat="50.084659", origin_coord_lon="8.785948", destination_coord_lat=50.1233048, destination_coord_lon=8.6129742, messages=True)[0]
trip_recon = client.trip_recon(trip)[0]
```
"""
```
"""
def __init__(self, access_id: str) -> None:
self.access_id = access_id
def board_arrival(self,
id: Union[str, None] = None,
id_ext: Union[str, None] = None,
direction: Union[str, Stop, StopTrip, None] = None,
date: Union[str, datetime, None] = None,
time: Union[str, datetime, None] = None,
duration: Union[int, timedelta] = 60,
journeys_max: int = -1,
operators: Union[str, list, None] = None,
lines: Union[str, list, None] = None,
passlist: bool = False,
board_type: Literal[BoardArrivalType.ARR, BoardArrivalType.ARR_EQUIVS, BoardArrivalType.ARR_MAST, BoardArrivalType.ARR_STATION] = BoardArrivalType.ARR,
retrieve_stops: bool = True
) -> BoardArrival:
"""Method returns a board with arriving transport.
def board_arrival(
self,
id: Union[str, None] = None,
id_ext: Union[str, None] = None,
direction: Union[str, Stop, StopTrip, None] = None,
date: Union[str, datetime, None] = None,
time: Union[str, datetime, None] = None,
duration: Union[int, timedelta] = 60,
journeys_max: int = -1,
operators: Union[str, list, None] = None,
lines: Union[str, list, None] = None,
passlist: bool = False,
board_type: Literal[
BoardArrivalType.ARR,
BoardArrivalType.ARR_EQUIVS,
BoardArrivalType.ARR_MAST,
BoardArrivalType.ARR_STATION,
] = BoardArrivalType.ARR,
retrieve_stops: bool = True,
) -> BoardArrival:
"""Method returns a board with arriving transport.
More detailed request is available as `raw.board_arrival()`, however returns `dict` instead of `Board`.
@@ -105,14 +113,14 @@ class Client():
* BoardArrival: Instance of `BoardArrival` object.
"""
if (isinstance(direction, Stop) or isinstance(direction, StopTrip)):
if isinstance(direction, Stop) or isinstance(direction, StopTrip):
direction = direction.id
board_raw = raw_board_arrival(
accessId=self.access_id,
id=id,
extId=id_ext,
direction=direction, # type: ignore
direction=direction, # type: ignore
date=date,
time=time,
duration=duration,
@@ -120,28 +128,34 @@ class Client():
operators=operators,
lines=lines,
passlist=passlist,
boardType=board_type.code
boardType=board_type.code,
)
find_exception(board_raw)
return BoardArrival(board_raw, self, retrieve_stops=retrieve_stops)
def board_departure(self,
id: Union[str, None] = None,
id_ext: Union[str, None] = None,
direction: Union[str, Stop, StopTrip, None] = None,
date: Union[str, datetime, None] = None,
time: Union[str, datetime, None] = None,
duration: Union[int, timedelta] = 60,
journeys_max: int = -1,
operators: Union[str, list, None] = None,
lines: Union[str, list, None] = None,
passlist: bool = False,
board_type: Literal[BoardDepartureType.DEP, BoardDepartureType.DEP_EQUIVS, BoardDepartureType.DEP_MAST, BoardDepartureType.DEP_STATION] = BoardDepartureType.DEP,
retrieve_stops: bool = True
) -> BoardDeparture:
"""Method returns a board with departing transport.
def board_departure(
self,
id: Union[str, None] = None,
id_ext: Union[str, None] = None,
direction: Union[str, Stop, StopTrip, None] = None,
date: Union[str, datetime, None] = None,
time: Union[str, datetime, None] = None,
duration: Union[int, timedelta] = 60,
journeys_max: int = -1,
operators: Union[str, list, None] = None,
lines: Union[str, list, None] = None,
passlist: bool = False,
board_type: Literal[
BoardDepartureType.DEP,
BoardDepartureType.DEP_EQUIVS,
BoardDepartureType.DEP_MAST,
BoardDepartureType.DEP_STATION,
] = BoardDepartureType.DEP,
retrieve_stops: bool = True,
) -> BoardDeparture:
"""Method returns a board with departing transport.
More detailed request is available as `raw.board_departure()`, however returns `dict` instead of `Board`.
@@ -163,14 +177,14 @@ class Client():
* BoardDeparture: Instance of `BoardDeparture` object.
"""
if (isinstance(direction, Stop) or isinstance(direction, StopTrip)):
if isinstance(direction, (Stop, StopTrip)):
direction = direction.id
board_raw = raw_board_departure(
accessId=self.access_id,
id=id,
extId=id_ext,
direction=direction, # type: ignore
direction=direction, # type: ignore
date=date,
time=time,
duration=duration,
@@ -178,41 +192,53 @@ class Client():
operators=operators,
lines=lines,
passlist=passlist,
boardType=board_type.code
boardType=board_type.code,
)
find_exception(board_raw)
return BoardDeparture(board_raw, self, retrieve_stops=retrieve_stops)
def him_search(self,
date_begin: Union[str, datetime, None] = None,
date_end: Union[str, datetime, None] = None,
time_begin: Union[str, datetime, None] = None,
time_end: Union[str, datetime, None] = None,
weekdays: Union[str, OrderedDict[str, bool], None] = None,
ids: Union[list, None] = None,
operators: Union[list, None] = None,
categories: Union[list, None] = None,
channels: Union[list, None] = None,
companies: Union[list, None] = None,
lines: Union[list, None] = None,
line_ids: Union[list, None] = None,
stations: Union[list, List[Stop], None] = None,
station_from: Union[str, Stop, None] = None,
station_to: Union[str, Stop, None] = None,
both_ways: Union[bool, None] = None,
train_names: Union[list, None] = None,
search_mode: Union[Literal[SearchMode.MATCH, SearchMode.NOMATCH, SearchMode.TFMATCH], None] = None,
affected_journey_mode: Union[Literal[AffectedJourneyMode.ALL, AffectedJourneyMode.OFF], None] = None,
affected_journey_stop_mode: Union[Literal[AffectedJourneyStopMode.ALL, AffectedJourneyStopMode.IMP, AffectedJourneyStopMode.OFF], None] = None,
priority_min: Union[int, None] = None,
priority_max: Union[int, None] = None
) -> List[Message]:
def him_search(
self,
date_begin: Union[str, datetime, None] = None,
date_end: Union[str, datetime, None] = None,
time_begin: Union[str, datetime, None] = None,
time_end: Union[str, datetime, None] = None,
weekdays: Union[str, OrderedDict[str, bool], None] = None,
ids: Union[list, None] = None,
operators: Union[list, None] = None,
categories: Union[list, None] = None,
channels: Union[list, None] = None,
companies: Union[list, None] = None,
lines: Union[list, None] = None,
line_ids: Union[list, None] = None,
stations: Union[list, List[Stop], None] = None,
station_from: Union[str, Stop, None] = None,
station_to: Union[str, Stop, None] = None,
both_ways: Union[bool, None] = None,
train_names: Union[list, None] = None,
search_mode: Union[
Literal[SearchMode.MATCH, SearchMode.NOMATCH, SearchMode.TFMATCH], None
] = None,
affected_journey_mode: Union[
Literal[AffectedJourneyMode.ALL, AffectedJourneyMode.OFF], None
] = None,
affected_journey_stop_mode: Union[
Literal[
AffectedJourneyStopMode.ALL,
AffectedJourneyStopMode.IMP,
AffectedJourneyStopMode.OFF,
],
None,
] = None,
priority_min: Union[int, None] = None,
priority_max: Union[int, None] = None,
) -> List[Message]:
"""The him_search method will deliver a list of HIM messages if matched by the given criteria as
well as affected products if any.
well as affected products if any.
More detailed request is available as `raw.him_search()`, however returns `dict` instead of `List[Message]`.
More detailed request is available as `raw.him_search()`, however returns `dict` instead of `List[Message]`.
### Args:
* date_begin (`Union[str, datetime]`, optional): Sets the event period start date. Defaults to `None`.
@@ -240,7 +266,7 @@ class Client():
### Returns:
* List[Message]: List of `Message` objects. Empty list if none found.
"""
"""
if isinstance(station_from, Stop):
station_from = station_from.ext_id
@@ -257,20 +283,13 @@ class Client():
new_stations.append(stop)
stations = new_stations
if search_mode == None:
search_mode = None
else:
search_mode = search_mode.code
if affected_journey_mode == None:
affected_journey_mode = None
else:
affected_journey_mode = affected_journey_mode.code
if affected_journey_stop_mode == None:
affected_journey_stop_mode = None
else:
affected_journey_stop_mode = affected_journey_stop_mode.code
search_mode = None if search_mode is None else search_mode.code
affected_journey_mode = (
None if affected_journey_mode is None else affected_journey_mode.code
)
affected_journey_stop_mode = (
None if affected_journey_stop_mode is None else affected_journey_stop_mode.code
)
messages = []
messages_raw = raw_him_search(
@@ -288,38 +307,47 @@ class Client():
lines=lines,
lineids=line_ids,
stations=stations,
fromstation=station_from, # type: ignore
tostation=station_to, # type: ignore
fromstation=station_from, # type: ignore
tostation=station_to, # type: ignore
bothways=both_ways,
trainnames=train_names,
searchmode=search_mode, # type: ignore
affectedJourneyMode=affected_journey_mode, # type: ignore
affectedJourneyStopMode=affected_journey_stop_mode, # type: ignore
searchmode=search_mode, # type: ignore
affectedJourneyMode=affected_journey_mode, # type: ignore
affectedJourneyStopMode=affected_journey_stop_mode, # type: ignore
maxprio=priority_max,
minprio=priority_min
minprio=priority_min,
)
find_exception(messages_raw)
if "Message" in messages_raw:
for message in messages_raw["Message"]:
messages.append(Message(message))
messages.extend(Message(message) for message in messages_raw["Message"])
return messages
def journey_detail(self,
id: str,
date: Union[str, datetime, None] = None,
real_time_mode: Union[Literal[RealTimeMode.FULL, RealTimeMode.INFOS, RealTimeMode.OFF, RealTimeMode.REALTIME, RealTimeMode.SERVER_DEFAULT], None] = None,
from_id: Union[str, None] = None,
from_index: Union[int, None] = None,
to_id: Union[str, None] = None,
to_index: Union[int, None] = None
) -> Journey:
def journey_detail(
self,
id: str,
date: Union[str, datetime, None] = None,
real_time_mode: Union[
Literal[
RealTimeMode.FULL,
RealTimeMode.INFOS,
RealTimeMode.OFF,
RealTimeMode.REALTIME,
RealTimeMode.SERVER_DEFAULT,
],
None,
] = None,
from_id: Union[str, None] = None,
from_index: Union[int, None] = None,
to_id: Union[str, None] = None,
to_index: Union[int, None] = None,
) -> Journey:
"""The journey_detail method will deliver information about the complete route of a vehicle. The journey
identifier is part of a trip or `board_departure()` response. It contains a list of all stops/stations of this journey
including all departure and arrival times (with real-time data if available) and additional information like
specific attributes about facilities and other texts.
specific attributes about facilities and other texts.
More detailed request is available as `raw.journey_detail()`, however returns `dict` instead of `Journey`.
@@ -334,40 +362,58 @@ class Client():
### Returns:
* Journey: Instance of `Journey` object.
"""
"""
if real_time_mode == None:
real_time_mode = None
else:
real_time_mode = real_time_mode.code
real_time_mode = None if real_time_mode is None else real_time_mode.code
journey_raw = raw_journey_detail(
accessId=self.access_id,
id=id,
date=date,
rtMode=real_time_mode, # type: ignore
rtMode=real_time_mode, # type: ignore
fromId=from_id,
fromIdx=from_index,
toId=to_id,
toIdx=to_index
toIdx=to_index,
)
find_exception(journey_raw)
return Journey(journey_raw)
def stop_by_coords(self,
coords_lat: Union[str, float],
coords_lon: Union[str, float],
lang: Literal[Language.DE, Language.DA, Language.EN, Language.ES, Language.FR, Language.HU, Language.IT, Language.NL, Language.NO, Language.PL, Language.SV, Language.TR] = Language.EN,
radius: Union[int, float] = 1000,
max_number: int = 10,
stop_type: Literal[LocationType.S, LocationType.P, LocationType.SP, LocationType.SE, LocationType.SPE] = LocationType.S,
selection_mode: Union[Literal[SelectionMode.SLCT_A, SelectionMode.SLCT_N], None] = None,
) -> List[Stop]:
def stop_by_coords(
self,
coords_lat: Union[str, float],
coords_lon: Union[str, float],
lang: Literal[
Language.DE,
Language.DA,
Language.EN,
Language.ES,
Language.FR,
Language.HU,
Language.IT,
Language.NL,
Language.NO,
Language.PL,
Language.SV,
Language.TR,
] = Language.EN,
radius: Union[int, float] = 1000,
max_number: int = 10,
stop_type: Literal[
LocationType.S,
LocationType.P,
LocationType.SP,
LocationType.SE,
LocationType.SPE,
] = LocationType.S,
selection_mode: Union[
Literal[SelectionMode.SLCT_A, SelectionMode.SLCT_N], None
] = None,
) -> List[Stop]:
"""Method returns a list of stops around a given center coordinate.
The returned results are ordered by their distance to the center coordinate.
The returned results are ordered by their distance to the center coordinate.
More detailed request is available as `raw.stop_by_coords()`, however returns `dict` instead of `List[Stop]`.
@@ -382,12 +428,9 @@ class Client():
### Returns:
* List[Stop]: List of `Stop` objects. Empty list if none found.
"""
"""
if selection_mode == None:
selection_mode = None
else:
selection_mode = selection_mode.code
selection_mode = None if selection_mode is None else selection_mode.code
stops = []
stops_raw = raw_stop_by_coords(
@@ -398,7 +441,7 @@ class Client():
radius=radius,
maxNo=max_number,
stopType=stop_type.code,
locationSelectionMode=selection_mode # type: ignore
locationSelectionMode=selection_mode, # type: ignore
)
find_exception(stops_raw)
@@ -412,59 +455,96 @@ class Client():
return stops
def stop_by_id(self,
query: str,
lang: Literal[Language.DE, Language.DA, Language.EN, Language.ES, Language.FR, Language.HU, Language.IT, Language.NL, Language.NO, Language.PL, Language.SV, Language.TR] = Language.EN,
) -> Union[Stop, None]:
def stop_by_id(
self,
query: str,
lang: Literal[
Language.DE,
Language.DA,
Language.EN,
Language.ES,
Language.FR,
Language.HU,
Language.IT,
Language.NL,
Language.NO,
Language.PL,
Language.SV,
Language.TR,
] = Language.EN,
) -> Union[Stop, None]:
"""Method can be used to get Stop object whilst only having id available.
### Args:
* query (`str`): Search for that token.
* lang (`Literal[Language.DE, Language.DA, Language.EN, Language.ES, Language.FR, Language.HU, Language.IT, Language.NL, Language.NO, Language.PL, Language.SV, Language.TR]`, **optional**): The language of response. Defaults to `Language.EN`.
### Returns:
* Union[Stop, None]: Instance of `Stop` object or `None` if not found.
"""
"""
stops_raw = raw_stop_by_name(
accessId=self.access_id,
inputString=query,
lang=lang.code,
maxNo=1
accessId=self.access_id, inputString=query, lang=lang.code, maxNo=1
)
find_exception(stops_raw)
if len(stops_raw["stopLocationOrCoordLocation"]) > 0:
stop = stops_raw["stopLocationOrCoordLocation"][0]
if len(stops_raw["stopLocationOrCoordLocation"]) <= 0:
return None
if "StopLocation" in stop:
return Stop(stop["StopLocation"])
elif "CoordLocation" in stop:
return Stop(stop["CoordLocation"])
else:
return None
stop = stops_raw["stopLocationOrCoordLocation"][0]
if "StopLocation" in stop:
return Stop(stop["StopLocation"])
elif "CoordLocation" in stop:
return Stop(stop["CoordLocation"])
else:
return None
def stop_by_name(self,
query: str,
lang: Literal[Language.DE, Language.DA, Language.EN, Language.ES, Language.FR, Language.HU, Language.IT, Language.NL, Language.NO, Language.PL, Language.SV, Language.TR] = Language.EN,
max_number: int = 10,
stop_type: Literal[LocationType.A, LocationType.ALL, LocationType.AP, LocationType.P, LocationType.S, LocationType.SA, LocationType.SP] = LocationType.ALL,
selection_mode: Union[Literal[SelectionMode.SLCT_A, SelectionMode.SLCT_N], None] = None,
coord_lat: Union[str, float, None] = None,
coord_lon: Union[str, float, None] = None,
radius: Union[int, float] = 1000,
refine_id: Union[str, None] = None,
stations: Union[str, list, None] = None,
filter_mode: Literal[FilterMode.DIST_PERI, FilterMode.EXCL_PERI, FilterMode.SLCT_PERI] = FilterMode.DIST_PERI
) -> List[Stop]:
def stop_by_name(
self,
query: str,
lang: Literal[
Language.DE,
Language.DA,
Language.EN,
Language.ES,
Language.FR,
Language.HU,
Language.IT,
Language.NL,
Language.NO,
Language.PL,
Language.SV,
Language.TR,
] = Language.EN,
max_number: int = 10,
stop_type: Literal[
LocationType.A,
LocationType.ALL,
LocationType.AP,
LocationType.P,
LocationType.S,
LocationType.SA,
LocationType.SP,
] = LocationType.ALL,
selection_mode: Union[
Literal[SelectionMode.SLCT_A, SelectionMode.SLCT_N], None
] = None,
coord_lat: Union[str, float, None] = None,
coord_lon: Union[str, float, None] = None,
radius: Union[int, float] = 1000,
refine_id: Union[str, None] = None,
stations: Union[str, list, None] = None,
filter_mode: Literal[
FilterMode.DIST_PERI, FilterMode.EXCL_PERI, FilterMode.SLCT_PERI
] = FilterMode.DIST_PERI,
) -> List[Stop]:
"""Method can be used to perform a pattern matching of a user input and to retrieve a list
of possible matches in the journey planner database. Possible matches might be stops/stations,
points of interest and addresses.
points of interest and addresses.
More detailed request is available as `raw.stop_by_name()`, however returns `dict` instead of `List[Stop]`.
More detailed request is available as `raw.stop_by_name()`, however returns `dict` instead of `List[Stop]`.
### Args:
* query (`str`): Search for that token.
@@ -481,12 +561,9 @@ class Client():
### Returns:
* List[Stop]: List of `Stop` objects. Empty list if none found.
"""
"""
if selection_mode == None:
selection_mode = None
else:
selection_mode = selection_mode.code
selection_mode = None if selection_mode is None else selection_mode.code
stops = []
stops_raw = raw_stop_by_name(
@@ -495,13 +572,13 @@ class Client():
lang=lang.code,
maxNo=max_number,
stopType=stop_type.code,
locationSelectionMode=selection_mode, # type: ignore
locationSelectionMode=selection_mode, # type: ignore
coordLat=coord_lat,
coordLong=coord_lon,
radius=radius,
refineId=refine_id,
stations=stations,
filterMode=filter_mode.code
filterMode=filter_mode.code,
)
find_exception(stops_raw)
@@ -515,71 +592,78 @@ class Client():
return stops
def trip_find(self,
lang: Literal[Language.DE, Language.DA, Language.EN, Language.ES, Language.FR, Language.HU, Language.IT, Language.NL, Language.NO, Language.PL, Language.SV, Language.TR] = Language.EN,
origin_id: Union[str, None] = None,
origin_id_ext: Union[str, None] = None,
origin_coord_lat: Union[str, float, None] = None,
origin_coord_lon: Union[str, float, None] = None,
origin_coord_name: Union[str, None] = None,
destination_id: Union[str, None] = None,
destination_id_ext: Union[str, None] = None,
destination_coord_lat: Union[str, float, None] = None,
destination_coord_lon: Union[str, float, None] = None,
destination_coord_name: Union[str, None] = None,
via: Union[str, None] = None,
via_id: Union[str, None] = None,
via_gis: Union[str, None] = None,
via_wait_time: int = 0,
avoid: Union[str, None] = None,
avoid_id: Union[str, None] = None,
change_time_percent: int = 100,
change_time_min: Union[int, None] = None,
change_time_max: Union[int, None] = None,
change_time_add: Union[int, None] = None,
change_max: Union[int, None] = None,
date: Union[str, datetime, None] = None,
time: Union[str, datetime, None] = None,
search_arrival: bool = False,
trips_after_time: Union[int, None] = None,
trips_before_time: Union[int, None] = None,
context: Union[str, None] = None,
passlist: bool = False,
operators: Union[str, list, None] = None,
lines: Union[str, list, None] = None,
lineids: Union[str, list, None] = None,
iv_include: bool = False,
iv_only: bool = False,
bike_carriage: bool = False,
passing_points: bool = False,
real_time_mode: Union[Literal[RealTimeMode.FULL, RealTimeMode.INFOS, RealTimeMode.OFF, RealTimeMode.REALTIME, RealTimeMode.SERVER_DEFAULT], None] = None,
include_earlier: bool = False,
ict_alternatives: bool = False,
tariff: Union[bool, None] = None,
messages: bool = False,
frequency: bool = True
) -> List[Trip]:
def trip_find(
self,
lang: Literal[
Language.DE,
Language.DA,
Language.EN,
Language.ES,
Language.FR,
Language.HU,
Language.IT,
Language.NL,
Language.NO,
Language.PL,
Language.SV,
Language.TR,
] = Language.EN,
origin_id: Union[str, None] = None,
origin_id_ext: Union[str, None] = None,
origin_coord_lat: Union[str, float, None] = None,
origin_coord_lon: Union[str, float, None] = None,
origin_coord_name: Union[str, None] = None,
destination_id: Union[str, None] = None,
destination_id_ext: Union[str, None] = None,
destination_coord_lat: Union[str, float, None] = None,
destination_coord_lon: Union[str, float, None] = None,
destination_coord_name: Union[str, None] = None,
via: Union[str, None] = None,
via_id: Union[str, None] = None,
via_gis: Union[str, None] = None,
via_wait_time: int = 0,
avoid: Union[str, None] = None,
avoid_id: Union[str, None] = None,
change_time_percent: int = 100,
change_time_min: Union[int, None] = None,
change_time_max: Union[int, None] = None,
change_time_add: Union[int, None] = None,
change_max: Union[int, None] = None,
date: Union[str, datetime, None] = None,
time: Union[str, datetime, None] = None,
search_arrival: bool = False,
trips_after_time: Union[int, None] = None,
trips_before_time: Union[int, None] = None,
context: Union[str, None] = None,
passlist: bool = False,
operators: Union[str, list, None] = None,
lines: Union[str, list, None] = None,
lineids: Union[str, list, None] = None,
iv_include: bool = False,
iv_only: bool = False,
bike_carriage: bool = False,
passing_points: bool = False,
real_time_mode: Union[
Literal[
RealTimeMode.FULL,
RealTimeMode.INFOS,
RealTimeMode.OFF,
RealTimeMode.REALTIME,
RealTimeMode.SERVER_DEFAULT,
],
None,
] = None,
include_earlier: bool = False,
ict_alternatives: bool = False,
tariff: Union[bool, None] = None,
messages: bool = False,
frequency: bool = True,
) -> List[Trip]:
"""The trip service calculates a trip from a specified origin to a specified destination. These might be
stop/station IDs or coordinates based on addresses and points of interest validated by the location service or
coordinates freely defined by the client.
More detailed request is available as `raw.trip_find()`, however returns `dict` instead of `List[Trip]`.
More detailed request is available as `raw.trip_find()`, however returns `dict` instead of `List[Trip]`.
### Args:
* lang (`Literal[Language.DE, Language.DA, Language.EN, Language.ES, Language.FR, Language.HU, Language.IT, Language.NL, Language.NO, Language.PL, Language.SV, Language.TR]`, **optional**): The language of response. Defaults to `Language.EN`.
@@ -627,105 +711,94 @@ class Client():
### Returns:
* List[Trip]: List of `Trip` objects. Empty list if none found.
"""
"""
if real_time_mode == None:
real_time_mode = None
else:
real_time_mode = real_time_mode.code
real_time_mode = None if real_time_mode is None else real_time_mode.code
trips = []
trips_raw = raw_trip_find(
accessId=self.access_id,
lang=lang.code,
originId=origin_id,
originExtId=origin_id_ext,
originCoordLat=origin_coord_lat,
originCoordLong=origin_coord_lon,
originCoordName=origin_coord_name,
destId=destination_id,
destExtId=destination_id_ext,
destCoordLat=destination_coord_lat,
destCoordLong=destination_coord_lon,
destCoordName=destination_coord_name,
via=via,
viaId=via_id,
viaGis=via_gis,
viaWaitTime=via_wait_time,
avoid=avoid,
avoidId=avoid_id,
changeTimePercent=change_time_percent,
minChangeTime=change_time_min,
maxChangeTime=change_time_max,
addChangeTime=change_time_add,
maxChange=change_max,
date=date,
time=time,
searchForArrival=search_arrival,
numF=trips_after_time,
numB=trips_before_time,
context=context,
passlist=passlist,
operators=operators,
lines=lines,
lineids=lineids,
includeIv=iv_include,
ivOnly=iv_only,
bikeCarriage=bike_carriage,
showPassingPoints=passing_points,
rtMode=real_time_mode, # type: ignore
rtMode=real_time_mode, # type: ignore
includeEarlier=include_earlier,
withICTAlternatives=ict_alternatives,
tariff=tariff,
trafficMessages=messages,
withFreq=frequency
withFreq=frequency,
)
find_exception(trips_raw)
if "Trip" in trips_raw:
for trip in trips_raw["Trip"]:
trips.append(Trip(trip))
trips.extend(Trip(trip) for trip in trips_raw["Trip"])
return trips
def trip_recon(self,
context: Union[str, Trip],
date: Union[str, datetime, None] = None,
match_real_time: Union[bool, None] = None,
enable_replacements: Union[bool, None] = None,
arrival_dev_lower: Union[int, None] = None,
arrival_dev_upper: Union[int, None] = None,
departure_dev_lower: Union[int, None] = None,
departure_dev_upper: Union[int, None] = None,
passlist: bool = False,
passing_points: bool = False,
real_time_mode: Union[Literal[RealTimeMode.FULL, RealTimeMode.INFOS, RealTimeMode.OFF, RealTimeMode.REALTIME, RealTimeMode.SERVER_DEFAULT], None] = None,
tariff: Union[bool, None] = None,
messages: bool = False
) -> List[Trip]:
def trip_recon(
self,
context: Union[str, Trip],
date: Union[str, datetime, None] = None,
match_real_time: Union[bool, None] = None,
enable_replacements: Union[bool, None] = None,
arrival_dev_lower: Union[int, None] = None,
arrival_dev_upper: Union[int, None] = None,
departure_dev_lower: Union[int, None] = None,
departure_dev_upper: Union[int, None] = None,
passlist: bool = False,
passing_points: bool = False,
real_time_mode: Union[
Literal[
RealTimeMode.FULL,
RealTimeMode.INFOS,
RealTimeMode.OFF,
RealTimeMode.REALTIME,
RealTimeMode.SERVER_DEFAULT,
],
None,
] = None,
tariff: Union[bool, None] = None,
messages: bool = False,
) -> List[Trip]:
"""Reconstructing a trip can be achieved using the reconstruction context provided by any trip result in the
`ctx_recon` attribute of `Trip` object. The result will be a true copy of the original trip search result given
that the underlying data did not change.
that the underlying data did not change.
More detailed request is available as `raw.trip_recon()`, however returns `dict` instead of `List[Trip]`.
More detailed request is available as `raw.trip_recon()`, however returns `dict` instead of `List[Trip]`.
### Args:
* context (`Union[str, Journey]`): Specifies the reconstruction context.
@@ -744,20 +817,20 @@ class Client():
### Returns:
* List[Trip]: List of `Trip` objects. Empty list if none found.
"""
"""
if real_time_mode == None:
real_time_mode = None
else:
real_time_mode = real_time_mode.code
if isinstance(context, Trip):
context = context.ctx_recon
trips = []
trips_raw = raw_trip_recon(
accessId=self.access_id,
ctx=context, # type: ignore
ctx=context, # type: ignore
date=date,
matchRtType=match_real_time,
enableReplacements=enable_replacements,
@@ -767,7 +840,7 @@ class Client():
depU=departure_dev_upper,
passlist=passlist,
showPassingPoints=passing_points,
rtMode=real_time_mode, # type: ignore
rtMode=real_time_mode, # type: ignore
tariff=tariff,
trafficMessages=messages,
)
@@ -775,7 +848,6 @@ class Client():
find_exception(trips_raw)
if "Trip" in trips_raw:
for trip in trips_raw["Trip"]:
trips.append(Trip(trip))
trips.extend(Trip(trip) for trip in trips_raw["Trip"])
return trips
return trips

View File

@@ -1,11 +1,13 @@
from typing import Any, Mapping
from isodate import parse_duration
class Gis():
"""Gis object."""
def __init__(self, ref: str, route: dict):
class Gis:
"""Gis object."""
def __init__(self, ref: str, route: Mapping[str, Any]):
self.ref = ref
self.dist = route["dist"]
self.duration = parse_duration(route["durS"])
self.geo = route["dirGeo"]
self.geo = route["dirGeo"]

View File

@@ -1,22 +1,22 @@
from pyrmv.classes.Stop import Stop
from typing import Any, Mapping
from pyrmv.classes.Message import Message
from pyrmv.classes.Stop import Stop
class Journey():
"""Journey object."""
def __init__(self, data: dict):
class Journey:
"""Journey object."""
def __init__(self, data: Mapping[str, Any]):
self.stops = []
self.ref = data["ref"]
self.direction = data["Directions"]["Direction"][0]["value"]
self.direction_flag = data["Directions"]["Direction"][0]["flag"]
self.stops.extend(Stop(stop) for stop in data["Stops"]["Stop"])
self.messages = []
self.stops.extend(Stop(stop) for stop in data["Stops"]["Stop"])
if "Messages" in data:
self.messages.extend(
Message(message) for message in data["Messages"]["Message"]
)
self.messages.extend(Message(message) for message in data["Messages"]["Message"])
def __str__(self) -> str:
return f"Journey with total of {len(self.stops)} stops and {len(self.messages)} messages heading {self.direction} ({self.direction_flag})"
return f"Journey with total of {len(self.stops)} stops and {len(self.messages)} messages heading {self.direction} ({self.direction_flag})"

View File

@@ -1,43 +1,29 @@
from typing import Any, Mapping
from isodate import parse_duration
from pyrmv.classes.Gis import Gis
from pyrmv.classes.Message import Message
from pyrmv.classes.Stop import StopTrip
from isodate import parse_duration
class Leg():
"""Trip leg object."""
def __init__(self, data: dict):
class Leg:
"""Trip leg object."""
def __init__(self, data: Mapping[str, Any]):
self.origin = StopTrip(data["Origin"])
self.destination = StopTrip(data["Destination"])
if "GisRef" in data:
self.gis = Gis(data["GisRef"]["ref"], data["GisRoute"])
else:
self.gis = None
self.gis = (
None if "GisRef" not in data else Gis(data["GisRef"]["ref"], data["GisRoute"])
)
self.messages = []
self.index = data["idx"]
self.name = data["name"]
self.type = data["type"]
if "direction" in data:
self.direction = data["direction"]
else:
self.direction = None
self.messages = []
if "Messages" in data:
for message in data["Messages"]["Message"]:
self.messages.append(Message(message))
if "number" in data:
self.number = data["number"]
else:
self.number = None
self.direction = data.get("direction")
self.number = data.get("number")
self.duration = parse_duration(data["duration"])
if "dist" in data:
self.distance = data["dist"]
else:
self.distance = None
self.distance = data.get("dist")
if "Messages" in data:
self.messages.extend(Message(message) for message in data["Messages"]["Message"])

View File

@@ -1,25 +1,29 @@
from pyrmv.classes.Stop import Stop
from datetime import datetime
from typing import Any, Mapping
from isodate import parse_duration
class Url():
"""Traffic message channel url object."""
from pyrmv.classes.Stop import Stop
def __init__(self, data: dict) -> None:
class Url:
"""Traffic message channel url object."""
def __init__(self, data: Mapping[str, Any]) -> None:
self.name = data["name"]
self.url = data["url"]
def __str__(self) -> str:
return f"{self.name}: {self.url}"
class Channel():
"""Traffic message channel object."""
def __init__(self, data: dict) -> None:
class Channel:
"""Traffic message channel object."""
def __init__(self, data: Mapping[str, Any]) -> None:
self.name = data["name"]
url = []
for link in url:
url.append(Url(link))
url.extend(Url(link) for link in url)
self.url = url
self.time_start = datetime.strptime(data["validFromTime"], "%H:%M:%S")
self.date_start = datetime.strptime(data["validFromDate"], "%Y-%m-%d")
@@ -30,30 +34,17 @@ class Channel():
return f"{self.name}: from {self.time_start} {self.date_start} until {self.time_end} {self.date_end}"
class Message():
"""Traffic message object."""
def __init__(self, data: dict) -> None:
class Message:
"""Traffic message object."""
def __init__(self, data: Mapping[str, Any]) -> None:
self.affected_stops = []
if "affectedStops" in data:
for stop in data["affectedStops"]["StopLocation"]:
self.affected_stops.append(Stop(stop))
if "validFromStop" in data:
self.valid_from_stop = Stop(data["validFromStop"])
else:
self.valid_from_stop = None
if "validToStop" in data:
self.valid_to_stop = Stop(data["validToStop"])
else:
self.valid_to_stop = None
self.valid_from_stop = (
None if "validFromStop" not in data else Stop(data["validFromStop"])
)
self.valid_to_stop = None if "validToStop" not in data else Stop(data["validToStop"])
self.channels = []
for channel in data["channel"]:
self.channels.append(Channel(channel))
self.channels.extend(Channel(channel) for channel in data["channel"])
self.id = data["id"]
self.active = data["act"]
self.head = data["head"]
@@ -74,11 +65,12 @@ class Message():
self.date_modified = datetime.strptime(data["modDate"], "%Y-%m-%d")
self.daily_start = datetime.strptime(data["dailyStartingAt"], "%H:%M:%S")
self.daily_duration = parse_duration(data["dailyDuration"])
self.base_type = data["baseType"] if "baseType" in data else None
if "baseType" in data:
self.base_type = data["baseType"]
else:
self.base_type = None
if "affectedStops" in data:
self.affected_stops.extend(
Stop(stop) for stop in data["affectedStops"]["StopLocation"]
)
def __str__(self) -> str:
return f"{self.base_type} message with priority {self.products} valid from {self.time_start.time()} {self.date_start.date()} until {self.time_end.time()} {self.date_end.date()}: {self.head} - {self.lead}"
return f"{self.base_type} message with priority {self.products} valid from {self.time_start.time()} {self.date_start.date()} until {self.time_end.time()} {self.date_end.date()}: {self.head} - {self.lead}"

View File

@@ -1,54 +1,32 @@
from datetime import datetime
class Stop():
"""Stop object."""
def __init__(self, data: dict):
class Stop:
"""Stop object."""
def __init__(self, data: dict):
self.name = data["name"]
self.id = data["id"]
if "extId" in data:
self.ext_id = data["extId"]
else:
self.ext_id = None
if "description" in data:
self.description = data["description"]
else:
self.description = None
self.ext_id = data.get("extId")
self.description = data.get("description")
self.lon = data["lon"]
self.lat = data["lat"]
if "routeIdx" in data:
self.route_index = data["routeIdx"]
else:
self.route_index = None
if "arrTrack" in data:
self.track_arrival = data["arrTrack"]
else:
self.track_arrival = None
if "depTrack" in data:
self.track_departure = data["depTrack"]
else:
self.track_departure = None
self.route_index = data.get("routeIdx")
self.track_arrival = data.get("arrTrack")
self.track_departure = data.get("depTrack")
def __str__(self) -> str:
return f"Stop {self.name} at {self.lon}, {self.lat}"
class StopTrip(Stop):
"""Trip stop object. It's like a Stop object, but with a date and time."""
"""Trip stop object. It's like a Stop object, but with a date and time."""
def __init__(self, data: dict):
self.type = data["type"]
self.date = datetime.strptime(data["date"], "%Y-%m-%d")
self.time = datetime.strptime(data["time"], "%H:%M:%S")
super().__init__(data)
def __str__(self) -> str:
return f"Stop {self.name} at {self.lon}, {self.lat} at {self.time.time()} {self.date.date()}"
return f"Stop {self.name} at {self.lon}, {self.lat} at {self.time.time()} {self.date.date()}"

View File

@@ -1,37 +1,28 @@
from pyrmv.classes.Leg import Leg
from pyrmv.classes.Stop import StopTrip
from isodate import parse_duration
class Trip():
"""Trip object."""
from pyrmv.classes.Leg import Leg
from pyrmv.classes.Stop import StopTrip
class Trip:
"""Trip object."""
def __init__(self, data: dict):
self.raw_data = data
self.origin = StopTrip(data["Origin"])
self.destination = StopTrip(data["Destination"])
self.legs = []
for leg in data["LegList"]["Leg"]:
self.legs.append(Leg(leg))
self.legs.extend(Leg(leg) for leg in data["LegList"]["Leg"])
self.calculation = data["calculation"]
self.index = data["idx"]
self.id = data["tripId"]
self.ctx_recon = data["ctxRecon"]
self.duration = parse_duration(data["duration"])
if "rtDuration" in data:
self.real_time_duration = parse_duration(data["rtDuration"])
else:
self.real_time_duration = None
self.real_time_duration = (
None if "rtDuration" not in data else parse_duration(data["rtDuration"])
)
self.checksum = data["checksum"]
if "transferCount" in data:
self.transfer_count = data["transferCount"]
else:
self.transfer_count = 0
self.transfer_count = data.get("transferCount", 0)
def __str__(self) -> str:
return f"Trip from {self.origin.name} to {self.destination.name} lasting {self.duration} ({self.real_time_duration}) with {len(self.legs)} legs and {self.transfer_count} transfers"
return f"Trip from {self.origin.name} to {self.destination.name} lasting {self.duration} ({self.real_time_duration}) with {len(self.legs)} legs and {self.transfer_count} transfers"

View File

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