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 2d0f0517c..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,67 +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 - - @property - def pretty_name(self) -> str: - """Return radio type name.""" - return self.description.split(" = ", 1)[0] - - UNKNOWN = "unknown" UNKNOWN_MANUFACTURER = "unk_manufacturer" UNKNOWN_MODEL = "unk_model" diff --git a/zha/application/gateway.py b/zha/application/gateway.py index 6bd93dce6..17b804f5d 100644 --- a/zha/application/gateway.py +++ b/zha/application/gateway.py @@ -4,9 +4,12 @@ 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 from typing import Any, Final, Self, TypeVar, cast @@ -43,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, @@ -188,11 +196,38 @@ 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 + + @property + def radio_library(self) -> RadioLibrary: + """Get the current radio library.""" + radio_type = self.radio_type + radio_libraries = self.radio_libraries + + if radio_type not in radio_libraries: + raise ValueError(f"Unknown radio type: {radio_type!r}") + + return radio_libraries[radio_type] + + @cached_property + def radio_libraries(self) -> dict[str, RadioLibrary]: + """Get all available radio libraries.""" + radio_libraries = {} - def get_application_controller_data(self) -> tuple[ControllerApplication, dict]: + 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) + + radio_libraries[library.radio_type] = dataclasses.replace( + library, controller=radio_cls + ) + + return radio_libraries + + 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] = { @@ -208,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 is RadioType.ezsp + and self.radio_type == "ezsp" and app_config[CONF_DEVICE][CONF_DEVICE_PATH].startswith("socket://") ): app_config[CONF_USE_THREAD] = False - return self.radio_type.controller, app_config + return app_config @classmethod async def async_from_config(cls, config: ZHAData) -> Self: @@ -244,7 +279,12 @@ 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() + # `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 f6ffa7638..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.""" @@ -372,6 +424,9 @@ class ZHAConfiguration: alarm_control_panel_options: AlarmControlPanelOptions = dataclasses.field( default_factory=AlarmControlPanelOptions ) + external_radio_libraries: list[RadioLibrary] = dataclasses.field( + default_factory=list + ) @dataclasses.dataclass(kw_only=True, slots=True) diff --git a/zha/zigbee/device.py b/zha/zigbee/device.py index 38264d337..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.radio_type.pretty_name})" + return f"Generic Zigbee Coordinator ({self.gateway.radio_library.description})" return model if (