pyurbantz/urbantz/delivery.py

from datetime import datetime, timedelta
from urbantz.base import JSONSerializable, Coordinates
from urbantz.utils import DictObject
from urbantz.exceptions import APIError
import requests


class Delivery(JSONSerializable):
    """
    A UrbanTZ delivery with a unique ID.
    """

    def __init__(self, tracking_code=None):
        """
        :param tracking_code: A delivery public tracking code.
        :type tracking_code: str or None
        """
        self.tracking_code = tracking_code
        """
        The delivery public tracking code.

        :type: str or None
        """

        self.last_updated = None
        """
        Last API update date/time. Is None if data has never been fetched
        from the API.

        :type: datetime or None
        """

        self.payload = None
        """
        Latest parsed JSON payload from the API. Is None if data has never
        been fetched from the API or loaded via :meth:`Delivery.use`.

        :type: dict or None
        """

        self.id = None
        """
        Identifier for this delivery.

        :type: str or None
        """

        self.date = None
        """
        Date of the delivery.

        :type: date or None
        """

        self.task_id = None
        """
        Task identifier for this delivery.

        :type: str or None
        """

        self.instructions = None
        """
        Delivery instructions given to the driver.

        :type: str or None
        """

        self.arrival_time = None
        """
        Estimated or actual arrival time.

        :type: datetime or None
        """

        self.eta_margin = None
        """
        Margin, in minutes, for the estimated time of arrival.

        :type: int or None
        """

        self.eta_rounding = None
        """
        Rounding, in minutes, for the estimated time of arrival.

        :type: int or None
        """

        self.time_window = None
        """
        Planned delivery time window (start and end datetimes)

        :type: list(datetime) or None
        """

        self.position = None
        """
        Coordinates of the delivery truck's position.

        :type: urbantz.base.Coordinates or None
        """

        self.destination = None
        """
        Coordinates of the delivery destination.

        :type: urbantz.base.Coordinates or None
        """

        self.recipient = None
        """
        Informations about the recipient (name, language, phone number, etc.)
        Does not contain the destination location information.

        :type: urbantz.utils.DictObject or None
        """

        self.features = None
        """
        Dictionary of booleans indicating which features of the UrbanTZ
        tracking software are enabled on this delivery.
        For example, ``consumerModifyInstructions`` will indicate whether
        the client is allowed to update the delivery instructions after the
        driver departs for its round.

        :type: urbantz.utils.DictObject or None
        """

    def __repr__(self):
        return '{}({})'.format(
            self.__class__.__name__, repr(self.tracking_code))

    @property
    def api_url(self):
        """
        URL pointing to the API endpoint to use for the specific delivery.

        :type: str
        """
        return 'https://backend.urbantz.com/public/task/tracking/{}'.format(
            self.tracking_code)

    @property
    def eta(self):
        """
        Estimated time of arrival: start and end datetimes, computed from the
        arrival time and rounding settings

        :type: list(datetime) or None
        """
        if not (self.eta_margin and self.eta_rounding and self.arrival_time):
            return
        start = self.arrival_time - timedelta(minutes=self.eta_margin / 2)
        start -= timedelta(
            minutes=start.minute % self.eta_rounding,
            seconds=start.second,
            microseconds=start.microsecond,
        )
        end = start + timedelta(minutes=self.eta_margin)
        return [start, end]

    def update(self):
        """
        Fetch the latest delivery information from the API.

        :raises urbantz.exceptions.APIError: If the API returned a JSON error.
        :raises requests.exceptions.HTTPError: If the response has an
           HTTP 4xx or 5xx code, or an empty payload.
        """
        resp = requests.get(self.api_url)
        data = {}
        try:
            data = resp.json()
        except Exception:
            pass

        if 'error' in data:
            raise APIError(data['error'])

        resp.raise_for_status()
        if not data:
            # If requests does not raise anything and there is no data,
            # raise our own error
            raise APIError({'message': 'API returned an empty payload'})

        self.use(data)

        # TODO: See if the payload holds a last update value
        self.last_updated = datetime.now()

    def use(self, payload):
        """
        Use a parsed JSON payload to update the properties.

        :param dict payload: A parsed JSON payload from the API.
        """
        self.payload = payload
        self.id = self.payload['_id']
        self.date = datetime.strptime(
            self.payload['date'][:10], "%Y-%m-%d").date()
        self.task_id = self.payload['taskId']
        self.by = self.payload['by']
        self.instructions = self.payload['instructions']

        self.arrival_time = datetime.strptime(
            self.payload['arriveTime'], '%Y-%m-%dT%H:%M:%S.000Z')
        self.eta_margin = self.payload['eta']['margin']
        self.eta_rounding = self.payload['eta']['rounding']
        self.time_window = [
            datetime.strptime(
                self.payload['timeWindow']['start'],
                '%Y-%m-%dT%H:%M:%S.000Z',
            ),
            datetime.strptime(
                self.payload['timeWindow']['stop'],
                '%Y-%m-%dT%H:%M:%S.000Z',
            ),
        ]

        self.position = Coordinates.fromJSON(self.payload['position'])
        self.destination = Coordinates.fromJSON(
            self.payload['location']['location']['geometry'])
        self.recipient = DictObject(self.payload['contact'])
        self.features = DictObject(self.payload['features'])

    @classmethod
    def fromJSON(cls, payload):
        """
        Create a Delivery instance from an existing payload.

        :param payload: A parsed JSON payload.
        :type payload: dict
        """
        instance = cls()
        instance.use(payload)
        return instance

    def toJSON(self):
        return self.payload