Fix time_date sensor to update at predictable intervals (#7644)
* Fix time_date sensor to update at predictable intervals * Delete automations.yaml
This commit is contained in:
parent
e3307fb1c2
commit
54c45f80c1
2 changed files with 133 additions and 6 deletions
|
@ -10,11 +10,13 @@ import logging
|
||||||
|
|
||||||
import voluptuous as vol
|
import voluptuous as vol
|
||||||
|
|
||||||
|
from homeassistant.core import callback
|
||||||
from homeassistant.components.sensor import PLATFORM_SCHEMA
|
from homeassistant.components.sensor import PLATFORM_SCHEMA
|
||||||
from homeassistant.const import CONF_DISPLAY_OPTIONS
|
from homeassistant.const import CONF_DISPLAY_OPTIONS
|
||||||
from homeassistant.helpers.entity import Entity
|
from homeassistant.helpers.entity import Entity
|
||||||
import homeassistant.helpers.config_validation as cv
|
import homeassistant.helpers.config_validation as cv
|
||||||
import homeassistant.util.dt as dt_util
|
import homeassistant.util.dt as dt_util
|
||||||
|
from homeassistant.helpers.event import async_track_point_in_utc_time
|
||||||
|
|
||||||
_LOGGER = logging.getLogger(__name__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
@ -44,7 +46,10 @@ def async_setup_platform(hass, config, async_add_devices, discovery_info=None):
|
||||||
|
|
||||||
devices = []
|
devices = []
|
||||||
for variable in config[CONF_DISPLAY_OPTIONS]:
|
for variable in config[CONF_DISPLAY_OPTIONS]:
|
||||||
devices.append(TimeDateSensor(variable))
|
device = TimeDateSensor(hass, variable)
|
||||||
|
async_track_point_in_utc_time(
|
||||||
|
hass, device.point_in_time_listener, device.get_next_interval())
|
||||||
|
devices.append(device)
|
||||||
|
|
||||||
async_add_devices(devices, True)
|
async_add_devices(devices, True)
|
||||||
return True
|
return True
|
||||||
|
@ -53,11 +58,14 @@ def async_setup_platform(hass, config, async_add_devices, discovery_info=None):
|
||||||
class TimeDateSensor(Entity):
|
class TimeDateSensor(Entity):
|
||||||
"""Implementation of a Time and Date sensor."""
|
"""Implementation of a Time and Date sensor."""
|
||||||
|
|
||||||
def __init__(self, option_type):
|
def __init__(self, hass, option_type):
|
||||||
"""Initialize the sensor."""
|
"""Initialize the sensor."""
|
||||||
self._name = OPTION_TYPES[option_type]
|
self._name = OPTION_TYPES[option_type]
|
||||||
self.type = option_type
|
self.type = option_type
|
||||||
self._state = None
|
self._state = None
|
||||||
|
self.hass = hass
|
||||||
|
|
||||||
|
self._update_internal_state(dt_util.utcnow())
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def name(self):
|
def name(self):
|
||||||
|
@ -79,10 +87,22 @@ class TimeDateSensor(Entity):
|
||||||
else:
|
else:
|
||||||
return 'mdi:clock'
|
return 'mdi:clock'
|
||||||
|
|
||||||
@asyncio.coroutine
|
def get_next_interval(self, now=None):
|
||||||
def async_update(self):
|
"""Compute next time an update should occur."""
|
||||||
"""Get the latest data and updates the states."""
|
if now is None:
|
||||||
time_date = dt_util.utcnow()
|
now = dt_util.utcnow()
|
||||||
|
if self.type == 'date':
|
||||||
|
now = dt_util.start_of_local_day(now)
|
||||||
|
return now + timedelta(seconds=86400)
|
||||||
|
elif self.type == 'beat':
|
||||||
|
interval = 86.4
|
||||||
|
else:
|
||||||
|
interval = 60
|
||||||
|
timestamp = int(dt_util.as_timestamp(now))
|
||||||
|
delta = interval - (timestamp % interval)
|
||||||
|
return now + timedelta(seconds=delta)
|
||||||
|
|
||||||
|
def _update_internal_state(self, time_date):
|
||||||
time = dt_util.as_local(time_date).strftime(TIME_STR_FORMAT)
|
time = dt_util.as_local(time_date).strftime(TIME_STR_FORMAT)
|
||||||
time_utc = time_date.strftime(TIME_STR_FORMAT)
|
time_utc = time_date.strftime(TIME_STR_FORMAT)
|
||||||
date = dt_util.as_local(time_date).date().isoformat()
|
date = dt_util.as_local(time_date).date().isoformat()
|
||||||
|
@ -106,3 +126,11 @@ class TimeDateSensor(Entity):
|
||||||
self._state = time_utc
|
self._state = time_utc
|
||||||
elif self.type == 'beat':
|
elif self.type == 'beat':
|
||||||
self._state = '@{0:03d}'.format(beat)
|
self._state = '@{0:03d}'.format(beat)
|
||||||
|
|
||||||
|
@callback
|
||||||
|
def point_in_time_listener(self, time_date):
|
||||||
|
"""Get the latest data and update state."""
|
||||||
|
self._update_internal_state(time_date)
|
||||||
|
self.hass.async_add_job(self.async_update_ha_state())
|
||||||
|
async_track_point_in_utc_time(
|
||||||
|
self.hass, self.point_in_time_listener, self.get_next_interval())
|
||||||
|
|
99
tests/components/sensor/test_time_date.py
Normal file
99
tests/components/sensor/test_time_date.py
Normal file
|
@ -0,0 +1,99 @@
|
||||||
|
"""The tests for Kira sensor platform."""
|
||||||
|
import unittest
|
||||||
|
|
||||||
|
from homeassistant.components.sensor import time_date as time_date
|
||||||
|
import homeassistant.util.dt as dt_util
|
||||||
|
|
||||||
|
from tests.common import get_test_home_assistant
|
||||||
|
|
||||||
|
|
||||||
|
class TestTimeDateSensor(unittest.TestCase):
|
||||||
|
"""Tests the Kira Sensor platform."""
|
||||||
|
|
||||||
|
# pylint: disable=invalid-name
|
||||||
|
DEVICES = []
|
||||||
|
|
||||||
|
def add_devices(self, devices):
|
||||||
|
"""Mock add devices."""
|
||||||
|
for device in devices:
|
||||||
|
self.DEVICES.append(device)
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
"""Initialize values for this testcase class."""
|
||||||
|
self.hass = get_test_home_assistant()
|
||||||
|
self.DEFAULT_TIME_ZONE = dt_util.DEFAULT_TIME_ZONE
|
||||||
|
|
||||||
|
def tearDown(self):
|
||||||
|
"""Stop everything that was started."""
|
||||||
|
dt_util.set_default_time_zone(self.DEFAULT_TIME_ZONE)
|
||||||
|
self.hass.stop()
|
||||||
|
|
||||||
|
# pylint: disable=protected-access
|
||||||
|
def test_intervals(self):
|
||||||
|
"""Test timing intervals of sensors."""
|
||||||
|
device = time_date.TimeDateSensor(self.hass, 'time')
|
||||||
|
now = dt_util.utc_from_timestamp(45)
|
||||||
|
next_time = device.get_next_interval(now)
|
||||||
|
assert next_time == dt_util.utc_from_timestamp(60)
|
||||||
|
|
||||||
|
device = time_date.TimeDateSensor(self.hass, 'date')
|
||||||
|
now = dt_util.utc_from_timestamp(12345)
|
||||||
|
next_time = device.get_next_interval(now)
|
||||||
|
assert next_time == dt_util.utc_from_timestamp(86400)
|
||||||
|
|
||||||
|
device = time_date.TimeDateSensor(self.hass, 'beat')
|
||||||
|
now = dt_util.utc_from_timestamp(29)
|
||||||
|
next_time = device.get_next_interval(now)
|
||||||
|
assert next_time == dt_util.utc_from_timestamp(86.4)
|
||||||
|
|
||||||
|
device = time_date.TimeDateSensor(self.hass, 'date_time')
|
||||||
|
now = dt_util.utc_from_timestamp(1495068899)
|
||||||
|
next_time = device.get_next_interval(now)
|
||||||
|
assert next_time == dt_util.utc_from_timestamp(1495068900)
|
||||||
|
|
||||||
|
now = dt_util.utcnow()
|
||||||
|
device = time_date.TimeDateSensor(self.hass, 'time_date')
|
||||||
|
next_time = device.get_next_interval()
|
||||||
|
assert next_time > now
|
||||||
|
|
||||||
|
def test_states(self):
|
||||||
|
"""Test states of sensors."""
|
||||||
|
now = dt_util.utc_from_timestamp(1495068856)
|
||||||
|
device = time_date.TimeDateSensor(self.hass, 'time')
|
||||||
|
device._update_internal_state(now)
|
||||||
|
assert device.state == "00:54"
|
||||||
|
|
||||||
|
device = time_date.TimeDateSensor(self.hass, 'date')
|
||||||
|
device._update_internal_state(now)
|
||||||
|
assert device.state == "2017-05-18"
|
||||||
|
|
||||||
|
device = time_date.TimeDateSensor(self.hass, 'time_utc')
|
||||||
|
device._update_internal_state(now)
|
||||||
|
assert device.state == "00:54"
|
||||||
|
|
||||||
|
device = time_date.TimeDateSensor(self.hass, 'beat')
|
||||||
|
device._update_internal_state(now)
|
||||||
|
assert device.state == "@079"
|
||||||
|
|
||||||
|
# pylint: disable=no-member
|
||||||
|
def test_timezone_intervals(self):
|
||||||
|
"""Test date sensor behavior in a timezone besides UTC."""
|
||||||
|
new_tz = dt_util.get_time_zone('America/New_York')
|
||||||
|
assert new_tz is not None
|
||||||
|
dt_util.set_default_time_zone(new_tz)
|
||||||
|
|
||||||
|
device = time_date.TimeDateSensor(self.hass, 'date')
|
||||||
|
now = dt_util.utc_from_timestamp(50000)
|
||||||
|
next_time = device.get_next_interval(now)
|
||||||
|
# start of local day in EST was 18000.0
|
||||||
|
# so the second day was 18000 + 86400
|
||||||
|
assert next_time.timestamp() == 104400
|
||||||
|
|
||||||
|
def test_icons(self):
|
||||||
|
"""Test attributes of sensors."""
|
||||||
|
device = time_date.TimeDateSensor(self.hass, 'time')
|
||||||
|
assert device.icon == "mdi:clock"
|
||||||
|
device = time_date.TimeDateSensor(self.hass, 'date')
|
||||||
|
assert device.icon == "mdi:calendar"
|
||||||
|
device = time_date.TimeDateSensor(self.hass, 'date_time')
|
||||||
|
assert device.icon == "mdi:calendar-clock"
|
Loading…
Add table
Reference in a new issue