Implement time triggers with offset for timestamp sensors (#120858)

* Implement time triggers with offset for timestamp sensors

* Fix bad change

* Add testcase for multiple conf_at with offsets

* Fix fixture rename

* Fix testcase - if no offset provided, it should be just the string of the entity id

* Get test to pass

* Simplify code

* Update the messaging and make the offset optional allowing specifying only the entity_id

* Move state tracking one level up

* Implement requesteed changes

homeassistant/components/homeassistant/triggers/time.py

@@ -1,7 +1,9 @@
 """Offer time listening automation rules."""
 
-from datetime import datetime
+from collections.abc import Callable
+from datetime import datetime, timedelta
 from functools import partial
+from typing import NamedTuple
 
 import voluptuous as vol
 
@@ -9,6 +11,8 @@ from homeassistant.components import sensor
 from homeassistant.const import (
     ATTR_DEVICE_CLASS,
     CONF_AT,
+    CONF_ENTITY_ID,
+    CONF_OFFSET,
     CONF_PLATFORM,
     STATE_UNAVAILABLE,
     STATE_UNKNOWN,
@@ -32,11 +36,22 @@ from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo
 from homeassistant.helpers.typing import ConfigType
 import homeassistant.util.dt as dt_util
 
+_TIME_TRIGGER_ENTITY = vol.All(str, cv.entity_domain(["input_datetime", "sensor"]))
+
+_TIME_TRIGGER_ENTITY_WITH_OFFSET = vol.Schema(
+    {
+        vol.Required(CONF_ENTITY_ID): cv.entity_domain(["sensor"]),
+        vol.Optional(CONF_OFFSET): cv.time_period,
+    }
+)
+
 _TIME_TRIGGER_SCHEMA = vol.Any(
     cv.time,
-    vol.All(str, cv.entity_domain(["input_datetime", "sensor"])),
+    _TIME_TRIGGER_ENTITY,
+    _TIME_TRIGGER_ENTITY_WITH_OFFSET,
     msg=(
-        "Expected HH:MM, HH:MM:SS or Entity ID with domain 'input_datetime' or 'sensor'"
+        "Expected HH:MM, HH:MM:SS, an Entity ID with domain 'input_datetime' or "
+        "'sensor', or a combination of a timestamp sensor entity and an offset."
     ),
 )
 
@@ -48,6 +63,13 @@ TRIGGER_SCHEMA = cv.TRIGGER_BASE_SCHEMA.extend(
 )
 
 
+class TrackEntity(NamedTuple):
+    """Represents a tracking entity for a time trigger."""
+
+    entity_id: str
+    callback: Callable
+
+
 async def async_attach_trigger(
     hass: HomeAssistant,
     config: ConfigType,
@@ -56,7 +78,7 @@ async def async_attach_trigger(
 ) -> CALLBACK_TYPE:
     """Listen for state changes based on configuration."""
     trigger_data = trigger_info["trigger_data"]
-    entities: dict[str, CALLBACK_TYPE] = {}
+    entities: dict[tuple[str, timedelta], CALLBACK_TYPE] = {}
     removes: list[CALLBACK_TYPE] = []
     job = HassJob(action, f"time trigger {trigger_info}")
 
@@ -79,15 +101,21 @@ async def async_attach_trigger(
         )
 
     @callback
-    def update_entity_trigger_event(event: Event[EventStateChangedData]) -> None:
+    def update_entity_trigger_event(
+        event: Event[EventStateChangedData], offset: timedelta = timedelta(0)
+    ) -> None:
         """update_entity_trigger from the event."""
-        return update_entity_trigger(event.data["entity_id"], event.data["new_state"])
+        return update_entity_trigger(
+            event.data["entity_id"], event.data["new_state"], offset
+        )
 
     @callback
-    def update_entity_trigger(entity_id: str, new_state: State | None = None) -> None:
+    def update_entity_trigger(
+        entity_id: str, new_state: State | None = None, offset: timedelta = timedelta(0)
+    ) -> None:
         """Update the entity trigger for the entity_id."""
         # If a listener was already set up for entity, remove it.
-        if remove := entities.pop(entity_id, None):
+        if remove := entities.pop((entity_id, offset), None):
             remove()
             remove = None
 
@@ -153,6 +181,9 @@ async def async_attach_trigger(
         ):
             trigger_dt = dt_util.parse_datetime(new_state.state)
 
+            if trigger_dt is not None:
+                trigger_dt += offset
+
             if trigger_dt is not None and trigger_dt > dt_util.utcnow():
                 remove = async_track_point_in_time(
                     hass,
@@ -166,15 +197,27 @@ async def async_attach_trigger(
 
         # Was a listener set up?
         if remove:
-            entities[entity_id] = remove
+            entities[(entity_id, offset)] = remove
 
-    to_track: list[str] = []
+    to_track: list[TrackEntity] = []
 
     for at_time in config[CONF_AT]:
         if isinstance(at_time, str):
             # entity
-            to_track.append(at_time)
             update_entity_trigger(at_time, new_state=hass.states.get(at_time))
+            to_track.append(TrackEntity(at_time, update_entity_trigger_event))
+        elif isinstance(at_time, dict) and CONF_OFFSET in at_time:
+            # entity with offset
+            entity_id: str = at_time.get(CONF_ENTITY_ID, "")
+            offset: timedelta = at_time.get(CONF_OFFSET, timedelta(0))
+            update_entity_trigger(
+                entity_id, new_state=hass.states.get(entity_id), offset=offset
+            )
+            to_track.append(
+                TrackEntity(
+                    entity_id, partial(update_entity_trigger_event, offset=offset)
+                )
+            )
         else:
             # datetime.time
             removes.append(
@@ -187,9 +230,10 @@ async def async_attach_trigger(
                 )
             )
 
-    # Track state changes of any entities.
+    # Besides time, we also track state changes of requested entities.
-    removes.append(
+    removes.extend(
-        async_track_state_change_event(hass, to_track, update_entity_trigger_event)
+        (async_track_state_change_event(hass, entry.entity_id, entry.callback))
+        for entry in to_track
     )
 
     @callback

tests/components/homeassistant/triggers/test_time.py

@@ -156,17 +156,40 @@ async def test_if_fires_using_at_input_datetime(
     )
 
 
+@pytest.mark.parametrize(
+    ("conf_at", "trigger_deltas"),
+    [
+        (["5:00:00", "6:00:00"], [timedelta(0), timedelta(hours=1)]),
+        (
+            [
+                "5:00:05",
+                {"entity_id": "sensor.next_alarm", "offset": "00:00:10"},
+                "sensor.next_alarm",
+            ],
+            [timedelta(seconds=5), timedelta(seconds=10), timedelta(0)],
+        ),
+    ],
+)
 async def test_if_fires_using_multiple_at(
     hass: HomeAssistant,
     freezer: FrozenDateTimeFactory,
     service_calls: list[ServiceCall],
+    conf_at: list[str | dict[str, int | str]],
+    trigger_deltas: list[timedelta],
 ) -> None:
-    """Test for firing at."""
+    """Test for firing at multiple trigger times."""
 
     now = dt_util.now()
 
-    trigger_dt = now.replace(hour=5, minute=0, second=0, microsecond=0) + timedelta(2)
+    start_dt = now.replace(hour=5, minute=0, second=0, microsecond=0) + timedelta(2)
-    time_that_will_not_match_right_away = trigger_dt - timedelta(minutes=1)
+
+    hass.states.async_set(
+        "sensor.next_alarm",
+        start_dt.isoformat(),
+        {ATTR_DEVICE_CLASS: SensorDeviceClass.TIMESTAMP},
+    )
+
+    time_that_will_not_match_right_away = start_dt - timedelta(minutes=1)
 
     freezer.move_to(dt_util.as_utc(time_that_will_not_match_right_away))
     assert await async_setup_component(
@@ -174,7 +197,7 @@ async def test_if_fires_using_multiple_at(
         automation.DOMAIN,
         {
             automation.DOMAIN: {
-                "trigger": {"platform": "time", "at": ["5:00:00", "6:00:00"]},
+                "trigger": {"platform": "time", "at": conf_at},
                 "action": {
                     "service": "test.automation",
                     "data_template": {
@@ -186,17 +209,14 @@ async def test_if_fires_using_multiple_at(
     )
     await hass.async_block_till_done()
 
-    async_fire_time_changed(hass, trigger_dt + timedelta(seconds=1))
+    for count, delta in enumerate(sorted(trigger_deltas)):
-    await hass.async_block_till_done()
+        async_fire_time_changed(hass, start_dt + delta + timedelta(seconds=1))
+        await hass.async_block_till_done()
 
-    assert len(service_calls) == 1
+        assert len(service_calls) == count + 1
-    assert service_calls[0].data["some"] == "time - 5"
+        assert (
-
+            service_calls[count].data["some"] == f"time - {5 + (delta.seconds // 3600)}"
-    async_fire_time_changed(hass, trigger_dt + timedelta(hours=1, seconds=1))
+        )
-    await hass.async_block_till_done()
-
-    assert len(service_calls) == 2
-    assert service_calls[1].data["some"] == "time - 6"
 
 
 async def test_if_not_fires_using_wrong_at(
@@ -518,12 +538,99 @@ async def test_if_fires_using_at_sensor(
     assert len(service_calls) == 2
 
 
+@pytest.mark.parametrize(
+    ("offset", "delta"),
+    [
+        ("00:00:10", timedelta(seconds=10)),
+        ("-00:00:10", timedelta(seconds=-10)),
+        ({"minutes": 5}, timedelta(minutes=5)),
+    ],
+)
+async def test_if_fires_using_at_sensor_with_offset(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    freezer: FrozenDateTimeFactory,
+    offset: str | dict[str, int],
+    delta: timedelta,
+) -> None:
+    """Test for firing at sensor time."""
+    now = dt_util.now()
+
+    start_dt = now.replace(hour=5, minute=0, second=0, microsecond=0) + timedelta(2)
+    trigger_dt = start_dt + delta
+
+    hass.states.async_set(
+        "sensor.next_alarm",
+        start_dt.isoformat(),
+        {ATTR_DEVICE_CLASS: SensorDeviceClass.TIMESTAMP},
+    )
+
+    time_that_will_not_match_right_away = trigger_dt - timedelta(minutes=1)
+
+    some_data = "{{ trigger.platform }}-{{ trigger.now.day }}-{{ trigger.now.hour }}-{{ trigger.now.minute }}-{{ trigger.now.second }}-{{trigger.entity_id}}"
+
+    freezer.move_to(dt_util.as_utc(time_that_will_not_match_right_away))
+    assert await async_setup_component(
+        hass,
+        automation.DOMAIN,
+        {
+            automation.DOMAIN: {
+                "trigger": {
+                    "platform": "time",
+                    "at": {
+                        "entity_id": "sensor.next_alarm",
+                        "offset": offset,
+                    },
+                },
+                "action": {
+                    "service": "test.automation",
+                    "data_template": {"some": some_data},
+                },
+            }
+        },
+    )
+    await hass.async_block_till_done()
+
+    async_fire_time_changed(hass, trigger_dt + timedelta(seconds=1))
+    await hass.async_block_till_done()
+
+    assert len(service_calls) == 1
+    assert (
+        service_calls[0].data["some"]
+        == f"time-{trigger_dt.day}-{trigger_dt.hour}-{trigger_dt.minute}-{trigger_dt.second}-sensor.next_alarm"
+    )
+
+    start_dt += timedelta(days=1, hours=1)
+    trigger_dt += timedelta(days=1, hours=1)
+
+    hass.states.async_set(
+        "sensor.next_alarm",
+        start_dt.isoformat(),
+        {ATTR_DEVICE_CLASS: SensorDeviceClass.TIMESTAMP},
+    )
+    await hass.async_block_till_done()
+
+    async_fire_time_changed(hass, trigger_dt + timedelta(seconds=1))
+    await hass.async_block_till_done()
+
+    assert len(service_calls) == 2
+    assert (
+        service_calls[1].data["some"]
+        == f"time-{trigger_dt.day}-{trigger_dt.hour}-{trigger_dt.minute}-{trigger_dt.second}-sensor.next_alarm"
+    )
+
+
 @pytest.mark.parametrize(
     "conf",
     [
         {"platform": "time", "at": "input_datetime.bla"},
         {"platform": "time", "at": "sensor.bla"},
         {"platform": "time", "at": "12:34"},
+        {"platform": "time", "at": {"entity_id": "sensor.bla", "offset": "-00:01"}},
+        {
+            "platform": "time",
+            "at": [{"entity_id": "sensor.bla", "offset": "-01:00:00"}],
+        },
     ],
 )
 def test_schema_valid(conf) -> None:
@@ -537,6 +644,11 @@ def test_schema_valid(conf) -> None:
         {"platform": "time", "at": "binary_sensor.bla"},
         {"platform": "time", "at": 745},
         {"platform": "time", "at": "25:00"},
+        {
+            "platform": "time",
+            "at": {"entity_id": "input_datetime.bla", "offset": "0:10"},
+        },
+        {"platform": "time", "at": {"entity_id": "13:00:00", "offset": "0:10"}},
     ],
 )
 def test_schema_invalid(conf) -> None: