From 7bb68b17245cb5768b817527cae72b7758d3815b Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Tue, 10 Jun 2025 11:25:02 -0400 Subject: [PATCH 1/3] Allow external radio types --- zha/application/const.py | 5 ----- zha/application/gateway.py | 43 +++++++++++++++++++++++++++++++++----- zha/application/helpers.py | 11 ++++++++++ zha/zigbee/device.py | 2 +- 4 files changed, 50 insertions(+), 11 deletions(-) diff --git a/zha/application/const.py b/zha/application/const.py index 2d0f0517c..3f8f391c2 100644 --- a/zha/application/const.py +++ b/zha/application/const.py @@ -151,11 +151,6 @@ def description(self) -> str: """Return radio type description.""" return self._desc - @property - def pretty_name(self) -> str: - """Return radio type name.""" - return self.description.split(" = ", 1)[0] - UNKNOWN = "unknown" UNKNOWN_MANUFACTURER = "unk_manufacturer" diff --git a/zha/application/gateway.py b/zha/application/gateway.py index 6bd93dce6..ce9d7ff43 100644 --- a/zha/application/gateway.py +++ b/zha/application/gateway.py @@ -7,6 +7,7 @@ from dataclasses import dataclass from datetime import timedelta from enum import Enum +import importlib import logging import time from typing import Any, Final, Self, TypeVar, cast @@ -188,9 +189,37 @@ def __init__(self, config: ZHAData) -> None: self.config.gateway = self @property - def radio_type(self) -> RadioType: + def radio_type(self) -> str: """Get the current radio type.""" - return RadioType[self.config.config.coordinator_configuration.radio_type] + return self.config.config.coordinator_configuration.radio_type + + def get_controller_description(self) -> str: + """Get the controller description for the current radio type.""" + external_radio_libraries = self.config.config.external_radio_libraries + + # Prefer external radio libraries + if self.radio_type in external_radio_libraries: + return external_radio_libraries[self.radio_type].description + + if self.radio_type in RadioType: + return RadioType[self.radio_type].description + + raise ValueError(f"Unknown radio type: {self.radio_type!r}") + + def get_controller_cls(self) -> type[ControllerApplication]: + """Get the controller class for the current radio type.""" + external_radio_libraries = self.config.config.external_radio_libraries + + if self.radio_type in external_radio_libraries: + module = importlib.import_module( + external_radio_libraries[self.radio_type].module + ) + return module.ControllerApplication + + if self.radio_type in RadioType: + return RadioType[self.radio_type].controller + + raise ValueError(f"Unknown radio type: {self.radio_type!r}") def get_application_controller_data(self) -> tuple[ControllerApplication, dict]: """Get an uninitialized instance of a zigpy `ControllerApplication`.""" @@ -208,12 +237,12 @@ def get_application_controller_data(self) -> tuple[ControllerApplication, dict]: # event loop, when a connection to a TCP coordinator fails in a specific way if ( CONF_USE_THREAD not in app_config - and self.radio_type is RadioType.ezsp + and self.radio_type == RadioType.ezsp.name and app_config[CONF_DEVICE][CONF_DEVICE_PATH].startswith("socket://") ): app_config[CONF_USE_THREAD] = False - return self.radio_type.controller, app_config + return self.get_controller_cls(), app_config @classmethod async def async_from_config(cls, config: ZHAData) -> Self: @@ -244,7 +273,11 @@ async def _async_initialize(self) -> None: """Initialize controller and connect radio.""" self.shutting_down = False - app_controller_cls, app_config = self.get_application_controller_data() + # `get_application_controller_data` can import packages and the blocking IO must + # be done in a separate thread + app_controller_cls, app_config = await self.async_add_executor_job( + self.get_application_controller_data + ) self.application_controller = await app_controller_cls.new( config=app_config, auto_form=False, diff --git a/zha/application/helpers.py b/zha/application/helpers.py index f6ffa7638..69ddb05f2 100644 --- a/zha/application/helpers.py +++ b/zha/application/helpers.py @@ -354,6 +354,14 @@ class DeviceOverridesConfiguration: type: Platform +@dataclass(kw_only=True, slots=True) +class ExternalRadioLibrary: + """ZHA external radio library configuration.""" + + module: str + description: str + + @dataclass(kw_only=True, slots=True) class ZHAConfiguration: """ZHA configuration.""" @@ -372,6 +380,9 @@ class ZHAConfiguration: alarm_control_panel_options: AlarmControlPanelOptions = dataclasses.field( default_factory=AlarmControlPanelOptions ) + external_radio_libraries: dict[str, ExternalRadioLibrary] = dataclasses.field( + default_factory=dict + ) @dataclasses.dataclass(kw_only=True, slots=True) diff --git a/zha/zigbee/device.py b/zha/zigbee/device.py index 38264d337..fc4b697ae 100644 --- a/zha/zigbee/device.py +++ b/zha/zigbee/device.py @@ -323,7 +323,7 @@ def model(self) -> str: if self.is_active_coordinator: model = self.gateway.application_controller.state.node_info.model if model is None: - return f"Generic Zigbee Coordinator ({self.gateway.radio_type.pretty_name})" + return f"Generic Zigbee Coordinator ({self.gateway.get_controller_description()})" return model if ( From 057d3ed150eb0071ef4abb150ddcaff15f66b6d8 Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Tue, 10 Jun 2025 17:59:55 -0400 Subject: [PATCH 2/3] Fix radio type member lookup --- zha/application/gateway.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/zha/application/gateway.py b/zha/application/gateway.py index ce9d7ff43..b4eb8b57a 100644 --- a/zha/application/gateway.py +++ b/zha/application/gateway.py @@ -201,7 +201,7 @@ def get_controller_description(self) -> str: if self.radio_type in external_radio_libraries: return external_radio_libraries[self.radio_type].description - if self.radio_type in RadioType: + if self.radio_type in RadioType.__members__: return RadioType[self.radio_type].description raise ValueError(f"Unknown radio type: {self.radio_type!r}") @@ -216,7 +216,7 @@ def get_controller_cls(self) -> type[ControllerApplication]: ) return module.ControllerApplication - if self.radio_type in RadioType: + if self.radio_type in RadioType.__members__: return RadioType[self.radio_type].controller raise ValueError(f"Unknown radio type: {self.radio_type!r}") From 0a08169ed3434d6c01403addc85dd0d91754d492 Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Tue, 10 Jun 2025 21:52:07 -0400 Subject: [PATCH 3/3] Migrate everything to `RADIO_LIBRARIES` --- tests/conftest.py | 16 ++++------ tests/test_device.py | 5 ++- tests/test_gateway.py | 43 ++----------------------- zha/application/const.py | 62 ------------------------------------ zha/application/gateway.py | 65 +++++++++++++++++++++----------------- zha/application/helpers.py | 64 +++++++++++++++++++++++++++++++------ zha/zigbee/device.py | 2 +- 7 files changed, 104 insertions(+), 153 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index c2621c1a7..aca1ea8bd 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -299,16 +299,12 @@ def __init__(self, data: ZHAData, app: ControllerApplication): async def __aenter__(self) -> Gateway: """Start the ZHA gateway.""" - with ( - patch( - "bellows.zigbee.application.ControllerApplication.new", - return_value=self.app, - ), - patch( - "bellows.zigbee.application.ControllerApplication", - return_value=self.app, - ), - ): + with patch( + "bellows.zigbee.application.ControllerApplication", + return_value=self.app, + ) as mock_app: + mock_app.new = AsyncMock(return_value=self.app) + self.zha_gateway = await Gateway.async_from_config(self.zha_data) await self.zha_gateway.async_initialize() await self.zha_gateway.async_block_till_done() diff --git a/tests/test_device.py b/tests/test_device.py index 398579512..0c6ef4eb1 100644 --- a/tests/test_device.py +++ b/tests/test_device.py @@ -387,7 +387,10 @@ async def test_coordinator_info_generic_name( current_coordinator = await join_zigpy_device(zha_gateway, current_coord_dev) assert current_coordinator.is_active_coordinator - assert current_coordinator.model == "Generic Zigbee Coordinator (EZSP)" + assert ( + current_coordinator.model + == "Generic Zigbee Coordinator (Silicon Labs EmberZNet)" + ) assert current_coordinator.manufacturer == "" diff --git a/tests/test_gateway.py b/tests/test_gateway.py index 347f88737..bcb1fcf55 100644 --- a/tests/test_gateway.py +++ b/tests/test_gateway.py @@ -27,7 +27,6 @@ CONF_USE_THREAD, ZHA_GW_MSG, ZHA_GW_MSG_CONNECTION_LOST, - RadioType, ) from zha.application.gateway import ( ConnectionLostEvent, @@ -192,10 +191,6 @@ async def test_gateway_starts_entity_exception( "bellows.zigbee.application.ControllerApplication.new", return_value=zigpy_app_controller, ), - patch( - "bellows.zigbee.application.ControllerApplication", - return_value=zigpy_app_controller, - ), patch( "zha.application.platforms.sensor.DeviceCounterSensor.__init__", side_effect=Exception, @@ -220,15 +215,9 @@ async def test_mains_devices_startup_polling_config( ) -> None: """Test mains powered device startup polling config is respected.""" - with ( - patch( - "bellows.zigbee.application.ControllerApplication.new", - return_value=zigpy_app_controller, - ), - patch( - "bellows.zigbee.application.ControllerApplication", - return_value=zigpy_app_controller, - ), + with patch( + "bellows.zigbee.application.ControllerApplication.new", + return_value=zigpy_app_controller, ): zha_data.config.device_options.enable_mains_startup_polling = enabled zha_gateway = await Gateway.async_from_config(zha_data) @@ -779,29 +768,3 @@ async def test_gateway_handle_message( assert zha_dev_basic.available is True assert zha_dev_basic.on_network is True - - -def test_radio_type(): - """Test radio type.""" - - assert RadioType.list() == [ - "EZSP = Silicon Labs EmberZNet protocol: Elelabs, HUSBZB-1, Telegesis", - "ZNP = Texas Instruments Z-Stack ZNP protocol: CC253x, CC26x2, CC13x2", - "deCONZ = dresden elektronik deCONZ protocol: ConBee I/II, RaspBee I/II", - "ZiGate = ZiGate Zigbee radios: PiZiGate, ZiGate USB-TTL, ZiGate WiFi", - "XBee = Digi XBee Zigbee radios: Digi XBee Series 2, 2C, 3", - ] - - assert ( - RadioType.get_by_description( - "EZSP = Silicon Labs EmberZNet protocol: Elelabs, HUSBZB-1, Telegesis" - ) - == RadioType.ezsp - ) - - assert RadioType.ezsp.description == ( - "EZSP = Silicon Labs EmberZNet protocol: Elelabs, HUSBZB-1, Telegesis" - ) - - with pytest.raises(ValueError): - RadioType.get_by_description("Invalid description") diff --git a/zha/application/const.py b/zha/application/const.py index 3f8f391c2..ff4b07a74 100644 --- a/zha/application/const.py +++ b/zha/application/const.py @@ -5,13 +5,7 @@ import enum from typing import Final -import bellows.zigbee.application -import zigpy.application import zigpy.types as t -import zigpy_deconz.zigbee.application -import zigpy_xbee.zigbee.application -import zigpy_zigate.zigbee.application -import zigpy_znp.zigbee.application ATTR_ACTIVE_COORDINATOR = "active_coordinator" ATTR_ARGS = "args" @@ -96,62 +90,6 @@ ZCL_INIT_ATTRS = "ZCL_INIT_ATTRS" REPORT_CONFIG = "REPORT_CONFIG" -_ControllerClsType = type[zigpy.application.ControllerApplication] - - -class RadioType(enum.Enum): - """Possible options for radio type.""" - - ezsp = ( - "EZSP = Silicon Labs EmberZNet protocol: Elelabs, HUSBZB-1, Telegesis", - bellows.zigbee.application.ControllerApplication, - ) - znp = ( - "ZNP = Texas Instruments Z-Stack ZNP protocol: CC253x, CC26x2, CC13x2", - zigpy_znp.zigbee.application.ControllerApplication, - ) - deconz = ( - "deCONZ = dresden elektronik deCONZ protocol: ConBee I/II, RaspBee I/II", - zigpy_deconz.zigbee.application.ControllerApplication, - ) - zigate = ( - "ZiGate = ZiGate Zigbee radios: PiZiGate, ZiGate USB-TTL, ZiGate WiFi", - zigpy_zigate.zigbee.application.ControllerApplication, - ) - xbee = ( - "XBee = Digi XBee Zigbee radios: Digi XBee Series 2, 2C, 3", - zigpy_xbee.zigbee.application.ControllerApplication, - ) - - @classmethod - def list(cls) -> list[str]: - """Return a list of descriptions.""" - return [e.description for e in RadioType] - - @classmethod - def get_by_description(cls, description: str) -> RadioType: - """Get radio by description.""" - for radio in cls: - if radio.description == description: - return radio - raise ValueError - - def __init__(self, description: str, controller_cls: _ControllerClsType) -> None: - """Init instance.""" - self._desc = description - self._ctrl_cls = controller_cls - - @property - def controller(self) -> _ControllerClsType: - """Return controller class.""" - return self._ctrl_cls - - @property - def description(self) -> str: - """Return radio type description.""" - return self._desc - - UNKNOWN = "unknown" UNKNOWN_MANUFACTURER = "unk_manufacturer" UNKNOWN_MODEL = "unk_model" diff --git a/zha/application/gateway.py b/zha/application/gateway.py index b4eb8b57a..17b804f5d 100644 --- a/zha/application/gateway.py +++ b/zha/application/gateway.py @@ -4,9 +4,11 @@ import asyncio from contextlib import suppress +import dataclasses from dataclasses import dataclass from datetime import timedelta from enum import Enum +from functools import cached_property import importlib import logging import time @@ -44,9 +46,14 @@ ZHA_GW_MSG_GROUP_MEMBER_REMOVED, ZHA_GW_MSG_GROUP_REMOVED, ZHA_GW_MSG_RAW_INIT, - RadioType, ) -from zha.application.helpers import DeviceAvailabilityChecker, GlobalUpdater, ZHAData +from zha.application.helpers import ( + RADIO_LIBRARIES, + DeviceAvailabilityChecker, + GlobalUpdater, + RadioLibrary, + ZHAData, +) from zha.async_ import ( AsyncUtilMixin, create_eager_task, @@ -193,35 +200,34 @@ def radio_type(self) -> str: """Get the current radio type.""" return self.config.config.coordinator_configuration.radio_type - def get_controller_description(self) -> str: - """Get the controller description for the current radio type.""" - external_radio_libraries = self.config.config.external_radio_libraries + @property + def radio_library(self) -> RadioLibrary: + """Get the current radio library.""" + radio_type = self.radio_type + radio_libraries = self.radio_libraries - # Prefer external radio libraries - if self.radio_type in external_radio_libraries: - return external_radio_libraries[self.radio_type].description + if radio_type not in radio_libraries: + raise ValueError(f"Unknown radio type: {radio_type!r}") - if self.radio_type in RadioType.__members__: - return RadioType[self.radio_type].description + return radio_libraries[radio_type] - raise ValueError(f"Unknown radio type: {self.radio_type!r}") + @cached_property + def radio_libraries(self) -> dict[str, RadioLibrary]: + """Get all available radio libraries.""" + radio_libraries = {} - def get_controller_cls(self) -> type[ControllerApplication]: - """Get the controller class for the current radio type.""" - external_radio_libraries = self.config.config.external_radio_libraries + for library in RADIO_LIBRARIES + self.config.config.external_radio_libraries: + import_path, cls_name = library.module_path.split(":", 1) + module = importlib.import_module(import_path) + radio_cls = getattr(module, cls_name) - if self.radio_type in external_radio_libraries: - module = importlib.import_module( - external_radio_libraries[self.radio_type].module + radio_libraries[library.radio_type] = dataclasses.replace( + library, controller=radio_cls ) - return module.ControllerApplication - if self.radio_type in RadioType.__members__: - return RadioType[self.radio_type].controller + return radio_libraries - raise ValueError(f"Unknown radio type: {self.radio_type!r}") - - def get_application_controller_data(self) -> tuple[ControllerApplication, dict]: + def get_application_controller_config(self) -> dict: """Get an uninitialized instance of a zigpy `ControllerApplication`.""" app_config = self.config.zigpy_config app_config[CONF_DEVICE] = { @@ -237,12 +243,12 @@ def get_application_controller_data(self) -> tuple[ControllerApplication, dict]: # event loop, when a connection to a TCP coordinator fails in a specific way if ( CONF_USE_THREAD not in app_config - and self.radio_type == RadioType.ezsp.name + and self.radio_type == "ezsp" and app_config[CONF_DEVICE][CONF_DEVICE_PATH].startswith("socket://") ): app_config[CONF_USE_THREAD] = False - return self.get_controller_cls(), app_config + return app_config @classmethod async def async_from_config(cls, config: ZHAData) -> Self: @@ -273,11 +279,12 @@ async def _async_initialize(self) -> None: """Initialize controller and connect radio.""" self.shutting_down = False - # `get_application_controller_data` can import packages and the blocking IO must - # be done in a separate thread - app_controller_cls, app_config = await self.async_add_executor_job( - self.get_application_controller_data + # `radio_library` imports packages and should be used in a separate thread + app_config = self.get_application_controller_config() + app_controller_cls = await self.async_add_executor_job( + lambda: self.radio_library.controller ) + self.application_controller = await app_controller_cls.new( config=app_config, auto_form=False, diff --git a/zha/application/helpers.py b/zha/application/helpers.py index 69ddb05f2..8f7ad6fa0 100644 --- a/zha/application/helpers.py +++ b/zha/application/helpers.py @@ -35,6 +35,8 @@ from zha.decorators import periodic if TYPE_CHECKING: + from zigpy.application import ControllerApplication + from zha.application.gateway import Gateway from zha.zigbee.cluster_handlers import ClusterHandler from zha.zigbee.device import Device @@ -46,6 +48,56 @@ _LOGGER = logging.getLogger(__name__) +@dataclass(kw_only=True, slots=True) +class RadioLibrary: + """ZHA external radio library configuration.""" + + radio_type: str + display_name: str + description: str + module_path: str + deprecated: bool = False + + # This module w + controller: type[ControllerApplication] = None # type: ignore[assignment] + + +RADIO_LIBRARIES = [ + RadioLibrary( + radio_type="ezsp", + display_name="EZSP", + description="Silicon Labs EmberZNet", + module_path="bellows.zigbee.application:ControllerApplication", + ), + RadioLibrary( + radio_type="znp", + display_name="ZNP", + description="Texas Instruments Z-Stack", + module_path="zigpy_znp.zigbee.application:ControllerApplication", + ), + RadioLibrary( + radio_type="deconz", + display_name="deCONZ", + description="dresden elektronik deCONZ", + module_path="zigpy_deconz.zigbee.application:ControllerApplication", + ), + RadioLibrary( + radio_type="zigate", + display_name="ZiGate", + description="ZiGate", + module_path="zigpy_zigate.zigbee.application:ControllerApplication", + deprecated=True, + ), + RadioLibrary( + radio_type="xbee", + display_name="XBee", + description="Digi XBee", + module_path="zigpy_xbee.zigbee.application:ControllerApplication", + deprecated=True, + ), +] + + @dataclass class BindingPair: """Information for binding.""" @@ -354,14 +406,6 @@ class DeviceOverridesConfiguration: type: Platform -@dataclass(kw_only=True, slots=True) -class ExternalRadioLibrary: - """ZHA external radio library configuration.""" - - module: str - description: str - - @dataclass(kw_only=True, slots=True) class ZHAConfiguration: """ZHA configuration.""" @@ -380,8 +424,8 @@ class ZHAConfiguration: alarm_control_panel_options: AlarmControlPanelOptions = dataclasses.field( default_factory=AlarmControlPanelOptions ) - external_radio_libraries: dict[str, ExternalRadioLibrary] = dataclasses.field( - default_factory=dict + external_radio_libraries: list[RadioLibrary] = dataclasses.field( + default_factory=list ) diff --git a/zha/zigbee/device.py b/zha/zigbee/device.py index fc4b697ae..64ab1acdf 100644 --- a/zha/zigbee/device.py +++ b/zha/zigbee/device.py @@ -323,7 +323,7 @@ def model(self) -> str: if self.is_active_coordinator: model = self.gateway.application_controller.state.node_info.model if model is None: - return f"Generic Zigbee Coordinator ({self.gateway.get_controller_description()})" + return f"Generic Zigbee Coordinator ({self.gateway.radio_library.description})" return model if (