Add update entity for second Zigbee radio (#136918)

* Add get_radio helper function

This is defined here primarily for use in simplifying otherwise repetitive
logic in the lambdas for entity descriptions.

* Get firmware manifests for second radio

* Create optional update entity for radio 2

* Add info fixture for SLZB-MR1

* Test for firmware updates of second radio

* Remove use of entity description creating entities

* Add idx to lambda functions

* Add latest_version lambda to ED

* Use Single zb_update description

* test radio2 update

* device type heading for release notes

* fix failing no internet test

* update release note tests

* assert radios

* fix return type installed_version

* refactor latest_version code

* update listener

* Dont create update entities for legacy firmware that can't upgrade

* Address review comments for update listener
This commit is contained in:
TimL 2025-02-05 13:34:18 +11:00 committed by GitHub
parent 369f897f41
commit 280f61dd77
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 273 additions and 83 deletions

View File

@ -4,7 +4,7 @@ from __future__ import annotations
from dataclasses import dataclass
from pysmlight import Api2
from pysmlight import Api2, Info, Radio
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, Platform
@ -61,3 +61,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: SmConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: SmConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
def get_radio(info: Info, idx: int) -> Radio:
"""Get the radio object from the info."""
assert info.radios is not None
return info.radios[idx]

View File

@ -9,7 +9,7 @@ from typing import TYPE_CHECKING
from pysmlight import Api2, Info, Sensors
from pysmlight.const import Settings, SettingsProp
from pysmlight.exceptions import SmlightAuthError, SmlightConnectionError
from pysmlight.web import Firmware
from pysmlight.models import FirmwareList
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
@ -38,8 +38,8 @@ class SmFwData:
"""SMLIGHT firmware data stored in the FirmwareUpdateCoordinator."""
info: Info
esp_firmware: list[Firmware] | None
zb_firmware: list[Firmware] | None
esp_firmware: FirmwareList
zb_firmware: list[FirmwareList]
class SmBaseDataUpdateCoordinator[_DataT](DataUpdateCoordinator[_DataT]):
@ -144,15 +144,30 @@ class SmFirmwareUpdateCoordinator(SmBaseDataUpdateCoordinator[SmFwData]):
async def _internal_update_data(self) -> SmFwData:
"""Fetch data from the SMLIGHT device."""
info = await self.client.get_info()
assert info.radios is not None
esp_firmware = None
zb_firmware = None
zb_firmware: list[FirmwareList] = []
try:
esp_firmware = await self.client.get_firmware_version(info.fw_channel)
zb_firmware = await self.client.get_firmware_version(
info.fw_channel, device=info.model, mode="zigbee"
zb_firmware.extend(
[
await self.client.get_firmware_version(
info.fw_channel,
device=info.model,
mode="zigbee",
zb_type=r.zb_type,
idx=idx,
)
for idx, r in enumerate(info.radios)
]
)
except SmlightConnectionError as err:
self.async_set_update_error(err)
return SmFwData(info=info, esp_firmware=esp_firmware, zb_firmware=zb_firmware)
return SmFwData(
info=info,
esp_firmware=esp_firmware,
zb_firmware=zb_firmware,
)

View File

@ -5,7 +5,7 @@ from __future__ import annotations
import asyncio
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any, Final
from typing import Any
from pysmlight.const import Events as SmEvents
from pysmlight.models import Firmware, Info
@ -22,34 +22,43 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import SmConfigEntry
from . import SmConfigEntry, get_radio
from .const import LOGGER
from .coordinator import SmFirmwareUpdateCoordinator, SmFwData
from .entity import SmEntity
def zigbee_latest_version(data: SmFwData, idx: int) -> Firmware | None:
"""Get the latest Zigbee firmware version."""
if idx < len(data.zb_firmware):
firmware_list = data.zb_firmware[idx]
if firmware_list:
return firmware_list[0]
return None
@dataclass(frozen=True, kw_only=True)
class SmUpdateEntityDescription(UpdateEntityDescription):
"""Describes SMLIGHT SLZB-06 update entity."""
installed_version: Callable[[Info], str | None]
fw_list: Callable[[SmFwData], list[Firmware] | None]
installed_version: Callable[[Info, int], str | None]
latest_version: Callable[[SmFwData, int], Firmware | None]
UPDATE_ENTITIES: Final = [
SmUpdateEntityDescription(
key="core_update",
translation_key="core_update",
installed_version=lambda x: x.sw_version,
fw_list=lambda x: x.esp_firmware,
),
SmUpdateEntityDescription(
key="zigbee_update",
translation_key="zigbee_update",
installed_version=lambda x: x.zb_version,
fw_list=lambda x: x.zb_firmware,
),
]
CORE_UPDATE_ENTITY = SmUpdateEntityDescription(
key="core_update",
translation_key="core_update",
installed_version=lambda x, idx: x.sw_version,
latest_version=lambda x, idx: x.esp_firmware[0] if x.esp_firmware else None,
)
ZB_UPDATE_ENTITY = SmUpdateEntityDescription(
key="zigbee_update",
translation_key="zigbee_update",
installed_version=lambda x, idx: get_radio(x, idx).zb_version,
latest_version=zigbee_latest_version,
)
async def async_setup_entry(
@ -58,10 +67,21 @@ async def async_setup_entry(
"""Set up the SMLIGHT update entities."""
coordinator = entry.runtime_data.firmware
async_add_entities(
SmUpdateEntity(coordinator, description) for description in UPDATE_ENTITIES
# updates not available for legacy API, user will get repair to update externally
if coordinator.legacy_api == 2:
return
entities = [SmUpdateEntity(coordinator, CORE_UPDATE_ENTITY)]
radios = coordinator.data.info.radios
assert radios is not None
entities.extend(
SmUpdateEntity(coordinator, ZB_UPDATE_ENTITY, idx)
for idx, _ in enumerate(radios)
)
async_add_entities(entities)
class SmUpdateEntity(SmEntity, UpdateEntity):
"""Representation for SLZB-06 update entities."""
@ -80,42 +100,46 @@ class SmUpdateEntity(SmEntity, UpdateEntity):
self,
coordinator: SmFirmwareUpdateCoordinator,
description: SmUpdateEntityDescription,
idx: int = 0,
) -> None:
"""Initialize the entity."""
super().__init__(coordinator)
self.entity_description = description
self._attr_unique_id = f"{coordinator.unique_id}-{description.key}"
device = description.key + (f"_{idx}" if idx else "")
self._attr_unique_id = f"{coordinator.unique_id}-{device}"
self._finished_event = asyncio.Event()
self._firmware: Firmware | None = None
self._unload: list[Callable] = []
self.idx = idx
async def async_added_to_hass(self) -> None:
"""When entity is added to hass."""
await super().async_added_to_hass()
self._handle_coordinator_update()
@callback
def _handle_coordinator_update(self) -> None:
"""Handle coordinator update callbacks."""
self._firmware = self.entity_description.latest_version(
self.coordinator.data, self.idx
)
if self._firmware:
self.async_write_ha_state()
@property
def installed_version(self) -> str | None:
"""Version installed.."""
data = self.coordinator.data
version = self.entity_description.installed_version(data.info)
return version if version != "-1" else None
return self.entity_description.installed_version(data.info, self.idx)
@property
def latest_version(self) -> str | None:
"""Latest version available for install."""
data = self.coordinator.data
if self.coordinator.legacy_api == 2:
return None
fw = self.entity_description.fw_list(data)
if fw and self.entity_description.key == "zigbee_update":
fw = [f for f in fw if f.type == data.info.zb_type]
if fw:
self._firmware = fw[0]
return self._firmware.ver
return None
return self._firmware.ver if self._firmware else None
def register_callbacks(self) -> None:
"""Register callbacks for SSE update events."""
@ -143,9 +167,14 @@ class SmUpdateEntity(SmEntity, UpdateEntity):
def release_notes(self) -> str | None:
"""Return release notes for firmware."""
if "zigbee" in self.entity_description.key:
notes = f"### {'ZNP' if self.idx else 'EZSP'} Firmware\n\n"
else:
notes = "### Core Firmware\n\n"
if self._firmware and self._firmware.notes:
return self._firmware.notes
notes += self._firmware.notes
return notes
return None
@ -192,7 +221,7 @@ class SmUpdateEntity(SmEntity, UpdateEntity):
self._attr_update_percentage = None
self.register_callbacks()
await self.coordinator.client.fw_update(self._firmware)
await self.coordinator.client.fw_update(self._firmware, self.idx)
# block until update finished event received
await self._finished_event.wait()

View File

@ -92,7 +92,10 @@ def mock_smlight_client(request: pytest.FixtureRequest) -> Generator[MagicMock]:
"""Return the firmware version."""
fw_list = []
if kwargs.get("mode") == "zigbee":
fw_list = load_json_array_fixture("zb_firmware.json", DOMAIN)
if kwargs.get("zb_type") == 0:
fw_list = load_json_array_fixture("zb_firmware.json", DOMAIN)
else:
fw_list = load_json_array_fixture("zb_firmware_router.json", DOMAIN)
else:
fw_list = load_json_array_fixture("esp_firmware.json", DOMAIN)

View File

@ -2,10 +2,10 @@
{
"mode": "ESP",
"type": null,
"notes": "CHANGELOG (Current 2.5.2 vs. Previous 2.3.6):\\r\\nFixed incorrect device type detection for some devices\\r\\nFixed web interface not working on some devices\\r\\nFixed disabled SSID/pass fields\\r\\n",
"notes": "CHANGELOG (Current 2.7.5 vs. Previous 2.3.6):\\r\\nFixed incorrect device type detection for some devices\\r\\nFixed web interface not working on some devices\\r\\nFixed disabled SSID/pass fields\\r\\n",
"rev": "20240830",
"link": "https://smlight.tech/flasher/firmware/bin/slzb06x/core/slzb-06-v2.5.2-ota.bin",
"ver": "v2.5.2",
"ver": "v2.7.5",
"dev": false,
"prod": true,
"baud": null

View File

@ -0,0 +1,19 @@
{
"coord_mode": 0,
"device_ip": "192.168.1.161",
"fs_total": 3456,
"fw_channel": "dev",
"legacy_api": 0,
"hostname": "SLZB-06p7",
"MAC": "AA:BB:CC:DD:EE:FF",
"model": "SLZB-06p7",
"ram_total": 296,
"sw_version": "v2.3.6",
"wifi_mode": 0,
"zb_flash_size": 704,
"zb_channel": 0,
"zb_hw": "CC2652P7",
"zb_ram_size": 152,
"zb_version": "20240314",
"zb_type": 0
}

View File

@ -0,0 +1,41 @@
{
"coord_mode": 0,
"device_ip": "192.168.1.161",
"fs_total": 3456,
"fw_channel": "dev",
"legacy_api": 0,
"hostname": "SLZB-MR1",
"MAC": "AA:BB:CC:DD:EE:FF",
"model": "SLZB-MR1",
"ram_total": 296,
"sw_version": "v2.7.3",
"wifi_mode": 0,
"zb_flash_size": 704,
"zb_channel": 0,
"zb_hw": "CC2652P7",
"zb_ram_size": 152,
"zb_version": "20240314",
"zb_type": 0,
"radios": [
{
"chip_index": 0,
"zb_hw": "EFR32MG21",
"zb_version": 20241127,
"zb_type": 0,
"zb_channel": 0,
"zb_ram_size": 152,
"zb_flash_size": 704,
"radioModes": [true, true, true, false, false]
},
{
"chip_index": 1,
"zb_hw": "CC2652P7",
"zb_version": 20240314,
"zb_type": 1,
"zb_channel": 0,
"zb_ram_size": 152,
"zb_flash_size": 704,
"radioModes": [true, true, true, false, false]
}
]
}

View File

@ -3,24 +3,13 @@
"mode": "ZB",
"type": 0,
"notes": "<b>SMLIGHT latest Coordinator release for CC2674P10 chips [16-Jul-2024]</b>:<br>- +20dB TRANSMIT POWER SUPPORT;<br>- SDK 7.41 based (latest);<br>",
"rev": "20240716",
"rev": "20250201",
"link": "https://smlight.tech/flasher/firmware/bin/slzb06x/zigbee/slzb06p10/znp-SLZB-06P10-20240716.bin",
"ver": "20240716",
"ver": "20250201",
"dev": false,
"prod": true,
"baud": 115200
},
{
"mode": "ZB",
"type": 1,
"notes": "<b>SMLIGHT latest ROUTER release for CC2674P10 chips [16-Jul-2024]</b>:<br>- SDK 7.41 based (latest);<br><a href='https://smlight.tech/legal/licenses/restrictive-smlight-slzb06-v20240328.txt' target='_blank' aria-current='true'><span>Terms of use</span></a><span>",
"rev": "20240716",
"link": "https://smlight.tech/flasher/firmware/bin/slzb06x/zigbee/slzb06p10/zr-ZR_SLZB-06P10-20240716.bin",
"ver": "20240716",
"dev": false,
"prod": true,
"baud": 0
},
{
"mode": "ZB",
"type": 0,

View File

@ -0,0 +1,13 @@
[
{
"mode": "ZB",
"type": 1,
"notes": "<b>SMLIGHT latest ROUTER release for CC2652P7 chips [16-Jul-2024]</b>:<br>- SDK 7.41 based (latest);<br><a href='https://smlight.tech/legal/licenses/restrictive-smlight-slzb06-v20240328.txt' target='_blank' aria-current='true'><span>Terms of use</span></a><span> - by downloading and installing this firmware, you agree to the aforementioned terms.",
"rev": "20240716",
"link": "https://smlight.tech/flasher/firmware/bin/slzb06x/zigbee/slzb06p10/znp-SLZB-06P10-20240716.bin",
"ver": "20240716",
"dev": false,
"prod": true,
"baud": 115200
}
]

View File

@ -42,7 +42,7 @@
'friendly_name': 'Mock Title Core firmware',
'in_progress': False,
'installed_version': 'v2.3.6',
'latest_version': 'v2.5.2',
'latest_version': 'v2.7.5',
'release_summary': None,
'release_url': None,
'skipped_version': None,
@ -101,7 +101,7 @@
'friendly_name': 'Mock Title Zigbee firmware',
'in_progress': False,
'installed_version': '20240314',
'latest_version': '20240716',
'latest_version': '20250201',
'release_summary': None,
'release_url': None,
'skipped_version': None,

View File

@ -85,6 +85,7 @@ async def test_async_setup_no_internet(
freezer: FrozenDateTimeFactory,
) -> None:
"""Test we still load integration when no internet is available."""
side_effect = mock_smlight_client.get_firmware_version.side_effect
mock_smlight_client.get_firmware_version.side_effect = SmlightConnectionError
await setup_integration(hass, mock_config_entry_host)
@ -101,7 +102,7 @@ async def test_async_setup_no_internet(
assert entity is not None
assert entity.state == STATE_UNKNOWN
mock_smlight_client.get_firmware_version.side_effect = None
mock_smlight_client.get_firmware_version.side_effect = side_effect
freezer.tick(SCAN_FIRMWARE_INTERVAL)
async_fire_time_changed(hass)

View File

@ -4,13 +4,13 @@ from datetime import timedelta
from unittest.mock import MagicMock, patch
from freezegun.api import FrozenDateTimeFactory
from pysmlight import Firmware, Info
from pysmlight import Firmware, Info, Radio
from pysmlight.const import Events as SmEvents
from pysmlight.sse import MessageEvent
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.smlight.const import SCAN_FIRMWARE_INTERVAL
from homeassistant.components.smlight.const import DOMAIN, SCAN_FIRMWARE_INTERVAL
from homeassistant.components.update import (
ATTR_IN_PROGRESS,
ATTR_INSTALLED_VERSION,
@ -27,7 +27,12 @@ from homeassistant.helpers import entity_registry as er
from . import get_mock_event_function
from .conftest import setup_integration
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
from tests.common import (
MockConfigEntry,
async_fire_time_changed,
load_json_object_fixture,
snapshot_platform,
)
from tests.typing import WebSocketGenerator
pytestmark = [
@ -62,12 +67,14 @@ MOCK_FIRMWARE_FAIL = MessageEvent(
MOCK_FIRMWARE_NOTES = [
Firmware(
ver="v2.3.6",
ver="v2.7.2",
mode="ESP",
notes=None,
)
]
MOCK_RADIO = Radio(chip_index=1, zb_channel=0, zb_type=0, zb_version="20240716")
@pytest.fixture
def platforms() -> list[Platform]:
@ -103,7 +110,7 @@ async def test_update_firmware(
state = hass.states.get(entity_id)
assert state.state == STATE_ON
assert state.attributes[ATTR_INSTALLED_VERSION] == "v2.3.6"
assert state.attributes[ATTR_LATEST_VERSION] == "v2.5.2"
assert state.attributes[ATTR_LATEST_VERSION] == "v2.7.5"
await hass.services.async_call(
PLATFORM,
@ -126,7 +133,7 @@ async def test_update_firmware(
event_function(MOCK_FIRMWARE_DONE)
mock_smlight_client.get_info.return_value = Info(
sw_version="v2.5.2",
sw_version="v2.7.5",
)
freezer.tick(timedelta(seconds=5))
@ -135,8 +142,50 @@ async def test_update_firmware(
state = hass.states.get(entity_id)
assert state.state == STATE_OFF
assert state.attributes[ATTR_INSTALLED_VERSION] == "v2.5.2"
assert state.attributes[ATTR_LATEST_VERSION] == "v2.5.2"
assert state.attributes[ATTR_INSTALLED_VERSION] == "v2.7.5"
assert state.attributes[ATTR_LATEST_VERSION] == "v2.7.5"
async def test_update_zigbee2_firmware(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
mock_config_entry: MockConfigEntry,
mock_smlight_client: MagicMock,
) -> None:
"""Test update of zigbee2 firmware where available."""
mock_smlight_client.get_info.return_value = Info.from_dict(
load_json_object_fixture("info-MR1.json", DOMAIN)
)
await setup_integration(hass, mock_config_entry)
entity_id = "update.mock_title_zigbee_firmware_2"
state = hass.states.get(entity_id)
assert state.state == STATE_ON
assert state.attributes[ATTR_INSTALLED_VERSION] == "20240314"
assert state.attributes[ATTR_LATEST_VERSION] == "20240716"
await hass.services.async_call(
PLATFORM,
SERVICE_INSTALL,
{ATTR_ENTITY_ID: entity_id},
blocking=False,
)
assert len(mock_smlight_client.fw_update.mock_calls) == 1
event_function = get_mock_event_function(mock_smlight_client, SmEvents.FW_UPD_done)
event_function(MOCK_FIRMWARE_DONE)
with patch(
"homeassistant.components.smlight.update.get_radio", return_value=MOCK_RADIO
):
freezer.tick(timedelta(seconds=5))
async_fire_time_changed(hass)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_OFF
assert state.attributes[ATTR_INSTALLED_VERSION] == "20240716"
assert state.attributes[ATTR_LATEST_VERSION] == "20240716"
async def test_update_legacy_firmware_v2(
@ -156,7 +205,7 @@ async def test_update_legacy_firmware_v2(
state = hass.states.get(entity_id)
assert state.state == STATE_ON
assert state.attributes[ATTR_INSTALLED_VERSION] == "v2.0.18"
assert state.attributes[ATTR_LATEST_VERSION] == "v2.5.2"
assert state.attributes[ATTR_LATEST_VERSION] == "v2.7.5"
await hass.services.async_call(
PLATFORM,
@ -172,7 +221,7 @@ async def test_update_legacy_firmware_v2(
event_function(MOCK_FIRMWARE_DONE)
mock_smlight_client.get_info.return_value = Info(
sw_version="v2.5.2",
sw_version="v2.7.5",
)
freezer.tick(SCAN_FIRMWARE_INTERVAL)
@ -181,8 +230,8 @@ async def test_update_legacy_firmware_v2(
state = hass.states.get(entity_id)
assert state.state == STATE_OFF
assert state.attributes[ATTR_INSTALLED_VERSION] == "v2.5.2"
assert state.attributes[ATTR_LATEST_VERSION] == "v2.5.2"
assert state.attributes[ATTR_INSTALLED_VERSION] == "v2.7.5"
assert state.attributes[ATTR_LATEST_VERSION] == "v2.7.5"
async def test_update_firmware_failed(
@ -196,7 +245,7 @@ async def test_update_firmware_failed(
state = hass.states.get(entity_id)
assert state.state == STATE_ON
assert state.attributes[ATTR_INSTALLED_VERSION] == "v2.3.6"
assert state.attributes[ATTR_LATEST_VERSION] == "v2.5.2"
assert state.attributes[ATTR_LATEST_VERSION] == "v2.7.5"
await hass.services.async_call(
PLATFORM,
@ -233,7 +282,7 @@ async def test_update_reboot_timeout(
state = hass.states.get(entity_id)
assert state.state == STATE_ON
assert state.attributes[ATTR_INSTALLED_VERSION] == "v2.3.6"
assert state.attributes[ATTR_LATEST_VERSION] == "v2.5.2"
assert state.attributes[ATTR_LATEST_VERSION] == "v2.7.5"
with (
patch(
@ -267,18 +316,29 @@ async def test_update_reboot_timeout(
mock_warning.assert_called_once()
@pytest.mark.parametrize(
"entity_id",
[
"update.mock_title_core_firmware",
"update.mock_title_zigbee_firmware",
"update.mock_title_zigbee_firmware_2",
],
)
async def test_update_release_notes(
hass: HomeAssistant,
entity_id: str,
freezer: FrozenDateTimeFactory,
mock_config_entry: MockConfigEntry,
mock_smlight_client: MagicMock,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test firmware release notes."""
mock_smlight_client.get_info.return_value = Info.from_dict(
load_json_object_fixture("info-MR1.json", DOMAIN)
)
await setup_integration(hass, mock_config_entry)
ws_client = await hass_ws_client(hass)
await hass.async_block_till_done()
entity_id = "update.mock_title_core_firmware"
state = hass.states.get(entity_id)
assert state
@ -294,16 +354,30 @@ async def test_update_release_notes(
result = await ws_client.receive_json()
assert result["result"] is not None
async def test_update_blank_release_notes(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_smlight_client: MagicMock,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test firmware missing release notes."""
entity_id = "update.mock_title_core_firmware"
mock_smlight_client.get_firmware_version.side_effect = None
mock_smlight_client.get_firmware_version.return_value = MOCK_FIRMWARE_NOTES
freezer.tick(SCAN_FIRMWARE_INTERVAL)
async_fire_time_changed(hass)
await setup_integration(hass, mock_config_entry)
ws_client = await hass_ws_client(hass)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_ON
await ws_client.send_json(
{
"id": 2,
"id": 1,
"type": "update/release_notes",
"entity_id": entity_id,
}