feat: add AlarmDevice component (#28)

#28

custom_components/econnect_alarm/__init__.py

@@ -6,7 +6,6 @@
 import async_timeout
 from elmo.api.client import ElmoClient
 from elmo.api.exceptions import InvalidToken
-from elmo.devices import AlarmDevice
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
 from homeassistant.core import HomeAssistant
@@ -22,6 +21,7 @@
     POLLING_TIMEOUT,
     SCAN_INTERVAL,
 )
+from .devices import AlarmDevice
 
 _LOGGER = logging.getLogger(__name__)
 

custom_components/econnect_alarm/devices.py

@@ -0,0 +1,150 @@
+import logging
+
+from elmo import query as q
+from elmo.api.exceptions import CodeError, CredentialError, LockError, ParseError
+from elmo.utils import _filter_data
+from homeassistant.const import (
+    STATE_ALARM_ARMED_AWAY,
+    STATE_ALARM_DISARMED,
+    STATE_UNAVAILABLE,
+)
+from requests.exceptions import HTTPError
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class AlarmDevice:
+    """AlarmDevice class represents an e-connect alarm system. This method wraps around
+    a connector object (e.g. `ElmoClient`) so that the client can be stateless and just
+    return data, while this class persists the status of the alarm.
+
+    Usage:
+        # Initialization
+        conn = ElmoClient()
+        device = AlarmDevice(conn)
+
+        # Connect automatically grab the latest status
+        device.connect("username", "password")
+        print(device.state)
+    """
+
+    def __init__(self, connection):
+        # Configuration and internals
+        self._connection = connection
+        self._lastIds = {
+            q.SECTORS: 0,
+            q.INPUTS: 0,
+        }
+
+        # Alarm state
+        self.state = STATE_UNAVAILABLE
+        self.sectors_armed = {}
+        self.sectors_disarmed = {}
+        self.inputs_alerted = {}
+        self.inputs_wait = {}
+
+    def connect(self, username, password):
+        """Establish a connection with the E-connect backend, to retrieve an access
+        token. This method stores the `session_id` within the `ElmoClient` object
+        and is used automatically when other methods are called.
+
+        When a connection is successfully established, the device automatically
+        updates the status calling `self.update()`.
+        """
+        try:
+            self._connection.auth(username, password)
+        except HTTPError as err:
+            _LOGGER.error(f"Device | Error while authenticating with e-Connect: {err}")
+            raise err
+        except CredentialError as err:
+            _LOGGER.error(f"Device | Username or password are not correct: {err}")
+            raise err
+
+    def has_updates(self):
+        """Use the connection to detect a possible change. This is a blocking call
+        that must not be called in the main thread. Check `ElmoClient.poll()` method
+        for more details.
+
+        Values passed to `ElmoClient.poll()` are the last known IDs for sectors and
+        inputs. A new dictionary is sent to avoid the underlying client to mutate
+        the device internal state.
+        """
+        try:
+            return self._connection.poll({x: y for x, y in self._lastIds.items()})
+        except HTTPError as err:
+            _LOGGER.error(f"Device | Error while checking if there are updates: {err}")
+            raise err
+        except ParseError as err:
+            _LOGGER.error(f"Device | Error parsing the poll response: {err}")
+            raise err
+
+    def update(self):
+        """Updates the internal state of the device based on the latest data.
+
+        This method performs the following actions:
+        1. Queries for the latest sectors and inputs using the internal connection.
+        2. Filters the retrieved sectors and inputs to categorize them based on their status.
+        3. Updates the last known IDs for sectors and inputs.
+        4. Updates internal state for sectors' and inputs' statuses.
+
+        Raises:
+            HTTPError: If there's an error while making the HTTP request.
+            ParseError: If there's an error while parsing the response.
+
+        Attributes updated:
+            sectors_armed (dict): A dictionary of sectors that are armed.
+            sectors_disarmed (dict): A dictionary of sectors that are disarmed.
+            inputs_alerted (dict): A dictionary of inputs that are in an alerted state.
+            inputs_wait (dict): A dictionary of inputs that are in a wait state.
+            _lastIds (dict): Updated last known IDs for sectors and inputs.
+            state (str): Updated internal state of the device.
+        """
+        # Retrieve sectors and inputs
+        try:
+            sectors = self._connection.query(q.SECTORS)
+            inputs = self._connection.query(q.INPUTS)
+        except (HTTPError, ParseError) as err:
+            _LOGGER.error(f"Device | Error while checking if there are updates: {err}")
+            raise
+
+        # Filter sectors and inputs
+        self.sectors_armed = _filter_data(sectors, "sectors", True)
+        self.sectors_disarmed = _filter_data(sectors, "sectors", False)
+        self.inputs_alerted = _filter_data(inputs, "inputs", True)
+        self.inputs_wait = _filter_data(inputs, "inputs", False)
+
+        self._lastIds[q.SECTORS] = sectors.get("last_id", 0)
+        self._lastIds[q.INPUTS] = inputs.get("last_id", 0)
+
+        # Update the internal state machine
+        self.state = STATE_ALARM_ARMED_AWAY if self.sectors_armed else STATE_ALARM_DISARMED
+
+    def arm(self, code, sectors=None):
+        try:
+            with self._connection.lock(code):
+                self._connection.arm(sectors=sectors)
+                self.state = STATE_ALARM_ARMED_AWAY
+        except HTTPError as err:
+            _LOGGER.error(f"Device | Error while arming the system: {err}")
+            raise err
+        except LockError as err:
+            _LOGGER.error(f"Device | Error while acquiring the system lock: {err}")
+            raise err
+        except CodeError as err:
+            _LOGGER.error(f"Device | Credentials (alarm code) is incorrect: {err}")
+            raise err
+
+    def disarm(self, code, sectors=None):
+        try:
+            with self._connection.lock(code):
+                self._connection.disarm(sectors=sectors)
+                self.state = STATE_ALARM_DISARMED
+        except HTTPError as err:
+            _LOGGER.error(f"Device | Error while disarming the system: {err}")
+            raise err
+        except LockError as err:
+            _LOGGER.error(f"Device | Error while acquiring the system lock: {err}")
+            raise err
+        except CodeError as err:
+            _LOGGER.error(f"Device | Credentials (alarm code) is incorrect: {err}")
+            raise err

custom_components/econnect_alarm/manifest.json

@@ -15,7 +15,7 @@
     "custom_components.econnect_alarm"
   ],
   "requirements": [
-    "econnect-python==0.5.1"
+    "econnect-python==0.6.0"
   ],
   "version": "0.2.0"
 }

pyproject.toml

@@ -28,7 +28,7 @@ classifiers = [
 ]
 dependencies = [
   "homeassistant",
-  "econnect-python==0.5.1",
+  "econnect-python==0.6.0",
 ]
 
 [project.optional-dependencies]
@@ -39,6 +39,7 @@ dev = [
   "pytest",
   "pytest-cov",
   "pytest-mock",
+  "responses",
   "tox",
   # Home Assistant fixtures
   "freezegun",

tests/conftest.py

@@ -1,4 +1,8 @@
 import pytest
+import responses
+from elmo.api.client import ElmoClient
+
+from .fixtures import responses as r
 
 pytest_plugins = ["tests.hass.fixtures"]
 
@@ -7,3 +11,30 @@
 async def hass(hass):
     hass.data["custom_components"] = None
     yield hass
+
+
+@pytest.fixture(scope="function")
+def client():
+    """Creates an instance of `ElmoClient` which emulates the behavior of a real client for
+    testing purposes.
+
+    Although this client instance operates with mocked calls, it is designed to function as
+    if it were genuine. This ensures that the client's usage in tests accurately mirrors how it
+    would be employed in real scenarios.
+
+    Use it for integration tests where a realistic interaction with the `ElmoClient` is required
+    without actual external calls.
+    """
+    client = ElmoClient(base_url="https://example.com", domain="domain")
+    with responses.RequestsMock(assert_all_requests_are_fired=False) as server:
+        server.add(responses.GET, "https://example.com/api/login", body=r.LOGIN, status=200)
+        server.add(responses.POST, "https://example.com/api/updates", body=r.UPDATES, status=200)
+        server.add(responses.POST, "https://example.com/api/panel/syncLogin", body=r.SYNC_LOGIN, status=200)
+        server.add(responses.POST, "https://example.com/api/panel/syncLogout", body=r.SYNC_LOGOUT, status=200)
+        server.add(
+            responses.POST, "https://example.com/api/panel/syncSendCommand", body=r.SYNC_SEND_COMMAND, status=200
+        )
+        server.add(responses.POST, "https://example.com/api/strings", body=r.STRINGS, status=200)
+        server.add(responses.POST, "https://example.com/api/areas", body=r.AREAS, status=200)
+        server.add(responses.POST, "https://example.com/api/inputs", body=r.INPUTS, status=200)
+        yield client