From 987c2d1612a69422fb0cf6e94cf8ed5b69b6fbd0 Mon Sep 17 00:00:00 2001
From: Matthias Alphart <farmio@alphart.net>
Date: Fri, 19 Mar 2021 10:12:55 +0100
Subject: [PATCH] Type check KNX integration expose (#48055)
---
homeassistant/components/knx/__init__.py | 2 +-
homeassistant/components/knx/expose.py | 64 +++++++++++++++---------
homeassistant/components/knx/schema.py | 18 ++++++-
3 files changed, 58 insertions(+), 26 deletions(-)
diff --git a/homeassistant/components/knx/__init__.py b/homeassistant/components/knx/__init__.py
index 348eac8f40e..c252572e28e 100644
--- a/homeassistant/components/knx/__init__.py
+++ b/homeassistant/components/knx/__init__.py
@@ -195,7 +195,7 @@ SERVICE_KNX_EVENT_REGISTER_SCHEMA = vol.Schema(
)
SERVICE_KNX_EXPOSURE_REGISTER_SCHEMA = vol.Any(
- ExposeSchema.SCHEMA.extend(
+ ExposeSchema.EXPOSE_SENSOR_SCHEMA.extend(
{
vol.Optional(SERVICE_KNX_ATTR_REMOVE, default=False): cv.boolean,
}
diff --git a/homeassistant/components/knx/expose.py b/homeassistant/components/knx/expose.py
index 8bdd3d1d1d1..3d28c394140 100644
--- a/homeassistant/components/knx/expose.py
+++ b/homeassistant/components/knx/expose.py
@@ -1,6 +1,8 @@
"""Exposures to KNX bus."""
from __future__ import annotations
+from typing import Callable
+
from xknx import XKNX
from xknx.devices import DateTime, ExposeSensor
@@ -11,9 +13,14 @@ from homeassistant.const import (
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import callback
from homeassistant.helpers.event import async_track_state_change_event
-from homeassistant.helpers.typing import ConfigType
+from homeassistant.helpers.typing import (
+ ConfigType,
+ EventType,
+ HomeAssistantType,
+ StateType,
+)
from .const import KNX_ADDRESS
from .schema import ExposeSchema
@@ -21,19 +28,22 @@ from .schema import ExposeSchema
@callback
def create_knx_exposure(
- hass: HomeAssistant, xknx: XKNX, config: ConfigType
+ hass: HomeAssistantType, xknx: XKNX, config: ConfigType
) -> KNXExposeSensor | KNXExposeTime:
"""Create exposures from config."""
address = config[KNX_ADDRESS]
+ expose_type = config[ExposeSchema.CONF_KNX_EXPOSE_TYPE]
attribute = config.get(ExposeSchema.CONF_KNX_EXPOSE_ATTRIBUTE)
- entity_id = config.get(CONF_ENTITY_ID)
- expose_type = config.get(ExposeSchema.CONF_KNX_EXPOSE_TYPE)
default = config.get(ExposeSchema.CONF_KNX_EXPOSE_DEFAULT)
exposure: KNXExposeSensor | KNXExposeTime
- if expose_type.lower() in ["time", "date", "datetime"]:
+ if (
+ isinstance(expose_type, str)
+ and expose_type.lower() in ExposeSchema.EXPOSE_TIME_TYPES
+ ):
exposure = KNXExposeTime(xknx, expose_type, address)
else:
+ entity_id = config[CONF_ENTITY_ID]
exposure = KNXExposeSensor(
hass,
xknx,
@@ -43,14 +53,22 @@ def create_knx_exposure(
default,
address,
)
- exposure.async_register()
return exposure
class KNXExposeSensor:
"""Object to Expose Home Assistant entity to KNX bus."""
- def __init__(self, hass, xknx, expose_type, entity_id, attribute, default, address):
+ def __init__(
+ self,
+ hass: HomeAssistantType,
+ xknx: XKNX,
+ expose_type: int | str,
+ entity_id: str,
+ attribute: str | None,
+ default: StateType,
+ address: str,
+ ) -> None:
"""Initialize of Expose class."""
self.hass = hass
self.xknx = xknx
@@ -59,17 +77,17 @@ class KNXExposeSensor:
self.expose_attribute = attribute
self.expose_default = default
self.address = address
- self.device = None
- self._remove_listener = None
+ self._remove_listener: Callable[[], None] | None = None
+ self.device: ExposeSensor = self.async_register()
@callback
- def async_register(self):
+ def async_register(self) -> ExposeSensor:
"""Register listener."""
if self.expose_attribute is not None:
_name = self.entity_id + "__" + self.expose_attribute
else:
_name = self.entity_id
- self.device = ExposeSensor(
+ device = ExposeSensor(
self.xknx,
name=_name,
group_address=self.address,
@@ -78,6 +96,7 @@ class KNXExposeSensor:
self._remove_listener = async_track_state_change_event(
self.hass, [self.entity_id], self._async_entity_changed
)
+ return device
@callback
def shutdown(self) -> None:
@@ -85,10 +104,9 @@ class KNXExposeSensor:
if self._remove_listener is not None:
self._remove_listener()
self._remove_listener = None
- if self.device is not None:
- self.device.shutdown()
+ self.device.shutdown()
- async def _async_entity_changed(self, event):
+ async def _async_entity_changed(self, event: EventType) -> None:
"""Handle entity change."""
new_state = event.data.get("new_state")
if new_state is None:
@@ -110,8 +128,9 @@ class KNXExposeSensor:
return
await self._async_set_knx_value(new_attribute)
- async def _async_set_knx_value(self, value):
+ async def _async_set_knx_value(self, value: StateType) -> None:
"""Set new value on xknx ExposeSensor."""
+ assert self.device is not None
if value is None:
if self.expose_default is None:
return
@@ -129,17 +148,17 @@ class KNXExposeSensor:
class KNXExposeTime:
"""Object to Expose Time/Date object to KNX bus."""
- def __init__(self, xknx: XKNX, expose_type: str, address: str):
+ def __init__(self, xknx: XKNX, expose_type: str, address: str) -> None:
"""Initialize of Expose class."""
self.xknx = xknx
self.expose_type = expose_type
self.address = address
- self.device = None
+ self.device: DateTime = self.async_register()
@callback
- def async_register(self):
+ def async_register(self) -> DateTime:
"""Register listener."""
- self.device = DateTime(
+ return DateTime(
self.xknx,
name=self.expose_type.capitalize(),
broadcast_type=self.expose_type.upper(),
@@ -148,7 +167,6 @@ class KNXExposeTime:
)
@callback
- def shutdown(self):
+ def shutdown(self) -> None:
"""Prepare for deletion."""
- if self.device is not None:
- self.device.shutdown()
+ self.device.shutdown()
diff --git a/homeassistant/components/knx/schema.py b/homeassistant/components/knx/schema.py
index 376e86cf90c..0e290311f28 100644
--- a/homeassistant/components/knx/schema.py
+++ b/homeassistant/components/knx/schema.py
@@ -254,16 +254,30 @@ class ExposeSchema:
CONF_KNX_EXPOSE_TYPE = CONF_TYPE
CONF_KNX_EXPOSE_ATTRIBUTE = "attribute"
CONF_KNX_EXPOSE_DEFAULT = "default"
+ EXPOSE_TIME_TYPES = [
+ "time",
+ "date",
+ "datetime",
+ ]
- SCHEMA = vol.Schema(
+ EXPOSE_TIME_SCHEMA = vol.Schema(
+ {
+ vol.Required(CONF_KNX_EXPOSE_TYPE): vol.All(
+ cv.string, str.lower, vol.In(EXPOSE_TIME_TYPES)
+ ),
+ vol.Required(KNX_ADDRESS): ga_validator,
+ }
+ )
+ EXPOSE_SENSOR_SCHEMA = vol.Schema(
{
vol.Required(CONF_KNX_EXPOSE_TYPE): sensor_type_validator,
vol.Required(KNX_ADDRESS): ga_validator,
- vol.Optional(CONF_ENTITY_ID): cv.entity_id,
+ vol.Required(CONF_ENTITY_ID): cv.entity_id,
vol.Optional(CONF_KNX_EXPOSE_ATTRIBUTE): cv.string,
vol.Optional(CONF_KNX_EXPOSE_DEFAULT): cv.match_all,
}
)
+ SCHEMA = vol.Any(EXPOSE_TIME_SCHEMA, EXPOSE_SENSOR_SCHEMA)
class FanSchema: