4 Commits

Author SHA1 Message Date
4fd566f7c4 Small docstring added 2022-10-06 12:09:30 +02:00
170b472ca4 Test added 2022-10-06 11:36:32 +02:00
d49196ff26 Message improved 2022-10-06 11:36:18 +02:00
ebb21aea8c Journey added 2022-10-06 11:36:08 +02:00
6 changed files with 91 additions and 10 deletions

4
.gitignore vendored
View File

@@ -156,8 +156,8 @@ cython_debug/
# Custom # Custom
.pypirc .pypirc
tests tests
test.* test.bat
test_all.py test_key.json
build.* build.*
cleanup.* cleanup.*
publish.* publish.*

View File

@@ -1,2 +1,20 @@
from pyrmv.classes.Stop import Stop
from pyrmv.classes.Message import Message
class Journey(): class Journey():
pass """Journey object."""
def __init__(self, data: dict):
self.stops = []
self.direction = data["Directions"]["Direction"][0]["value"]
self.direction_flag = data["Directions"]["Direction"][0]["flag"]
self.messages = []
for stop in data["Stops"]["Stop"]:
self.stops.append(Stop(stop))
for message in data["Messages"]["Message"]:
self.messages.append(Message(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})"

View File

@@ -36,11 +36,19 @@ class Message():
def __init__(self, data: dict) -> None: def __init__(self, data: dict) -> None:
self.affected_stops = [] self.affected_stops = []
for stop in data["affectedStops"]["StopLocation"]: if "affectedStops" in data:
self.affected_stops.append(Stop(stop)) for stop in data["affectedStops"]["StopLocation"]:
self.affected_stops.append(Stop(stop))
self.valid_from_stop = Stop(data["validFromStop"]) if "validFromStop" in data:
self.valid_to_stop = Stop(data["validToStop"]) 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.channels = [] self.channels = []
for channel in data["channel"]: for channel in data["channel"]:

View File

@@ -8,17 +8,35 @@ class Stop():
self.name = data["name"] self.name = data["name"]
self.id = data["id"] self.id = data["id"]
if "extId" in data: if "extId" in data:
self.ext_id = data["extId"] self.ext_id = data["extId"]
else: else:
self.ext_id = None self.ext_id = None
if "description" in data: if "description" in data:
self.description = data["description"] self.description = data["description"]
else: else:
self.description = None self.description = None
self.lon = data["lon"] self.lon = data["lon"]
self.lat = data["lat"] 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
def __str__(self) -> str: def __str__(self) -> str:
return f"Stop {self.name} at {self.lon}, {self.lat}" return f"Stop {self.name} at {self.lon}, {self.lat}"

View File

@@ -1,5 +1,39 @@
from datetime import datetime
from typing import Union
from pyrmv.classes.Journey import Journey
from pyrmv.errors.not_ready import NotReadyYetError from pyrmv.errors.not_ready import NotReadyYetError
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(): def journey_detail(
raise NotReadyYetError()
access_id: str,
id: str,
date: Union[str, datetime] = None,
real_time_mode: Literal[RealTimeMode.FULL, RealTimeMode.INFOS, RealTimeMode.OFF, RealTimeMode.REALTIME, RealTimeMode.SERVER_DEFAULT] = None,
from_id: str = None,
from_index: int = None,
to_id: str = None,
to_index: int = None
):
journey_raw = raw_journey_detail(
accessId=access_id,
id=id,
date=date,
rtMode=real_time_mode.code,
fromId=from_id,
fromIdx=from_index,
toId=to_id,
toIdx=to_index
)
find_exception(journey_raw)
return Journey(journey_raw)

View File

@@ -57,7 +57,10 @@ def journey_detail(accessId: str,
payload = {} payload = {}
for var, val in locals().items(): for var, val in locals().items():
if str(var) not in ["json", "headers", "payload"]: if str(var) == "rtMode":
if val != None:
payload["rtMode"] = val.upper()
elif str(var) not in ["json", "headers", "payload"]:
if val != None: if val != None:
payload[str(var)] = val payload[str(var)] = val