Refactor onewire hub (#135186)

* Improve type hints in onewire hub

* More cleanups

* Improve

* Get host/port from entry data

* Use DeviceInfo object

homeassistant/components/onewire/__init__.py

@@ -10,7 +10,7 @@ from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.helpers import device_registry as dr
 
 from .const import DOMAIN
-from .onewirehub import CannotConnect, OneWireConfigEntry, OneWireHub
+from .onewirehub import OneWireConfigEntry, OneWireHub
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -24,11 +24,11 @@ _PLATFORMS = [
 
 async def async_setup_entry(hass: HomeAssistant, entry: OneWireConfigEntry) -> bool:
     """Set up a 1-Wire proxy for a config entry."""
-    onewire_hub = OneWireHub(hass)
+    onewire_hub = OneWireHub(hass, entry)
     try:
-        await onewire_hub.initialize(entry)
+        await onewire_hub.initialize()
     except (
-        CannotConnect,  # Failed to connect to the server
+        protocol.ConnError,  # Failed to connect to the server
         protocol.OwnetError,  # Connected to server, but failed to list the devices
     ) as exc:
         raise ConfigEntryNotReady from exc

homeassistant/components/onewire/config_flow.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 from copy import deepcopy
 from typing import Any
 
+from pyownet import protocol
 import voluptuous as vol
 
 from homeassistant.config_entries import ConfigFlow, ConfigFlowResult, OptionsFlow
@@ -24,7 +25,7 @@ from .const import (
     OPTION_ENTRY_SENSOR_PRECISION,
     PRECISION_MAPPING_FAMILY_28,
 )
-from .onewirehub import CannotConnect, OneWireConfigEntry, OneWireHub
+from .onewirehub import OneWireConfigEntry
 
 DATA_SCHEMA = vol.Schema(
     {
@@ -38,11 +39,11 @@ async def validate_input(
     hass: HomeAssistant, data: dict[str, Any], errors: dict[str, str]
 ) -> None:
     """Validate the user input allows us to connect."""
-
-    hub = OneWireHub(hass)
     try:
-        await hub.connect(data[CONF_HOST], data[CONF_PORT])
+        await hass.async_add_executor_job(
-    except CannotConnect:
+            protocol.proxy, data[CONF_HOST], data[CONF_PORT]
+        )
+    except protocol.ConnError:
         errors["base"] = "cannot_connect"
 
 

homeassistant/components/onewire/onewirehub.py

@@ -4,7 +4,6 @@ from __future__ import annotations
 
 import logging
 import os
-from typing import TYPE_CHECKING
 
 from pyownet import protocol
 
@@ -19,7 +18,6 @@ from homeassistant.const import (
     CONF_PORT,
 )
 from homeassistant.core import HomeAssistant
-from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers import device_registry as dr
 from homeassistant.helpers.device_registry import DeviceInfo
 
@@ -57,36 +55,34 @@ def _is_known_device(device_family: str, device_type: str | None) -> bool:
 class OneWireHub:
     """Hub to communicate with server."""
 
-    def __init__(self, hass: HomeAssistant) -> None:
+    owproxy: protocol._Proxy
+    devices: list[OWDeviceDescription]
+
+    def __init__(self, hass: HomeAssistant, config_entry: OneWireConfigEntry) -> None:
         """Initialize."""
-        self.hass = hass
+        self._hass = hass
-        self.owproxy: protocol._Proxy | None = None
+        self._config_entry = config_entry
-        self.devices: list[OWDeviceDescription] | None = None
 
-    async def connect(self, host: str, port: int) -> None:
+    def _initialize(self) -> None:
-        """Connect to the server."""
+        """Connect to the server, and discover connected devices.
-        try:
-            self.owproxy = await self.hass.async_add_executor_job(
-                protocol.proxy, host, port
-            )
-        except protocol.ConnError as exc:
-            raise CannotConnect from exc
 
-    async def initialize(self, config_entry: OneWireConfigEntry) -> None:
+        Needs to be run in executor.
-        """Initialize a config entry."""
+        """
-        host = config_entry.data[CONF_HOST]
+        host = self._config_entry.data[CONF_HOST]
-        port = config_entry.data[CONF_PORT]
+        port = self._config_entry.data[CONF_PORT]
         _LOGGER.debug("Initializing connection to %s:%s", host, port)
-        await self.connect(host, port)
+        self.owproxy = protocol.proxy(host, port)
-        await self.discover_devices()
+        self.devices = _discover_devices(self.owproxy)
-        if TYPE_CHECKING:
+
-            assert self.devices
+    async def initialize(self) -> None:
-        # Register discovered devices on Hub
+        """Initialize a config entry."""
-        device_registry = dr.async_get(self.hass)
+        await self._hass.async_add_executor_job(self._initialize)
+        # Populate the device registry
+        device_registry = dr.async_get(self._hass)
         for device in self.devices:
-            device_info: DeviceInfo = device.device_info
+            device_info = device.device_info
             device_registry.async_get_or_create(
-                config_entry_id=config_entry.entry_id,
+                config_entry_id=self._config_entry.entry_id,
                 identifiers=device_info[ATTR_IDENTIFIERS],
                 manufacturer=device_info[ATTR_MANUFACTURER],
                 model=device_info[ATTR_MODEL],
@@ -94,79 +90,59 @@ class OneWireHub:
                 via_device=device_info.get(ATTR_VIA_DEVICE),
             )
 
-    async def discover_devices(self) -> None:
-        """Discover all devices."""
-        if self.devices is None:
-            self.devices = await self.hass.async_add_executor_job(
-                self._discover_devices
-            )
 
-    def _discover_devices(
+def _discover_devices(
-        self, path: str = "/", parent_id: str | None = None
+    owproxy: protocol._Proxy, path: str = "/", parent_id: str | None = None
-    ) -> list[OWDeviceDescription]:
+) -> list[OWDeviceDescription]:
-        """Discover all server devices."""
+    """Discover all server devices."""
-        devices: list[OWDeviceDescription] = []
+    devices: list[OWDeviceDescription] = []
-        assert self.owproxy
+    for device_path in owproxy.dir(path):
-        for device_path in self.owproxy.dir(path):
+        device_id = os.path.split(os.path.split(device_path)[0])[1]
-            device_id = os.path.split(os.path.split(device_path)[0])[1]
+        device_family = owproxy.read(f"{device_path}family").decode()
-            device_family = self.owproxy.read(f"{device_path}family").decode()
+        _LOGGER.debug("read `%sfamily`: %s", device_path, device_family)
-            _LOGGER.debug("read `%sfamily`: %s", device_path, device_family)
+        device_type = _get_device_type(owproxy, device_path)
-            device_type = self._get_device_type(device_path)
+        if not _is_known_device(device_family, device_type):
-            if not _is_known_device(device_family, device_type):
+            _LOGGER.warning(
-                _LOGGER.warning(
+                "Ignoring unknown device family/type (%s/%s) found for device %s",
-                    "Ignoring unknown device family/type (%s/%s) found for device %s",
+                device_family,
-                    device_family,
+                device_type,
-                    device_type,
+                device_id,
-                    device_id,
+            )
+            continue
+        device_info = DeviceInfo(
+            identifiers={(DOMAIN, device_id)},
+            manufacturer=DEVICE_MANUFACTURER.get(device_family, MANUFACTURER_MAXIM),
+            model=device_type,
+            name=device_id,
+        )
+        if parent_id:
+            device_info[ATTR_VIA_DEVICE] = (DOMAIN, parent_id)
+        device = OWDeviceDescription(
+            device_info=device_info,
+            id=device_id,
+            family=device_family,
+            path=device_path,
+            type=device_type,
+        )
+        devices.append(device)
+        if device_branches := DEVICE_COUPLERS.get(device_family):
+            for branch in device_branches:
+                devices += _discover_devices(
+                    owproxy, f"{device_path}{branch}", device_id
                 )
-                continue
-            device_info: DeviceInfo = {
-                ATTR_IDENTIFIERS: {(DOMAIN, device_id)},
-                ATTR_MANUFACTURER: DEVICE_MANUFACTURER.get(
-                    device_family, MANUFACTURER_MAXIM
-                ),
-                ATTR_MODEL: device_type,
-                ATTR_NAME: device_id,
-            }
-            if parent_id:
-                device_info[ATTR_VIA_DEVICE] = (DOMAIN, parent_id)
-            device = OWDeviceDescription(
-                device_info=device_info,
-                id=device_id,
-                family=device_family,
-                path=device_path,
-                type=device_type,
-            )
-            devices.append(device)
-            if device_branches := DEVICE_COUPLERS.get(device_family):
-                for branch in device_branches:
-                    devices += self._discover_devices(
-                        f"{device_path}{branch}", device_id
-                    )
 
-        return devices
+    return devices
-
-    def _get_device_type(self, device_path: str) -> str | None:
-        """Get device model."""
-        if TYPE_CHECKING:
-            assert self.owproxy
-        try:
-            device_type = self.owproxy.read(f"{device_path}type").decode()
-        except protocol.ProtocolError as exc:
-            _LOGGER.debug("Unable to read `%stype`: %s", device_path, exc)
-            return None
-        _LOGGER.debug("read `%stype`: %s", device_path, device_type)
-        if device_type == "EDS":
-            device_type = self.owproxy.read(f"{device_path}device_type").decode()
-            _LOGGER.debug("read `%sdevice_type`: %s", device_path, device_type)
-        if TYPE_CHECKING:
-            assert isinstance(device_type, str)
-        return device_type
 
 
-class CannotConnect(HomeAssistantError):
+def _get_device_type(owproxy: protocol._Proxy, device_path: str) -> str | None:
-    """Error to indicate we cannot connect."""
+    """Get device model."""
-
+    try:
-
+        device_type: str = owproxy.read(f"{device_path}type").decode()
-class InvalidPath(HomeAssistantError):
+    except protocol.ProtocolError as exc:
-    """Error to indicate the path is invalid."""
+        _LOGGER.debug("Unable to read `%stype`: %s", device_path, exc)
+        return None
+    _LOGGER.debug("read `%stype`: %s", device_path, device_type)
+    if device_type == "EDS":
+        device_type = owproxy.read(f"{device_path}device_type").decode()
+        _LOGGER.debug("read `%sdevice_type`: %s", device_path, device_type)
+    return device_type

homeassistant/components/onewire/sensor.py

@@ -373,7 +373,6 @@ def get_entities(
         return []
 
     entities: list[OneWireSensor] = []
-    assert onewire_hub.owproxy
     for device in onewire_hub.devices:
         family = device.family
         device_type = device.type