Add Tasmota firmware update availability support

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

homeassistant/components/tasmota/__init__.py

@@ -31,11 +31,12 @@ from .const import (
     DATA_UNSUB,
     PLATFORMS,
 )
+from .coordinator import TasmotaConfigEntry, TasmotaLatestReleaseUpdateCoordinator
 
 _LOGGER = logging.getLogger(__name__)
 
 
-async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+async def async_setup_entry(hass: HomeAssistant, entry: TasmotaConfigEntry) -> bool:
     """Set up Tasmota from a config entry."""
     hass.data[DATA_UNSUB] = []
 
@@ -63,6 +64,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 
     device_registry = dr.async_get(hass)
 
+    entry.runtime_data = TasmotaLatestReleaseUpdateCoordinator(hass, entry)
+
     async def async_discover_device(config: TasmotaDeviceConfig, mac: str) -> None:
         """Discover and add a Tasmota device."""
         await async_setup_device(

homeassistant/components/tasmota/const.py

@@ -19,6 +19,7 @@ PLATFORMS = [
     Platform.LIGHT,
     Platform.SENSOR,
     Platform.SWITCH,
+    Platform.UPDATE,
 ]
 
 TASMOTA_EVENT = "tasmota_event"

homeassistant/components/tasmota/coordinator.py

@@ -0,0 +1,51 @@
+"""Data update coordinators for Tasmota."""
+
+from datetime import timedelta
+import logging
+from typing import override
+
+from aiogithubapi import (
+    GitHubAPI,
+    GitHubConnectionException,
+    GitHubException,
+    GitHubRatelimitException,
+    GitHubReleaseModel,
+)
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+type TasmotaConfigEntry = ConfigEntry[TasmotaLatestReleaseUpdateCoordinator]
+
+
+class TasmotaLatestReleaseUpdateCoordinator(DataUpdateCoordinator[GitHubReleaseModel]):
+    """Data update coordinator for Tasmota latest release info."""
+
+    def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry) -> None:
+        """Initialize the coordinator."""
+        self.client = GitHubAPI(session=async_get_clientsession(hass))
+        super().__init__(
+            hass,
+            logger=logging.getLogger(__name__),
+            config_entry=config_entry,
+            name="Tasmota latest release",
+            update_interval=timedelta(days=1),
+        )
+
+    @override
+    async def _async_update_data(self) -> GitHubReleaseModel:
+        """Get new data."""
+        try:
+            response = await self.client.repos.releases.latest("arendst/Tasmota")
+            if response.data is None:
+                raise UpdateFailed("No data received")
+        except (GitHubConnectionException, GitHubRatelimitException) as ex:
+            # Expected/transient, just wrap as failure
+            raise UpdateFailed(ex) from ex
+        except GitHubException as ex:
+            self.logger.exception("Unexpected GitHub exception")
+            raise UpdateFailed(ex) from ex
+        else:
+            return response.data

homeassistant/components/tasmota/discovery.py

@@ -39,6 +39,7 @@ ALREADY_DISCOVERED = "tasmota_discovered_components"
 DISCOVERY_DATA = "tasmota_discovery_data"
 TASMOTA_DISCOVERY_ENTITY_NEW = "tasmota_discovery_entity_new_{}"
 TASMOTA_DISCOVERY_ENTITY_UPDATED = "tasmota_discovery_entity_updated_{}_{}_{}_{}"
+TASMOTA_DISCOVERY_DEVICE_DISCOVERED = "tasmota_discovery_device_discovered"
 TASMOTA_DISCOVERY_INSTANCE = "tasmota_discovery_instance"
 
 MQTT_TOPIC_URL = "https://tasmota.github.io/docs/Home-Assistant/#tasmota-integration"
@@ -293,6 +294,8 @@ async def async_start(  # noqa: C901
             for tasmota_entity_config, discovery_hash in tasmota_entities:
                 _discover_entity(tasmota_entity_config, discovery_hash, platform)
 
+        async_dispatcher_send(hass, TASMOTA_DISCOVERY_DEVICE_DISCOVERED, mac)
+
     async def async_sensors_discovered(
         sensors: list[tuple[TasmotaBaseSensorConfig, DiscoveryHashType]], mac: str
     ) -> None:

homeassistant/components/tasmota/manifest.json

@@ -8,5 +8,5 @@
   "iot_class": "local_push",
   "loggers": ["hatasmota"],
   "mqtt": ["tasmota/discovery/#"],
-  "requirements": ["HATasmota==0.10.1"]
+  "requirements": ["HATasmota==0.10.1", "aiogithubapi==26.0.0"]
 }

homeassistant/components/tasmota/update.py

@@ -0,0 +1,126 @@
+"""Update entity for Tasmota."""
+
+import re
+from typing import override
+
+from homeassistant.components.update import (
+    UpdateDeviceClass,
+    UpdateEntity,
+    UpdateEntityFeature,
+)
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers import device_registry as dr
+from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, DeviceEntry
+from homeassistant.helpers.dispatcher import async_dispatcher_connect
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DATA_REMOVE_DISCOVER_COMPONENT
+from .coordinator import TasmotaConfigEntry, TasmotaLatestReleaseUpdateCoordinator
+from .discovery import TASMOTA_DISCOVERY_DEVICE_DISCOVERED
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: TasmotaConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up Tasmota update entities."""
+    coordinator = config_entry.runtime_data
+
+    device_registry = dr.async_get(hass)
+    added_macs: set[str] = set()
+
+    @callback
+    def async_device_discovered(mac: str) -> None:
+        """Create update entity for a newly discovered Tasmota device."""
+        if mac not in added_macs and (
+            device := device_registry.async_get_device(
+                connections={(CONNECTION_NETWORK_MAC, mac)}
+            )
+        ):
+            added_macs.add(mac)
+            async_add_entities([TasmotaUpdateEntity(coordinator, device)])
+
+    hass.data[DATA_REMOVE_DISCOVER_COMPONENT.format(Platform.UPDATE)] = (
+        async_dispatcher_connect(
+            hass,
+            TASMOTA_DISCOVERY_DEVICE_DISCOVERED,
+            async_device_discovered,
+        )
+    )
+
+    await coordinator.async_request_refresh()
+
+
+class TasmotaUpdateEntity(
+    CoordinatorEntity[TasmotaLatestReleaseUpdateCoordinator], UpdateEntity
+):
+    """Representation of a Tasmota update entity."""
+
+    _attr_device_class = UpdateDeviceClass.FIRMWARE
+    _attr_has_entity_name = True
+    _attr_name = "Firmware"
+    _attr_title = "Tasmota firmware"
+    _attr_supported_features = UpdateEntityFeature.RELEASE_NOTES
+
+    def __init__(
+        self,
+        coordinator: TasmotaLatestReleaseUpdateCoordinator,
+        device_entry: DeviceEntry,
+    ) -> None:
+        """Initialize the Tasmota update entity."""
+        super().__init__(coordinator=coordinator)
+        self._connections = device_entry.connections
+        self._attr_device_info = dr.DeviceInfo(connections=self._connections)
+        for connection_type, connection_value in self._connections:
+            if connection_type == dr.CONNECTION_NETWORK_MAC:
+                self._attr_unique_id = connection_value
+                break
+
+    @property
+    @override
+    def installed_version(self) -> str | None:
+        """Return the installed version."""
+        if self.hass and (
+            device := dr.async_get(self.hass).async_get_device(
+                connections=self._connections
+            )
+        ):
+            return device.sw_version
+        return None
+
+    @property
+    @override
+    def latest_version(self) -> str | None:
+        """Return the latest version."""
+        if not self.coordinator.data:
+            return None
+        return self.coordinator.data.tag_name.removeprefix("v")
+
+    @property
+    @override
+    def release_url(self) -> str | None:
+        """Return the release URL."""
+        if not self.coordinator.data:
+            return None
+        return self.coordinator.data.html_url
+
+    @property
+    @override
+    def release_summary(self) -> str | None:
+        """Return the release summary."""
+        if not self.coordinator.data:
+            return None
+        return self.coordinator.data.name
+
+    @override
+    def release_notes(self) -> str | None:
+        """Return the release notes."""
+        if not self.coordinator.data or not self.coordinator.data.body:
+            return None
+        # Remove the picture tag, it uses relative URLs that won't work in the UI
+        return re.sub(
+            r"^<picture>.*?</picture>", "", self.coordinator.data.body, flags=re.DOTALL
+        )

requirements_all.txt

@@ -276,6 +276,7 @@ aioftp==0.21.3
 aioghost==0.4.16
 
 # homeassistant.components.github
+# homeassistant.components.tasmota
 aiogithubapi==26.0.0
 
 # homeassistant.components.guardian

script/hassfest/requirements.py

@@ -216,6 +216,7 @@ FORBIDDEN_PACKAGE_EXCEPTIONS: dict[str, dict[str, set[str]]] = {
     "slimproto": {"aioslimproto": {"async-timeout"}},
     "surepetcare": {"surepy": {"async-timeout"}},
     "tailwind": {"gotailwind": {"backoff"}},
+    "tasmota": {"aiogithubapi": {"backoff"}},
     "technove": {"python-technove": {"backoff"}},
     "tibber": {"gql": {"backoff"}},
     "toon": {"toonapi": {"backoff"}},

tests/components/tasmota/conftest.py

@@ -1,7 +1,8 @@
 """Test fixtures for Tasmota component."""
 
-from unittest.mock import patch
+from unittest.mock import AsyncMock, patch
 
+from aiogithubapi import GitHubReleaseModel
 from hatasmota.discovery import get_status_sensor_entities
 import pytest
 
@@ -37,6 +38,27 @@ def disable_status_sensor(status_sensor_disabled):
         yield
 
 
+@pytest.fixture(autouse=True)
+def mock_github_api():
+    """Mock the GitHub release API to prevent network requests."""
+    mock_response = AsyncMock(
+        data=GitHubReleaseModel(
+            {
+                "tag_name": "v14.6.0",
+                "name": "Tasmota 14.6.0",
+                "html_url": "https://github.com/arendst/Tasmota/releases/tag/v14.6.0",
+                "body": "",
+            }
+        )
+    )
+
+    with patch(
+        "aiogithubapi.namespaces.releases.GitHubReleasesNamespace.latest",
+        new=AsyncMock(return_value=mock_response),
+    ):
+        yield
+
+
 async def setup_tasmota_helper(hass: HomeAssistant) -> None:
     """Set up Tasmota."""
     hass.config.components.add("tasmota")

tests/components/tasmota/test_discovery.py

@@ -644,7 +644,7 @@ async def test_same_topic(
     device_entry = device_registry.async_get_device(
         connections={(dr.CONNECTION_NETWORK_MAC, configs[0]["mac"])}
     )
-    assert len(er.async_entries_for_device(entity_registry, device_entry.id, True)) == 1
+    assert len(er.async_entries_for_device(entity_registry, device_entry.id, True)) == 2
     device_entry = device_registry.async_get_device(
         connections={(dr.CONNECTION_NETWORK_MAC, configs[1]["mac"])}
     )
@@ -697,7 +697,7 @@ async def test_same_topic(
     device_entry = device_registry.async_get_device(
         connections={(dr.CONNECTION_NETWORK_MAC, configs[2]["mac"])}
     )
-    assert len(er.async_entries_for_device(entity_registry, device_entry.id, True)) == 1
+    assert len(er.async_entries_for_device(entity_registry, device_entry.id, True)) == 2
 
     # Verify the repairs issue has been updated
     issue = issue_registry.async_get_issue("tasmota", issue_id)
@@ -716,7 +716,7 @@ async def test_same_topic(
     device_entry = device_registry.async_get_device(
         connections={(dr.CONNECTION_NETWORK_MAC, configs[1]["mac"])}
     )
-    assert len(er.async_entries_for_device(entity_registry, device_entry.id, True)) == 1
+    assert len(er.async_entries_for_device(entity_registry, device_entry.id, True)) == 2
 
     # Verify the repairs issue has been removed
     assert issue_registry.async_get_issue("tasmota", issue_id) is None
@@ -776,7 +776,7 @@ async def test_topic_no_prefix(
     device_entry = device_registry.async_get_device(
         connections={(dr.CONNECTION_NETWORK_MAC, config["mac"])}
     )
-    assert len(er.async_entries_for_device(entity_registry, device_entry.id, True)) == 1
+    assert len(er.async_entries_for_device(entity_registry, device_entry.id, True)) == 2
 
     # Verify the repairs issue has been removed
     assert ("tasmota", issue_id) not in issue_registry.issues

tests/components/tasmota/test_update.py

@@ -0,0 +1,116 @@
+"""Tests for the Tasmota update platform."""
+
+import copy
+import json
+from unittest.mock import AsyncMock, patch
+
+from aiogithubapi import GitHubReleaseModel
+import pytest
+
+from homeassistant.components.tasmota.const import DEFAULT_PREFIX, DOMAIN
+from homeassistant.components.update import (
+    ATTR_INSTALLED_VERSION,
+    ATTR_LATEST_VERSION,
+    ATTR_RELEASE_URL,
+    DATA_COMPONENT,
+)
+from homeassistant.const import STATE_OFF, STATE_ON, Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import device_registry as dr, entity_registry as er
+
+from .conftest import setup_tasmota_helper
+from .test_common import DEFAULT_CONFIG
+
+from tests.common import async_fire_mqtt_message
+from tests.typing import MqttMockHAClient
+
+
+@pytest.fixture
+def mock_github_latest_release(request: pytest.FixtureRequest):
+    """Mock the GitHub release API to return a specific version."""
+    tag_name: str = request.param
+
+    mock_response = AsyncMock(
+        data=GitHubReleaseModel(
+            {
+                "tag_name": tag_name,
+                "name": f"Tasmota {tag_name.removeprefix('v')}",
+                "html_url": f"https://github.com/arendst/Tasmota/releases/tag/{tag_name}",
+                "body": """\
+<picture>
+  <source media="(prefers-color-scheme: dark)" srcset="./tools/logo/TASMOTA_FullLogo_Vector_White.svg">
+  <img alt="Logo" src="./tools/logo/TASMOTA_FullLogo_Vector.svg" align="right" height="76">
+</picture>
+
+# RELEASE NOTES
+
+...        """,
+            }
+        )
+    )
+
+    with patch(
+        "aiogithubapi.namespaces.releases.GitHubReleasesNamespace.latest",
+        new=AsyncMock(return_value=mock_response),
+    ):
+        yield tag_name
+
+
+@pytest.mark.parametrize(
+    ("mock_github_latest_release", "expected_update_state"),
+    [
+        ("v0.0.0", STATE_OFF),
+        ("v" + DEFAULT_CONFIG["sw"], STATE_OFF),
+        (
+            "v" + ".".join(str(int(x) + 1) for x in DEFAULT_CONFIG["sw"].split(".")),
+            STATE_ON,
+        ),
+    ],
+    indirect=["mock_github_latest_release"],
+)
+async def test_device_update_entity(
+    hass: HomeAssistant,
+    mqtt_mock: MqttMockHAClient,
+    device_registry: dr.DeviceRegistry,
+    mock_github_latest_release: str,
+    expected_update_state: str,
+) -> None:
+    """Test that an update entity is created and reports correct version state."""
+    await setup_tasmota_helper(hass)
+
+    config = copy.deepcopy(DEFAULT_CONFIG)
+    mac = config["mac"]
+
+    async_fire_mqtt_message(
+        hass,
+        f"{DEFAULT_PREFIX}/{mac}/config",
+        json.dumps(config),
+    )
+    await hass.async_block_till_done()
+
+    device_entry = device_registry.async_get_device(
+        connections={(dr.CONNECTION_NETWORK_MAC, mac)}
+    )
+    assert device_entry is not None
+    assert device_entry.sw_version == DEFAULT_CONFIG["sw"]
+
+    entity_id = er.async_get(hass).async_get_entity_id(
+        Platform.UPDATE, DOMAIN, dr.format_mac(mac)
+    )
+    assert entity_id is not None, "Update entity should exist for each Tasmota device"
+
+    state = hass.states.get(entity_id)
+    assert state is not None
+    assert state.attributes[ATTR_INSTALLED_VERSION] == DEFAULT_CONFIG["sw"]
+    assert state.attributes[
+        ATTR_LATEST_VERSION
+    ] == mock_github_latest_release.removeprefix("v")
+    assert state.attributes[ATTR_RELEASE_URL]
+    assert state.state == expected_update_state
+
+    update_entity = hass.data[DATA_COMPONENT].get_entity(entity_id)
+    assert update_entity is not None
+    result = await update_entity.async_release_notes()
+    assert result
+    assert "<picture>" not in result
+    assert "# RELEASE NOTES" in result