Type check KNX integration expose (#48055)

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,
         }

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()

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: