refactor: add device.item to query the inventory

custom_components/econnect_metronet/devices.py

@@ -62,17 +62,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 input 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()
+        """Generate a sequence of alerts from the device's inventory.
+
+        This function yields key-value pairs representing alerts obtained from the device's inventory.
+
+        Yields:
+            tuple: A tuple where the first item is the alert name and the second item is the alert status.
+
+        Example:
+            >>> device = AlarmDevice()
+            >>> list(device.alerts_v2)
+            [{"alarm_led": 0, "anomalies_led": 1}]
+        """
+        yield from self._inventory.get(q.ALERTS, {}).items()
+
+    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 +159,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 +226,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 +235,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)

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,7 +35,7 @@ async def test_coordinator_async_update_with_data(mocker, coordinator):
     # Test
     await coordinator.async_refresh()
     assert coordinator.data == {
-        "alerts": {
+        q.ALERTS: {
             "alarm_led": 0,
             "anomalies_led": 1,
             "device_failure": 0,
@@ -61,12 +62,12 @@ async def test_coordinator_async_update_with_data(mocker, coordinator):
             "system_test": 0,
             "tamper_led": 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},

tests/test_devices.py

@@ -60,6 +60,36 @@ def test_device_constructor_with_config(client):
     assert device.inputs_wait == {}
 
 
+def test_item_query_without_status(alarm_device):
+    """Verify that querying items without specifying a status works correctly"""
+    alarm_device.connect("username", "password")
+    # Test
+    alarm_device.update()
+    assert dict(alarm_device.items(q.SECTORS)) == alarm_device._inventory.get(q.SECTORS)
+
+
+def test_item_query_with_status(alarm_device):
+    """Verify that querying items with specifying a status works correctly"""
+    alarm_device.connect("username", "password")
+    # Test
+    alarm_device.update()
+    items = {2: {"element": 3, "excluded": False, "id": 3, "index": 2, "name": "S3 Outdoor", "status": False}}
+    assert dict(alarm_device.items(q.SECTORS, status=False)) == items
+
+
+def test_item_query_without_inventory(alarm_device):
+    """Verify that querying items without inventory populated works correctly"""
+    alarm_device._inventory = {q.SECTORS: {}, q.INPUTS: {}, q.ALERTS: {}}
+    assert dict(alarm_device.items(q.SECTORS, status=False)) == {}
+
+
+def test_item_with_empty_query(alarm_device):
+    """Verify that querying items with empty query works correctly"""
+    alarm_device._inventory = {q.SECTORS: {}}
+    # Test
+    assert dict(alarm_device.items(q.SECTORS, status=False)) == {}
+
+
 def test_device_connect(client, mocker):
     """Should call authentication endpoints and update internal state."""
     device = AlarmDevice(client)
@@ -198,7 +228,7 @@ def test_device_update_success(client, mocker):
     device.connect("username", "password")
     # Test
     device.update()
-    assert device._connection.query.call_count == 2
+    assert device._connection.query.call_count == 3
     assert device.sectors_armed == sectors_armed
     assert device.sectors_disarmed == sectors_disarmed
     assert device.inputs_alerted == inputs_alerted
@@ -215,17 +245,17 @@ def test_device_inventory_update_success(client, mocker):
     device = AlarmDevice(client)
     mocker.spy(device._connection, "query")
     inventory = {
-        "sectors": {
+        q.SECTORS: {
             0: {"id": 1, "index": 0, "element": 1, "excluded": False, "status": True, "name": "S1 Living Room"},
             1: {"id": 2, "index": 1, "element": 2, "excluded": False, "status": True, "name": "S2 Bedroom"},
             2: {"id": 3, "index": 2, "element": 3, "excluded": False, "status": False, "name": "S3 Outdoor"},
         },
-        "inputs": {
+        q.INPUTS: {
             0: {"id": 1, "index": 0, "element": 1, "excluded": False, "status": True, "name": "Entryway Sensor"},
             1: {"id": 2, "index": 1, "element": 2, "excluded": False, "status": True, "name": "Outdoor Sensor 1"},
             2: {"id": 3, "index": 2, "element": 3, "excluded": True, "status": False, "name": "Outdoor Sensor 2"},
         },
-        "alerts": {
+        q.ALERTS: {
             "alarm_led": 0,
             "anomalies_led": 1,
             "device_failure": 0,
@@ -409,6 +439,36 @@ def test_device_update_state_machine_armed(client, mocker):
                 2: {"id": 3, "index": 2, "element": 3, "excluded": True, "status": False, "name": "Outdoor Sensor 2"},
             },
         },
+        {
+            "last_id": 3,
+            "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,
+            },
+        },
     ]
     # Test
     device.update()
@@ -437,6 +497,36 @@ def test_device_update_state_machine_disarmed(client, mocker):
                 2: {"id": 3, "index": 2, "element": 3, "excluded": True, "status": False, "name": "Outdoor Sensor 2"},
             },
         },
+        {
+            "last_id": 3,
+            "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,
+            },
+        },
     ]
     # Test
     device.update()