diff --git a/homeassistant/components/netatmo/binary_sensor.py b/homeassistant/components/netatmo/binary_sensor.py index 81498c3d767..c550c31c4a6 100644 --- a/homeassistant/components/netatmo/binary_sensor.py +++ b/homeassistant/components/netatmo/binary_sensor.py @@ -1,8 +1,12 @@ """Support for Netatmo binary sensors.""" +from collections.abc import Callable from dataclasses import dataclass +from functools import partial import logging -from typing import Final, cast +from typing import Any, Final, cast + +from pyatmo.modules.device_types import DeviceCategory as NetatmoDeviceCategory from homeassistant.components.binary_sensor import ( BinarySensorDeviceClass, @@ -13,34 +17,166 @@ from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback -from homeassistant.helpers.typing import StateType -from .const import NETATMO_CREATE_WEATHER_BINARY_SENSOR -from .data_handler import NetatmoDevice -from .entity import NetatmoWeatherModuleEntity +from .const import ( + CONF_URL_SECURITY, + DOORTAG_CATEGORY_DOOR, + DOORTAG_CATEGORY_FURNITURE, + DOORTAG_CATEGORY_GARAGE, + DOORTAG_CATEGORY_GATE, + DOORTAG_CATEGORY_OTHER, + DOORTAG_CATEGORY_WINDOW, + DOORTAG_STATUS_CALIBRATING, + DOORTAG_STATUS_CALIBRATION_FAILED, + DOORTAG_STATUS_CLOSED, + DOORTAG_STATUS_MAINTENANCE, + DOORTAG_STATUS_NO_NEWS, + DOORTAG_STATUS_OPEN, + DOORTAG_STATUS_UNDEFINED, + DOORTAG_STATUS_WEAK_SIGNAL, + NETATMO_CREATE_CONNECTIVITY_BINARY_SENSOR, + NETATMO_CREATE_OPENING_BINARY_SENSOR, + NETATMO_CREATE_WEATHER_BINARY_SENSOR, +) +from .data_handler import SIGNAL_NAME, NetatmoDevice +from .entity import NetatmoModuleEntity, NetatmoWeatherModuleEntity _LOGGER = logging.getLogger(__name__) +DEFAULT_OPENING_SENSOR_KEY = "opening_sensor" + +OPENING_STATUS_TO_BINARY_SENSOR_STATE: Final[dict[str, bool | None]] = { + DOORTAG_STATUS_NO_NEWS: None, + DOORTAG_STATUS_CALIBRATING: None, + DOORTAG_STATUS_UNDEFINED: None, + DOORTAG_STATUS_CLOSED: False, + DOORTAG_STATUS_OPEN: True, + DOORTAG_STATUS_CALIBRATION_FAILED: None, + DOORTAG_STATUS_MAINTENANCE: None, + DOORTAG_STATUS_WEAK_SIGNAL: None, +} + + +OPENING_CATEGORY_TO_DEVICE_CLASS: Final[dict[str | None, BinarySensorDeviceClass]] = { + DOORTAG_CATEGORY_DOOR: BinarySensorDeviceClass.DOOR, + DOORTAG_CATEGORY_FURNITURE: BinarySensorDeviceClass.OPENING, + DOORTAG_CATEGORY_GARAGE: BinarySensorDeviceClass.GARAGE_DOOR, + DOORTAG_CATEGORY_GATE: BinarySensorDeviceClass.OPENING, + DOORTAG_CATEGORY_OTHER: BinarySensorDeviceClass.OPENING, + DOORTAG_CATEGORY_WINDOW: BinarySensorDeviceClass.WINDOW, +} + + +def get_opening_category(netatmo_device: NetatmoDevice) -> str: + """Helper function to get opening category from Netatmo API raw data.""" + + # Iterate through each home in the raw data. + for home in netatmo_device.data_handler.account.raw_data["homes"]: + # Check if the modules list exists for the current home. + if "modules" in home: + # Iterate through each module to find a matching ID. + for module in home["modules"]: + if module["id"] == netatmo_device.device.entity_id: + # We found the matching device. Get its category. + if module.get("category") is not None: + return cast(str, module["category"]) + raise ValueError( + f"Device {netatmo_device.device.entity_id} found, " + "but 'category' is missing in raw data." + ) + + raise ValueError( + f"Device {netatmo_device.device.entity_id} not found in Netatmo raw data." + ) + + +OPENING_CATEGORY_TO_KEY: Final[dict[str, str | None]] = { + DOORTAG_CATEGORY_DOOR: None, + DOORTAG_CATEGORY_FURNITURE: DOORTAG_CATEGORY_FURNITURE, + DOORTAG_CATEGORY_GARAGE: None, + DOORTAG_CATEGORY_GATE: DOORTAG_CATEGORY_GATE, + DOORTAG_CATEGORY_OTHER: DEFAULT_OPENING_SENSOR_KEY, + DOORTAG_CATEGORY_WINDOW: None, +} + @dataclass(frozen=True, kw_only=True) class NetatmoBinarySensorEntityDescription(BinarySensorEntityDescription): """Describes Netatmo binary sensor entity.""" - name: str | None = None # The default name of the sensor - netatmo_name: str # The name used by Netatmo API for this sensor + netatmo_name: str | None = ( + None # The name used by Netatmo API for this sensor (exposed feature as attribute) if different than key + ) + value_fn: Callable[[str], str | bool | None] = lambda x: x -NETATMO_WEATHER_BINARY_SENSOR_DESCRIPTIONS: Final[ +NETATMO_CONNECTIVITY_BINARY_SENSOR_DESCRIPTIONS: Final[ list[NetatmoBinarySensorEntityDescription] ] = [ NetatmoBinarySensorEntityDescription( key="reachable", - name="Connectivity", - netatmo_name="reachable", device_class=BinarySensorDeviceClass.CONNECTIVITY, ), ] +# Assuming a Module object with the following attributes: +# {'battery_level': 5780, +# 'battery_percent': None, +# 'battery_state': 'full', +# 'bridge': 'XX:XX:XX:XX:XX:XX', +# 'device_category': , +# 'device_type': , +# 'entity_id': 'NN:NN:NN:NN:NN:NN', +# 'features': {'status', 'battery', 'rf_strength', 'reachable'}, +# 'firmware_name': None, +# 'firmware_revision': 58, +# 'history_features': set(), +# 'history_features_values': {}, +# 'home': , +# 'modules': None, +# 'name': 'YYYYYY', +# 'reachable': True, +# 'rf_strength': 74, +# 'room_id': 'ZZZZZZZZ', +# 'status': 'open'} + +NETATMO_OPENING_BINARY_SENSOR_DESCRIPTIONS: Final[ + list[NetatmoBinarySensorEntityDescription] +] = [ + NetatmoBinarySensorEntityDescription( + key="opening", + netatmo_name="status", + value_fn=OPENING_STATUS_TO_BINARY_SENSOR_STATE.get, + ), +] + +DEVICE_CATEGORY_BINARY_URLS: Final[dict[NetatmoDeviceCategory, str]] = { + NetatmoDeviceCategory.opening: CONF_URL_SECURITY, +} + +DEVICE_CATEGORY_WEATHER_BINARY_SENSORS: Final[ + dict[NetatmoDeviceCategory, list[NetatmoBinarySensorEntityDescription]] +] = { + NetatmoDeviceCategory.air_care: NETATMO_CONNECTIVITY_BINARY_SENSOR_DESCRIPTIONS, + NetatmoDeviceCategory.weather: NETATMO_CONNECTIVITY_BINARY_SENSOR_DESCRIPTIONS, +} + +DEVICE_CATEGORY_CONNECTIVITY_BINARY_SENSORS: Final[ + dict[NetatmoDeviceCategory, list[NetatmoBinarySensorEntityDescription]] +] = { + NetatmoDeviceCategory.opening: NETATMO_CONNECTIVITY_BINARY_SENSOR_DESCRIPTIONS, +} + +DEVICE_CATEGORY_OPENING_BINARY_SENSORS: Final[ + dict[NetatmoDeviceCategory, list[NetatmoBinarySensorEntityDescription]] +] = { + NetatmoDeviceCategory.opening: NETATMO_OPENING_BINARY_SENSOR_DESCRIPTIONS, +} + +DEVICE_CATEGORY_BINARY_PUBLISHERS: Final[list[NetatmoDeviceCategory]] = [ + NetatmoDeviceCategory.opening, +] + async def async_setup_entry( hass: HomeAssistant, @@ -50,25 +186,47 @@ async def async_setup_entry( """Set up Netatmo weather binary sensors based on a config entry.""" @callback - def _create_weather_binary_sensor_entity(netatmo_device: NetatmoDevice) -> None: - """Create weather binary sensor entities for a Netatmo weather device.""" + def _create_binary_sensor_entity( + binarySensorClass: type[ + NetatmoWeatherBinarySensor + | NetatmoOpeningBinarySensor + | NetatmoConnectivityBinarySensor + ], + descriptions: dict[ + NetatmoDeviceCategory, list[NetatmoBinarySensorEntityDescription] + ], + netatmo_device: NetatmoDevice, + ) -> None: + """Create binary sensor entities for a Netatmo device.""" - descriptions_to_add = NETATMO_WEATHER_BINARY_SENSOR_DESCRIPTIONS + if netatmo_device.device.device_category is None: + return - entities: list[NetatmoWeatherBinarySensor] = [] + descriptions_to_add = descriptions.get( + netatmo_device.device.device_category, [] + ) + + entities: list[ + NetatmoWeatherBinarySensor + | NetatmoOpeningBinarySensor + | NetatmoConnectivityBinarySensor + ] = [] # Create binary sensors for module for description in descriptions_to_add: - # Actual check is simple for reachable - feature_check = description.key + if description.netatmo_name is None: + feature_check = description.key + else: + feature_check = description.netatmo_name if feature_check in netatmo_device.device.features: _LOGGER.debug( - 'Adding "%s" weather binary sensor for device %s', + 'Adding "%s" (native: "%s") binary sensor for device %s', + description.key, feature_check, netatmo_device.device.name, ) entities.append( - NetatmoWeatherBinarySensor( + binarySensorClass( netatmo_device, description, ) @@ -81,12 +239,100 @@ async def async_setup_entry( async_dispatcher_connect( hass, NETATMO_CREATE_WEATHER_BINARY_SENSOR, - _create_weather_binary_sensor_entity, + partial( + _create_binary_sensor_entity, + NetatmoWeatherBinarySensor, + DEVICE_CATEGORY_WEATHER_BINARY_SENSORS, + ), + ) + ) + + entry.async_on_unload( + async_dispatcher_connect( + hass, + NETATMO_CREATE_OPENING_BINARY_SENSOR, + partial( + _create_binary_sensor_entity, + NetatmoOpeningBinarySensor, + DEVICE_CATEGORY_OPENING_BINARY_SENSORS, + ), + ) + ) + + entry.async_on_unload( + async_dispatcher_connect( + hass, + NETATMO_CREATE_CONNECTIVITY_BINARY_SENSOR, + partial( + _create_binary_sensor_entity, + NetatmoConnectivityBinarySensor, + DEVICE_CATEGORY_CONNECTIVITY_BINARY_SENSORS, + ), ) ) -class NetatmoWeatherBinarySensor(NetatmoWeatherModuleEntity, BinarySensorEntity): +class NetatmoBinarySensor(NetatmoModuleEntity, BinarySensorEntity): + """Implementation of a Netatmo binary sensor.""" + + entity_description: NetatmoBinarySensorEntityDescription + _attr_has_entity_name = True + + def __init__( + self, + netatmo_device: NetatmoDevice, + description: NetatmoBinarySensorEntityDescription, + **kwargs: Any, # Add this to capture extra args from super() + ) -> None: + """Initialize a Netatmo binary sensor.""" + + # To prevent exception about missing URL we need to set it explicitly + if netatmo_device.device.device_category is not None: + if ( + DEVICE_CATEGORY_BINARY_URLS.get(netatmo_device.device.device_category) + is not None + ): + self._attr_configuration_url = DEVICE_CATEGORY_BINARY_URLS[ + netatmo_device.device.device_category + ] + + super().__init__(netatmo_device, **kwargs) + + self.entity_description = description + self._attr_unique_id = f"{self.device.entity_id}-{description.key}" + + # Register publishers for the entity if needed (not already done in parent class - weather and air_care) + # We need to keep this here because we have two classes depending on it and we want to avoid adding publishers for all binary sensors + if self.device.device_category in DEVICE_CATEGORY_BINARY_PUBLISHERS: + self._publishers.extend( + [ + { + "name": self.home.entity_id, + "home_id": self.home.entity_id, + SIGNAL_NAME: netatmo_device.signal_name, + }, + ] + ) + + @callback + def async_update_callback(self) -> None: + """Update the entity's state.""" + + # Should be the connectivity (reachable) sensor only here as we have update for opening in its class + + # Setting reachable sensor, so we just get it directly (backward compatibility to weather binary sensor) + value = getattr(self.device, self.entity_description.key, None) + + if value is None: + self._attr_available = False + self._attr_is_on = False + else: + self._attr_available = True + self._attr_is_on = cast(bool, value) + self.async_write_ha_state() + + +class NetatmoWeatherBinarySensor(NetatmoWeatherModuleEntity, NetatmoBinarySensor): """Implementation of a Netatmo weather binary sensor.""" entity_description: NetatmoBinarySensorEntityDescription @@ -98,24 +344,68 @@ class NetatmoWeatherBinarySensor(NetatmoWeatherModuleEntity, BinarySensorEntity) ) -> None: """Initialize a Netatmo weather binary sensor.""" - super().__init__(netatmo_device) + super().__init__(netatmo_device, description=description) - self.entity_description = description - self._attr_unique_id = f"{self.device.entity_id}-{description.key}" + +class NetatmoOpeningBinarySensor(NetatmoBinarySensor): + """Implementation of a Netatmo opening binary sensor.""" + + entity_description: NetatmoBinarySensorEntityDescription + _attr_has_entity_name = True + + def __init__( + self, + netatmo_device: NetatmoDevice, + description: NetatmoBinarySensorEntityDescription, + ) -> None: + """Initialize a Netatmo binary sensor.""" + + super().__init__(netatmo_device, description) + + # Apply Dynamic Device Class override + self._attr_device_class = OPENING_CATEGORY_TO_DEVICE_CLASS.get( + get_opening_category(netatmo_device), BinarySensorDeviceClass.OPENING + ) + + # Apply Dynamic Translation Key override if needed + translation_key = OPENING_CATEGORY_TO_KEY.get( + get_opening_category(netatmo_device), DEFAULT_OPENING_SENSOR_KEY + ) + if translation_key is not None: + self._attr_translation_key = translation_key @callback def async_update_callback(self) -> None: """Update the entity's state.""" - value: StateType | None = None - - value = getattr(self.device, self.entity_description.netatmo_name, None) - - if value is None: + if not self.device.reachable: + # If reachable is None or False we set availability to False self._attr_available = False - self._attr_is_on = False + self._attr_is_on = None + else: + # If reachable is True, we get the actual value + if self.entity_description.netatmo_name is None: + raw_value = getattr(self.device, self.entity_description.key, None) + else: + raw_value = getattr( + self.device, self.entity_description.netatmo_name, None + ) + + if raw_value is not None: + value = self.entity_description.value_fn(raw_value) + else: + value = None + + # Set sensor state self._attr_available = True - self._attr_is_on = cast(bool, value) + self._attr_is_on = cast(bool, value) if value is not None else None self.async_write_ha_state() + + +class NetatmoConnectivityBinarySensor(NetatmoBinarySensor): + """Implementation of a Netatmo connectivity binary sensor.""" + + entity_description: NetatmoBinarySensorEntityDescription + _attr_has_entity_name = True diff --git a/homeassistant/components/netatmo/const.py b/homeassistant/components/netatmo/const.py index e789885f56b..9a95cd36fed 100644 --- a/homeassistant/components/netatmo/const.py +++ b/homeassistant/components/netatmo/const.py @@ -46,9 +46,11 @@ NETATMO_CREATE_CAMERA = "netatmo_create_camera" NETATMO_CREATE_CAMERA_LIGHT = "netatmo_create_camera_light" NETATMO_CREATE_CLIMATE = "netatmo_create_climate" NETATMO_CREATE_COVER = "netatmo_create_cover" +NETATMO_CREATE_CONNECTIVITY_BINARY_SENSOR = "netatmo_create_connectivity_binary_sensor" NETATMO_CREATE_BUTTON = "netatmo_create_button" NETATMO_CREATE_FAN = "netatmo_create_fan" NETATMO_CREATE_LIGHT = "netatmo_create_light" +NETATMO_CREATE_OPENING_BINARY_SENSOR = "netatmo_create_opening_binary_sensor" NETATMO_CREATE_ROOM_SENSOR = "netatmo_create_room_sensor" NETATMO_CREATE_SELECT = "netatmo_create_select" NETATMO_CREATE_SENSOR = "netatmo_create_sensor" @@ -191,6 +193,23 @@ MODE_LIGHT_OFF = "off" MODE_LIGHT_ON = "on" CAMERA_LIGHT_MODES = [MODE_LIGHT_ON, MODE_LIGHT_OFF, MODE_LIGHT_AUTO] +# Door tag categories +DOORTAG_CATEGORY_DOOR = "door" +DOORTAG_CATEGORY_FURNITURE = "furniture" +DOORTAG_CATEGORY_GARAGE = "garage" +DOORTAG_CATEGORY_GATE = "gate" +DOORTAG_CATEGORY_OTHER = "other" +DOORTAG_CATEGORY_WINDOW = "window" +# Door tag statuses +DOORTAG_STATUS_CALIBRATING = "calibrating" +DOORTAG_STATUS_CALIBRATION_FAILED = "calibration_failed" +DOORTAG_STATUS_CLOSED = "closed" +DOORTAG_STATUS_MAINTENANCE = "maintenance" +DOORTAG_STATUS_NO_NEWS = "no_news" +DOORTAG_STATUS_OPEN = "open" +DOORTAG_STATUS_UNDEFINED = "undefined" +DOORTAG_STATUS_WEAK_SIGNAL = "weak_signal" + # Webhook push_types MUST follow exactly Netatmo's naming on products! # See https://dev.netatmo.com/apidocumentation # e.g. cameras: NACamera, NOC, etc. diff --git a/homeassistant/components/netatmo/data_handler.py b/homeassistant/components/netatmo/data_handler.py index ee1b369c58c..31845e1c0c7 100644 --- a/homeassistant/components/netatmo/data_handler.py +++ b/homeassistant/components/netatmo/data_handler.py @@ -38,9 +38,11 @@ from .const import ( NETATMO_CREATE_CAMERA, NETATMO_CREATE_CAMERA_LIGHT, NETATMO_CREATE_CLIMATE, + NETATMO_CREATE_CONNECTIVITY_BINARY_SENSOR, NETATMO_CREATE_COVER, NETATMO_CREATE_FAN, NETATMO_CREATE_LIGHT, + NETATMO_CREATE_OPENING_BINARY_SENSOR, NETATMO_CREATE_ROOM_SENSOR, NETATMO_CREATE_SELECT, NETATMO_CREATE_SENSOR, @@ -367,6 +369,10 @@ class NetatmoDataHandler: ], NetatmoDeviceCategory.meter: [NETATMO_CREATE_SENSOR], NetatmoDeviceCategory.fan: [NETATMO_CREATE_FAN], + NetatmoDeviceCategory.opening: [ + NETATMO_CREATE_CONNECTIVITY_BINARY_SENSOR, + NETATMO_CREATE_OPENING_BINARY_SENSOR, + ], } for module in home.modules.values(): if not module.device_category: diff --git a/homeassistant/components/netatmo/entity.py b/homeassistant/components/netatmo/entity.py index b519c75ae55..2d12631a3db 100644 --- a/homeassistant/components/netatmo/entity.py +++ b/homeassistant/components/netatmo/entity.py @@ -93,9 +93,11 @@ class NetatmoBaseEntity(Entity): class NetatmoDeviceEntity(NetatmoBaseEntity): """Netatmo entity base class.""" - def __init__(self, data_handler: NetatmoDataHandler, device: NetatmoBase) -> None: + def __init__( + self, data_handler: NetatmoDataHandler, device: NetatmoBase, **kwargs: Any + ) -> None: """Set up Netatmo entity base.""" - super().__init__(data_handler) + super().__init__(data_handler, **kwargs) self.device = device @property @@ -153,9 +155,9 @@ class NetatmoModuleEntity(NetatmoDeviceEntity): device: Module _attr_configuration_url: str - def __init__(self, device: NetatmoDevice) -> None: + def __init__(self, device: NetatmoDevice, **kwargs: Any) -> None: """Set up a Netatmo module entity.""" - super().__init__(device.data_handler, device.device) + super().__init__(device.data_handler, device.device, **kwargs) self._attr_device_info = DeviceInfo( identifiers={(DOMAIN, device.device.entity_id)}, name=device.device.name, @@ -175,9 +177,9 @@ class NetatmoWeatherModuleEntity(NetatmoModuleEntity): _attr_configuration_url = CONF_URL_WEATHER - def __init__(self, device: NetatmoDevice) -> None: + def __init__(self, device: NetatmoDevice, **kwargs: Any) -> None: """Set up a Netatmo weather module entity.""" - super().__init__(device) + super().__init__(device, **kwargs) assert self.device.device_category category = self.device.device_category.name self._publishers.extend( diff --git a/homeassistant/components/netatmo/strings.json b/homeassistant/components/netatmo/strings.json index 3307dcb6d45..0aadcbfea13 100644 --- a/homeassistant/components/netatmo/strings.json +++ b/homeassistant/components/netatmo/strings.json @@ -54,6 +54,17 @@ } }, "entity": { + "binary_sensor": { + "furniture": { + "name": "Furniture" + }, + "gate": { + "name": "Gate" + }, + "opening_sensor": { + "name": "Opening" + } + }, "button": { "preferred_position": { "name": "Preferred position" diff --git a/tests/components/netatmo/common.py b/tests/components/netatmo/common.py index 617a0070f5c..90c3bcb55d1 100644 --- a/tests/components/netatmo/common.py +++ b/tests/components/netatmo/common.py @@ -83,6 +83,10 @@ async def fake_post_request(hass: HomeAssistant, *args: Any, **kwargs: Any): else: payload = json.loads(await async_load_fixture(hass, f"{endpoint}.json", DOMAIN)) + # Apply test-specific modifications to the payload + if "msg_callback" in kwargs: + kwargs["msg_callback"](payload) + return AiohttpClientMockResponse( method="POST", url=kwargs["endpoint"], diff --git a/tests/components/netatmo/snapshots/test_binary_sensor.ambr b/tests/components/netatmo/snapshots/test_binary_sensor.ambr index fe50b59f183..9558edc0431 100644 --- a/tests/components/netatmo/snapshots/test_binary_sensor.ambr +++ b/tests/components/netatmo/snapshots/test_binary_sensor.ambr @@ -572,3 +572,105 @@ 'state': 'on', }) # --- +# name: test_entity[binary_sensor.window_hall_connectivity-entry] + EntityRegistryEntrySnapshot({ + 'aliases': set({ + }), + 'area_id': None, + 'capabilities': None, + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'binary_sensor', + 'entity_category': None, + 'entity_id': 'binary_sensor.window_hall_connectivity', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Connectivity', + 'options': dict({ + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Connectivity', + 'platform': 'netatmo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': None, + 'unique_id': '12:34:56:00:86:99-reachable', + 'unit_of_measurement': None, + }) +# --- +# name: test_entity[binary_sensor.window_hall_connectivity-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + 'attribution': 'Data provided by Netatmo', + 'device_class': 'connectivity', + 'friendly_name': 'Window Hall Connectivity', + }), + 'context': , + 'entity_id': 'binary_sensor.window_hall_connectivity', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': 'off', + }) +# --- +# name: test_entity[binary_sensor.window_hall_window-entry] + EntityRegistryEntrySnapshot({ + 'aliases': set({ + }), + 'area_id': None, + 'capabilities': None, + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'binary_sensor', + 'entity_category': None, + 'entity_id': 'binary_sensor.window_hall_window', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Window', + 'options': dict({ + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Window', + 'platform': 'netatmo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': None, + 'unique_id': '12:34:56:00:86:99-opening', + 'unit_of_measurement': None, + }) +# --- +# name: test_entity[binary_sensor.window_hall_window-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + 'attribution': 'Data provided by Netatmo', + 'device_class': 'window', + 'friendly_name': 'Window Hall Window', + }), + 'context': , + 'entity_id': 'binary_sensor.window_hall_window', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': 'unavailable', + }) +# --- diff --git a/tests/components/netatmo/test_binary_sensor.py b/tests/components/netatmo/test_binary_sensor.py index 91d2b3ad63b..f9dc4aaadcb 100644 --- a/tests/components/netatmo/test_binary_sensor.py +++ b/tests/components/netatmo/test_binary_sensor.py @@ -1,17 +1,27 @@ """Support for Netatmo binary sensors.""" -from unittest.mock import AsyncMock +from datetime import timedelta +from typing import Any +from unittest.mock import AsyncMock, patch +from freezegun.api import FrozenDateTimeFactory import pytest from syrupy.assertion import SnapshotAssertion -from homeassistant.const import Platform +from homeassistant.components.binary_sensor import BinarySensorDeviceClass +from homeassistant.const import CONF_WEBHOOK_ID, Platform from homeassistant.core import HomeAssistant from homeassistant.helpers import entity_registry as er +import homeassistant.util.dt as dt_util -from .common import snapshot_platform_entities +from .common import ( + FAKE_WEBHOOK_ACTIVATION, + fake_post_request, + simulate_webhook, + snapshot_platform_entities, +) -from tests.common import MockConfigEntry +from tests.common import MockConfigEntry, async_fire_time_changed @pytest.mark.usefixtures("entity_registry_enabled_by_default") @@ -30,3 +40,310 @@ async def test_entity( entity_registry, snapshot, ) + + +async def test_doortag_setup( + hass: HomeAssistant, config_entry: MockConfigEntry, netatmo_auth: AsyncMock +) -> None: + """Test doortag setup.""" + fake_post_hits = 0 + + async def fake_post(*args: Any, **kwargs: Any): + """Fake error during requesting backend data.""" + nonlocal fake_post_hits + fake_post_hits += 1 + return await fake_post_request(hass, *args, **kwargs) + + with ( + patch( + "homeassistant.components.netatmo.api.AsyncConfigEntryNetatmoAuth" + ) as mock_auth, + patch( + "homeassistant.components.netatmo.data_handler.PLATFORMS", + ["camera", "binary_sensor"], + ), + patch( + "homeassistant.components.netatmo.async_get_config_entry_implementation", + return_value=AsyncMock(), + ), + patch( + "homeassistant.components.netatmo.webhook_generate_url", + ) as mock_webhook, + ): + mock_auth.return_value.async_post_api_request.side_effect = fake_post + mock_auth.return_value.async_addwebhook.side_effect = AsyncMock() + mock_auth.return_value.async_dropwebhook.side_effect = AsyncMock() + mock_webhook.return_value = "https://example.com" + assert await hass.config_entries.async_setup(config_entry.entry_id) + + await hass.async_block_till_done() + + webhook_id = config_entry.data[CONF_WEBHOOK_ID] + + # Fake webhook activation + await simulate_webhook(hass, webhook_id, FAKE_WEBHOOK_ACTIVATION) + await hass.async_block_till_done() + + # Define variables for the test + _doortag_entity = "window_hall" + _doortag_entity_opening = f"binary_sensor.{_doortag_entity}_window" + _doortag_entity_connectivity = f"binary_sensor.{_doortag_entity}_connectivity" + + # Check opening creation + assert hass.states.get(_doortag_entity_opening) is not None + # Check connectivity creation + assert hass.states.get(_doortag_entity_connectivity) is not None + + # Check opening initial state + assert hass.states.get(_doortag_entity_opening).state == "unavailable" + # Check connectivity initial state + assert hass.states.get(_doortag_entity_connectivity).state == "off" + + +@pytest.mark.parametrize( + ("doortag_status", "expected"), + [ + ("no_news", "unknown"), + ("calibrating", "unknown"), + ("undefined", "unknown"), + ("closed", "off"), + ("open", "on"), + ("calibration_failed", "unknown"), + ("maintenance", "unknown"), + ("weak_signal", "unknown"), + ("invalid_value", "unknown"), + ], +) +async def test_doortag_opening_status_change( + hass: HomeAssistant, + config_entry: MockConfigEntry, + netatmo_auth: AsyncMock, + freezer: FrozenDateTimeFactory, + doortag_status: str, + expected: str, +) -> None: + """Test doortag opening status changes.""" + fake_post_hits = 0 + # Repeatedly used variables for the test and initial value from fixture + # Use nonexistent ID to prevent matching during initial setup + doortag_entity_id = "aa:bb:cc:dd:ee:ff" + doortag_connectivity = False + doortag_opening = "no_news" + doortag_timestamp = None + + def tag_modifier(payload): + """This function will be called by common.py during ANY homestatus call.""" + nonlocal doortag_connectivity, doortag_opening, doortag_timestamp + + if doortag_timestamp is not None: + payload["time_server"] = doortag_timestamp + body = payload.get("body", {}) + + # Handle both structures: {"home": {...}} AND {"homes": [{...}]} + homes_to_check = [] + if "home" in body and isinstance(body["home"], dict): + homes_to_check.append(body["home"]) + elif "homes" in body and isinstance(body["homes"], list): + homes_to_check.extend(body["homes"]) + + for home_data in homes_to_check: + # Safety check: ensure home_data is actually a dictionary + if not isinstance(home_data, dict): + continue + + modules = home_data.get("modules", []) + for module in modules: + if isinstance(module, dict) and module.get("id") == doortag_entity_id: + module["reachable"] = doortag_connectivity + module["status"] = doortag_opening + if doortag_timestamp is not None: + module["last_seen"] = doortag_timestamp + break + + async def fake_tag_post(*args, **kwargs): + """Fake tag status during requesting backend data.""" + nonlocal fake_post_hits + fake_post_hits += 1 + return await fake_post_request(hass, *args, msg_callback=tag_modifier, **kwargs) + + with ( + patch( + "homeassistant.components.netatmo.api.AsyncConfigEntryNetatmoAuth" + ) as mock_auth, + patch( + "homeassistant.components.netatmo.data_handler.PLATFORMS", + ["camera", "binary_sensor"], + ), + patch( + "homeassistant.components.netatmo.async_get_config_entry_implementation", + return_value=AsyncMock(), + ), + patch( + "homeassistant.components.netatmo.webhook_generate_url", + ) as mock_webhook, + ): + mock_auth.return_value.async_post_api_request.side_effect = fake_tag_post + mock_auth.return_value.async_addwebhook.side_effect = AsyncMock() + mock_auth.return_value.async_dropwebhook.side_effect = AsyncMock() + mock_webhook.return_value = "https://example.com" + assert await hass.config_entries.async_setup(config_entry.entry_id) + + await hass.async_block_till_done() + + webhook_id = config_entry.data[CONF_WEBHOOK_ID] + + # Fake webhook activation + await simulate_webhook(hass, webhook_id, FAKE_WEBHOOK_ACTIVATION) + await hass.async_block_till_done() + + # Define the variables for the test + _doortag_entity = "window_hall" + _doortag_entity_opening = f"binary_sensor.{_doortag_entity}_window" + _doortag_entity_connectivity = f"binary_sensor.{_doortag_entity}_connectivity" + + # Check connectivity creation + assert hass.states.get(_doortag_entity_connectivity) is not None + # Check opening creation + assert hass.states.get(_doortag_entity_opening) is not None + + # Check connectivity initial state + assert hass.states.get(_doortag_entity_connectivity).state == "off" + # Check opening initial state + assert hass.states.get(_doortag_entity_opening).state == "unavailable" + + # Trigger some polling cycle to let API throttling work + for _ in range(11): + freezer.tick(timedelta(seconds=30)) + async_fire_time_changed(hass) + await hass.async_block_till_done() + await hass.async_block_till_done() + await hass.async_block_till_done() + await hass.async_block_till_done() + + # Change mocked status + doortag_entity_id = "12:34:56:00:86:99" + doortag_connectivity = True + doortag_opening = doortag_status + doortag_timestamp = int(dt_util.utcnow().timestamp()) + + # Trigger some polling cycle to let status change be picked up + + for _ in range(11): + freezer.tick(timedelta(seconds=30)) + async_fire_time_changed(hass) + await hass.async_block_till_done() + await hass.async_block_till_done() + await hass.async_block_till_done() + await hass.async_block_till_done() + + # Check connectivity mocked state + assert hass.states.get(_doortag_entity_connectivity).state == "on" + # Check opening mocked state + assert hass.states.get(_doortag_entity_opening).state == expected + + +@pytest.mark.parametrize( + ("doortag_category", "expected_key", "expected_class"), + [ + ("door", "door", BinarySensorDeviceClass.DOOR), + ("furniture", "furniture", BinarySensorDeviceClass.OPENING), + ("garage", "garage_door", BinarySensorDeviceClass.GARAGE_DOOR), + ("gate", "gate", BinarySensorDeviceClass.OPENING), + ("other", "opening", BinarySensorDeviceClass.OPENING), + ("window", "window", BinarySensorDeviceClass.WINDOW), + ("invalid_value", "opening", BinarySensorDeviceClass.OPENING), + ], +) +async def test_doortag_opening_category( + hass: HomeAssistant, + config_entry: MockConfigEntry, + netatmo_auth: AsyncMock, + doortag_category: str, + expected_key: str, + expected_class: BinarySensorDeviceClass, +) -> None: + """Test doortag opening status changes.""" + fake_post_hits = 0 + # Repeatedly used variables for the test and initial value from fixture + doortag_entity_id = "12:34:56:00:86:99" + doortag_connectivity = False + doortag_opening = "no_news" + + def tag_modifier(payload): + """This function will be called by common.py during ANY homestatus call.""" + nonlocal doortag_connectivity, doortag_opening + payload["time_server"] = int(dt_util.utcnow().timestamp()) + body = payload.get("body", {}) + + # Handle both structures: {"home": {...}} AND {"homes": [{...}]} + homes_to_check = [] + if "home" in body and isinstance(body["home"], dict): + homes_to_check.append(body["home"]) + elif "homes" in body and isinstance(body["homes"], list): + homes_to_check.extend(body["homes"]) + + for home_data in homes_to_check: + # Safety check: ensure home_data is actually a dictionary + if not isinstance(home_data, dict): + continue + + modules = home_data.get("modules", []) + for module in modules: + if isinstance(module, dict) and module.get("id") == doortag_entity_id: + module["category"] = doortag_category + break + + async def fake_tag_post(*args, **kwargs): + """Fake tag status during requesting backend data.""" + nonlocal fake_post_hits + fake_post_hits += 1 + return await fake_post_request(hass, *args, msg_callback=tag_modifier, **kwargs) + + with ( + patch( + "homeassistant.components.netatmo.api.AsyncConfigEntryNetatmoAuth" + ) as mock_auth, + patch( + "homeassistant.components.netatmo.data_handler.PLATFORMS", + ["camera", "binary_sensor"], + ), + patch( + "homeassistant.components.netatmo.async_get_config_entry_implementation", + return_value=AsyncMock(), + ), + patch( + "homeassistant.components.netatmo.webhook_generate_url", + ) as mock_webhook, + ): + mock_auth.return_value.async_post_api_request.side_effect = fake_tag_post + mock_auth.return_value.async_addwebhook.side_effect = AsyncMock() + mock_auth.return_value.async_dropwebhook.side_effect = AsyncMock() + mock_webhook.return_value = "https://example.com" + assert await hass.config_entries.async_setup(config_entry.entry_id) + + await hass.async_block_till_done() + + webhook_id = config_entry.data[CONF_WEBHOOK_ID] + + # Fake webhook activation + await simulate_webhook(hass, webhook_id, FAKE_WEBHOOK_ACTIVATION) + await hass.async_block_till_done() + + # Define the variables for the test + _doortag_entity = "window_hall" + _doortag_entity_opening = f"binary_sensor.{_doortag_entity}_{expected_key}" + + # Check opening creation with right key + assert hass.states.get(_doortag_entity_opening) is not None + # Check opening device class + assert ( + hass.states.get(_doortag_entity_opening).attributes.get("device_class") + == expected_class.value + ) + # Check opening device name + assert ( + hass.states.get(_doortag_entity_opening).attributes.get("friendly_name") + == _doortag_entity.replace("_", " ").title() + + " " + + expected_key.replace("_", " ").capitalize() + )