onboardapis.train.it.ti

Implementation of the Italian operator Trenitalia S.p.A.

Contains the following APIs:


 1"""
 2Implementation of the Italian operator Trenitalia S.p.A.
 3
 4Contains the following APIs:
 5
 6- ``PortaleRegionale``
 7
 8---
 9"""
10from __future__ import annotations
11
12from .apis import PortaleRegionale
13
14__all__ = [
15    "PortaleRegionale",
16]
class PortaleRegionale(onboardapis.train.Train):
17class PortaleRegionale(Train):
18    """
19    Wrapper for interacting with the Trenitalia PortaleRegionale API
20    """
21
22    _api: PortaleRegionaleConnector
23
24    # noinspection PyTypeHints
25    _stations: dict[ID, TrainStation]
26    """A dict that contains the known stations (origin, destination and passed stations)"""
27
28    def __init__(self):
29        self._api = PortaleRegionaleConnector()
30        self._stations = dict()
31        Train.__init__(self)
32
33    @property
34    def now(self) -> datetime:
35        return datetime.fromisoformat(self._api['infovaggio']['datetime'])
36
37    @property
38    def id(self) -> ID:
39        return self.line_number
40
41    @property
42    def line_number(self) -> str:
43        return self._api['infovaggio']['infos']['trackNumber']
44
45    def delay(self) -> timedelta:
46        delay_minutes = self._api['infovaggio']['infos'].get('delay', '')
47        return timedelta() if delay_minutes in ('0', '') else timedelta(minutes=int(delay_minutes))
48
49    @property
50    def speed(self) -> float:
51        if self._api['infovaggio'].get('isGpsValid', 'false').lower() == 'true':
52            return meters_per_second(kilometers_per_hour=(float(self._api['infovaggio']['infos']['speed'])))
53        raise DataInvalidError('GPS data invalid.')
54
55    @property
56    def stations_dict(self) -> dict[ID, TrainStation]:
57        first = self._api['infovaggio']['infos']['stazionePartenza']
58        current = self._api['infovaggio']['nextStation']
59        last = self._api['infovaggio']['infos']['stazioneArrivo']
60
61        if len(self._stations) == 0:
62            self._stations = {
63                first: TrainStation(
64                    id=first, name=first,
65                    arrival=None, departure=None, position=None, distance=None, _connections=(), platform=None
66                ),
67                current: TrainStation(
68                    id=current, name=current,
69                    arrival=None, departure=None, position=None, distance=None, _connections=(), platform=None
70                ),  # if current == first Python will just overwrite the entry
71                last: TrainStation(
72                    id=last, name=last,
73                    arrival=None, departure=None, position=None, distance=None, _connections=(), platform=None
74                ),
75            }
76            return self._stations.copy()
77
78        self._stations = {
79            first: TrainStation(
80                id=first, name=first,
81                arrival=None, departure=None, position=None, distance=None, _connections=(), platform=None
82            ),
83            **{key: value for key, value in tuple(self._stations.items())[1:-1]},
84            current: TrainStation(
85                id=current, name=current,
86                arrival=None, departure=None, position=None, distance=None, _connections=(), platform=None
87            ),
88            last: TrainStation(
89                id=last, name=last,
90                arrival=None, departure=None, position=None, distance=None, _connections=(), platform=None
91            ),  # if current == last Python will just overwrite the entry
92        }
93        return self._stations.copy()
94
95    @property
96    def current_station(self) -> TrainStation:
97        return self.stations_dict[self._api['infovaggio']['nextStation']]

Wrapper for interacting with the Trenitalia PortaleRegionale API

now: datetime.datetime
33    @property
34    def now(self) -> datetime:
35        return datetime.fromisoformat(self._api['infovaggio']['datetime'])

Get the current time as seen by the vehicle

Returns

The current time

id: ~ID
37    @property
38    def id(self) -> ID:
39        return self.line_number
Returns

The ID of the vehicle

Raises
  • DataInvalidError: If the ID could not be fetched from the server
line_number: str
41    @property
42    def line_number(self) -> str:
43        return self._api['infovaggio']['infos']['trackNumber']
Returns

The line number

Raises
  • DataInvalidError: If the line number could not be fetched from the server
def delay(self) -> datetime.timedelta:
45    def delay(self) -> timedelta:
46        delay_minutes = self._api['infovaggio']['infos'].get('delay', '')
47        return timedelta() if delay_minutes in ('0', '') else timedelta(minutes=int(delay_minutes))
speed: float
49    @property
50    def speed(self) -> float:
51        if self._api['infovaggio'].get('isGpsValid', 'false').lower() == 'true':
52            return meters_per_second(kilometers_per_hour=(float(self._api['infovaggio']['infos']['speed'])))
53        raise DataInvalidError('GPS data invalid.')
stations_dict: dict[~ID, onboardapis.train.TrainStation]
55    @property
56    def stations_dict(self) -> dict[ID, TrainStation]:
57        first = self._api['infovaggio']['infos']['stazionePartenza']
58        current = self._api['infovaggio']['nextStation']
59        last = self._api['infovaggio']['infos']['stazioneArrivo']
60
61        if len(self._stations) == 0:
62            self._stations = {
63                first: TrainStation(
64                    id=first, name=first,
65                    arrival=None, departure=None, position=None, distance=None, _connections=(), platform=None
66                ),
67                current: TrainStation(
68                    id=current, name=current,
69                    arrival=None, departure=None, position=None, distance=None, _connections=(), platform=None
70                ),  # if current == first Python will just overwrite the entry
71                last: TrainStation(
72                    id=last, name=last,
73                    arrival=None, departure=None, position=None, distance=None, _connections=(), platform=None
74                ),
75            }
76            return self._stations.copy()
77
78        self._stations = {
79            first: TrainStation(
80                id=first, name=first,
81                arrival=None, departure=None, position=None, distance=None, _connections=(), platform=None
82            ),
83            **{key: value for key, value in tuple(self._stations.items())[1:-1]},
84            current: TrainStation(
85                id=current, name=current,
86                arrival=None, departure=None, position=None, distance=None, _connections=(), platform=None
87            ),
88            last: TrainStation(
89                id=last, name=last,
90                arrival=None, departure=None, position=None, distance=None, _connections=(), platform=None
91            ),  # if current == last Python will just overwrite the entry
92        }
93        return self._stations.copy()
current_station: onboardapis.train.TrainStation
95    @property
96    def current_station(self) -> TrainStation:
97        return self.stations_dict[self._api['infovaggio']['nextStation']]