Enable strict type checks for onewire (home-assistant#50422)

.strict-typing

@@ -31,6 +31,7 @@ homeassistant.components.media_player.*
 homeassistant.components.nam.*
 homeassistant.components.notify.*
 homeassistant.components.number.*
+homeassistant.components.onewire.*
 homeassistant.components.persistent_notification.*
 homeassistant.components.proximity.*
 homeassistant.components.recorder.purge

homeassistant/components/onewire/__init__.py

@@ -13,7 +13,7 @@
 _LOGGER = logging.getLogger(__name__)
 
 
-async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry):
+async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
     """Set up a 1-Wire proxy for a config entry."""
     hass.data.setdefault(DOMAIN, {})
 
@@ -65,7 +65,7 @@ async def start_platforms() -> None:
     return True
 
 
-async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry):
+async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
     """Unload a config entry."""
     unload_ok = await hass.config_entries.async_unload_platforms(
         config_entry, PLATFORMS

homeassistant/components/onewire/binary_sensor.py

@@ -1,14 +1,21 @@
 """Support for 1-Wire binary sensors."""
+from __future__ import annotations
+
 import os
 
 from homeassistant.components.binary_sensor import BinarySensorEntity
+from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_TYPE
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
 from .const import CONF_TYPE_OWSERVER, DOMAIN, SENSOR_TYPE_SENSED
-from .onewire_entities import OneWireProxyEntity
+from .model import DeviceComponentDescription
+from .onewire_entities import OneWireBaseEntity, OneWireProxyEntity
 from .onewirehub import OneWireHub
 
-DEVICE_BINARY_SENSORS = {
+DEVICE_BINARY_SENSORS: dict[str, list[DeviceComponentDescription]] = {
     # Family : { path, sensor_type }
     "12": [
         {
@@ -77,7 +84,11 @@
 }
 
 
-async def async_setup_entry(hass, config_entry, async_add_entities):
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
     """Set up 1-Wire platform."""
     # Only OWServer implementation works with binary sensors
     if config_entry.data[CONF_TYPE] == CONF_TYPE_OWSERVER:
@@ -87,9 +98,12 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
         async_add_entities(entities, True)
 
 
-def get_entities(onewirehub: OneWireHub):
+def get_entities(onewirehub: OneWireHub) -> list[OneWireBaseEntity]:
     """Get a list of entities."""
-    entities = []
+    if not onewirehub.devices:
+        return []
+
+    entities: list[OneWireBaseEntity] = []
 
     for device in onewirehub.devices:
         family = device["family"]
@@ -98,7 +112,7 @@ def get_entities(onewirehub: OneWireHub):
 
         if family not in DEVICE_BINARY_SENSORS:
             continue
-        device_info = {
+        device_info: DeviceInfo = {
             "identifiers": {(DOMAIN, device_id)},
             "manufacturer": "Maxim Integrated",
             "model": device_type,
@@ -126,6 +140,6 @@ class OneWireProxyBinarySensor(OneWireProxyEntity, BinarySensorEntity):
     """Implementation of a 1-Wire binary sensor."""
 
     @property
-    def is_on(self):
+    def is_on(self) -> bool:
         """Return true if sensor is on."""
-        return self._state
+        return bool(self._state)

homeassistant/components/onewire/config_flow.py

@@ -1,9 +1,14 @@
 """Config flow for 1-Wire component."""
+from __future__ import annotations
+
+from typing import Any
+
 import voluptuous as vol
 
 from homeassistant.config_entries import ConfigFlow
 from homeassistant.const import CONF_HOST, CONF_PORT, CONF_TYPE
 from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResult
 
 from .const import (
     CONF_MOUNT_DIR,
@@ -32,7 +37,9 @@
 )
 
 
-async def validate_input_owserver(hass: HomeAssistant, data):
+async def validate_input_owserver(
+    hass: HomeAssistant, data: dict[str, Any]
+) -> dict[str, str]:
     """Validate the user input allows us to connect.
 
     Data has the keys from DATA_SCHEMA_OWSERVER with values provided by the user.
@@ -49,7 +56,9 @@ async def validate_input_owserver(hass: HomeAssistant, data):
     return {"title": host}
 
 
-def is_duplicate_owserver_entry(hass: HomeAssistant, user_input):
+def is_duplicate_owserver_entry(
+    hass: HomeAssistant, user_input: dict[str, Any]
+) -> bool:
     """Check existing entries for matching host and port."""
     for config_entry in hass.config_entries.async_entries(DOMAIN):
         if (
@@ -61,7 +70,9 @@ def is_duplicate_owserver_entry(hass: HomeAssistant, user_input):
     return False
 
 
-async def validate_input_mount_dir(hass: HomeAssistant, data):
+async def validate_input_mount_dir(
+    hass: HomeAssistant, data: dict[str, Any]
+) -> dict[str, str]:
     """Validate the user input allows us to connect.
 
     Data has the keys from DATA_SCHEMA_MOUNTDIR with values provided by the user.
@@ -82,16 +93,18 @@ class OneWireFlowHandler(ConfigFlow, domain=DOMAIN):
 
     VERSION = 1
 
-    def __init__(self):
+    def __init__(self) -> None:
         """Initialize 1-Wire config flow."""
-        self.onewire_config = {}
+        self.onewire_config: dict[str, Any] = {}
 
-    async def async_step_user(self, user_input=None):
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
         """Handle 1-Wire config flow start.
 
         Let user manually input configuration.
         """
-        errors = {}
+        errors: dict[str, str] = {}
         if user_input is not None:
             self.onewire_config.update(user_input)
             if CONF_TYPE_OWSERVER == user_input[CONF_TYPE]:
@@ -105,7 +118,9 @@ async def async_step_user(self, user_input=None):
             errors=errors,
         )
 
-    async def async_step_owserver(self, user_input=None):
+    async def async_step_owserver(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
         """Handle OWServer configuration."""
         errors = {}
         if user_input:
@@ -130,7 +145,9 @@ async def async_step_owserver(self, user_input=None):
             errors=errors,
         )
 
-    async def async_step_mount_dir(self, user_input=None):
+    async def async_step_mount_dir(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
         """Handle SysBus configuration."""
         errors = {}
         if user_input:
@@ -157,7 +174,7 @@ async def async_step_mount_dir(self, user_input=None):
             errors=errors,
         )
 
-    async def async_step_import(self, platform_config):
+    async def async_step_import(self, platform_config: dict[str, Any]) -> FlowResult:
         """Handle import configuration from YAML."""
         # OWServer
         if platform_config[CONF_TYPE] == CONF_TYPE_OWSERVER:

homeassistant/components/onewire/const.py

@@ -1,4 +1,6 @@
 """Constants for 1-Wire component."""
+from __future__ import annotations
+
 from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
 from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
 from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
@@ -44,7 +46,7 @@
 SWITCH_TYPE_LATCH = "latch"
 SWITCH_TYPE_PIO = "pio"
 
-SENSOR_TYPES = {
+SENSOR_TYPES: dict[str, list[str | None]] = {
     # SensorType: [ Unit, DeviceClass ]
     SENSOR_TYPE_TEMPERATURE: [TEMP_CELSIUS, DEVICE_CLASS_TEMPERATURE],
     SENSOR_TYPE_HUMIDITY: [PERCENTAGE, DEVICE_CLASS_HUMIDITY],

homeassistant/components/onewire/model.py

@@ -0,0 +1,21 @@
+"""Type definitions for 1-Wire integration."""
+from __future__ import annotations
+
+from typing import TypedDict
+
+
+class DeviceComponentDescription(TypedDict, total=False):
+    """Device component description class."""
+
+    path: str
+    name: str
+    type: str
+    default_disabled: bool
+
+
+class OWServerDeviceDescription(TypedDict):
+    """OWServer device description class."""
+
+    path: str
+    family: str
+    type: str

homeassistant/components/onewire/onewire_entities.py

@@ -7,6 +7,7 @@
 from pyownet import protocol
 
 from homeassistant.helpers.entity import DeviceInfo, Entity
+from homeassistant.helpers.typing import StateType
 
 from .const import (
     SENSOR_TYPE_COUNT,
@@ -15,6 +16,7 @@
     SWITCH_TYPE_LATCH,
     SWITCH_TYPE_PIO,
 )
+from .model import DeviceComponentDescription
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -24,13 +26,13 @@ class OneWireBaseEntity(Entity):
 
     def __init__(
         self,
-        name,
-        device_file,
+        name: str,
+        device_file: str,
         entity_type: str,
-        entity_name: str = None,
-        device_info: DeviceInfo | None = None,
-        default_disabled: bool = False,
-        unique_id: str = None,
+        entity_name: str,
+        device_info: DeviceInfo,
+        default_disabled: bool,
+        unique_id: str,
     ):
         """Initialize the entity."""
         self._name = f"{name} {entity_name or entity_type.capitalize()}"
@@ -39,10 +41,10 @@ def __init__(
         self._device_class = SENSOR_TYPES[entity_type][1]
         self._unit_of_measurement = SENSOR_TYPES[entity_type][0]
         self._device_info = device_info
-        self._state = None
-        self._value_raw = None
+        self._state: StateType = None
+        self._value_raw: float | None = None
         self._default_disabled = default_disabled
-        self._unique_id = unique_id or device_file
+        self._unique_id = unique_id
 
     @property
     def name(self) -> str | None:
@@ -84,7 +86,7 @@ def __init__(
         device_name: str,
         device_info: DeviceInfo,
         entity_path: str,
-        entity_specs: dict[str, Any],
+        entity_specs: DeviceComponentDescription,
         owproxy: protocol._Proxy,
     ):
         """Initialize the sensor."""
@@ -99,31 +101,30 @@ def __init__(
         )
         self._owproxy = owproxy
 
-    def _read_value_ownet(self):
+    def _read_value_ownet(self) -> str:
         """Read a value from the owserver."""
-        return self._owproxy.read(self._device_file).decode().lstrip()
+        read_bytes: bytes = self._owproxy.read(self._device_file)
+        return read_bytes.decode().lstrip()
 
-    def _write_value_ownet(self, value: bytes):
+    def _write_value_ownet(self, value: bytes) -> None:
         """Write a value to the owserver."""
-        return self._owproxy.write(self._device_file, value)
+        self._owproxy.write(self._device_file, value)
 
-    def update(self):
+    def update(self) -> None:
         """Get the latest data from the device."""
-        value = None
         try:
             self._value_raw = float(self._read_value_ownet())
         except protocol.Error as exc:
             _LOGGER.error("Owserver failure in read(), got: %s", exc)
+            self._state = None
         else:
             if self._entity_type == SENSOR_TYPE_COUNT:
-                value = int(self._value_raw)
+                self._state = int(self._value_raw)
             elif self._entity_type in [
                 SENSOR_TYPE_SENSED,
                 SWITCH_TYPE_LATCH,
                 SWITCH_TYPE_PIO,
             ]:
-                value = int(self._value_raw) == 1
+                self._state = int(self._value_raw) == 1
             else:
-                value = round(self._value_raw, 1)
-
-        self._state = value
+                self._state = round(self._value_raw, 1)

homeassistant/components/onewire/onewirehub.py

@@ -1,4 +1,6 @@
 """Hub for communication with 1-Wire server or mount_dir."""
+from __future__ import annotations
+
 import os
 
 from pi1wire import Pi1Wire
@@ -10,6 +12,7 @@
 from homeassistant.exceptions import HomeAssistantError
 
 from .const import CONF_MOUNT_DIR, CONF_TYPE_OWSERVER, CONF_TYPE_SYSBUS
+from .model import OWServerDeviceDescription
 
 DEVICE_COUPLERS = {
     # Family : [branches]
@@ -23,10 +26,10 @@ class OneWireHub:
     def __init__(self, hass: HomeAssistant):
         """Initialize."""
         self.hass = hass
-        self.type: str = None
-        self.pi1proxy: Pi1Wire = None
-        self.owproxy: protocol._Proxy = None
-        self.devices = None
+        self.type: str | None = None
+        self.pi1proxy: Pi1Wire | None = None
+        self.owproxy: protocol._Proxy | None = None
+        self.devices: list | None = None
 
     async def connect(self, host: str, port: int) -> None:
         """Connect to the owserver host."""
@@ -54,22 +57,25 @@ async def initialize(self, config_entry: ConfigEntry) -> None:
             await self.connect(host, port)
         await self.discover_devices()
 
-    async def discover_devices(self):
+    async def discover_devices(self) -> None:
         """Discover all devices."""
         if self.devices is None:
             if self.type == CONF_TYPE_SYSBUS:
+                assert self.pi1proxy
                 self.devices = await self.hass.async_add_executor_job(
                     self.pi1proxy.find_all_sensors
                 )
             if self.type == CONF_TYPE_OWSERVER:
                 self.devices = await self.hass.async_add_executor_job(
                     self._discover_devices_owserver
                 )
-        return self.devices
 
-    def _discover_devices_owserver(self, path="/"):
+    def _discover_devices_owserver(
+        self, path: str = "/"
+    ) -> list[OWServerDeviceDescription]:
         """Discover all owserver devices."""
         devices = []
+        assert self.owproxy
         for device_path in self.owproxy.dir(path):
             device_family = self.owproxy.read(f"{device_path}family").decode()
             device_type = self.owproxy.read(f"{device_path}type").decode()