ledgrab-haos-integration/custom_components/ledgrab/update.py

"""Update platform for LED Screen Controller.

Surfaces the led-grab server's auto-update service (``/api/v1/system/update``)
as a Home Assistant Update entity attached to the synthetic Server device.

The entity is only registered when the server actually exposes the endpoint —
older servers return 404 and the coordinator stores ``None`` in
``data["system"]["update"]``.
"""

from __future__ import annotations

import logging
from typing import Any

from homeassistant.components.update import (
    UpdateDeviceClass,
    UpdateEntity,
    UpdateEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity

from .const import DATA_COORDINATOR, DOMAIN
from .coordinator import LedGrabCoordinator

_LOGGER = logging.getLogger(__name__)

# release_summary is capped at 255 chars by HA's frontend; the full markdown
# body is exposed via async_release_notes() and rendered in the notes dialog.
_RELEASE_SUMMARY_MAX = 255


async def async_setup_entry(
    hass: HomeAssistant,
    entry: ConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    """Set up the server update entity, when supported by the server."""
    coordinator: LedGrabCoordinator = hass.data[DOMAIN][entry.entry_id][
        DATA_COORDINATOR
    ]

    update_data = (coordinator.data or {}).get("system", {}).get("update")
    if update_data is None:
        # Server doesn't expose /api/v1/system/update/status — skip silently.
        return

    async_add_entities([LedGrabServerUpdate(coordinator, entry.entry_id)])


class LedGrabServerUpdate(CoordinatorEntity, UpdateEntity):
    """Update entity for the led-grab server itself."""

    _attr_has_entity_name = True
    _attr_translation_key = "server_update"
    _attr_device_class = UpdateDeviceClass.FIRMWARE

    def __init__(self, coordinator: LedGrabCoordinator, entry_id: str) -> None:
        super().__init__(coordinator)
        self._entry_id = entry_id
        self._attr_unique_id = f"{entry_id}_server_update"

    @property
    def device_info(self) -> dict[str, Any]:
        return {"identifiers": {(DOMAIN, f"{self._entry_id}_server")}}

    def _data(self) -> dict[str, Any] | None:
        if not self.coordinator.data:
            return None
        return (self.coordinator.data.get("system") or {}).get("update")

    def _release(self) -> dict[str, Any]:
        return (self._data() or {}).get("release") or {}

    @property
    def supported_features(self) -> UpdateEntityFeature:
        # The server's apply endpoint refuses to run for install types that
        # can't auto-update (e.g. docker-managed containers). Hide the install
        # button in those cases instead of letting users click it and 400.
        features = UpdateEntityFeature.PROGRESS | UpdateEntityFeature.RELEASE_NOTES
        data = self._data() or {}
        if data.get("can_auto_update"):
            features |= UpdateEntityFeature.INSTALL
        return features

    @property
    def installed_version(self) -> str | None:
        data = self._data() or {}
        return data.get("current_version")

    @property
    def latest_version(self) -> str | None:
        data = self._data()
        if not data:
            return None
        if data.get("has_update"):
            return self._release().get("version") or data.get("current_version")
        # No update — match installed so HA shows "Up to date".
        return data.get("current_version")

    @property
    def release_url(self) -> str | None:
        data = self._data() or {}
        base = data.get("releases_url") or None
        tag = self._release().get("tag")
        if base and tag:
            # Works for both Gitea (`<base>/<repo>/releases/tag/<tag>`)
            # and GitHub — `releases_url` already ends in `/releases`.
            return f"{base.rstrip('/')}/tag/{tag}"
        return base or None

    @property
    def release_summary(self) -> str | None:
        body = self._release().get("body")
        if not body:
            return None
        return body[:_RELEASE_SUMMARY_MAX]

    async def async_release_notes(self) -> str | None:
        """Return the full release body for the notes dialog."""
        return self._release().get("body") or None

    @property
    def in_progress(self) -> bool | int:
        data = self._data() or {}
        if data.get("applying"):
            # Apply phase has no granular progress reporting from the server.
            return True
        if data.get("downloading"):
            # Server reports 0..1; HA wants 0..100.
            progress = float(data.get("download_progress") or 0.0)
            return max(0, min(100, int(round(progress * 100))))
        return False

    @property
    def available(self) -> bool:
        return self._data() is not None

    @property
    def extra_state_attributes(self) -> dict[str, Any]:
        data = self._data() or {}
        return {
            "install_type": data.get("install_type"),
            "can_auto_update": data.get("can_auto_update"),
            "checking": data.get("checking"),
            "last_check": data.get("last_check"),
            "last_error": data.get("last_error"),
            "dismissed_version": data.get("dismissed_version"),
            "prerelease": self._release().get("prerelease"),
            "published_at": self._release().get("published_at"),
        }

    async def async_install(
        self,
        version: str | None,
        backup: bool,
        **kwargs: Any,
    ) -> None:
        """Apply the available update.

        The server picks up whichever release it has cached; the ``version``
        and ``backup`` arguments from HA aren't separately addressable on the
        server side, so we ignore them.
        """
        await self.coordinator.apply_update()