Allow external radio types

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

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 == ""
 
 

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

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"

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 @@
         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 @@
         # 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 @@
         """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,

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)

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 (