Bump pypx800v5 to version 1.3.1 and add access control support

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## 1.8.0
+
+- bump pypx800v5
+- Add Access Control support
+- Fix deprecated code
+
 ## 1.7.1
 
 - Bump pypx800v5 to fix security issue

custom_components/ipx800v5/__init__.py

@@ -1,4 +1,5 @@
 """Support for the GCE IPX800 V5."""
+
 from datetime import timedelta
 import logging
 
@@ -209,12 +210,12 @@ async def async_update_data():
 
     for platform in PLATFORMS:
         _LOGGER.debug("Load platform %s", platform)
-        hass.data[DOMAIN][entry.entry_id][CONF_DEVICES][
-            platform
-        ] = filter_entities_by_platform(entities, platform)
-        hass.async_create_task(
-            hass.config_entries.async_forward_entry_setup(entry, platform)
+        hass.data[DOMAIN][entry.entry_id][CONF_DEVICES][platform] = (
+            filter_entities_by_platform(entities, platform)
         )
+    hass.async_create_task(
+        hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+    )
 
     # Provide endpoints for the IPX to call to push states
     if CONF_PUSH_PASSWORD in config:

custom_components/ipx800v5/binary_sensor.py

@@ -1,4 +1,5 @@
 """Support for IPX800 V5 binary sensors."""
+
 import logging
 
 from pypx800v5 import (
@@ -7,12 +8,14 @@
     EXT_X24D,
     IPX,
     IPX800,
+    OBJECT_ACCESS_CONTROL,
     OBJECT_TEMPO,
     OBJECT_THERMOSTAT,
     TYPE_IO,
     X8D,
     X8R,
     X24D,
+    AccessControl,
     IPX800DigitalInput,
     IPX800OptoInput,
     Tempo,
@@ -75,6 +78,15 @@ async def async_setup_entry(
             entities.append(
                 ThermostatFaultStateBinarySensor(device, controller, coordinator)
             )
+        elif device[CONF_EXT_TYPE] == OBJECT_ACCESS_CONTROL:
+            entities.append(
+                AccessControlBinarySensor(device, controller, coordinator, "io_out_id")
+            )
+            entities.append(
+                AccessControlBinarySensor(
+                    device, controller, coordinator, "io_fault_id"
+                )
+            )
 
     async_add_entities(entities, True)
 
@@ -85,7 +97,7 @@ class IOBinarySensor(IpxEntity, BinarySensorEntity):
     @property
     def is_on(self) -> bool:
         """Return the current value."""
-        return self.coordinator.data[self._io_id]["on"] is True
+        return self.coordinator.data[str(self._io_id)]["on"] is True
 
 
 class IpxDigitalInputBinarySensor(IpxEntity, BinarySensorEntity):
@@ -198,3 +210,24 @@ def __init__(
     def is_on(self) -> bool:
         """Return the current value."""
         return self.coordinator.data[self.control.io_fault_id]["on"] is True
+
+
+class AccessControlBinarySensor(IpxEntity, BinarySensorEntity):
+    """Representation the Access Control state as a binary sensor."""
+
+    def __init__(
+        self,
+        device_config: dict,
+        ipx: IPX800,
+        coordinator: DataUpdateCoordinator,
+        id_name: str,
+    ) -> None:
+        """Initialize the sensor of the tempo."""
+        super().__init__(device_config, ipx, coordinator)
+        self.control = AccessControl(ipx, self._ext_number)
+        self._id_name = id_name
+
+    @property
+    def is_on(self) -> bool:
+        """Return the current value."""
+        return self.coordinator.data[getattr(self.control, self._id_name)]["on"] is True

custom_components/ipx800v5/button.py

@@ -1,4 +1,5 @@
 """Support for IPX800 V5 button."""
+
 import logging
 
 from pypx800v5 import IPX
@@ -28,7 +29,7 @@ async def async_setup_entry(
 
     for device in devices:
         if device[CONF_EXT_TYPE] == IPX:
-            entities.append(RebootButton(device, controller, coordinator))
+            entities.append(RebootButton(device, controller, coordinator))  # noqa: PERF401
 
     async_add_entities(entities, True)
 

custom_components/ipx800v5/climate.py

@@ -1,4 +1,5 @@
 """Support for IPX800 V5 climates."""
+
 import logging
 from typing import Any
 

custom_components/ipx800v5/config_flow.py

@@ -1,4 +1,5 @@
 """Config flow to configure the ipx800v5 integration."""
+
 from itertools import groupby
 import logging
 from typing import Any
@@ -16,7 +17,7 @@
 from voluptuous.util import Upper
 
 from homeassistant import config_entries
-from homeassistant.config_entries import ConfigEntry, OptionsFlow
+from homeassistant.config_entries import ConfigEntry, ConfigFlowResult, OptionsFlow
 from homeassistant.const import (
     CONF_API_KEY,
     CONF_DEVICES,
@@ -26,7 +27,6 @@
     CONF_SCAN_INTERVAL,
 )
 from homeassistant.core import callback
-from homeassistant.data_entry_flow import FlowResult
 from homeassistant.helpers.aiohttp_client import async_get_clientsession
 
 from .const import (
@@ -65,7 +65,7 @@ def __init__(self) -> None:
         """Initialize class variables."""
         self.base_config: dict[str, Any] = {}
 
-    async def async_step_import(self, import_info) -> FlowResult:
+    async def async_step_import(self, import_info) -> ConfigFlowResult:
         """Import an advanced configuration from YAML config."""
         entry = await self.async_set_unique_id(f"{DOMAIN}, {import_info[CONF_HOST]}")
 
@@ -83,7 +83,7 @@ async def async_step_import(self, import_info) -> FlowResult:
             title=f"{import_info[CONF_NAME]} (from yaml)", data=import_info
         )
 
-    async def async_step_user(self, user_input=None) -> FlowResult:
+    async def async_step_user(self, user_input=None) -> ConfigFlowResult:
         """Get configuration from the user."""
         errors: dict[str, str] = {}
         if user_input is None:
@@ -107,7 +107,7 @@ async def async_step_user(self, user_input=None) -> FlowResult:
         self.base_config = user_input
         return await self.async_step_params()
 
-    async def async_step_params(self, user_input=None) -> FlowResult:
+    async def async_step_params(self, user_input=None) -> ConfigFlowResult:
         """Handle the param flow to customize according to device config."""
         if user_input is None:
             session = async_get_clientsession(self.hass, False)
@@ -137,7 +137,7 @@ def __init__(self, config_entry: ConfigEntry) -> None:
         """Initialize."""
         self.config_entry = config_entry
 
-    async def async_step_init(self, user_input=None) -> FlowResult:
+    async def async_step_init(self, user_input=None) -> ConfigFlowResult:
         """Manage the options."""
         if user_input is None:
             session = async_get_clientsession(self.hass, False)
@@ -269,7 +269,7 @@ async def _build_param_schema(
                                 ): vol.All(str, vol.Lower, vol.In(["switch", "light"])),
                             }
                         )
-                ext_number += 1
+                ext_number += 1  # noqa: SIM113
 
     return schema
 

custom_components/ipx800v5/const.py

@@ -1,4 +1,5 @@
 """Constants for the ipx800v5 integration."""
+
 DOMAIN = "ipx800v5"
 
 CONTROLLER = "controller"
@@ -31,13 +32,13 @@
 TYPE_XPWM_RGBW = "xpwm_rgbw"
 
 PLATFORMS = [
-    "light",
-    "switch",
-    "sensor",
     "binary_sensor",
-    "cover",
+    "button",
     "climate",
+    "cover",
+    "light",
     "number",
-    "button",
     "select",
+    "sensor",
+    "switch",
 ]

custom_components/ipx800v5/cover.py

@@ -1,4 +1,5 @@
 """Support for IPX800 V5 covers."""
+
 import logging
 from typing import Any
 
@@ -35,7 +36,7 @@ async def async_setup_entry(
 
     for device in devices:
         if device[CONF_EXT_TYPE] == EXT_X4VR:
-            entities.append(X4VRCover(device, controller, coordinator))
+            entities.append(X4VRCover(device, controller, coordinator))  # noqa: PERF401
 
     async_add_entities(entities, True)
 
@@ -60,7 +61,7 @@ def __init__(
             | CoverEntityFeature.SET_POSITION
         )
         if self.control.mode in [2, 3]:
-            self._attr_supported_features += (
+            self._attr_supported_features |= (
                 CoverEntityFeature.CLOSE_TILT | CoverEntityFeature.OPEN_TILT
             )
 

custom_components/ipx800v5/light.py

@@ -1,4 +1,5 @@
 """Support for IPX800 V5 lights."""
+
 from asyncio import gather as async_gather
 import logging
 from typing import Any

custom_components/ipx800v5/manifest.json

@@ -12,7 +12,7 @@
   "iot_class": "local_polling",
   "issue_tracker": "https://github.com/Aohzan/ipx800v5/issues",
   "requirements": [
-    "pypx800v5==1.2.2"
+    "pypx800v5==1.3.1"
   ],
-  "version": "1.7.1"
+  "version": "1.8.0"
 }

custom_components/ipx800v5/number.py

@@ -1,4 +1,5 @@
 """Support for IPX800 V5 numbers."""
+
 import logging
 
 from pypx800v5 import (
@@ -64,7 +65,7 @@ class AnalogNumber(IpxEntity, NumberEntity):
     @property
     def native_value(self) -> float:
         """Return the current value."""
-        return float(self.coordinator.data[self._io_id]["value"])
+        return float(self.coordinator.data[str(self._io_id)]["value"])
 
     async def async_set_native_value(self, value: float) -> None:
         """Update the current value."""
@@ -121,13 +122,13 @@ def __init__(
         coordinator: DataUpdateCoordinator,
         param: str,
     ) -> None:
-        """Initialize the RelaySwitch."""
+        """Initialize the ThermostatParamNumber."""
         super().__init__(
             device_config, ipx, coordinator, suffix_name=f"{param} Temperature"
         )
         self.control = Thermostat(ipx, self._ext_number)
         self._param = param
-        self._value = self.control._config[f"setPoint{param}"]
+        self._value = self.control.init_config[f"setPoint{param}"]
 
     @property
     def native_value(self) -> float:

custom_components/ipx800v5/request_views.py

@@ -1,4 +1,5 @@
 """IPX800V5 request views to handle push information."""
+
 from base64 import b64decode
 from http import HTTPStatus
 import logging
@@ -13,24 +14,29 @@
 _LOGGER = logging.getLogger(__name__)
 
 
+def api_call_not_authorized(msg: str):
+    """Raise an API call not authorized."""
+    raise ApiCallNotAuthorized(msg)
+
+
 def check_api_auth(request, host, push_password) -> bool:
     """Check authentication on API call."""
     try:
         if request.remote != host:
-            raise ApiCallNotAuthorized("API call not coming from IPX800 IP.")
+            api_call_not_authorized("API call not coming from IPX800 IP.")
         if "Authorization" not in request.headers:
-            raise ApiCallNotAuthorized("API call no authentication provided.")
+            api_call_not_authorized("API call no authentication provided.")
         header_auth = request.headers["Authorization"]
         split = header_auth.strip().split(" ")
         if len(split) != 2 or split[0].strip().lower() != "basic":
-            raise ApiCallNotAuthorized("Malformed Authorization header")
+            api_call_not_authorized("Malformed Authorization header")
         username, password = b64decode(split[1]).decode().split(":", 1)
         if username != PUSH_USERNAME or password != push_password:
-            raise ApiCallNotAuthorized("API call authentication invalid.")
-        return True
+            api_call_not_authorized("API call authentication invalid.")
     except ApiCallNotAuthorized as err:
         _LOGGER.warning(err)
         return False
+    return True
 
 
 class IpxRequestView(HomeAssistantView):
@@ -58,6 +64,7 @@ async def get(self, request, entity_id, state):
                 return web.Response(status=HTTPStatus.OK, text="OK")
             _LOGGER.warning("Entity not found for state updating: %s", entity_id)
         _LOGGER.warning("Authentication for PUSH invalid")
+        return None
 
 
 class IpxRequestDataView(HomeAssistantView):
@@ -96,6 +103,7 @@ async def get(self, request, data):
 
             return web.Response(status=HTTPStatus.OK, text="OK")
         _LOGGER.warning("Authentication for PUSH invalid")
+        return None
 
 
 class IpxRequestRefreshView(HomeAssistantView):
@@ -121,6 +129,7 @@ async def get(self, request, data):
             await self.coordinator.async_request_refresh()
             return web.Response(status=HTTPStatus.OK, text="OK")
         _LOGGER.warning("Authentication for PUSH invalid")
+        return None
 
 
 class ApiCallNotAuthorized(BaseException):

custom_components/ipx800v5/select.py

@@ -1,4 +1,5 @@
 """Support for IPX800 V5 select."""
+
 from collections.abc import Mapping
 import logging
 from typing import Any
@@ -63,7 +64,7 @@ def current_option(self) -> str | None:
         )
         if len(self.control.screens) < screen_id + 1:
             _LOGGER.warning(
-                "X-Display current screen #%s is not recognize, please reload the integration to refresh screens",
+                "X-Display current screen #%s is not recognized, please reload the integration to refresh screens, or update if it's persist",
                 screen_id,
             )
             return None