refactor: add device.item to query the inventory

custom_components/econnect_metronet/binary_sensor.py

@@ -38,11 +38,10 @@ async def async_setup_entry(
         unique_id = f"{entry.entry_id}_{DOMAIN}_{query.INPUTS}_{sensor_id}"
         sensors.append(InputSensor(unique_id, sensor_id, entry, name, coordinator, device))
 
-    # Retrieve alarm system global status
-    alerts = await hass.async_add_executor_job(device._connection.get_status)
-    for alert_id, _ in alerts.items():
-        unique_id = f"{entry.entry_id}_{DOMAIN}_{alert_id}"
-        sensors.append(AlertSensor(unique_id, alert_id, entry, coordinator, device))
+    # Iterate through the alerts of the provided device and create AlertSensor objects
+    for alert_id, name in device.alerts:
+        unique_id = f"{entry.entry_id}_{DOMAIN}_{query.ALERTS}_{alert_id}"
+        sensors.append(AlertSensor(unique_id, alert_id, entry, name, coordinator, device))
 
     async_add_entities(sensors)
 
@@ -55,17 +54,19 @@ class AlertSensor(CoordinatorEntity, BinarySensorEntity):
     def __init__(
         self,
         unique_id: str,
-        alert_id: str,
+        alert_id: int,
         config: ConfigEntry,
+        name: str,
         coordinator: DataUpdateCoordinator,
         device: AlarmDevice,
     ) -> None:
         """Construct."""
         super().__init__(coordinator)
-        self.entity_id = generate_entity_id(config, alert_id)
+        self.entity_id = generate_entity_id(config, name)
+        self._name = name
+        self._device = device
         self._unique_id = unique_id
         self._alert_id = alert_id
-        self._device = device
 
     @property
     def unique_id(self) -> str:
@@ -75,7 +76,7 @@ def unique_id(self) -> str:
     @property
     def translation_key(self) -> str:
         """Return the translation key to translate the entity's name and states."""
-        return self._alert_id
+        return self._name
 
     @property
     def icon(self) -> str:
@@ -89,9 +90,14 @@ def device_class(self) -> str:
 
     @property
     def is_on(self) -> bool:
-        """Return the sensor status (on/off)."""
-        state = self._device.alerts.get(self._alert_id)
-        return state > 1 if self._alert_id == "anomalies_led" else state > 0
+        """Return the binary sensor status (on/off)."""
+        for item_id, item in self._device.items(query.ALERTS):
+            if item_id == self._alert_id:
+                if item_id == 1:
+                    return True if item.get("status") > 1 else False
+                else:
+                    return bool(item.get("status", False))
+        return False
 
 
 class InputSensor(CoordinatorEntity, BinarySensorEntity):

custom_components/econnect_metronet/devices.py

@@ -53,7 +53,6 @@ def __init__(self, connection, config=None):
 
         # Alarm state
         self.state = STATE_UNAVAILABLE
-        self.alerts = {}
         self.sectors_armed = {}
         self.sectors_disarmed = {}
         self.inputs_alerted = {}
@@ -62,17 +61,69 @@ def __init__(self, connection, config=None):
 
     @property
     def inputs(self):
-        for input_id, item in self._inventory.get("inputs", {}).items():
+        """Iterate over the device's inventory of inputs.
+        This property provides an iterator over the device's inventory, where each item is a tuple
+        containing the input's ID and its name.
+        Yields:
+            tuple: A tuple where the first item is the input ID and the second item is the input name.
+        Example:
+            >>> device = AlarmDevice()
+            >>> list(device.inputs)
+            [(1, 'Front Door'), (2, 'Back Door')]
+        """
+        for input_id, item in self._inventory.get(q.INPUTS, {}).items():
             yield input_id, item["name"]
 
     @property
     def sectors(self):
-        for sector_id, item in self._inventory.get("sectors", {}).items():
+        """Iterate over the device's inventory of sectors.
+        This property provides an iterator over the device's inventory, where each item is a tuple
+        containing the sectors's ID and its name.
+        Yields:
+            tuple: A tuple where the first item is the sector ID and the second item is the sector name.
+        Example:
+            >>> device = AlarmDevice()
+            >>> list(device.sectors)
+            [(1, 'S1 Living Room'), (2, 'S2 Bedroom')]
+        """
+        for sector_id, item in self._inventory.get(q.SECTORS, {}).items():
             yield sector_id, item["name"]
 
     @property
-    def alerts_v2(self):
-        yield from self._inventory.get("alerts", {}).items()
+    def alerts(self):
+        """Iterate over the device's inventory of alerts.
+        This property provides an iterator over the device's inventory, where each item is a tuple
+        containing the alerts's ID and its name.
+        Yields:
+            tuple: A tuple where the first item is the alert ID and the second item is the alert name.
+        Example:
+            >>> device = AlarmDevice()
+            >>> list(device.alerts)
+            [{1: "alarm_led", 2: "anomalies_led"}]
+        """
+        # yield from self._inventory.get(q.ALERTS, {}).items()
+        for alert_id, item in self._inventory.get(q.ALERTS, {}).items():
+            yield alert_id, item["name"]
+
+    def items(self, query, status=None):
+        """Iterate over items from the device's inventory based on a query and optional status filter.
+        This method provides an iterator over the items in the device's inventory that match the given query.
+        If a status is provided, only items with that status will be yielded.
+        Args:
+            query (str): The query string to match against items in the inventory.
+            status (Optional[Any]): An optional status filter. If provided, only items with this status
+                                    will be yielded. Defaults to None, which means all items matching the query
+                                    will be yielded regardless of their status.
+        Yields:
+            tuple: A tuple where the first item is the item ID and the second item is the item dictionary.
+        Example:
+            >>> device = AlarmDevice()
+            >>> list(device.items('door', status=True))
+            [(1, {'name': 'Front Door', 'status': True, ...})]
+        """
+        for item_id, item in self._inventory.get(query, {}).items():
+            if status is None or item.get("status") == status:
+                yield item_id, item
 
     def connect(self, username, password):
         """Establish a connection with the E-connect backend, to retrieve an access
@@ -107,7 +158,7 @@ def has_updates(self):
             dict: Dictionary with the updates if any, based on the last known IDs.
         """
         try:
-            self._connection.get_status()
+            self._connection.query(q.ALERTS)
             return self._connection.poll({key: value for key, value in self._last_ids.items()})
         except HTTPError as err:
             _LOGGER.error(f"Device | Error while polling for updates: {err.response.text}")
@@ -174,7 +225,7 @@ def update(self):
         try:
             sectors = self._connection.query(q.SECTORS)
             inputs = self._connection.query(q.INPUTS)
-            alerts = self._connection.get_status()
+            alerts = self._connection.query(q.ALERTS)
         except HTTPError as err:
             _LOGGER.error(f"Device | Error during the update: {err.response.text}")
             raise err
@@ -183,9 +234,9 @@ def update(self):
             raise err
 
         # Update the _inventory
-        self._inventory.update({"inputs": inputs["inputs"]})
-        self._inventory.update({"sectors": sectors["sectors"]})
-        self._inventory.update({"alerts": alerts})
+        self._inventory.update({q.SECTORS: sectors["sectors"]})
+        self._inventory.update({q.INPUTS: inputs["inputs"]})
+        self._inventory.update({q.ALERTS: alerts})
 
         # Filter sectors and inputs
         self.sectors_armed = _filter_data(sectors, "sectors", True)
@@ -196,9 +247,6 @@ def update(self):
         self._last_ids[q.SECTORS] = sectors.get("last_id", 0)
         self._last_ids[q.INPUTS] = inputs.get("last_id", 0)
 
-        # Update system alerts
-        self.alerts = alerts
-
         # Update the internal state machine (mapping state)
         self.state = self.get_state()
 

pyproject.toml

@@ -31,7 +31,7 @@ classifiers = [
   "Programming Language :: Python :: Implementation :: CPython",
 ]
 dependencies = [
-  "econnect-python==0.7.0",
+  "econnect-python @ git+https://github.com/palazzem/econnect-python@26b4d8a01e0f29998d98d3f3dc28a509eebf44aa",
   "homeassistant",
 ]
 

tests/fixtures/responses.py

@@ -265,7 +265,7 @@ def test_client_get_sectors_status(server):
             "InputBypass": 0,
             "InputLowBattery": 0,
             "InputNoSupervision": 0,
-            "DeviceTamper": 0,
+            "DeviceTamper": 1,
             "DeviceFailure": 0,
             "DeviceNoPower": 0,
             "DeviceLowBattery": 0,

tests/test_binary_sensors.py

@@ -10,36 +10,64 @@
 
 
 class TestAlertSensor:
+    def test_binary_sensor_is_on(self, hass, config_entry, alarm_device):
+        alarm_device.connect("username", "password")
+        alarm_device.update()
+        coordinator = DataUpdateCoordinator(hass, logging.getLogger(__name__), name="econnect_metronet")
+        entity = AlertSensor("device_tamper", 7, config_entry, "device_tamper", coordinator, alarm_device)
+        assert entity.is_on is True
+
+    def test_binary_sensor_is_off(self, hass, config_entry, alarm_device):
+        alarm_device.connect("username", "password")
+        alarm_device.update()
+        coordinator = DataUpdateCoordinator(hass, logging.getLogger(__name__), name="econnect_metronet")
+        entity = AlertSensor("device_failure", 2, config_entry, "device_failure", coordinator, alarm_device)
+        assert entity.is_on is False
+
+    def test_binary_sensor_missing(self, hass, config_entry, alarm_device):
+        alarm_device.connect("username", "password")
+        alarm_device.update()
+        coordinator = DataUpdateCoordinator(hass, logging.getLogger(__name__), name="econnect_metronet")
+        entity = AlertSensor("test_id", 1000, config_entry, "test_id", coordinator, alarm_device)
+        assert entity.is_on is False
+
+    def test_binary_sensor_anomalies_led_is_off(self, hass, config_entry, alarm_device):
+        alarm_device.connect("username", "password")
+        alarm_device.update()
+        coordinator = DataUpdateCoordinator(hass, logging.getLogger(__name__), name="econnect_metronet")
+        entity = AlertSensor("anomalies_led", 1, config_entry, "anomalies_led", coordinator, alarm_device)
+        assert entity.is_on is False
+
     def test_binary_sensor_name(hass, config_entry, alarm_device):
         # Ensure the alert has the right translation key
         coordinator = DataUpdateCoordinator(hass, logging.getLogger(__name__), name="econnect_metronet")
-        entity = AlertSensor("test_id", "has_anomalies", config_entry, coordinator, alarm_device)
+        entity = AlertSensor("test_id", 0, config_entry, "has_anomalies", coordinator, alarm_device)
         assert entity.translation_key == "has_anomalies"
 
     def test_binary_sensor_name_with_system_name(hass, config_entry, alarm_device):
         # The system name doesn't change the translation key
         config_entry.data["system_name"] = "Home"
         coordinator = DataUpdateCoordinator(hass, logging.getLogger(__name__), name="econnect_metronet")
-        entity = AlertSensor("test_id", "has_anomalies", config_entry, coordinator, alarm_device)
+        entity = AlertSensor("test_id", 0, config_entry, "has_anomalies", coordinator, alarm_device)
         assert entity.translation_key == "has_anomalies"
 
     def test_binary_sensor_entity_id(hass, config_entry, alarm_device):
         # Ensure the alert has a valid Entity ID
         coordinator = DataUpdateCoordinator(hass, logging.getLogger(__name__), name="econnect_metronet")
-        entity = AlertSensor("test_id", "has_anomalies", config_entry, coordinator, alarm_device)
+        entity = AlertSensor("test_id", 0, config_entry, "has_anomalies", coordinator, alarm_device)
         assert entity.entity_id == "econnect_metronet.econnect_metronet_test_user_has_anomalies"
 
     def test_binary_sensor_entity_id_with_system_name(hass, config_entry, alarm_device):
         # Ensure the Entity ID takes into consideration the system name
         config_entry.data["system_name"] = "Home"
         coordinator = DataUpdateCoordinator(hass, logging.getLogger(__name__), name="econnect_metronet")
-        entity = AlertSensor("test_id", "has_anomalies", config_entry, coordinator, alarm_device)
+        entity = AlertSensor("test_id", 0, config_entry, "has_anomalies", coordinator, alarm_device)
         assert entity.entity_id == "econnect_metronet.econnect_metronet_home_has_anomalies"
 
     def test_binary_sensor_icon(hass, config_entry, alarm_device):
         # Ensure the sensor has the right icon
         coordinator = DataUpdateCoordinator(hass, logging.getLogger(__name__), name="econnect_metronet")
-        entity = AlertSensor("test_id", "has_anomalies", config_entry, coordinator, alarm_device)
+        entity = AlertSensor("test_id", 0, config_entry, "has_anomalies", coordinator, alarm_device)
         assert entity.icon == "hass:alarm-light"
 
 

tests/test_coordinator.py

@@ -1,6 +1,7 @@
 from datetime import timedelta
 
 import pytest
+from elmo import query as q
 from elmo.api.exceptions import CredentialError, InvalidToken
 from homeassistant.exceptions import ConfigEntryNotReady
 from requests.exceptions import HTTPError
@@ -34,39 +35,39 @@ async def test_coordinator_async_update_with_data(mocker, coordinator):
     # Test
     await coordinator.async_refresh()
     assert coordinator.data == {
-        "alerts": {
-            "alarm_led": 0,
-            "anomalies_led": 1,
-            "device_failure": 0,
-            "device_low_battery": 0,
-            "device_no_power": 0,
-            "device_no_supervision": 0,
-            "device_system_block": 0,
-            "device_tamper": 0,
-            "gsm_anomaly": 0,
-            "gsm_low_balance": 0,
-            "has_anomaly": False,
-            "input_alarm": 0,
-            "input_bypass": 0,
-            "input_failure": 0,
-            "input_low_battery": 0,
-            "input_no_supervision": 0,
-            "inputs_led": 2,
-            "module_registration": 0,
-            "panel_low_battery": 0,
-            "panel_no_power": 0,
-            "panel_tamper": 0,
-            "pstn_anomaly": 0,
-            "rf_interference": 0,
-            "system_test": 0,
-            "tamper_led": 0,
+        q.ALERTS: {
+            0: {"name": "alarm_led", "status": 0},
+            1: {"name": "anomalies_led", "status": 1},
+            2: {"name": "device_failure", "status": 0},
+            3: {"name": "device_low_battery", "status": 0},
+            4: {"name": "device_no_power", "status": 0},
+            5: {"name": "device_no_supervision", "status": 0},
+            6: {"name": "device_system_block", "status": 0},
+            7: {"name": "device_tamper", "status": 1},
+            8: {"name": "gsm_anomaly", "status": 0},
+            9: {"name": "gsm_low_balance", "status": 0},
+            10: {"name": "has_anomaly", "status": False},
+            11: {"name": "input_alarm", "status": 0},
+            12: {"name": "input_bypass", "status": 0},
+            13: {"name": "input_failure", "status": 0},
+            14: {"name": "input_low_battery", "status": 0},
+            15: {"name": "input_no_supervision", "status": 0},
+            16: {"name": "inputs_led", "status": 2},
+            17: {"name": "module_registration", "status": 0},
+            18: {"name": "panel_low_battery", "status": 0},
+            19: {"name": "panel_no_power", "status": 0},
+            20: {"name": "panel_tamper", "status": 0},
+            21: {"name": "pstn_anomaly", "status": 0},
+            22: {"name": "rf_interference", "status": 0},
+            23: {"name": "system_test", "status": 0},
+            24: {"name": "tamper_led", "status": 0},
         },
-        "inputs": {
+        q.INPUTS: {
             0: {"element": 1, "excluded": False, "id": 1, "index": 0, "name": "Entryway Sensor", "status": True},
             1: {"element": 2, "excluded": False, "id": 2, "index": 1, "name": "Outdoor Sensor 1", "status": True},
             2: {"element": 3, "excluded": True, "id": 3, "index": 2, "name": "Outdoor Sensor 2", "status": False},
         },
-        "sectors": {
+        q.SECTORS: {
             0: {"element": 1, "excluded": False, "id": 1, "index": 0, "name": "S1 Living Room", "status": True},
             1: {"element": 2, "excluded": False, "id": 2, "index": 1, "name": "S2 Bedroom", "status": True},
             2: {"element": 3, "excluded": False, "id": 3, "index": 2, "name": "S3 Outdoor", "status": False},