From 1a148d4e3fd4ac2e343c46913f5504405a6029d8 Mon Sep 17 00:00:00 2001 From: Pierre Gironde Date: Sun, 25 Oct 2026 20:39:52 +0100 Subject: [PATCH] New time period virtual sensor --- unit_tests/test_time_period.py | 101 +++++++++++++++++++++++++++++++++ virtualsensors.py | 87 ++++++++++++++++++++++++++++ 2 files changed, 188 insertions(+) create mode 100644 unit_tests/test_time_period.py diff --git a/unit_tests/test_time_period.py b/unit_tests/test_time_period.py new file mode 100644 index 0000000..0c4a8fb --- /dev/null +++ b/unit_tests/test_time_period.py @@ -0,0 +1,101 @@ +"""Time-period boundaries, source updates and timer lifecycle without AppDaemon.""" +import ast +import datetime +import re +import unittest +from pathlib import Path +from unittest.mock import Mock + +source = Path(__file__).resolve().parents[1] / 'virtualsensors.py' +nodes = [node for node in ast.parse(source.read_text()).body + if isinstance(node, ast.ClassDef) and node.name in ('VirtualSensorBase', 'TimePeriod', 'VirtualSensors')] +from kwargsparser import kwargsParser +namespace = {'datetime': datetime, 're': re, 'kwargsParser': kwargsParser, 'ParsingException': ValueError} +exec(compile(ast.Module(body=nodes, type_ignores=[]), str(source), 'exec'), namespace) +TimePeriod = namespace['TimePeriod'] + + +class TimePeriodTests(unittest.TestCase): + def make_sensor(self, start='08:00:00', end='18:00:00', now='2026-10-25 12:00:00'): + self.states = {'input_datetime.start': start, 'input_datetime.end': end} + self.api = Mock() + self.api.datetime.return_value = datetime.datetime.fromisoformat(now) + self.api.get_state.side_effect = self.states.get + self.api.entity_exists.return_value = True + self.api.get_entity.return_value.get_state.return_value = 'unknown' + self.sensor = TimePeriod(self.api, Mock(), virtual_sensor_name='period', + yaml_block={'start': 'input_datetime.start', 'end': 'input_datetime.end'}) + self.sensor._initialize() + return self.sensor + + def assert_state(self, expected): + self.assertEqual(self.sensor.entity.set_state.call_args.kwargs['state'], expected) + + def test_factory_registration(self): + self.make_sensor() + factory = namespace['VirtualSensors'](ad_api=self.api, logger_interface=Mock(), + yaml_block={'sensors': {'time_period.example': { + 'start': 'input_datetime.start', 'end': 'input_datetime.end'}}}) + self.assertIsInstance(factory.virtual_sensors['binary_sensor.example'], TimePeriod) + + def test_day_boundaries_and_timer(self): + self.make_sensor(now='2026-10-25 07:59:50') + self.assert_state('off') + self.assertEqual(self.api.run_in.call_args.args[1], 10) + for hour, expected in [('08:00:00', 'on'), ('17:59:59', 'on'), ('18:00:00', 'off')]: + self.api.datetime.return_value = datetime.datetime.fromisoformat('2026-10-25 ' + hour) + self.sensor.on_time_elapsed({'generation': self.sensor._timer_generation}) + self.assert_state(expected) + + def test_overnight_and_equal_boundaries(self): + for hour, expected in [('23:00:00', 'on'), ('02:00:00', 'on'), ('06:00:00', 'off'), ('12:00:00', 'off')]: + self.make_sensor('22:00:00', '06:00:00', '2026-10-25 ' + hour) + self.assert_state(expected) + self.make_sensor('08:00:00', '08:00:00') + self.assert_state('off') + + def test_source_change_cancels_timer_and_ignores_stale_callback(self): + self.make_sensor() + self.assert_state('on') + previous_generation = self.sensor._timer_generation + previous_handle = self.sensor.cb_handle + self.states['input_datetime.end'] = '10:00:00' + self.sensor.on_input_change('input_datetime.end', 'state', '18:00:00', '10:00:00', {}) + self.assert_state('off') + self.api.cancel_timer.assert_called_with(previous_handle, silent=True) + count = self.api.run_in.call_count + self.sensor.on_time_elapsed({'generation': previous_generation}) + self.assertEqual(self.api.run_in.call_count, count) + self.assertEqual(self.api.listen_state.call_count, 2) + + def test_unavailable_and_recovery(self): + for invalid in [None, 'unknown', 'unavailable', '25:00:00', '2026-10-25']: + self.make_sensor(start=invalid) + self.assert_state('unavailable') + self.states['input_datetime.start'] = '08:00:00' + self.sensor.update_period() + self.assert_state('on') + + def test_dated_periods(self): + for start, end, expected in [ + ('2026-10-24 22:00:00', '2026-10-26 06:00:00', 'on'), + ('2026-10-24 22:00:00', '2026-10-25 12:00:00', 'off'), + ('2026-10-26 22:00:00', '2026-10-24 06:00:00', 'off'), + ('2026-10-24', '2026-10-26', 'on'), + ]: + self.make_sensor(start, end) + self.assert_state(expected) + + def test_dependencies_constants_and_validation(self): + self.make_sensor() + self.assertEqual(self.sensor.sensor_name, 'binary_sensor.period') + self.sensor.constants['begin'] = 'input_datetime.start' + self.sensor.yaml_block['start'] = 'begin' + self.assertEqual(self.sensor.generate_dependencies_list(), list(self.states)) + self.sensor.yaml_block['end'] = 'sensor.invalid' + with self.assertRaises(ValueError): + self.sensor.generate_dependencies_list() + + +if __name__ == '__main__': + unittest.main() diff --git a/virtualsensors.py b/virtualsensors.py index cbb5421..e967e3f 100644 --- a/virtualsensors.py +++ b/virtualsensors.py @@ -4,6 +4,7 @@ import smartvalue as SmartValue from kwargsparser import kwargsParser from expressionparser import ParsingException import time +import datetime import re from logger_interface import LoggerInterface from ad_toolbox.eventhandler import EventHandler @@ -618,6 +619,90 @@ class RetainCondition(VirtualSensorBase): if self.output_sensor.get_state() != 'off': self.output_sensor.set_state(state = 'off') +class TimePeriod(VirtualSensorBase): + """Active in [start, end), using two input_datetime entities. + + Time-only inputs repeat daily and may cross midnight. Dated inputs + describe a single interval. Equal boundaries describe an empty interval. + """ + + def generate_sensor_name(self, virtual_sensor_name): + return f"binary_sensor.{virtual_sensor_name}" + + def generate_dependencies_list(self): + inputs = [] + for key in ('start', 'end'): + value = self.yaml_block.get(key) + value = self.constants.get(value, value) if isinstance(value, str) else value + if not self.is_entity_reference(value) or not value.startswith('input_datetime.'): + raise ValueError(f"{self.sensor_name}: {key} must reference an input_datetime") + inputs.append(value) + return inputs + + def initialize(self): + self.start_entity, self.end_entity = self.generate_dependencies_list() + self.cb_handle = None + self._timer_generation = 0 + for entity in set((self.start_entity, self.end_entity)): + self.ad_api.listen_state(self.on_input_change, entity) + self.update_period() + + @staticmethod + def parse_boundary(value): + for fmt, kind in (('%H:%M:%S', 'time'), + ('%Y-%m-%d %H:%M:%S', 'datetime'), + ('%Y-%m-%d', 'date')): + try: + parsed = datetime.datetime.strptime(value, fmt) + return kind, parsed.time() if kind == 'time' else parsed + except (TypeError, ValueError): + continue + raise ValueError('Unavailable or invalid input_datetime state') + + def on_input_change(self, entity, attribute, old, new, kwargs): + self.update_period() + + def on_time_elapsed(self, kwargs): + if kwargs.get('generation') != self._timer_generation: + return + self.cb_handle = None + self.update_period() + + def update_period(self): + self._timer_generation += 1 + if self.cb_handle is not None: + self.ad_api.cancel_timer(self.cb_handle, silent=True) + self.cb_handle = None + now = self.ad_api.datetime() + delay = 60.0 + try: + start_kind, start = self.parse_boundary(self.ad_api.get_state(self.start_entity)) + end_kind, end = self.parse_boundary(self.ad_api.get_state(self.end_entity)) + if start_kind != end_kind: + raise ValueError('Both inputs must have the same date/time format') + if start_kind == 'time': + current = now.time() + active = (start <= current < end if start <= end + else current >= start or current < end) + boundaries = [datetime.datetime.combine(now.date(), value) + for value in (start, end)] + boundaries = [value if value > now else value + datetime.timedelta(days=1) + for value in boundaries] + else: + active = start <= now < end + boundaries = [value for value in (start, end) if value > now] + state = 'on' if active else 'off' + if boundaries: + delay = min(delay, min((value - now).total_seconds() for value in boundaries)) + except ValueError: + state = 'unavailable' + if self.entity.get_state() != state: + self.entity.set_state(state=state) + # Refresh at boundaries, and at least every minute to follow clock/DST changes. + self.cb_handle = self.ad_api.run_in( + self.on_time_elapsed, delay, generation=self._timer_generation) + + class BinarySensor(VirtualSensorBase): def initialize(self): self.output_sensor = self.ad_api.get_entity(self.sensor_name) @@ -764,6 +849,8 @@ class VirtualSensors(): if splitted_sensor[0] == 'binary_sensor': self.virtual_sensors[f"binary_sensor.{splitted_sensor[1]}"] = BinarySensor(self.ad_api,self.logger_interface,splitted_sensor[1],super_entity_id = super_entity_id,yaml_block = yaml_block['sensors'][sensor],constants = constants,templates_library = templates_library,app_name = app_name) + elif splitted_sensor[0] == 'time_period': + self.virtual_sensors[f"binary_sensor.{splitted_sensor[1]}"] = TimePeriod(self.ad_api,self.logger_interface,splitted_sensor[1],super_entity_id = super_entity_id,yaml_block = yaml_block['sensors'][sensor],constants = constants,templates_library = templates_library,app_name = app_name) elif splitted_sensor[0] == 'sensor': self.virtual_sensors[f"sensor.{splitted_sensor[1]}"] = ValueSensor(self.ad_api,self.logger_interface,splitted_sensor[1],super_entity_id = super_entity_id,yaml_block = yaml_block['sensors'][sensor],constants = constants,templates_library = templates_library,app_name = app_name) elif splitted_sensor[0] == 'continuous_condition':