95 lines
2.8 KiB
Python
95 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
from homeassistant.components.update import UpdateEntity, UpdateEntityFeature
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
|
|
|
from . import DOMAIN
|
|
from .store import BahmcloudStore, Package
|
|
|
|
|
|
async def async_setup_platform(
|
|
hass: HomeAssistant,
|
|
config,
|
|
async_add_entities: AddEntitiesCallback,
|
|
discovery_info=None,
|
|
):
|
|
store: BahmcloudStore = hass.data[DOMAIN]
|
|
entities: dict[str, BahmcloudPackageUpdate] = {}
|
|
|
|
def should_have_update_entity(pkg: Package) -> bool:
|
|
# Store selbst immer als Update
|
|
if pkg.type == "store":
|
|
return True
|
|
# Andere Pakete nur, wenn installiert
|
|
return store.is_installed(pkg.domain)
|
|
|
|
def rebuild_entities() -> None:
|
|
# Create entities for packages that qualify
|
|
for pkg in store.packages.values():
|
|
if not should_have_update_entity(pkg):
|
|
continue
|
|
|
|
uid = f"{DOMAIN}:{pkg.id}"
|
|
if uid not in entities:
|
|
ent = BahmcloudPackageUpdate(store, pkg.id)
|
|
entities[uid] = ent
|
|
async_add_entities([ent], update_before_add=True)
|
|
|
|
# Refresh states
|
|
for ent in entities.values():
|
|
ent.async_write_ha_state()
|
|
|
|
store.add_listener(rebuild_entities)
|
|
rebuild_entities()
|
|
|
|
|
|
class BahmcloudPackageUpdate(UpdateEntity):
|
|
_attr_supported_features = UpdateEntityFeature.INSTALL
|
|
|
|
def __init__(self, store: BahmcloudStore, package_id: str) -> None:
|
|
self.store = store
|
|
self.package_id = package_id
|
|
|
|
self._attr_unique_id = f"{DOMAIN}_{package_id}"
|
|
self._attr_name = f"{package_id} update"
|
|
|
|
@property
|
|
def _pkg(self) -> Package | None:
|
|
return self.store.packages.get(self.package_id)
|
|
|
|
@property
|
|
def title(self) -> str | None:
|
|
pkg = self._pkg
|
|
return pkg.name if pkg else None
|
|
|
|
@property
|
|
def installed_version(self) -> str | None:
|
|
pkg = self._pkg
|
|
if not pkg:
|
|
return None
|
|
if not self.store.is_installed(pkg.domain):
|
|
return None
|
|
return self.store.installed_version(pkg.domain)
|
|
|
|
@property
|
|
def latest_version(self) -> str | None:
|
|
pkg = self._pkg
|
|
return pkg.latest_version if pkg else None
|
|
|
|
@property
|
|
def release_summary(self) -> str | None:
|
|
pkg = self._pkg
|
|
if not pkg:
|
|
return None
|
|
if pkg.release_url:
|
|
return f"Release: {pkg.release_url}"
|
|
return f"Repo: {pkg.repo}"
|
|
|
|
async def async_install(self, version: str | None, backup: bool, **kwargs) -> None:
|
|
pkg = self._pkg
|
|
if not pkg:
|
|
return
|
|
await self.store.install_from_zip(pkg)
|
|
self.async_write_ha_state()
|