Filter out forced updates in live logbook when the state has not changed (#73335)

homeassistant/components/logbook/helpers.py

@@ -189,9 +189,10 @@ def async_subscribe_events(
     def _forward_state_events_filtered(event: Event) -> None:
         if event.data.get("old_state") is None or event.data.get("new_state") is None:
             return
-        state: State = event.data["new_state"]
+        new_state: State = event.data["new_state"]
-        if _is_state_filtered(ent_reg, state) or (
+        old_state: State = event.data["old_state"]
-            entities_filter and not entities_filter(state.entity_id)
+        if _is_state_filtered(ent_reg, new_state, old_state) or (
+            entities_filter and not entities_filter(new_state.entity_id)
         ):
             return
         target(event)
@@ -229,17 +230,20 @@ def is_sensor_continuous(ent_reg: er.EntityRegistry, entity_id: str) -> bool:
     )
 
 
-def _is_state_filtered(ent_reg: er.EntityRegistry, state: State) -> bool:
+def _is_state_filtered(
+    ent_reg: er.EntityRegistry, new_state: State, old_state: State
+) -> bool:
     """Check if the logbook should filter a state.
 
     Used when we are in live mode to ensure
     we only get significant changes (state.last_changed != state.last_updated)
     """
     return bool(
-        split_entity_id(state.entity_id)[0] in ALWAYS_CONTINUOUS_DOMAINS
+        new_state.state == old_state.state
-        or state.last_changed != state.last_updated
+        or split_entity_id(new_state.entity_id)[0] in ALWAYS_CONTINUOUS_DOMAINS
-        or ATTR_UNIT_OF_MEASUREMENT in state.attributes
+        or new_state.last_changed != new_state.last_updated
-        or is_sensor_continuous(ent_reg, state.entity_id)
+        or ATTR_UNIT_OF_MEASUREMENT in new_state.attributes
+        or is_sensor_continuous(ent_reg, new_state.entity_id)
     )
 
 

tests/components/logbook/test_websocket_api.py

@@ -2404,3 +2404,117 @@ async def test_subscribe_entities_some_have_uom_multiple(
 
     # Check our listener got unsubscribed
     assert sum(hass.bus.async_listeners().values()) == init_count
+
+
+@patch("homeassistant.components.logbook.websocket_api.EVENT_COALESCE_TIME", 0)
+async def test_logbook_stream_ignores_forced_updates(
+    hass, recorder_mock, hass_ws_client
+):
+    """Test logbook live stream ignores forced updates."""
+    now = dt_util.utcnow()
+    await asyncio.gather(
+        *[
+            async_setup_component(hass, comp, {})
+            for comp in ("homeassistant", "logbook", "automation", "script")
+        ]
+    )
+
+    await hass.async_block_till_done()
+    init_count = sum(hass.bus.async_listeners().values())
+
+    hass.states.async_set("binary_sensor.is_light", STATE_ON)
+    hass.states.async_set("binary_sensor.is_light", STATE_OFF)
+    state: State = hass.states.get("binary_sensor.is_light")
+    await hass.async_block_till_done()
+
+    await async_wait_recording_done(hass)
+    websocket_client = await hass_ws_client()
+    await websocket_client.send_json(
+        {"id": 7, "type": "logbook/event_stream", "start_time": now.isoformat()}
+    )
+
+    msg = await asyncio.wait_for(websocket_client.receive_json(), 2)
+    assert msg["id"] == 7
+    assert msg["type"] == TYPE_RESULT
+    assert msg["success"]
+
+    msg = await asyncio.wait_for(websocket_client.receive_json(), 2)
+    assert msg["id"] == 7
+    assert msg["type"] == "event"
+    assert msg["event"]["events"] == [
+        {
+            "entity_id": "binary_sensor.is_light",
+            "state": "off",
+            "when": state.last_updated.timestamp(),
+        }
+    ]
+    assert msg["event"]["start_time"] == now.timestamp()
+    assert msg["event"]["end_time"] > msg["event"]["start_time"]
+    assert msg["event"]["partial"] is True
+
+    await hass.async_block_till_done()
+
+    msg = await asyncio.wait_for(websocket_client.receive_json(), 2)
+    assert msg["id"] == 7
+    assert msg["type"] == "event"
+    assert "partial" not in msg["event"]["events"]
+    assert msg["event"]["events"] == []
+
+    hass.states.async_set("binary_sensor.is_light", STATE_ON)
+    hass.states.async_set("binary_sensor.is_light", STATE_OFF)
+
+    msg = await asyncio.wait_for(websocket_client.receive_json(), 2)
+    assert msg["id"] == 7
+    assert msg["type"] == "event"
+    assert "partial" not in msg["event"]["events"]
+    assert msg["event"]["events"] == [
+        {
+            "entity_id": "binary_sensor.is_light",
+            "state": STATE_ON,
+            "when": ANY,
+        },
+        {
+            "entity_id": "binary_sensor.is_light",
+            "state": STATE_OFF,
+            "when": ANY,
+        },
+    ]
+
+    # Now we force an update to make sure we ignore
+    # forced updates when the state has not actually changed
+
+    hass.states.async_set("binary_sensor.is_light", STATE_ON)
+    for _ in range(3):
+        hass.states.async_set("binary_sensor.is_light", STATE_OFF, force_update=True)
+
+    msg = await asyncio.wait_for(websocket_client.receive_json(), 2)
+    assert msg["id"] == 7
+    assert msg["type"] == "event"
+    assert "partial" not in msg["event"]["events"]
+    assert msg["event"]["events"] == [
+        {
+            "entity_id": "binary_sensor.is_light",
+            "state": STATE_ON,
+            "when": ANY,
+        },
+        # We should only get the first one and ignore
+        # the other forced updates since the state
+        # has not actually changed
+        {
+            "entity_id": "binary_sensor.is_light",
+            "state": STATE_OFF,
+            "when": ANY,
+        },
+    ]
+
+    await websocket_client.send_json(
+        {"id": 8, "type": "unsubscribe_events", "subscription": 7}
+    )
+    msg = await asyncio.wait_for(websocket_client.receive_json(), 2)
+
+    assert msg["id"] == 8
+    assert msg["type"] == TYPE_RESULT
+    assert msg["success"]
+
+    # Check our listener got unsubscribed
+    assert sum(hass.bus.async_listeners().values()) == init_count