Add occupancy binary sensor triggers (home-assistant#157631)

Author: emontnemery

homeassistant/components/automation/__init__.py

@@ -124,6 +124,7 @@
 _EXPERIMENTAL_TRIGGER_PLATFORMS = {
     "alarm_control_panel",
     "assist_satellite",
+    "binary_sensor",
     "climate",
     "cover",
     "fan",

homeassistant/components/binary_sensor/icons.json

@@ -174,5 +174,13 @@
         "on": "mdi:window-open"
       }
     }
+  },
+  "triggers": {
+    "occupancy_cleared": {
+      "trigger": "mdi:home-outline"
+    },
+    "occupancy_detected": {
+      "trigger": "mdi:home"
+    }
   }
 }

homeassistant/components/binary_sensor/strings.json

@@ -1,4 +1,8 @@
 {
+  "common": {
+    "trigger_behavior_description_presence": "The behavior of the targeted presence sensors to trigger on.",
+    "trigger_behavior_name": "Behavior"
+  },
   "device_automation": {
     "condition_type": {
       "is_bat_low": "{entity_name} battery is low",
@@ -317,5 +321,36 @@
       }
     }
   },
-  "title": "Binary sensor"
+  "selector": {
+    "trigger_behavior": {
+      "options": {
+        "any": "Any",
+        "first": "First",
+        "last": "Last"
+      }
+    }
+  },
+  "title": "Binary sensor",
+  "triggers": {
+    "occupancy_cleared": {
+      "description": "Triggers after one or more occupancy sensors stop detecting occupancy.",
+      "fields": {
+        "behavior": {
+          "description": "[%key:component::binary_sensor::common::trigger_behavior_description_presence%]",
+          "name": "[%key:component::binary_sensor::common::trigger_behavior_name%]"
+        }
+      },
+      "name": "Occupancy cleared"
+    },
+    "occupancy_detected": {
+      "description": "Triggers after one ore more occupancy sensors start detecting occupancy.",
+      "fields": {
+        "behavior": {
+          "description": "[%key:component::binary_sensor::common::trigger_behavior_description_presence%]",
+          "name": "[%key:component::binary_sensor::common::trigger_behavior_name%]"
+        }
+      },
+      "name": "Occupancy detected"
+    }
+  }
 }

homeassistant/components/binary_sensor/trigger.py

@@ -0,0 +1,67 @@
+"""Provides triggers for binary sensors."""
+
+from homeassistant.const import STATE_OFF, STATE_ON
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.entity import get_device_class
+from homeassistant.helpers.trigger import EntityStateTriggerBase, Trigger
+from homeassistant.helpers.typing import UNDEFINED, UndefinedType
+
+from . import DOMAIN, BinarySensorDeviceClass
+
+
+def get_device_class_or_undefined(
+    hass: HomeAssistant, entity_id: str
+) -> str | None | UndefinedType:
+    """Get the device class of an entity or UNDEFINED if not found."""
+    try:
+        return get_device_class(hass, entity_id)
+    except HomeAssistantError:
+        return UNDEFINED
+
+
+class BinarySensorOnOffTrigger(EntityStateTriggerBase):
+    """Class for binary sensor on/off triggers."""
+
+    _device_class: BinarySensorDeviceClass | None
+    _domain: str = DOMAIN
+
+    def entity_filter(self, entities: set[str]) -> set[str]:
+        """Filter entities of this domain."""
+        entities = super().entity_filter(entities)
+        return {
+            entity_id
+            for entity_id in entities
+            if get_device_class_or_undefined(self._hass, entity_id)
+            == self._device_class
+        }
+
+
+def make_binary_sensor_trigger(
+    device_class: BinarySensorDeviceClass | None,
+    to_state: str,
+) -> type[BinarySensorOnOffTrigger]:
+    """Create an entity state trigger class."""
+
+    class CustomTrigger(BinarySensorOnOffTrigger):
+        """Trigger for entity state changes."""
+
+        _device_class = device_class
+        _to_state = to_state
+
+    return CustomTrigger
+
+
+TRIGGERS: dict[str, type[Trigger]] = {
+    "occupancy_detected": make_binary_sensor_trigger(
+        BinarySensorDeviceClass.OCCUPANCY, STATE_ON
+    ),
+    "occupancy_cleared": make_binary_sensor_trigger(
+        BinarySensorDeviceClass.OCCUPANCY, STATE_OFF
+    ),
+}
+
+
+async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
+    """Return the triggers for binary sensors."""
+    return TRIGGERS

homeassistant/components/binary_sensor/triggers.yaml

@@ -0,0 +1,25 @@
+.trigger_common_fields: &trigger_common_fields
+  behavior:
+    required: true
+    default: any
+    selector:
+      select:
+        translation_key: trigger_behavior
+        options:
+          - first
+          - last
+          - any
+
+occupancy_cleared:
+  fields: *trigger_common_fields
+  target:
+    entity:
+      domain: binary_sensor
+      device_class: presence
+
+occupancy_detected:
+  fields: *trigger_common_fields
+  target:
+    entity:
+      domain: binary_sensor
+      device_class: presence

tests/components/__init__.py

@@ -29,8 +29,15 @@
 from tests.common import MockConfigEntry, mock_device_registry
 
 
-async def target_entities(hass: HomeAssistant, domain: str) -> list[str]:
-    """Create multiple entities associated with different targets."""
+async def target_entities(
+    hass: HomeAssistant, domain: str
+) -> tuple[list[str], list[str]]:
+    """Create multiple entities associated with different targets.
+
+    Returns a dict with the following keys:
+    - included: List of entity_ids meant to be targeted.
+    - excluded: List of entity_ids not meant to be targeted.
+    """
     await async_setup_component(hass, domain, {})
 
     config_entry = MockConfigEntry(domain="test")
@@ -55,40 +62,71 @@ async def target_entities(hass: HomeAssistant, domain: str) -> list[str]:
     mock_device_registry(hass, {device.id: device})
 
     entity_reg = er.async_get(hass)
-    # Entity associated with area
+    # Entities associated with area
     entity_area = entity_reg.async_get_or_create(
         domain=domain,
         platform="test",
         unique_id=f"{domain}_area",
         suggested_object_id=f"area_{domain}",
     )
     entity_reg.async_update_entity(entity_area.entity_id, area_id=area.id)
+    entity_area_excluded = entity_reg.async_get_or_create(
+        domain=domain,
+        platform="test",
+        unique_id=f"{domain}_area_excluded",
+        suggested_object_id=f"area_{domain}_excluded",
+    )
+    entity_reg.async_update_entity(entity_area_excluded.entity_id, area_id=area.id)
 
-    # Entity associated with device
+    # Entities associated with device
     entity_reg.async_get_or_create(
         domain=domain,
         platform="test",
         unique_id=f"{domain}_device",
         suggested_object_id=f"device_{domain}",
         device_id=device.id,
     )
+    entity_reg.async_get_or_create(
+        domain=domain,
+        platform="test",
+        unique_id=f"{domain}_device_excluded",
+        suggested_object_id=f"device_{domain}_excluded",
+        device_id=device.id,
+    )
 
-    # Entity associated with label
+    # Entities associated with label
     entity_label = entity_reg.async_get_or_create(
         domain=domain,
         platform="test",
         unique_id=f"{domain}_label",
         suggested_object_id=f"label_{domain}",
     )
     entity_reg.async_update_entity(entity_label.entity_id, labels={label.label_id})
+    entity_label_excluded = entity_reg.async_get_or_create(
+        domain=domain,
+        platform="test",
+        unique_id=f"{domain}_label_excluded",
+        suggested_object_id=f"label_{domain}_excluded",
+    )
+    entity_reg.async_update_entity(
+        entity_label_excluded.entity_id, labels={label.label_id}
+    )
 
     # Return all available entities
-    return [
-        f"{domain}.standalone_{domain}",
-        f"{domain}.label_{domain}",
-        f"{domain}.area_{domain}",
-        f"{domain}.device_{domain}",
-    ]
+    return {
+        "included": [
+            f"{domain}.standalone_{domain}",
+            f"{domain}.label_{domain}",
+            f"{domain}.area_{domain}",
+            f"{domain}.device_{domain}",
+        ],
+        "excluded": [
+            f"{domain}.standalone_{domain}_excluded",
+            f"{domain}.label_{domain}_excluded",
+            f"{domain}.area_{domain}_excluded",
+            f"{domain}.device_{domain}_excluded",
+        ],
+    }
 
 
 def parametrize_target_entities(domain: str) -> list[tuple[dict, str, int]]:
@@ -112,11 +150,18 @@ def parametrize_target_entities(domain: str) -> list[tuple[dict, str, int]]:
     ]
 
 
-class StateDescription(TypedDict):
+class _StateDescription(TypedDict):
     """Test state and expected service call count."""
 
     state: str | None
     attributes: dict
+
+
+class StateDescription(TypedDict):
+    """Test state and expected service call count."""
+
+    included: _StateDescription
+    excluded: _StateDescription
     count: int
 
 
@@ -147,10 +192,26 @@ def state_with_attributes(
     ) -> dict:
         """Return (state, attributes) dict."""
         if isinstance(state, str) or state is None:
-            return {"state": state, "attributes": additional_attributes, "count": count}
+            return {
+                "included": {
+                    "state": state,
+                    "attributes": additional_attributes,
+                },
+                "excluded": {
+                    "state": state,
+                    "attributes": {},
+                },
+                "count": count,
+            }
         return {
-            "state": state[0],
-            "attributes": state[1] | additional_attributes,
+            "included": {
+                "state": state[0],
+                "attributes": state[1] | additional_attributes,
+            },
+            "excluded": {
+                "state": state[0],
+                "attributes": state[1],
+            },
             "count": count,
         }
 

tests/components/alarm_control_panel/test_trigger.py

@@ -42,7 +42,7 @@ def enable_experimental_triggers_conditions() -> Generator[None]:
 @pytest.fixture
 async def target_alarm_control_panels(hass: HomeAssistant) -> list[str]:
     """Create multiple alarm control panel entities associated with different targets."""
-    return await target_entities(hass, "alarm_control_panel")
+    return (await target_entities(hass, "alarm_control_panel"))["included"]
 
 
 @pytest.mark.parametrize(
@@ -160,13 +160,14 @@ async def test_alarm_control_panel_state_trigger_behavior_any(
 
     # Set all alarm control panels, including the tested one, to the initial state
     for eid in target_alarm_control_panels:
-        set_or_remove_state(hass, eid, states[0])
+        set_or_remove_state(hass, eid, states[0]["included"])
         await hass.async_block_till_done()
 
     await arm_trigger(hass, trigger, {}, trigger_target_config)
 
     for state in states[1:]:
-        set_or_remove_state(hass, entity_id, state)
+        included_state = state["included"]
+        set_or_remove_state(hass, entity_id, included_state)
         await hass.async_block_till_done()
         assert len(service_calls) == state["count"]
         for service_call in service_calls:
@@ -175,7 +176,7 @@ async def test_alarm_control_panel_state_trigger_behavior_any(
 
         # Check if changing other alarm control panels also triggers
         for other_entity_id in other_entity_ids:
-            set_or_remove_state(hass, other_entity_id, state)
+            set_or_remove_state(hass, other_entity_id, included_state)
             await hass.async_block_till_done()
         assert len(service_calls) == (entities_in_target - 1) * state["count"]
         service_calls.clear()
@@ -271,13 +272,14 @@ async def test_alarm_control_panel_state_trigger_behavior_first(
 
     # Set all alarm control panels, including the tested one, to the initial state
     for eid in target_alarm_control_panels:
-        set_or_remove_state(hass, eid, states[0])
+        set_or_remove_state(hass, eid, states[0]["included"])
         await hass.async_block_till_done()
 
     await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
 
     for state in states[1:]:
-        set_or_remove_state(hass, entity_id, state)
+        included_state = state["included"]
+        set_or_remove_state(hass, entity_id, included_state)
         await hass.async_block_till_done()
         assert len(service_calls) == state["count"]
         for service_call in service_calls:
@@ -286,7 +288,7 @@ async def test_alarm_control_panel_state_trigger_behavior_first(
 
         # Triggering other alarm control panels should not cause the trigger to fire again
         for other_entity_id in other_entity_ids:
-            set_or_remove_state(hass, other_entity_id, state)
+            set_or_remove_state(hass, other_entity_id, included_state)
             await hass.async_block_till_done()
         assert len(service_calls) == 0
 
@@ -381,18 +383,19 @@ async def test_alarm_control_panel_state_trigger_behavior_last(
 
     # Set all alarm control panels, including the tested one, to the initial state
     for eid in target_alarm_control_panels:
-        set_or_remove_state(hass, eid, states[0])
+        set_or_remove_state(hass, eid, states[0]["included"])
         await hass.async_block_till_done()
 
     await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
 
     for state in states[1:]:
+        included_state = state["included"]
         for other_entity_id in other_entity_ids:
-            set_or_remove_state(hass, other_entity_id, state)
+            set_or_remove_state(hass, other_entity_id, included_state)
             await hass.async_block_till_done()
         assert len(service_calls) == 0
 
-        set_or_remove_state(hass, entity_id, state)
+        set_or_remove_state(hass, entity_id, included_state)
         await hass.async_block_till_done()
         assert len(service_calls) == state["count"]
         for service_call in service_calls: