New time period virtual sensor

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()

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':