Water valve support

pyalarmdotcomajax/devices/__init__.py

@@ -38,6 +38,7 @@ class DeviceType(ExtendedEnumMixin):
     SYSTEM = "systems"
     THERMOSTAT = "thermostats"
     WATER_SENSOR = "waterSensors"
+    WATER_VALVE = "waterValves"
 
     # Unsupported
     ACCESS_CONTROL = "accessControlAccessPointDevices"
@@ -56,7 +57,6 @@ class DeviceType(ExtendedEnumMixin):
     SWITCH = "switches"
     VALVE_SWITCH = "valveSwitches"
     WATER_METER = "waterMeters"
-    WATER_VALVE = "waterValves"
     X10_LIGHT = "x10Lights"
 
 

pyalarmdotcomajax/devices/registry.py

@@ -6,7 +6,6 @@
 from dataclasses import dataclass, field
 from typing import TypedDict
 
-from pyalarmdotcomajax.devices import DeviceType
 from pyalarmdotcomajax.devices.camera import Camera
 from pyalarmdotcomajax.devices.garage_door import GarageDoor
 from pyalarmdotcomajax.devices.gate import Gate
@@ -18,9 +17,12 @@
 from pyalarmdotcomajax.devices.system import System
 from pyalarmdotcomajax.devices.thermostat import Thermostat
 from pyalarmdotcomajax.devices.water_sensor import WaterSensor
+from pyalarmdotcomajax.devices.water_valve import WaterValve
 from pyalarmdotcomajax.exceptions import UnkonwnDevice, UnsupportedDeviceType
 from pyalarmdotcomajax.helpers import classproperty
 
+from . import DeviceType
+
 log = logging.getLogger(__name__)
 
 AllDevices_t = (
@@ -35,6 +37,7 @@
     | System
     | Thermostat
     | WaterSensor
+    | WaterValve
 )
 
 AllDeviceTypes_t = (
@@ -49,6 +52,7 @@
     | type[System]
     | type[Thermostat]
     | type[WaterSensor]
+    | type[WaterValve]
 )
 
 # AllCommands_t = (
@@ -78,6 +82,7 @@
     | list[System]
     | list[Thermostat]
     | list[WaterSensor]
+    | list[WaterValve]
 )
 
 AllDevicesDicts_t = (
@@ -92,6 +97,7 @@
     | dict[str, System]
     | dict[str, Thermostat]
     | dict[str, WaterSensor]
+    | dict[str, WaterValve]
 )
 
 ATTRIBUTES: dict[DeviceType, AttributeRegistryEntry] = {
@@ -164,6 +170,12 @@
         "rel_id": "devices/water-sensor",
         "device_registry_property": "water_sensors",
     },
+    DeviceType.WATER_VALVE: {
+        "endpoints": {"primary": "{}web/api/devices/waterValves/{}"},
+        "class_": WaterValve,
+        "rel_id": "devices/water-valve",
+        "device_registry_property": "water_valve",
+    },
     DeviceType.ACCESS_CONTROL: {
         "endpoints": {"primary": "{}web/api/devices/accessControlAccessPointDevices/{}"},
         "rel_id": "devices/access-control-access-point-device",
@@ -228,10 +240,6 @@
         "endpoints": {"primary": "{}web/api/devices/waterMeters/{}"},
         "rel_id": "devices/water-meter",
     },
-    DeviceType.WATER_VALVE: {
-        "endpoints": {"primary": "{}web/api/devices/waterValves/{}"},
-        "rel_id": "devices/water-valve",
-    },
     DeviceType.X10_LIGHT: {
         "endpoints": {"primary": "{}web/api/devices/x10Lights/{}"},
         "rel_id": "devices/x10-light",
@@ -342,6 +350,11 @@ def water_sensors(self) -> dict[str, WaterSensor]:
         """Return water sensors."""
         return {device_id: device for device_id, device in self._devices.items() if type(device) == WaterSensor}
 
+    @property
+    def water_valves(self) -> dict[str, WaterValve]:
+        """Return water sensors."""
+        return {device_id: device for device_id, device in self._devices.items() if type(device) == WaterValve}
+
 
 class AttributeRegistry:
     """Device registry."""

pyalarmdotcomajax/devices/thermostat.py

@@ -131,7 +131,7 @@ class ThermostatAttributes(BaseDevice.DeviceAttributes):
 
     @property
     def models(self) -> dict:
-        """Return mapping of known ADC model IDs to manufacturer and model name. To be overridden by children."""
+        """Return mapping of known ADC model IDs to manufacturer and model name."""
         return {
             4293: {"manufacturer": "Honeywell", "model": "T6 Pro"},
             10023: {"manufacturer": "ecobee", "model": "ecobee3 lite"},

pyalarmdotcomajax/devices/water_valve.py

@@ -0,0 +1,62 @@
+"""Alarm.com water valve."""
+
+from __future__ import annotations
+
+import logging
+
+from pyalarmdotcomajax.devices import DeviceType
+
+from . import BaseDevice
+
+log = logging.getLogger(__name__)
+
+
+class WaterValve(BaseDevice):
+    """Represent Alarm.com sensor element."""
+
+    class DeviceState(BaseDevice.DeviceState):
+        """Enum of water valve states."""
+
+        # https://www.alarm.com/web/system/assets/customer-site/enums/WaterValveStatus.js
+
+        UNKNOWN = 0
+        CLOSED = 1
+        OPEN = 2
+
+    class Command(BaseDevice.Command):
+        """Commands for ADC water valves."""
+
+        OPEN = "open"
+        CLOSE = "close"
+
+    @property
+    def models(self) -> dict:
+        """Return mapping of known ADC model IDs to manufacturer and model name."""
+        return {
+            9361: {
+                "manufacturer": "Qolsys",
+                "model": "IQ Water Valve",
+            }  # OEM is Custos - Z-Wave Ball Valve Servo US/CA
+        }
+
+    async def async_open(self) -> None:
+        """Send open command."""
+
+        await self.async_handle_external_desired_state_change(self.DeviceState.OPEN)
+
+        await self._send_action(
+            device_type=DeviceType.WATER_VALVE,
+            event=self.Command.OPEN,
+            device_id=self.id_,
+        )
+
+    async def async_close(self) -> None:
+        """Send close command."""
+
+        await self.async_handle_external_desired_state_change(self.DeviceState.CLOSED)
+
+        await self._send_action(
+            device_type=DeviceType.WATER_VALVE,
+            event=self.Command.CLOSE,
+            device_id=self.id_,
+        )

pyalarmdotcomajax/websockets/client.py

@@ -21,6 +21,7 @@
 from pyalarmdotcomajax.devices.sensor import Sensor
 from pyalarmdotcomajax.devices.thermostat import Thermostat
 from pyalarmdotcomajax.devices.water_sensor import WaterSensor
+from pyalarmdotcomajax.devices.water_valve import WaterValve
 from pyalarmdotcomajax.exceptions import (
     AuthenticationFailed,
     UnexpectedResponse,
@@ -35,6 +36,9 @@
 from pyalarmdotcomajax.websockets.handler.water_sensor import (
     WaterSensorWebSocketHandler,
 )
+from pyalarmdotcomajax.websockets.handler.water_valve import (
+    WaterValveWebSocketHandler,
+)
 from pyalarmdotcomajax.websockets.messages import (
     MonitoringEventMessage,
     process_raw_message,
@@ -197,6 +201,8 @@ async def _async_handle_message(self, raw_message: dict) -> None:
                     await ThermostatWebSocketHandler().process_message(message)
                 case WaterSensor():
                     await WaterSensorWebSocketHandler().process_message(message)
+                case WaterValve():
+                    await WaterValveWebSocketHandler().process_message(message)
                 case _:
                     log.debug(
                         f"WebSocket support not yet implemented for {message.device.__class__.__name__.lower()}s."

pyalarmdotcomajax/websockets/handler/water_valve.py

@@ -0,0 +1,57 @@
+"""WaterValve websocket message handler."""
+
+from __future__ import annotations
+
+import logging
+
+from pyalarmdotcomajax.devices.water_valve import WaterValve
+from pyalarmdotcomajax.websockets.const import EventType
+from pyalarmdotcomajax.websockets.handler import BaseWebSocketHandler
+from pyalarmdotcomajax.websockets.messages import (
+    EventMessage,
+    StatusChangeMessage,
+    WebSocketMessage,
+)
+
+log = logging.getLogger(__name__)
+
+EVENT_STATE_MAP = {
+    EventType.Opened: WaterValve.DeviceState.OPEN,
+    EventType.Closed: WaterValve.DeviceState.CLOSED,
+}
+
+
+class WaterValveWebSocketHandler(BaseWebSocketHandler):
+    """Base class for device-type-specific websocket message handler."""
+
+    SUPPORTED_DEVICE_TYPE = WaterValve
+
+    async def process_message(self, message: WebSocketMessage) -> None:
+        """Handle websocket message."""
+
+        # https://www.alarm.com/web/system/assets/customer-site/websockets/handlers/water-valves.js
+
+        if type(message.device) != WaterValve:
+            return
+
+        match message:
+            case StatusChangeMessage():
+                if message.new_state:
+                    await message.device.async_handle_external_dual_state_change(message.new_state)
+
+            case EventMessage():
+                match message.event_type:
+                    case EventType.Opened | EventType.Closed:
+                        await message.device.async_handle_external_dual_state_change(
+                            EVENT_STATE_MAP[message.event_type]
+                        )
+                    case _:
+                        log.debug(
+                            f"Support for event {message.event_type} ({message.event_type_id}) not yet implemented"
+                            f" by {self.SUPPORTED_DEVICE_TYPE.__name__}."
+                        )
+
+            case _:
+                log.debug(
+                    f"Support for {type(message)} not yet implemented by {self.SUPPORTED_DEVICE_TYPE.__name__}."
+                )

tests/conftest.py

@@ -166,3 +166,31 @@ def device_catalog_no_permissions(response_mocker: aioresponses, all_base_ok_res
     )
 
     all_base_ok_responses_callable()
+
+
+@pytest.fixture
+def system_without_partition(response_mocker: aioresponses, all_base_ok_responses_callable: Callable) -> None:
+    """No permission to view devices."""
+
+    response_mocker.get(
+        url=AlarmController.ALL_DEVICES_URL_TEMPLATE.format(c.URL_BASE, "id-system"),
+        status=200,
+        body=get_http_body_json("device_catalog_no_partitions"),
+        repeat=True,
+    )
+
+    response_mocker.get(
+        url=AttributeRegistry.get_endpoints(DeviceType.SYSTEM)["primary"].format(c.URL_BASE, "id-system"),
+        status=200,
+        body=get_http_body_json("system_no_partitions"),
+        repeat=True,
+    )
+
+    response_mocker.get(
+        url=AlarmController.ALL_SYSTEMS_URL_TEMPLATE.format(c.URL_BASE),
+        status=200,
+        body=get_http_body_json("available_systems_ok"),
+        repeat=True,
+    )
+
+    all_base_ok_responses_callable()

tests/responses/__init__.py

@@ -6,10 +6,10 @@
 def get_http_body_json(name: str) -> str:
     """Get server/client response/request body from JSON file."""
 
-    return resources.read_text(__package__, f"{name}.json")
+    return resources.files(__package__).joinpath(f"{name}.json").read_text()
 
 
 def get_http_body_html(name: str) -> str:
     """Get server/client response/request body from HTML file."""
 
-    return resources.read_text(__package__, f"{name}.html")
+    return resources.files(__package__).joinpath(f"{name}.html").read_text()