Force TX after CCA failure on EZSP v7+

bellows/ezsp/config.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
 import dataclasses
+import typing
 
 import bellows.ezsp.v4.types as types_v4
 import bellows.ezsp.v6.types as types_v6
+import bellows.ezsp.v7.types as types_v7
 import bellows.types as t
 
 
@@ -14,6 +16,12 @@ class RuntimeConfig:
  minimum: bool = False
 
 
+@dataclasses.dataclass(frozen=True)
+class ValueConfig:
+ value_id: t.enum8
+ value: typing.Any
+
+
 DEFAULT_CONFIG_COMMON = [
  RuntimeConfig(
  config_id=types_v4.EzspConfigId.CONFIG_INDIRECT_TRANSMISSION_TIMEOUT,
@@ -105,6 +113,10 @@ class RuntimeConfig:
  ),
  value=90,
  ),
+ ValueConfig(
+ value_id=types_v7.EzspValueId.VALUE_FORCE_TX_AFTER_FAILED_CCA_ATTEMPTS,
+ value=t.uint8_t(1),
+ ),
 ] + DEFAULT_CONFIG_COMMON
 
 

bellows/ezsp/protocol.py

@@ -37,13 +37,6 @@ def __init__(self, cb_handler: Callable, gateway: GatewayType) -> None:
  }
  self.tc_policy = 0
 
- async def _cfg(self, config_id: int, value: Any) -> None:
- (status,) = await self.setConfigurationValue(config_id, value)
- if status != self.types.EmberStatus.SUCCESS:
- LOGGER.warning(
- "Couldn't set %s=%s configuration value: %s", config_id, value, status
- )
-
  def _ezsp_frame(self, name: str, *args: Tuple[Any, ...]) -> bytes:
  """Serialize the named frame and data."""
  c = self.COMMANDS[name]
@@ -66,16 +59,21 @@ async def initialize(self, zigpy_config: Dict) -> None:
  """Initialize EmberZNet Stack."""
 
  # Prevent circular import
- from bellows.ezsp.config import DEFAULT_CONFIG, RuntimeConfig
+ from bellows.ezsp.config import DEFAULT_CONFIG, RuntimeConfig, ValueConfig
 
  # Not all config will be present in every EZSP version so only use valid keys
  ezsp_config = {}
+ ezsp_values = {}
 
  for cfg in DEFAULT_CONFIG[self.VERSION]:
- config_id = self.types.EzspConfigId[cfg.config_id.name]
- ezsp_config[cfg.config_id.name] = dataclasses.replace(
- cfg, config_id=config_id
- )
+ if isinstance(cfg, RuntimeConfig):
+ ezsp_config[cfg.config_id.name] = dataclasses.replace(
+ cfg, config_id=self.types.EzspConfigId[cfg.config_id.name]
+ )
+ elif isinstance(cfg, ValueConfig):
+ ezsp_values[cfg.value_id.name] = dataclasses.replace(
+ cfg, value_id=self.types.EzspValueId[cfg.value_id.name]
+ )
 
  # Override the defaults with user-specified values (or `None` for deletions)
  for name, value in self.SCHEMAS[CONF_EZSP_CONFIG](
@@ -99,6 +97,34 @@ async def initialize(self, zigpy_config: Dict) -> None:
  ],
  }
 
+ # First, set the values
+ for cfg in ezsp_values.values():
+ # XXX: A read failure does not mean the value is not writeable!
+ status, current_value = await self.getValue(cfg.value_id)
+
+ if status == self.types.EmberStatus.SUCCESS:
+ current_value, _ = type(cfg.value).deserialize(current_value)
+ else:
+ current_value = None
+
+ LOGGER.debug(
+ "Setting value %s = %s (old value %s)",
+ cfg.value_id.name,
+ cfg.value,
+ current_value,
+ )
+
+ (status,) = await self.setValue(cfg.value_id, cfg.value.serialize())
+
+ if status != self.types.EmberStatus.SUCCESS:
+ LOGGER.debug(
+ "Could not set value %s = %s: %s",
+ cfg.value_id.name,
+ cfg.value,
+ status,
+ )
+ continue
+
  # Finally, set the config
  for cfg in ezsp_config.values():
  (status, current_value) = await self.getConfigurationValue(cfg.config_id)
@@ -123,7 +149,16 @@ async def initialize(self, zigpy_config: Dict) -> None:
  cfg.value,
  current_value,
  )
- await self._cfg(cfg.config_id, cfg.value)
+
+ (status,) = await self.setConfigurationValue(cfg.config_id, cfg.value)
+ if status != self.types.EmberStatus.SUCCESS:
+ LOGGER.debug(
+ "Could not set config %s = %s: %s",
+ cfg.config_id,
+ cfg.value,
+ status,
+ )
+ continue
 
  async def get_free_buffers(self) -> Optional[int]:
  status, value = await self.getValue(self.types.EzspValueId.VALUE_FREE_BUFFERS)

tests/test_ezsp_protocol.py

@@ -62,26 +62,6 @@ def test_receive_reply_invalid_command(prot_hndl):
  assert prot_hndl._handle_callback.call_count == 0
 
 
-async def test_cfg_initialize(prot_hndl, caplog):
- """Test initialization."""
-
- p1 = patch.object(prot_hndl, "setConfigurationValue", new=AsyncMock())
- p2 = patch.object(
- prot_hndl,
- "getConfigurationValue",
- new=AsyncMock(return_value=(t.EzspStatus.SUCCESS, 22)),
- )
- p3 = patch.object(prot_hndl, "get_free_buffers", new=AsyncMock(22))
- with p1 as cfg_mock, p2, p3:
- cfg_mock.return_value = (t.EzspStatus.SUCCESS,)
- await prot_hndl.initialize({"ezsp_config": {}, "source_routing": True})
-
- cfg_mock.return_value = (t.EzspStatus.ERROR_OUT_OF_MEMORY,)
- with caplog.at_level(logging.WARNING):
- await prot_hndl.initialize({"ezsp_config": {}, "source_routing": False})
- assert "Couldn't set" in caplog.text
-
-
 async def test_config_initialize_husbzb1(prot_hndl):
  """Test timeouts are properly set for HUSBZB-1."""
 
@@ -117,15 +97,48 @@ async def test_config_initialize_husbzb1(prot_hndl):
 
 
 @pytest.mark.parametrize("prot_hndl_cls", EZSP._BY_VERSION.values())
-async def test_config_initialize(prot_hndl_cls):
+async def test_config_initialize(prot_hndl_cls, caplog):
  """Test config initialization for all protocol versions."""
 
  prot_hndl = prot_hndl_cls(MagicMock(), MagicMock())
  prot_hndl.getConfigurationValue = AsyncMock(return_value=(t.EzspStatus.SUCCESS, 0))
  prot_hndl.setConfigurationValue = AsyncMock(return_value=(t.EzspStatus.SUCCESS,))
 
+ prot_hndl.setValue = AsyncMock(return_value=(t.EzspStatus.SUCCESS,))
+ prot_hndl.getValue = AsyncMock(return_value=(t.EzspStatus.SUCCESS, b"\xFF"))
+
  await prot_hndl.initialize({"ezsp_config": {}})
 
+ with caplog.at_level(logging.DEBUG):
+ prot_hndl.setConfigurationValue.return_value = (
+ t.EzspStatus.ERROR_OUT_OF_MEMORY,
+ )
+ await prot_hndl.initialize({"ezsp_config": {}})
+
+ assert "Could not set config" in caplog.text
+ prot_hndl.setConfigurationValue.return_value = (t.EzspStatus.SUCCESS,)
+ caplog.clear()
+
+ # EZSPv6 does not set any values on startup
+ if prot_hndl_cls.VERSION < 7:
+ return
+
+ prot_hndl.setValue.reset_mock()
+ prot_hndl.getValue.return_value = (t.EzspStatus.ERROR_INVALID_ID, b"")
+ await prot_hndl.initialize({"ezsp_config": {}})
+ assert len(prot_hndl.setValue.mock_calls) == 1
+
+ prot_hndl.getValue = AsyncMock(return_value=(t.EzspStatus.SUCCESS, b"\xFF"))
+ caplog.clear()
+
+ with caplog.at_level(logging.DEBUG):
+ prot_hndl.setValue.return_value = (t.EzspStatus.ERROR_INVALID_ID,)
+ await prot_hndl.initialize({"ezsp_config": {}})
+
+ assert "Could not set value" in caplog.text
+ prot_hndl.setValue.return_value = (t.EzspStatus.SUCCESS,)
+ caplog.clear()
+
 
 async def test_cfg_initialize_skip(prot_hndl):
  """Test initialization."""