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

New date and time behavior, improved typing

Browse Source
This commit is contained in:
Profitroll
2023-11-27 21:44:17 +01:00
parent 5690080a6a
commit 3ee90c7a41
17 changed files with 598 additions and 415 deletions
Download Patch File Download Diff File Expand all files Collapse all files

174
src/pyrmv/classes/board.py
View File

@@ -1,41 +1,57 @@
from datetime import datetime
from typing import Any, Mapping
from typing import Any, List, Mapping, Union
from pyrmv.classes.journey import Journey
from pyrmv.classes.message import Message
from pyrmv.classes.stop import Stop
from pyrmv.utility import ref_upgrade
class LineArrival:
def __init__(self, data: Mapping[str, Any], client, retrieve_stops: bool = True):
def __init__(
self,
data: Mapping[str, Any],
retrieve_stops: bool = False,
retrieve_journey: bool = False,
client: Union["pyrmv.Client", None] = None,
):
if (retrieve_stops or retrieve_journey) and client is None:
raise KeyError(
"Stops/journey retrieval requested but client argument was not provided."
)
# Upgrade is temporarily used due to RMV API mismatch
# self.journey = client.journey_detail(data["JourneyDetailRef"]["ref"])
self.journey = client.journey_detail(ref_upgrade(data["JourneyDetailRef"]["ref"]))
self.journey: Union[Journey, None] = (
client.journey_detail(ref_upgrade(data["JourneyDetailRef"]["ref"]))
if retrieve_journey
else None
)
self.status = data["JourneyStatus"]
self.messages = []
self.name = data["name"]
self.type = data["type"]
self.stop_name = data["stop"]
self.stop_id = data["stopid"]
self.stop_id_ext = data["stopExtId"]
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"]
self.status: str = data["JourneyStatus"]
self.messages: List[Message] = []
self.name: str = data["name"]
self.type: str = data["type"]
self.stop_name: str = data["stop"]
self.stop_id: str = data["stopid"]
self.stop_id_ext: Union[str, None] = data.get("stopExtId")
self.stop: Union[Stop, None] = (
client.stop_by_id(self.stop_id) if retrieve_stops else None
)
self.time: datetime = datetime.strptime(
f"{data['date']} {data['time']}", "%Y-%m-%d %H:%M:%S"
)
self.time_real: datetime = (
datetime.strptime(f"{data['rtDate']} {data['rtTime']}", "%Y-%m-%d %H:%M:%S")
if data.get("rtTime") and data.get("rtDate")
else None
)
self.reachable: bool = data["reachable"]
self.origin: Union[str, None] = data.get("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")
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()}"
@@ -43,35 +59,51 @@ class LineArrival:
class LineDeparture:
def __init__(self, data: Mapping[str, Any], client, retrieve_stops: bool = True):
def __init__(
self,
data: Mapping[str, Any],
retrieve_stops: bool = False,
retrieve_journey: bool = False,
client: Union["pyrmv.Client", None] = None,
):
if (retrieve_stops or retrieve_journey) and client is None:
raise KeyError(
"Stops/journey retrieval requested but client argument was not provided."
)
# Upgrade is temporarily used due to RMV API mismatch
# self.journey = client.journey_detail(data["JourneyDetailRef"]["ref"])
self.journey = client.journey_detail(ref_upgrade(data["JourneyDetailRef"]["ref"]))
self.journey: Union[Journey, None] = (
client.journey_detail(ref_upgrade(data["JourneyDetailRef"]["ref"]))
if retrieve_journey
else None
)
self.status = data["JourneyStatus"]
self.messages = []
self.name = data["name"]
self.type = data["type"]
self.stop_name = data["stop"]
self.stop_id = data["stopid"]
self.stop_id_ext = data["stopExtId"]
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")
self.reachable = data["reachable"]
self.direction = data["direction"]
self.direction_flag = data["directionFlag"]
self.status: str = data["JourneyStatus"]
self.messages: List[Message] = []
self.name: str = data["name"]
self.type: str = data["type"]
self.stop_name: str = data["stop"]
self.stop_id: str = data["stopid"]
self.stop_id_ext: Union[str, None] = data.get("stopExtId")
self.stop: Union[Stop, None] = (
client.stop_by_id(self.stop_id) if retrieve_stops else None
)
self.time: datetime = datetime.strptime(
f"{data['date']} {data['time']}", "%Y-%m-%d %H:%M:%S"
)
self.time_real: datetime = (
datetime.strptime(f"{data['rtDate']} {data['rtTime']}", "%Y-%m-%d %H:%M:%S")
if data.get("rtTime") and data.get("rtDate")
else None
)
self.reachable: bool = data["reachable"]
self.direction: Union[str, None] = data.get("direction")
self.direction_flag: Union[str, None] = data.get("directionFlag")
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()}"
@@ -79,28 +111,70 @@ class LineDeparture:
class BoardArrival(list):
def __init__(self, data: Mapping[str, Any], client, retrieve_stops: bool = True):
def __init__(
self,
data: Mapping[str, Any],
retrieve_stops: bool = False,
retrieve_journey: bool = False,
client: Union["pyrmv.Client", None] = None,
):
"""Arrival board representation
### Args:
* data (`Mapping[str, Any]`): Dictionary from RMV to be parsed.
* retrieve_stops (`bool`, *optional*): Retrieve `Stop` objects for each line of the board. Defaults to `False`.
* retrieve_journey (`bool`, *optional*): Retrieve `Journey` object for each line of the board. Defaults to `False`.
* client (`Union[Client, None]`, *optional*): Client to be used if `retrieve_stops` or `retrieve_journey` are set to `True`. Defaults to `None`.
"""
super().__init__([])
if "Arrival" not in data:
return
for line in data["Arrival"]:
self.append(LineArrival(line, client, retrieve_stops=retrieve_stops))
self.append(
LineArrival(
line,
retrieve_stops=retrieve_stops,
retrieve_journey=retrieve_journey,
client=client,
)
)
def __str__(self) -> str:
return "Arrival board\n" + "\n".join([str(line) for line in self])
class BoardDeparture(list):
def __init__(self, data: Mapping[str, Any], client, retrieve_stops: bool = True):
def __init__(
self,
data: Mapping[str, Any],
retrieve_stops: bool = False,
retrieve_journey: bool = False,
client: Union["pyrmv.Client", None] = None,
):
"""Departure board representation
### Args:
* data (`Mapping[str, Any]`): Dictionary from RMV to be parsed.
* retrieve_stops (`bool`, *optional*): Retrieve `Stop` objects for each line of the board. Defaults to `False`.
* retrieve_journey (`bool`, *optional*): Retrieve `Journey` object for each line of the board. Defaults to `False`.
* client (`Union[Client, None]`, *optional*): Client to be used if `retrieve_stops` or `retrieve_journey` are set to `True`. Defaults to `None`.
"""
super().__init__([])
if "Departure" not in data:
return
for line in data["Departure"]:
self.append(LineDeparture(line, client, retrieve_stops=retrieve_stops))
self.append(
LineDeparture(
line,
retrieve_stops=retrieve_stops,
retrieve_journey=retrieve_journey,
client=client,
)
)
def __str__(self) -> str:
return "Departure board\n" + "\n".join([str(line) for line in self])