Merge pull request #30 from imhotep/open-support-refactor

adding support for OPEN and small refactor

custom_components/unifi_access/binary_sensor.py

@@ -1,4 +1,5 @@
 """Platform for sensor integration."""
+
 from __future__ import annotations
 
 import logging
@@ -14,7 +15,8 @@
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
 from .const import DOMAIN
-from .hub import UnifiAccessCoordinator, UnifiAccessHub
+from .coordinator import UnifiAccessCoordinator
+from .hub import UnifiAccessHub
 
 _LOGGER = logging.getLogger(__name__)
 

custom_components/unifi_access/config_flow.py

@@ -1,4 +1,5 @@
 """Config flow for Unifi Access integration."""
+
 from __future__ import annotations
 
 import logging
@@ -7,8 +8,8 @@
 import voluptuous as vol
 
 from homeassistant import config_entries
+from homeassistant.config_entries import ConfigFlowResult
 from homeassistant.core import HomeAssistant
-from homeassistant.data_entry_flow import FlowResult
 from homeassistant.exceptions import HomeAssistantError
 
 from .const import DOMAIN
@@ -63,7 +64,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
 
     async def async_step_user(
         self, user_input: dict[str, Any] | None = None
-    ) -> FlowResult:
+    ) -> ConfigFlowResult:
         """Handle the initial step."""
         errors: dict[str, str] = {}
         if user_input is not None:

custom_components/unifi_access/coordinator.py

@@ -0,0 +1,42 @@
+"""Unifi Access Coordinator.
+
+This module has the Unifi Access Coordinator to be used by entities.
+"""
+
+import asyncio
+from datetime import timedelta
+import logging
+
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryAuthFailed
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .errors import ApiAuthError, ApiError
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class UnifiAccessCoordinator(DataUpdateCoordinator):
+    """Unifi Access Coordinator. This is mostly used for local polling."""
+
+    def __init__(self, hass: HomeAssistant, hub) -> None:
+        """Initialize Unifi Access Coordinator."""
+        update_interval = timedelta(seconds=3) if hub.use_polling is True else None
+
+        super().__init__(
+            hass,
+            _LOGGER,
+            name="Unifi Access Coordinator",
+            update_interval=update_interval,
+        )
+        self.hub = hub
+
+    async def _async_update_data(self):
+        """Handle Unifi Access Coordinator updates."""
+        try:
+            async with asyncio.timeout(10):
+                return await self.hass.async_add_executor_job(self.hub.update)
+        except ApiAuthError as err:
+            raise ConfigEntryAuthFailed from err
+        except ApiError as err:
+            raise UpdateFailed("Error communicating with API") from err

custom_components/unifi_access/door.py

@@ -1,4 +1,5 @@
 """Unifi Access Door Module."""
+
 from collections.abc import Callable
 import logging
 
@@ -61,6 +62,10 @@ def is_unlocking(self):
         """Solely used for unlocking state when calling unlock."""
         return self._is_unlocking
 
+    def open(self) -> None:
+        """Open door."""
+        self.unlock()
+
     def unlock(self) -> None:
         """Unlock door."""
         if self.is_locked:

custom_components/unifi_access/errors.py

@@ -0,0 +1,12 @@
+"""Unifi Access Errors.
+
+This module lists Unifi Access API errors.
+"""
+
+
+class ApiAuthError(Exception):
+    """Raised when we can't authenticate with the API Token."""
+
+
+class ApiError(Exception):
+    """Raised when we have some trouble using the API."""

custom_components/unifi_access/event.py

@@ -1,4 +1,5 @@
 """Platform for sensor integration."""
+
 from __future__ import annotations
 
 import logging
@@ -16,8 +17,9 @@
     DOORBELL_START_EVENT,
     DOORBELL_STOP_EVENT,
 )
+from .coordinator import UnifiAccessCoordinator
 from .door import UnifiAccessDoor
-from .hub import UnifiAccessCoordinator, UnifiAccessHub
+from .hub import UnifiAccessHub
 
 _LOGGER = logging.getLogger(__name__)
 

custom_components/unifi_access/hub.py

@@ -2,8 +2,8 @@
 
 This module interacts with the Unifi Access API server.
 """
+
 import asyncio
-from datetime import timedelta
 import json
 import logging
 import ssl
@@ -15,10 +15,6 @@
 import urllib3
 import websocket
 
-from homeassistant.core import HomeAssistant
-from homeassistant.exceptions import ConfigEntryAuthFailed
-from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
-
 from .const import (
     ACCESS_EVENT,
     DEVICE_NOTIFICATIONS_URL,
@@ -29,18 +25,11 @@
     UNIFI_ACCESS_API_PORT,
 )
 from .door import UnifiAccessDoor
+from .errors import ApiAuthError, ApiError
 
 _LOGGER = logging.getLogger(__name__)
 
 
-class ApiAuthError(Exception):
-    """Raised when we can't authenticate with the API Token."""
-
-
-class ApiError(Exception):
-    """Raised when we have some trouble using the API."""
-
-
 class UnifiAccessHub:
     """UnifiAccessHub.
 
@@ -373,29 +362,3 @@ def listen_for_updates(self):
         if self.verify_ssl is False:
             sslopt = {"cert_reqs": ssl.CERT_NONE}
         ws.run_forever(sslopt=sslopt, reconnect=5)
-
-
-class UnifiAccessCoordinator(DataUpdateCoordinator):
-    """Unifi Access Coordinator. This is mostly used for local polling."""
-
-    def __init__(self, hass: HomeAssistant, hub) -> None:
-        """Initialize Unifi Access Coordinator."""
-        update_interval = timedelta(seconds=3) if hub.use_polling is True else None
-
-        super().__init__(
-            hass,
-            _LOGGER,
-            name="Unifi Access Coordinator",
-            update_interval=update_interval,
-        )
-        self.hub = hub
-
-    async def _async_update_data(self):
-        """Handle Unifi Access Coordinator updates."""
-        try:
-            async with asyncio.timeout(10):
-                return await self.hass.async_add_executor_job(self.hub.update)
-        except ApiAuthError as err:
-            raise ConfigEntryAuthFailed from err
-        except ApiError as err:
-            raise UpdateFailed("Error communicating with API") from err

custom_components/unifi_access/lock.py

@@ -1,19 +1,21 @@
 """Platform for sensor integration."""
+
 from __future__ import annotations
 
 import logging
 from typing import Any
 
-from homeassistant.components.lock import LockEntity
+from homeassistant.components.lock import LockEntity, LockEntityFeature
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.device_registry import DeviceInfo
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
 from .const import DOMAIN
+from .coordinator import UnifiAccessCoordinator
 from .door import UnifiAccessDoor
-from .hub import UnifiAccessCoordinator, UnifiAccessHub
+from .hub import UnifiAccessHub
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -40,19 +42,15 @@ class UnifiDoorLockEntity(CoordinatorEntity, LockEntity):
 
     should_poll = False
 
+    supported_features = LockEntityFeature.OPEN
+
     def __init__(self, coordinator, door_id) -> None:
         """Initialize Unifi Access Door Lock."""
         super().__init__(coordinator, context=id)
-        self.id = door_id
         self.door: UnifiAccessDoor = self.coordinator.data[door_id]
         self._attr_unique_id = self.door.id
         self._attr_name = self.door.name
 
-    @property
-    def available(self) -> bool:
-        """Gray out lock when it's unlocked."""
-        return self.door.is_locked
-
     @property
     def device_info(self) -> DeviceInfo:
         """Get Unifi Access Door Lock device information."""
@@ -77,6 +75,14 @@ async def async_unlock(self, **kwargs: Any) -> None:
         """Unlock all or specified locks. A code to unlock the lock with may optionally be specified."""
         await self.hass.async_add_executor_job(self.door.unlock)
 
+    async def async_open(self, **kwargs: Any) -> None:
+        """Unlock all or specified locks. A code to unlock the lock with may optionally be specified."""
+        await self.hass.async_add_executor_job(self.door.open)
+
+    def lock(self, **kwargs: Any) -> None:
+        """Lock all or specified locks. A code to lock the lock with may optionally be specified."""
+        _LOGGER.warning("Locking is not supported by Unifi Access API")
+
     @property
     def is_locked(self) -> bool | None:
         """Get Unifi Access Door Lock locked status."""