"""Support for Tibber sensors.""" import asyncio import logging from datetime import timedelta import aiohttp from homeassistant.components.tibber import DOMAIN as TIBBER_DOMAIN from homeassistant.exceptions import PlatformNotReady from homeassistant.helpers.entity import Entity from homeassistant.util import dt as dt_util from homeassistant.util import Throttle _LOGGER = logging.getLogger(__name__) ICON = 'mdi:currency-usd' ICON_RT = 'mdi:power-plug' SCAN_INTERVAL = timedelta(minutes=1) MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=5) async def async_setup_platform( hass, config, async_add_entities, discovery_info=None): """Set up the Tibber sensor.""" if discovery_info is None: return tibber_connection = hass.data.get(TIBBER_DOMAIN) dev = [] for home in tibber_connection.get_homes(only_active=False): try: await home.update_info() except asyncio.TimeoutError as err: _LOGGER.error("Timeout connecting to Tibber home: %s ", err) raise PlatformNotReady() except aiohttp.ClientError as err: _LOGGER.error("Error connecting to Tibber home: %s ", err) raise PlatformNotReady() if home.has_active_subscription: dev.append(TibberSensorElPrice(home)) if home.has_real_time_consumption: dev.append(TibberSensorRT(home)) async_add_entities(dev, True) class TibberSensorElPrice(Entity): """Representation of an Tibber sensor for el price.""" def __init__(self, tibber_home): """Initialize the sensor.""" self._tibber_home = tibber_home self._last_updated = None self._last_data_timestamp = None self._state = None self._is_available = False self._device_state_attributes = {} self._unit_of_measurement = self._tibber_home.price_unit self._name = 'Electricity price {}'.format(tibber_home.info['viewer'] ['home']['appNickname']) async def async_update(self): """Get the latest data and updates the states.""" now = dt_util.now() if self._tibber_home.current_price_total and self._last_updated and \ self._last_updated.hour == now.hour and self._last_data_timestamp: return if (not self._last_data_timestamp or (self._last_data_timestamp - now).total_seconds()/3600 < 12 or not self._is_available): _LOGGER.debug("Asking for new data.") await self._fetch_data() self._is_available = self._update_current_price() @property def device_state_attributes(self): """Return the state attributes.""" return self._device_state_attributes @property def available(self): """Return True if entity is available.""" return self._is_available @property def name(self): """Return the name of the sensor.""" return self._name @property def state(self): """Return the state of the device.""" return self._state @property def icon(self): """Return the icon to use in the frontend.""" return ICON @property def unit_of_measurement(self): """Return the unit of measurement of this entity.""" return self._unit_of_measurement @property def unique_id(self): """Return a unique ID.""" home = self._tibber_home.info['viewer']['home'] return home['meteringPointData']['consumptionEan'] @Throttle(MIN_TIME_BETWEEN_UPDATES) async def _fetch_data(self): try: await self._tibber_home.update_info() await self._tibber_home.update_price_info() except (asyncio.TimeoutError, aiohttp.ClientError): return data = self._tibber_home.info['viewer']['home'] self._device_state_attributes['app_nickname'] = data['appNickname'] self._device_state_attributes['grid_company'] = \ data['meteringPointData']['gridCompany'] self._device_state_attributes['estimated_annual_consumption'] = \ data['meteringPointData']['estimatedAnnualConsumption'] def _update_current_price(self): state = None max_price = 0 min_price = 10000 sum_price = 0 num = 0 now = dt_util.now() for key, price_total in self._tibber_home.price_total.items(): price_time = dt_util.as_local(dt_util.parse_datetime(key)) price_total = round(price_total, 3) time_diff = (now - price_time).total_seconds()/60 if (not self._last_data_timestamp or price_time > self._last_data_timestamp): self._last_data_timestamp = price_time if 0 <= time_diff < 60: state = price_total self._last_updated = price_time if now.date() == price_time.date(): max_price = max(max_price, price_total) min_price = min(min_price, price_total) num += 1 sum_price += price_total self._state = state self._device_state_attributes['max_price'] = max_price self._device_state_attributes['avg_price'] = round(sum_price / num, 3) self._device_state_attributes['min_price'] = min_price return state is not None class TibberSensorRT(Entity): """Representation of an Tibber sensor for real time consumption.""" def __init__(self, tibber_home): """Initialize the sensor.""" self._tibber_home = tibber_home self._state = None self._device_state_attributes = {} self._unit_of_measurement = 'W' nickname = tibber_home.info['viewer']['home']['appNickname'] self._name = 'Real time consumption {}'.format(nickname) async def async_added_to_hass(self): """Start unavailability tracking.""" await self._tibber_home.rt_subscribe(self.hass.loop, self._async_callback) async def _async_callback(self, payload): """Handle received data.""" errors = payload.get('errors') if errors: _LOGGER.error(errors[0]) return data = payload.get('data') if data is None: return live_measurement = data.get('liveMeasurement') if live_measurement is None: return self._state = live_measurement.pop('power', None) for key, value in live_measurement.items(): if value is None: continue self._device_state_attributes[key] = value self.async_schedule_update_ha_state() @property def device_state_attributes(self): """Return the state attributes.""" return self._device_state_attributes @property def available(self): """Return True if entity is available.""" return self._tibber_home.rt_subscription_running @property def name(self): """Return the name of the sensor.""" return self._name @property def should_poll(self): """Return the polling state.""" return False @property def state(self): """Return the state of the device.""" return self._state @property def icon(self): """Return the icon to use in the frontend.""" return ICON_RT @property def unit_of_measurement(self): """Return the unit of measurement of this entity.""" return self._unit_of_measurement @property def unique_id(self): """Return a unique ID.""" home = self._tibber_home.info['viewer']['home'] _id = home['meteringPointData']['consumptionEan'] return'{}_rt_consumption'.format(_id)