feat: Add ZigBee contact sensors support (#1754)

* Add ZigBee contact sensors support

* Respect unavailable state of binary_sensor

* Add safe get of applianceTypes

Co-authored-by: Cosimo Meli <[email protected]>

custom_components/alexa_media/__init__.py

@@ -263,13 +263,15 @@ async def login_success(event=None) -> None:
                 "switch": {},
                 "guard": [],
                 "light": [],
+                "binary_sensor": [],
                 "temperature": [],
             },
             "entities": {
                 "media_player": {},
                 "switch": {},
                 "sensor": {},
                 "light": [],
+                "binary_sensor": [],
                 "alarm_control_panel": {},
             },
             "excluded": {},
@@ -401,6 +403,10 @@ async def async_update_data() -> Optional[AlexaEntityData]:
             if light.enabled:
                 entities_to_monitor.add(light.alexa_entity_id)
 
+        for binary_sensor in hass.data[DATA_ALEXAMEDIA]["accounts"][email]["entities"]["binary_sensor"]:
+            if binary_sensor.enabled:
+                entities_to_monitor.add(binary_sensor.alexa_entity_id)
+
         for guard in hass.data[DATA_ALEXAMEDIA]["accounts"][email]["entities"][
             "alarm_control_panel"
         ].values():

custom_components/alexa_media/alexa_entity.py

@@ -94,10 +94,17 @@ def is_light(appliance: Dict[Text, Any]) -> bool:
     """Is the given appliance a light controlled locally by an Echo."""
     return (
         is_local(appliance)
-        and "LIGHT" in appliance["applianceTypes"]
+        and "LIGHT" in appliance.get("applianceTypes", [])
         and has_capability(appliance, "Alexa.PowerController", "powerState")
     )
 
+def is_contact_sensor(appliance: Dict[Text, Any]) -> bool:
+    """Is the given appliance a contact sensor controlled locally by an Echo."""
+    return (
+        is_local(appliance)
+        and "CONTACT_SENSOR" in appliance.get("applianceTypes", [])
+        and has_capability(appliance, "Alexa.ContactSensor", "detectionState")
+    )
 
 def get_friendliest_name(appliance: Dict[Text, Any]) -> Text:
     """Find the best friendly name. Alexa seems to store manual renames in aliases. Prefer that one."""
@@ -140,20 +147,27 @@ class AlexaTemperatureEntity(AlexaEntity):
 
     device_serial: Text
 
+class AlexaBinaryEntity(AlexaEntity):
+    """Class for AlexaBinaryEntity."""
+
+    battery_level: bool
+
 
 class AlexaEntities(TypedDict):
     """Class for holding entities."""
 
     light: List[AlexaLightEntity]
     guard: List[AlexaEntity]
     temperature: List[AlexaTemperatureEntity]
+    binary_sensor: List[AlexaBinaryEntity]
 
 
 def parse_alexa_entities(network_details: Optional[Dict[Text, Any]]) -> AlexaEntities:
     """Turn the network details into a list of useful entities with the important details extracted."""
     lights = []
     guards = []
     temperature_sensors = []
+    contact_sensors = []
     location_details = network_details["locationDetails"]["locationDetails"]
     for location in location_details.values():
         amazon_bridge_details = location["amazonBridgeDetails"]["amazonBridgeDetails"]
@@ -187,8 +201,15 @@ def parse_alexa_entities(network_details: Optional[Dict[Text, Any]]) -> AlexaEnt
                         "colorTemperatureInKelvin",
                     )
                     lights.append(processed_appliance)
+                elif is_contact_sensor(appliance):
+                    processed_appliance["battery_level"] = has_capability(
+                        appliance, "Alexa.BatteryLevelSensor", "batteryLevel"
+                    )
+                    contact_sensors.append(processed_appliance)
+                else:
+                    _LOGGER.debug("Found unsupported device %s", appliance)
 
-    return {"light": lights, "guard": guards, "temperature": temperature_sensors}
+    return {"light": lights, "guard": guards, "temperature": temperature_sensors, "binary_sensor": contact_sensors}
 
 
 class AlexaCapabilityState(TypedDict):
@@ -286,6 +307,14 @@ def parse_guard_state_from_coordinator(
     )
 
 
+def parse_detection_state_from_coordinator(
+    coordinator: DataUpdateCoordinator, entity_id: Text
+) -> Optional[bool]:
+    """Get the detection state from the coordinator data."""
+    return parse_value_from_coordinator(
+        coordinator, entity_id, "Alexa.ContactSensor", "detectionState"
+    )
+
 def parse_value_from_coordinator(
     coordinator: DataUpdateCoordinator,
     entity_id: Text,

custom_components/alexa_media/binary_sensor.py

@@ -0,0 +1,109 @@
+"""
+Alexa Devices Sensors.
+
+SPDX-License-Identifier: Apache-2.0
+
+For more details about this platform, please refer to the documentation at
+https://community.home-assistant.io/t/echo-devices-alexa-as-media-player-testers-needed/58639
+"""
+
+import logging
+from typing import List  # noqa pylint: disable=unused-import
+
+from alexapy import hide_serial
+from homeassistant.components.binary_sensor import (
+    BinarySensorDeviceClass,
+    BinarySensorEntity,
+)
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from . import (
+    CONF_EMAIL,
+    CONF_EXCLUDE_DEVICES,
+    CONF_INCLUDE_DEVICES,
+    DATA_ALEXAMEDIA,
+    hide_email,
+)
+from .alexa_entity import parse_detection_state_from_coordinator
+from .const import CONF_EXTENDED_ENTITY_DISCOVERY
+from .helpers import add_devices
+
+_LOGGER = logging.getLogger(__name__)
+
+async def async_setup_platform(hass, config, add_devices_callback, discovery_info=None):
+    """Set up the Alexa sensor platform."""
+    devices: List[BinarySensorEntity] = []
+    account = config[CONF_EMAIL] if config else discovery_info["config"][CONF_EMAIL]
+    account_dict = hass.data[DATA_ALEXAMEDIA]["accounts"][account]
+    include_filter = config.get(CONF_INCLUDE_DEVICES, [])
+    exclude_filter = config.get(CONF_EXCLUDE_DEVICES, [])
+    coordinator = account_dict["coordinator"]
+    binary_entities = account_dict.get("devices", {}).get("binary_sensor", [])
+    if binary_entities and account_dict["options"].get(CONF_EXTENDED_ENTITY_DISCOVERY):
+        for be in binary_entities:
+            _LOGGER.debug(
+                "Creating entity %s for a binary_sensor with name %s",
+                hide_serial(be["id"]),
+                be["name"],
+            )
+            contact_sensor = AlexaContact(coordinator, be)
+            account_dict["entities"]["binary_sensor"].append(contact_sensor)
+            devices.append(contact_sensor)
+
+    return await add_devices(
+        hide_email(account),
+        devices,
+        add_devices_callback,
+        include_filter,
+        exclude_filter,
+    )
+
+
+async def async_setup_entry(hass, config_entry, async_add_devices):
+    """Set up the Alexa sensor platform by config_entry."""
+    return await async_setup_platform(
+        hass, config_entry.data, async_add_devices, discovery_info=None
+    )
+
+
+async def async_unload_entry(hass, entry) -> bool:
+    """Unload a config entry."""
+    account = entry.data[CONF_EMAIL]
+    account_dict = hass.data[DATA_ALEXAMEDIA]["accounts"][account]
+    _LOGGER.debug("Attempting to unload binary sensors")
+    for binary_sensor in account_dict["entities"]["binary_sensor"]:
+        await binary_sensor.async_remove()
+    return True
+
+class AlexaContact(CoordinatorEntity, BinarySensorEntity):
+    """A contact sensor controlled by an Echo."""
+
+    _attr_device_class = BinarySensorDeviceClass.DOOR
+
+    def __init__(self, coordinator, details):
+        super().__init__(coordinator)
+        self.alexa_entity_id = details["id"]
+        self._name = details["name"]
+
+    @property
+    def name(self):
+        return self._name
+
+    @property
+    def unique_id(self):
+        return self.alexa_entity_id
+
+    @property
+    def is_on(self):
+        detection = parse_detection_state_from_coordinator(
+            self.coordinator, self.alexa_entity_id
+        )
+
+        return detection == 'DETECTED' if detection is not None else None
+
+    @property
+    def assumed_state(self) -> bool:
+        last_refresh_success = (
+            self.coordinator.data and self.alexa_entity_id in self.coordinator.data
+        )
+        return not last_refresh_success

custom_components/alexa_media/const.py

@@ -30,6 +30,7 @@
     "sensor",
     "alarm_control_panel",
     "light",
+    "binary_sensor"
 ]
 
 HTTP_COOKIE_HEADER = "# HTTP Cookie File"