core/homeassistant/components/sensor/uber.py

"""
Support for the Uber API.

For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/sensor.uber/
"""
import logging
from datetime import timedelta

from homeassistant.helpers.entity import Entity
from homeassistant.util import Throttle

_LOGGER = logging.getLogger(__name__)
REQUIREMENTS = ['uber_rides==0.2.1']

ICON = 'mdi:taxi'

# Return cached results if last scan was less then this time ago.
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=60)


def setup_platform(hass, config, add_devices, discovery_info=None):
    """Set up the Uber sensor."""
    if None in (config.get('start_latitude'), config.get('start_longitude')):
        _LOGGER.error(
            "You must set start latitude and longitude to use the Uber sensor!"
        )
        return False

    if config.get('server_token') is None:
        _LOGGER.error("You must set a server_token to use the Uber sensor!")
        return False

    from uber_rides.session import Session

    session = Session(server_token=config.get('server_token'))

    wanted_product_ids = config.get('product_ids')

    dev = []
    timeandpriceest = UberEstimate(session, config['start_latitude'],
                                   config['start_longitude'],
                                   config.get('end_latitude'),
                                   config.get('end_longitude'))
    for product_id, product in timeandpriceest.products.items():
        if (wanted_product_ids is not None) and \
           (product_id not in wanted_product_ids):
            continue
        dev.append(UberSensor('time', timeandpriceest, product_id, product))
        is_metered = (product['price_details']['estimate'] == "Metered")
        if 'price_details' in product and is_metered is False:
            dev.append(UberSensor('price', timeandpriceest,
                                  product_id, product))
    add_devices(dev)


# pylint: disable=too-few-public-methods
class UberSensor(Entity):
    """Implementation of an Uber sensor."""

    def __init__(self, sensorType, products, product_id, product):
        """Initialize the Uber sensor."""
        self.data = products
        self._product_id = product_id
        self._product = product
        self._sensortype = sensorType
        self._name = "{} {}".format(self._product['display_name'],
                                    self._sensortype)
        if self._sensortype == "time":
            self._unit_of_measurement = "min"
            time_estimate = self._product.get('time_estimate_seconds', 0)
            self._state = int(time_estimate / 60)
        elif self._sensortype == "price":
            if 'price_details' in self._product:
                price_details = self._product['price_details']
                self._unit_of_measurement = price_details.get('currency_code',
                                                              'N/A')
                if 'low_estimate' in price_details:
                    statekey = 'minimum'
                else:
                    statekey = 'low_estimate'
                self._state = int(price_details.get(statekey, 0))
            else:
                self._unit_of_measurement = 'N/A'
                self._state = 0
        self.update()

    @property
    def name(self):
        """Return the name of the sensor."""
        return self._name

    @property
    def state(self):
        """Return the state of the sensor."""
        return self._state

    @property
    def unit_of_measurement(self):
        """Return the unit of measurement of this entity, if any."""
        return self._unit_of_measurement

    @property
    def device_state_attributes(self):
        """Return the state attributes."""
        time_estimate = self._product.get('time_estimate_seconds', 'N/A')
        params = {
            'Product ID': self._product['product_id'],
            'Product short description': self._product['short_description'],
            'Product display name': self._product['display_name'],
            'Product description': self._product['description'],
            'Pickup time estimate (in seconds)': time_estimate,
            'Trip duration (in seconds)': self._product.get('duration', 'N/A'),
            'Vehicle Capacity': self._product['capacity']
        }

        if 'price_details' in self._product:
            price_details = self._product['price_details']
            distance_key = 'Trip distance (in {}s)'.format(price_details[
                'distance_unit'])
            distance_val = self._product.get('distance')
            params['Minimum price'] = price_details['minimum'],
            params['Cost per minute'] = price_details['cost_per_minute'],
            params['Distance units'] = price_details['distance_unit'],
            params['Cancellation fee'] = price_details['cancellation_fee'],
            params['Cost per distance'] = price_details['cost_per_distance'],
            params['Base price'] = price_details['base'],
            params['Price estimate'] = price_details.get('estimate', 'N/A'),
            params['Price currency code'] = price_details.get('currency_code'),
            params['High price estimate'] = price_details.get('high_estimate',
                                                              'N/A'),
            params['Low price estimate'] = price_details.get('low_estimate',
                                                             'N/A'),
            params['Surge multiplier'] = price_details.get('surge_multiplier',
                                                           'N/A')
        else:
            distance_key = 'Trip distance (in miles)'
            distance_val = self._product.get('distance', 'N/A')

        params[distance_key] = distance_val

        return params

    @property
    def icon(self):
        """Icon to use in the frontend, if any."""
        return ICON

    # pylint: disable=too-many-branches
    def update(self):
        """Get the latest data from the Uber API and update the states."""
        self.data.update()
        self._product = self.data.products[self._product_id]
        if self._sensortype == "time":
            time_estimate = self._product.get('time_estimate_seconds', 0)
            self._state = int(time_estimate / 60)
        elif self._sensortype == "price":
            price_details = self._product.get('price_details')
            if price_details is not None:
                min_price = price_details.get('minimum')
                self._state = int(price_details.get('low_estimate', min_price))
            else:
                self._state = 0


# pylint: disable=too-few-public-methods
class UberEstimate(object):
    """The class for handling the time and price estimate."""

    # pylint: disable=too-many-arguments
    def __init__(self, session, start_latitude, start_longitude,
                 end_latitude=None, end_longitude=None):
        """Initialize the UberEstimate object."""
        self._session = session
        self.start_latitude = start_latitude
        self.start_longitude = start_longitude
        self.end_latitude = end_latitude
        self.end_longitude = end_longitude
        self.products = None
        self.update()

    @Throttle(MIN_TIME_BETWEEN_UPDATES)
    def update(self):
        """Get the latest product info and estimates from the Uber API."""
        from uber_rides.client import UberRidesClient
        client = UberRidesClient(self._session)

        self.products = {}

        products_response = client.get_products(
            self.start_latitude, self.start_longitude)

        products = products_response.json.get('products')

        for product in products:
            self.products[product['product_id']] = product

        if self.end_latitude is not None and self.end_longitude is not None:
            price_response = client.get_price_estimates(
                self.start_latitude,
                self.start_longitude,
                self.end_latitude,
                self.end_longitude)

            prices = price_response.json.get('prices', [])

            for price in prices:
                product = self.products[price['product_id']]
                price_details = product.get("price_details")
                product["duration"] = price.get('duration', '0')
                product["distance"] = price.get('distance', '0')
                if price_details is not None:
                    price_details["estimate"] = price.get('estimate',
                                                          '0')
                    price_details["high_estimate"] = price.get('high_estimate',
                                                               '0')
                    price_details["low_estimate"] = price.get('low_estimate',
                                                              '0')
                    surge_multiplier = price.get('surge_multiplier', '0')
                    price_details["surge_multiplier"] = surge_multiplier

        estimate_response = client.get_pickup_time_estimates(
            self.start_latitude, self.start_longitude)

        estimates = estimate_response.json.get('times')

        for estimate in estimates:
            self.products[estimate['product_id']][
                "time_estimate_seconds"] = estimate.get('estimate', '0')