evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Migrate to voluptuous and upgrade uber_rides to 0.2.5 (#3181)

Browse source
This commit is contained in:
Fabian Affolter 2016-09-05 19:33:35 +02:00 • committed by Teagan Glenn
parent 428db4a644
commit 4638696f8c
2 changed files with 100 additions and 89 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

187
homeassistant/components/sensor/uber.py
View file

@ -7,50 +7,61 @@ https://home-assistant.io/components/sensor.uber/
import logging import logging
from datetime import timedelta from datetime import timedelta
import voluptuous as vol
from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.helpers.entity import Entity from homeassistant.helpers.entity import Entity
from homeassistant.util import Throttle from homeassistant.util import Throttle
import homeassistant.helpers.config_validation as cv
REQUIREMENTS = ['uber_rides==0.2.5']
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
REQUIREMENTS = ["uber_rides==0.2.4"]
ICON = "mdi:taxi" CONF_END_LATITUDE = 'end_latitude'
CONF_END_LONGITUDE = 'end_longitude'
CONF_PRODUCT_IDS = 'product_ids'
CONF_SERVER_TOKEN = 'server_token'
CONF_START_LATITUDE = 'start_latitude'
CONF_START_LONGITUDE = 'start_longitude'
ICON = 'mdi:taxi'
# Return cached results if last scan was less then this time ago.
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=60) MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=60)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_SERVER_TOKEN): cv.string,
vol.Required(CONF_START_LATITUDE): cv.latitude,
vol.Required(CONF_START_LONGITUDE): cv.longitude,
vol.Optional(CONF_END_LATITUDE): cv.latitude,
vol.Optional(CONF_END_LONGITUDE): cv.longitude,
vol.Optional(CONF_PRODUCT_IDS, default=[]):
vol.All(cv.ensure_list, [cv.string]),
})
def setup_platform(hass, config, add_devices, discovery_info=None): def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up the Uber sensor.""" """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 from uber_rides.session import Session
session = Session(server_token=config.get("server_token")) session = Session(server_token=config.get(CONF_SERVER_TOKEN))
wanted_product_ids = config.get("product_ids") wanted_product_ids = config.get(CONF_PRODUCT_IDS)
dev = [] dev = []
timeandpriceest = UberEstimate(session, config["start_latitude"], timeandpriceest = UberEstimate(session, config[CONF_START_LATITUDE],
config["start_longitude"], config[CONF_START_LONGITUDE],
config.get("end_latitude"), config.get(CONF_END_LATITUDE),
config.get("end_longitude")) config.get(CONF_END_LONGITUDE))
for product_id, product in timeandpriceest.products.items(): for product_id, product in timeandpriceest.products.items():
if (wanted_product_ids is not None) and \ if (wanted_product_ids is not None) and \
(product_id not in wanted_product_ids): (product_id not in wanted_product_ids):
continue continue
dev.append(UberSensor("time", timeandpriceest, product_id, product)) dev.append(UberSensor('time', timeandpriceest, product_id, product))
if (product.get("price_details") is not None) and \ if (product.get('price_details') is not None) and \
product["price_details"]["estimate"] is not "Metered": product['price_details']['estimate'] is not 'Metered':
dev.append(UberSensor("price", timeandpriceest, dev.append(UberSensor(
product_id, product)) 'price', timeandpriceest, product_id, product))
add_devices(dev) add_devices(dev)
@ -64,20 +75,20 @@ class UberSensor(Entity):
self._product_id = product_id self._product_id = product_id
self._product = product self._product = product
self._sensortype = sensorType self._sensortype = sensorType
self._name = "{} {}".format(self._product["display_name"], self._name = '{} {}'.format(self._product['display_name'],
self._sensortype) self._sensortype)
if self._sensortype == "time": if self._sensortype == 'time':
self._unit_of_measurement = "min" self._unit_of_measurement = 'min'
time_estimate = self._product.get("time_estimate_seconds", 0) time_estimate = self._product.get('time_estimate_seconds', 0)
self._state = int(time_estimate / 60) self._state = int(time_estimate / 60)
elif self._sensortype == "price": elif self._sensortype == 'price':
if self._product.get("price_details") is not None: if self._product.get('price_details') is not None:
price_details = self._product["price_details"] price_details = self._product['price_details']
self._unit_of_measurement = price_details.get("currency_code") self._unit_of_measurement = price_details.get('currency_code')
if price_details.get("low_estimate") is not None: if price_details.get('low_estimate') is not None:
statekey = "minimum" statekey = 'minimum'
else: else:
statekey = "low_estimate" statekey = 'low_estimate'
self._state = int(price_details.get(statekey, 0)) self._state = int(price_details.get(statekey, 0))
else: else:
self._state = 0 self._state = 0
@ -86,8 +97,8 @@ class UberSensor(Entity):
@property @property
def name(self): def name(self):
"""Return the name of the sensor.""" """Return the name of the sensor."""
if "uber" not in self._name.lower(): if 'uber' not in self._name.lower():
self._name = "Uber{}".format(self._name) self._name = 'Uber{}'.format(self._name)
return self._name return self._name
@property @property
@ -105,35 +116,35 @@ class UberSensor(Entity):
"""Return the state attributes.""" """Return the state attributes."""
time_estimate = self._product.get("time_estimate_seconds") time_estimate = self._product.get("time_estimate_seconds")
params = { params = {
"Product ID": self._product["product_id"], 'Product ID': self._product['product_id'],
"Product short description": self._product["short_description"], 'Product short description': self._product['short_description'],
"Product display name": self._product["display_name"], 'Product display name': self._product['display_name'],
"Product description": self._product["description"], 'Product description': self._product['description'],
"Pickup time estimate (in seconds)": time_estimate, 'Pickup time estimate (in seconds)': time_estimate,
"Trip duration (in seconds)": self._product.get("duration"), 'Trip duration (in seconds)': self._product.get('duration'),
"Vehicle Capacity": self._product["capacity"] 'Vehicle Capacity': self._product['capacity']
} }
if self._product.get("price_details") is not None: if self._product.get('price_details') is not None:
price_details = self._product["price_details"] price_details = self._product['price_details']
dunit = price_details.get("distance_unit") dunit = price_details.get('distance_unit')
distance_key = "Trip distance (in {}s)".format(dunit) distance_key = 'Trip distance (in {}s)'.format(dunit)
distance_val = self._product.get("distance") distance_val = self._product.get('distance')
params["Cost per minute"] = price_details.get("cost_per_minute") params['Cost per minute'] = price_details.get('cost_per_minute')
params["Distance units"] = price_details.get("distance_unit") params['Distance units'] = price_details.get('distance_unit')
params["Cancellation fee"] = price_details.get("cancellation_fee") params['Cancellation fee'] = price_details.get('cancellation_fee')
cpd = price_details.get("cost_per_distance") cpd = price_details.get('cost_per_distance')
params["Cost per distance"] = cpd params['Cost per distance'] = cpd
params["Base price"] = price_details.get("base") params['Base price'] = price_details.get('base')
params["Minimum price"] = price_details.get("minimum") params['Minimum price'] = price_details.get('minimum')
params["Price estimate"] = price_details.get("estimate") params['Price estimate'] = price_details.get('estimate')
params["Price currency code"] = price_details.get("currency_code") params['Price currency code'] = price_details.get('currency_code')
params["High price estimate"] = price_details.get("high_estimate") params['High price estimate'] = price_details.get('high_estimate')
params["Low price estimate"] = price_details.get("low_estimate") params['Low price estimate'] = price_details.get('low_estimate')
params["Surge multiplier"] = price_details.get("surge_multiplier") params['Surge multiplier'] = price_details.get('surge_multiplier')
else: else:
distance_key = "Trip distance (in miles)" distance_key = 'Trip distance (in miles)'
distance_val = self._product.get("distance") distance_val = self._product.get('distance')
params[distance_key] = distance_val params[distance_key] = distance_val
@ -149,14 +160,14 @@ class UberSensor(Entity):
"""Get the latest data from the Uber API and update the states.""" """Get the latest data from the Uber API and update the states."""
self.data.update() self.data.update()
self._product = self.data.products[self._product_id] self._product = self.data.products[self._product_id]
if self._sensortype == "time": if self._sensortype == 'time':
time_estimate = self._product.get("time_estimate_seconds", 0) time_estimate = self._product.get('time_estimate_seconds', 0)
self._state = int(time_estimate / 60) self._state = int(time_estimate / 60)
elif self._sensortype == "price": elif self._sensortype == 'price':
price_details = self._product.get("price_details") price_details = self._product.get('price_details')
if price_details is not None: if price_details is not None:
min_price = price_details.get("minimum") min_price = price_details.get('minimum')
self._state = int(price_details.get("low_estimate", min_price)) self._state = int(price_details.get('low_estimate', min_price))
else: else:
self._state = 0 self._state = 0
@ -188,39 +199,39 @@ class UberEstimate(object):
products_response = client.get_products( products_response = client.get_products(
self.start_latitude, self.start_longitude) self.start_latitude, self.start_longitude)
products = products_response.json.get("products") products = products_response.json.get('products')
for product in products: for product in products:
self.products[product["product_id"]] = product self.products[product['product_id']] = product
if self.end_latitude is not None and self.end_longitude is not None: if self.end_latitude is not None and self.end_longitude is not None:
price_response = client.get_price_estimates( price_response = client.get_price_estimates(
self.start_latitude, self.start_longitude, self.start_latitude, self.start_longitude,
self.end_latitude, self.end_longitude) self.end_latitude, self.end_longitude)
prices = price_response.json.get("prices", []) prices = price_response.json.get('prices', [])
for price in prices: for price in prices:
product = self.products[price["product_id"]] product = self.products[price['product_id']]
product["duration"] = price.get("duration", "0") product['duration'] = price.get('duration', '0')
product["distance"] = price.get("distance", "0") product['distance'] = price.get('distance', '0')
price_details = product.get("price_details") price_details = product.get('price_details')
if product.get("price_details") is None: if product.get('price_details') is None:
price_details = {} price_details = {}
price_details["estimate"] = price.get("estimate", "0") price_details['estimate'] = price.get('estimate', '0')
price_details["high_estimate"] = price.get("high_estimate", price_details['high_estimate'] = price.get('high_estimate',
"0") '0')
price_details["low_estimate"] = price.get("low_estimate", "0") price_details['low_estimate'] = price.get('low_estimate', '0')
price_details["currency_code"] = price.get("currency_code") price_details['currency_code'] = price.get('currency_code')
surge_multiplier = price.get("surge_multiplier", "0") surge_multiplier = price.get('surge_multiplier', '0')
price_details["surge_multiplier"] = surge_multiplier price_details['surge_multiplier'] = surge_multiplier
product["price_details"] = price_details product['price_details'] = price_details
estimate_response = client.get_pickup_time_estimates( estimate_response = client.get_pickup_time_estimates(
self.start_latitude, self.start_longitude) self.start_latitude, self.start_longitude)
estimates = estimate_response.json.get("times") estimates = estimate_response.json.get('times')
for estimate in estimates: for estimate in estimates:
self.products[estimate["product_id"]][ self.products[estimate['product_id']][
"time_estimate_seconds"] = estimate.get("estimate", "0") 'time_estimate_seconds'] = estimate.get('estimate', '0')

2
requirements_all.txt
View file

@ -471,7 +471,7 @@ transmissionrpc==0.11
twilio==5.4.0 twilio==5.4.0
# homeassistant.components.sensor.uber # homeassistant.components.sensor.uber
uber_rides==0.2.4 uber_rides==0.2.5
# homeassistant.components.device_tracker.unifi # homeassistant.components.device_tracker.unifi
unifi==1.2.5 unifi==1.2.5