# -*- coding: utf-8 -*-
import datetime
import re
from functools import total_ordering
from math import floor
from dateutil.parser import parse
from bvggrabber.utils.format import timeformat
from bvggrabber.utils.json import ObjectJSONEncoder
[docs]def compute_remaining(start, end):
"""Compute the number of seconds between ``start`` and ``end`` and return
the number of seconds rounded down to entire minutes. That means:
* [0, 59] => 0
* [60, 119] => 60
* [120, 179] => 120
* [-59, -1] => -60
* [-119, -60] => -120
:param datetime.datetime start: The start to compute the remaining number
of minutes.
:param datetime.datetime end: The end of the interval
:raises: TypeError if either of ``start`` or ``end`` is not a
``datetime.datetime``.
:return: The number of remaining seconds rounded to entire minutes
:rtype: int
"""
if not isinstance(start, datetime.datetime):
raise TypeError("start needs to be a datetime.datetime")
if not isinstance(end, datetime.datetime):
raise TypeError("start needs to be a datetime.datetime")
seconds = (end - start).total_seconds()
return datetime.timedelta(minutes=floor(seconds / 60)).total_seconds()
[docs]class QueryApi(object):
def __init__(self):
pass
[docs] def call(self):
"""Needs to be implemented by inheriting classes!"""
raise NotImplementedError("The inheriting class needs to implement "
"the call() method!")
[docs]class Response(object):
[docs] def __init__(self, state, station=None, departures=None, error=None):
"""Creates a new response. Returned by :meth:`QueryApi.call`
:param bool state: ``True`` iff the request and parsing was successful,
``False`` otherwise.
:param station: If a ``list``, the station name is ambiguous. If a
string the full qualified name of the station.
:param departures: A list of :class:`Departure`` objects
:param Exception error: In case an unexpected error occurred, this
contains the original exception.
If ``state`` is ``True``, ``station`` must be a ``str`` and
``departures`` must be a list of :class:`Departure`` objects. If
``state`` is ``False`` there must be several reasons for that:
1. The provided station name during the :meth:`QueryApi.call`
returned multiple possible departing stations. You have to
specify the name in an unambiguous way.
2. The station does not exist at all.
3. An exception occurred during the :meth:`QueryApi.call`
.. deprecated:: 0.1b3
The ``state`` argument will be removed in the future and will be
computed automatically based on ``station``, ``departures`` and
``error``.
"""
self._state = state
self._departures = [(station, departures)]
self._error = error
if self._error is None:
if isinstance(station, list):
self._state = False
msg = ', '.join(station)
self._error = Exception("Station is ambiguous: %s" % msg)
elif station is None:
self._state = False
self._error = Exception("Station does not exist")
elif isinstance(self._error, str):
self._state = False
self._error = Exception(self._error)
[docs] def merge(self, other):
"""Checks that ``other`` is a :class:`Response` and extends
:attr:`departures` by the departures given in ``other`` iff neither
response object has a invalid state.
"""
if isinstance(other, Response):
if not other.state:
raise ValueError("The response contains errors: " +
str(other.error))
elif not self.state:
raise ValueError("The response contains errors: " +
str(self.error))
else:
self.departures.extend(other.departures)
else:
raise TypeError("The given object is not a response object")
@property
def to_json(self):
""".. deprecated:: 0.0.1
Use :attr:`json` instead.
"""
return ObjectJSONEncoder(ensure_ascii=False).encode(self.departures)
@property
[docs] def departures(self):
"""A list of 2-tuple in the form (:class:`str`, :class:`Departure`).
The first element in the tuple defines the departing station where the
second element holds a list of departure objects.
"""
if self.state:
return self._departures
return str(self.error)
@property
[docs] def error(self):
"""The error that occurred during creation or ``None``"""
return self._error
@property
[docs] def json(self):
"""Uses :class:`~bvggrabber.utils.json.ObjectJSONEncoder` to encode
the :attr:`departures` to a JSON format.
"""
if self.state:
return ObjectJSONEncoder(ensure_ascii=False).encode(self.departures)
return ObjectJSONEncoder(ensure_ascii=False).encode(str(self.error))
@property
[docs] def state(self):
"""``True`` iff the request and parsing was successful, ``False``
otherwise.
"""
return self._state
@total_ordering
[docs]class Departure(object):
[docs] def __init__(self, start, end, when, line, since=None, no_add_day=False):
"""
:param str start: The start station
:param str end: The end station
:param when: The leaving time of the public transport at the given
``start`` station. Might be an :class:`int` (timestamp), a
:class:`datetime.datetime` instance or a :class:`str` accepted by
``dateutil.parse()``. If ``when`` is smaller than ``since`` and the
difference between both times is larger than 12 hours (43200sec),
the API will add another day unless ``no_add_day`` is ``True``.
:param str line: The line of the public transport
:param since: Either ``None`` or :class:`datetime.datetime`. Defines
the temporal start for searching. ``None`` will internally be
resolved as :meth:`datetime.datetime.now`.
:param bool no_add_day: If true, there no additional day will be added
if ``when`` is smaller than ``since``. Default ``False``.
:raises: :exc:`TypeError` if ``when`` is invalid or cannot be parsed.
"""
if since is None:
self.now = datetime.datetime.now()
else:
self.now = since
self.start = start
self.end = end
self.line = line
if isinstance(when, (int, float)):
# We assume to get a UNIX / POSIX timestamp
self.when = datetime.datetime.fromtimestamp(when)
elif isinstance(when, str):
# We need to parse a string. But we need to remove trailing
# whitespaces and *
self.when = parse(re.sub('[\s*]$', '', when))
elif isinstance(when, datetime.datetime):
# Everything's fine, we can just take the parameter as is
self.when = when
else:
raise TypeError("when must be a valid datetime, timestamp or "
"string!")
diff = abs((self.when - self.now).total_seconds())
if not no_add_day and self.when < self.now and diff > 43200:
# 43200 are 12 hours in seconds So we accept a offset of 12 hours
# that is still counted as "time gone" for the current day.
self.when = self.when + datetime.timedelta(days=1)
@property
[docs] def remaining(self):
""".. seealso:: bvggrabber.api.compute_remaining"""
return int(compute_remaining(self.now, self.when))
[docs] def __eq__(self, other):
"""Two departures are assumed to be equal iff their remaining time
and their destination are equal.
Right now we do **not** considering the start or line, since that would
require some kind of geo location in order to define a *total order*.
"""
return ((self.remaining, self.end.lower()) ==
(other.remaining, other.end.lower()))
[docs] def __lt__(self, other):
"""A departure is assumed to be less than another iff its remaining
time is less than the remaining time of the other departure.
Right now we do **not** considering the start, end or line, since that
would require some kind of geo location in order to define a *total
order*.
"""
return (self.remaining < other.remaining)
def __str__(self):
return "Start: %s, End: %s, when: %s, now: %s, line: %s" % (
self.start, self.end, timeformat(self.when),
timeformat(self.now), self.line)
def __repr__(self):
return self.__str__()