onboardapis.mixins

  1"""
  2Mixins for vehicles.
  3
  4The mixins are used to indicate and add functionality to the vehicle classes.
  5"""
  6from __future__ import annotations
  7
  8from abc import ABCMeta, abstractmethod
  9from datetime import timedelta
 10from typing import Generic
 11
 12from .exceptions import DataInvalidError
 13from .data import Position, ID, StationType, API
 14
 15
 16class PositionMixin(metaclass=ABCMeta):
 17    @property
 18    @abstractmethod
 19    def position(self) -> Position:
 20        """
 21        :return: The current position of the vehicle
 22        :raises DataInvalidError: If the position could not be fetched from the server
 23        """
 24        raise NotImplementedError
 25
 26
 27class SpeedMixin(metaclass=ABCMeta):
 28    """
 29    Functionality for a vehicle that provides information about its speed.
 30    """
 31
 32    @property
 33    @abstractmethod
 34    def speed(self) -> float:
 35        r"""
 36        :return: The current speed of the vehicle in $\frac{meters}{second}$
 37        :raises DataInvalidError: If the speed could not be fetched from the server
 38        """
 39        raise NotImplementedError
 40
 41
 42class StationsMixin(Generic[StationType], metaclass=ABCMeta):
 43    """
 44    Functionality for a vehicle that provides information on the journey.
 45    """
 46
 47    def calculate_distance(self, station: StationType) -> float:
 48        """
 49        :return: The distance in meters between the vehicle and a station
 50        :param station: The station to calculate the distance to
 51        :raises DataInvalidError: If the distance between the vehicle and the station could not be calculated
 52                                  due to missing information from the server
 53        :raises NotImplementedError: If the vehicle does not implement ``position`` or ``distance``
 54        """
 55        if hasattr(self, 'position'):
 56            position: Position = getattr(self, 'position')
 57            return station.calculate_distance(position)
 58
 59        if hasattr(self, 'distance'):
 60            distance: float = getattr(self, 'distance')
 61            return station.calculate_distance(distance)
 62
 63        raise NotImplementedError
 64
 65    @property
 66    @abstractmethod
 67    def stations_dict(self) -> dict[ID, StationType]:
 68        """
 69        :return: The stations as a dict of station ID to station instance
 70        :raises DataInvalidError: If the stations could not be fetched from the server
 71        """
 72        raise NotImplementedError
 73
 74    @property
 75    def stations(self) -> list[StationType]:
 76        """
 77        :return: A list that contains every station from `onboardapis.mixins.StationsMixin.stations_dict`.
 78        :raises DataInvalidError: If the stations could not be fetched from the server
 79        """
 80        return list(self.stations_dict.values())
 81
 82    @property
 83    def origin(self) -> StationType:
 84        """
 85        :return: The first station on this trip.
 86        :raises DataInvalidError: If the origin station could not be fetched from the server
 87        """
 88        if len(self.stations) > 0:
 89            return self.stations[0]
 90        raise DataInvalidError("No origin station found!")
 91
 92    @property
 93    @abstractmethod
 94    def current_station(self) -> StationType:
 95        """
 96        :return: The station where this vehicle will arrive next or is currently at
 97        :raises DataInvalidError: If the current station could not be fetched from the server
 98        """
 99        raise NotImplementedError
100
101    @property
102    def destination(self) -> StationType:
103        """
104        :return: The station where this vehicle terminates the current journey
105        :raises DataInvalidError: If the destination station could not be fetched from the server
106        """
107        if len(self.stations) > 0:
108            return self.stations[-1]
109        raise DataInvalidError("No destination station found!")
110
111    @property
112    def delay(self) -> timedelta:
113        """
114        :return: The current delay of the vehicle as a `datetime.timedelta` object.
115        :raises DataInvalidError: If the delay could not be fetched from the server
116        """
117        return timedelta(seconds=(
118            self.current_station.arrival.actual - self.current_station.arrival.scheduled
119        ).total_seconds())
120
121    @property
122    def is_delayed(self) -> bool:
123        """
124        :returns: Whether ``delay`` `> timedelta(seconds=0)`
125        :raises DataInvalidError: If the delay could not be fetched from the server
126        """
127        return self.delay > timedelta()
128
129    @property
130    def distance(self) -> float:
131        """
132        :return: The distance from the start in meters
133        :raises DataInvalidError: If the distance could not be fetched from the server
134        """
135        return self.calculate_distance(self.origin)
136
137
138class InternetAccessMixin(metaclass=ABCMeta):
139    """Adds the internet_access property to a class
140    that defines an :class:`InternetAccessInterface` as ``_internet_access``."""
141    _internet_access: InternetAccessInterface
142
143    @property
144    def internet_access(self) -> InternetAccessInterface:
145        """
146        :return: An interface to manage the internet access for this device
147        """
148        return self._internet_access
149
150
151class InternetAccessInterface(metaclass=ABCMeta):
152    """Interface adding functions for connecting and disconnecting to the internet
153    as well as viewing the current status."""
154
155    _is_enabled: bool = False
156    """Cached information on connection status"""
157    _api: API
158
159    def __init__(self, api: API) -> None:
160        self._api = api
161
162    @abstractmethod
163    def enable(self) -> None:
164        """Enable the internet access for this device.
165
166        **IMPORTANT**:
167            By using this method you automatically agree to the terms of service of the internet access provider.
168
169        :raises ConnectionError: If the internet access is (temporarily) not available.
170        """
171        self._is_enabled = True
172
173    @abstractmethod
174    def disable(self) -> None:
175        """Disable the internet access for this device.
176
177        Disable the internet access for this device by signing out of the captive portal.
178
179        :raises ConnectionError: If the internet access is (temporarily) not available.
180        """
181        if not self.is_enabled:
182            return
183
184        self._is_enabled = False
185
186    @property
187    def is_enabled(self) -> bool:
188        """
189        :return: Whether the internet access is enabled for this device
190        """
191        return self._is_enabled
192
193
194class InternetMetricsInterface(metaclass=ABCMeta):
195    """Interface for information on limited internet access."""
196
197    @property
198    @abstractmethod
199    def limit(self) -> int | float | None:
200        """
201        :return: The total internet access quota in MB or `None` if there is none
202        :raises DataInvalidError: If the quota could not be fetched from the server
203        """
204        raise NotImplementedError
205
206    @property
207    @abstractmethod
208    def used(self) -> int | float | None:
209        """
210        :return: The amount used of the quota in MB or `None` if there is none
211        :raises DataInvalidError: If the usage information could not be fetched from the server
212        """
213        raise NotImplementedError