feat: add OutputSwitch to control CU outputs (#110)

#110

custom_components/econnect_metronet/__init__.py

@@ -22,7 +22,7 @@
 
 _LOGGER = logging.getLogger(__name__)
 
-PLATFORMS = ["alarm_control_panel", "binary_sensor", "sensor"]
+PLATFORMS = ["alarm_control_panel", "binary_sensor", "sensor", "switch"]
 
 
 async def async_migrate_entry(hass, config: ConfigEntry):

custom_components/econnect_metronet/const.py

@@ -14,6 +14,14 @@
 CONF_AREAS_ARM_VACATION = "areas_arm_vacation"
 CONF_SCAN_INTERVAL = "scan_interval"
 DOMAIN = "econnect_metronet"
+NOTIFICATION_MESSAGE = (
+    "The switch cannot be used because it requires two settings to be configured in the Alarm Panel: "
+    "'manual control' and 'activation without authentication'. "
+    "While these settings can be enabled by your installer, this may not always be the case. "
+    "Please contact your installer for further assistance"
+)
+NOTIFICATION_TITLE = "Unable to toggle the switch"
+NOTIFICATION_IDENTIFIER = "econnect_metronet_output_fail"
 KEY_DEVICE = "device"
 KEY_COORDINATOR = "coordinator"
 KEY_UNSUBSCRIBER = "options_unsubscriber"

custom_components/econnect_metronet/devices.py

@@ -18,6 +18,7 @@
     CONF_AREAS_ARM_HOME,
     CONF_AREAS_ARM_NIGHT,
     CONF_AREAS_ARM_VACATION,
+    NOTIFICATION_MESSAGE,
 )
 
 _LOGGER = logging.getLogger(__name__)
@@ -45,6 +46,7 @@ def __init__(self, connection, config=None):
         self._last_ids = {
             q.SECTORS: 0,
             q.INPUTS: 0,
+            q.OUTPUTS: 0,
             q.ALERTS: 0,
         }
 
@@ -73,6 +75,21 @@ def inputs(self):
         for input_id, item in self._inventory.get(q.INPUTS, {}).items():
             yield input_id, item["name"]
 
+    @property
+    def outputs(self):
+        """Iterate over the device's inventory of outputs.
+        This property provides an iterator over the device's inventory, where each item is a tuple
+        containing the output's ID and its name.
+        Yields:
+            tuple: A tuple where the first item is the output ID and the second item is the output name.
+        Example:
+            >>> device = AlarmDevice()
+            >>> list(device.outputs)
+            [(1, 'Output 1'), (2, 'Output 2')]
+        """
+        for input_id, item in self._inventory.get(q.OUTPUTS, {}).items():
+            yield input_id, item["name"]
+
     @property
     def sectors(self):
         """Iterate over the device's inventory of sectors.
@@ -232,6 +249,7 @@ def update(self):
         try:
             sectors = self._connection.query(q.SECTORS)
             inputs = self._connection.query(q.INPUTS)
+            outputs = self._connection.query(q.OUTPUTS)
             alerts = self._connection.query(q.ALERTS)
         except HTTPError as err:
             _LOGGER.error(f"Device | Error during the update: {err.response.text}")
@@ -243,11 +261,13 @@ def update(self):
         # Update the _inventory
         self._inventory.update({q.SECTORS: sectors["sectors"]})
         self._inventory.update({q.INPUTS: inputs["inputs"]})
+        self._inventory.update({q.OUTPUTS: outputs["outputs"]})
         self._inventory.update({q.ALERTS: alerts["alerts"]})
 
         # Update the _last_ids
         self._last_ids[q.SECTORS] = sectors.get("last_id", 0)
         self._last_ids[q.INPUTS] = inputs.get("last_id", 0)
+        self._last_ids[q.OUTPUTS] = outputs.get("last_id", 0)
         self._last_ids[q.ALERTS] = alerts.get("last_id", 0)
 
         # Update the internal state machine (mapping state)
@@ -284,3 +304,101 @@ def disarm(self, code, sectors=None):
         except CodeError as err:
             _LOGGER.error(f"Device | Credentials (alarm code) is incorrect: {err}")
             raise err
+
+    def turn_off(self, output):
+        """
+        Turn off a specified output.
+
+        Args:
+            output: The ID of the output to be turned off.
+
+        Raises:
+            HTTPError: If there is an error in the HTTP request to turn off the output.
+
+        Notes:
+            - The `element` is the sector ID used to arm/disarm the sector.
+            - This method checks for authentication requirements and control permissions
+                before attempting to turn off the output.
+            - If the output can't be manually controlled or if authentication is required but not provided,
+                appropriate error messages are logged.
+
+        Example:
+            To turn off an output with ID '1', use:
+            >>> device_instance.turn_off(1)
+        """
+        for id, item in self.items(q.OUTPUTS):
+            # Skip if it's not matching the output
+            if id != output:
+                continue
+
+            # If the output isn't manual controllable by users write an error il log
+            if item.get("control_denied_to_users"):
+                _LOGGER.warning(
+                    f"Device | Error while turning off output: {item.get('name')}, Can't be manual controlled"
+                )
+                _LOGGER.warning(NOTIFICATION_MESSAGE)
+                break
+
+            # If the output require authentication for control write an error il log
+            if not item.get("do_not_require_authentication"):
+                _LOGGER.warning(f"Device | Error while turning off output: {item.get('name')}, Required authentication")
+                _LOGGER.warning(NOTIFICATION_MESSAGE)
+                break
+
+            try:
+                element_id = item.get("element")
+                self._connection.turn_off(element_id)
+                return True
+            except HTTPError as err:
+                _LOGGER.error(f"Device | Error while turning off output: {err.response.text}")
+                raise err
+        return False
+
+    def turn_on(self, output):
+        """
+        Turn on a specified output.
+
+        Args:
+            output: The ID of the output to be turned on.
+
+        Raises:
+            HTTPError: If there is an error in the HTTP request to turn on the output.
+
+        Notes:
+            - The `element` is the sector ID used to arm/disarm the sector.
+            - This method checks for authentication requirements and control permissions
+                before attempting to turn on the output.
+            - If the output can't be manually controlled or if authentication is required but not provided,
+                appropriate error messages are logged.
+
+        Example:
+            To turn on an output with ID '1', use:
+            >>> device_instance.turn_on(1)
+        """
+        for id, item in self.items(q.OUTPUTS):
+            # Skip if it's not matching the output
+            if id != output:
+                continue
+
+            # If the output isn't manual controllable by users write an error log
+            if item.get("control_denied_to_users"):
+                _LOGGER.warning(
+                    f"Device | Error while turning on output: {item.get('name')}, Can't be manual controlled"
+                )
+                _LOGGER.warning(NOTIFICATION_MESSAGE)
+                break
+
+            # If the output require authentication for control write an error log
+            if not item.get("do_not_require_authentication"):
+                _LOGGER.warning(f"Device | Error while turning on output: {item.get('name')}, Required authentication")
+                _LOGGER.warning(NOTIFICATION_MESSAGE)
+                break
+
+            try:
+                element_id = item.get("element")
+                self._connection.turn_on(element_id)
+                return True
+            except HTTPError as err:
+                _LOGGER.error(f"Device | Error while turning on outputs: {err.response.text}")
+                raise err
+        return False

custom_components/econnect_metronet/switch.py

@@ -0,0 +1,101 @@
+from elmo import query as q
+from homeassistant.components import persistent_notification
+from homeassistant.components.switch import SwitchEntity
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import (
+    CoordinatorEntity,
+    DataUpdateCoordinator,
+)
+
+from .const import (
+    CONF_EXPERIMENTAL,
+    CONF_FORCE_UPDATE,
+    DOMAIN,
+    KEY_COORDINATOR,
+    KEY_DEVICE,
+    NOTIFICATION_IDENTIFIER,
+    NOTIFICATION_MESSAGE,
+    NOTIFICATION_TITLE,
+)
+from .devices import AlarmDevice
+from .helpers import generate_entity_id
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    device = hass.data[DOMAIN][entry.entry_id][KEY_DEVICE]
+    coordinator = hass.data[DOMAIN][entry.entry_id][KEY_COORDINATOR]
+    outputs = []
+
+    # Iterate through the outputs of the provided device and create OutputSwitch objects
+    for output_id, name in device.outputs:
+        unique_id = f"{entry.entry_id}_{DOMAIN}_{q.OUTPUTS}_{output_id}"
+        outputs.append(OutputSwitch(unique_id, output_id, entry, name, coordinator, device))
+
+    async_add_entities(outputs)
+
+
+class OutputSwitch(CoordinatorEntity, SwitchEntity):
+    """Representation of a e-connect output switch."""
+
+    _attr_has_entity_name = True
+
+    def __init__(
+        self,
+        unique_id: str,
+        output_id: int,
+        config: ConfigEntry,
+        name: str,
+        coordinator: DataUpdateCoordinator,
+        device: AlarmDevice,
+    ) -> None:
+        """Construct."""
+        # Enable experimental settings from the configuration file
+        experimental = coordinator.hass.data[DOMAIN].get(CONF_EXPERIMENTAL, {})
+        self._attr_force_update = experimental.get(CONF_FORCE_UPDATE, False)
+
+        super().__init__(coordinator)
+        self.entity_id = generate_entity_id(config, name)
+        self._name = name
+        self._device = device
+        self._unique_id = unique_id
+        self._output_id = output_id
+
+    @property
+    def unique_id(self) -> str:
+        """Return the unique identifier."""
+        return self._unique_id
+
+    @property
+    def name(self) -> str:
+        """Return the name of this entity."""
+        return self._name
+
+    @property
+    def icon(self) -> str:
+        """Return the icon used by this entity."""
+        return "hass:toggle-switch-variant"
+
+    @property
+    def is_on(self) -> bool:
+        """Return the switch status (on/off)."""
+        return bool(self._device.get_status(q.OUTPUTS, self._output_id))
+
+    async def async_turn_off(self):
+        """Turn the entity off."""
+        if not await self.hass.async_add_executor_job(self._device.turn_off, self._output_id):  # pragma: no cover
+            persistent_notification.async_create(
+                self.hass, NOTIFICATION_MESSAGE, NOTIFICATION_TITLE, NOTIFICATION_IDENTIFIER
+            )
+
+    async def async_turn_on(self):
+        """Turn the entity off."""
+        if not await self.hass.async_add_executor_job(self._device.turn_on, self._output_id):  # pragma: no cover
+            persistent_notification.async_create(
+                self.hass, NOTIFICATION_MESSAGE, NOTIFICATION_TITLE, NOTIFICATION_IDENTIFIER
+            )

pyproject.toml

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

tests/conftest.py

@@ -127,6 +127,7 @@ def client(socket_enabled):
         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)
+        server.add(responses.POST, "https://example.com/api/outputs", body=r.OUTPUTS, status=200)
         server.add(responses.POST, "https://example.com/api/statusadv", body=r.STATUS, status=200)
         yield client