96 lines
2.7 KiB
Python
96 lines
2.7 KiB
Python
"""TrueNAS binary sensor platform."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from logging import getLogger
|
|
from typing import Any
|
|
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
|
|
|
from homeassistant.components.update import (
|
|
UpdateDeviceClass,
|
|
UpdateEntity,
|
|
UpdateEntityFeature,
|
|
)
|
|
|
|
from .coordinator import TrueNASCoordinator
|
|
from .entity import TrueNASEntity, async_add_entities
|
|
from .update_types import SENSOR_SERVICES, SENSOR_TYPES
|
|
|
|
_LOGGER = getLogger(__name__)
|
|
DEVICE_UPDATE = "device_update"
|
|
|
|
|
|
# ---------------------------
|
|
# async_setup_entry
|
|
# ---------------------------
|
|
async def async_setup_entry(
|
|
hass: HomeAssistant,
|
|
config_entry: ConfigEntry,
|
|
_async_add_entities: AddEntitiesCallback,
|
|
) -> None:
|
|
"""Set up device tracker for OpenMediaVault component."""
|
|
dispatcher = {
|
|
"TrueNASUpdate": TrueNASUpdate,
|
|
}
|
|
await async_add_entities(hass, config_entry, dispatcher)
|
|
|
|
|
|
# ---------------------------
|
|
# TrueNASUpdate
|
|
# ---------------------------
|
|
class TrueNASUpdate(TrueNASEntity, UpdateEntity):
|
|
"""Define an TrueNAS Update Sensor."""
|
|
|
|
TYPE = DEVICE_UPDATE
|
|
_attr_device_class = UpdateDeviceClass.FIRMWARE
|
|
|
|
def __init__(
|
|
self,
|
|
coordinator: TrueNASCoordinator,
|
|
entity_description,
|
|
uid: str | None = None,
|
|
):
|
|
"""Set up device update entity."""
|
|
super().__init__(coordinator, entity_description, uid)
|
|
|
|
self._attr_supported_features = UpdateEntityFeature.INSTALL
|
|
self._attr_supported_features |= UpdateEntityFeature.PROGRESS
|
|
self._attr_title = self.entity_description.title
|
|
|
|
@property
|
|
def installed_version(self) -> str:
|
|
"""Version installed and in use."""
|
|
return self._data["version"]
|
|
|
|
@property
|
|
def latest_version(self) -> str:
|
|
"""Latest version available for install."""
|
|
return self._data["update_version"]
|
|
|
|
async def options_updated(self) -> None:
|
|
"""No action needed."""
|
|
|
|
async def async_install(self, version: str, backup: bool, **kwargs: Any) -> None:
|
|
"""Install an update."""
|
|
self._data["update_jobid"] = await self.hass.async_add_executor_job(
|
|
self.coordinator.api.query,
|
|
"update/update",
|
|
"post",
|
|
{"reboot": True},
|
|
)
|
|
await self.coordinator.async_refresh()
|
|
|
|
@property
|
|
def in_progress(self) -> int:
|
|
"""Update installation progress."""
|
|
if self._data["update_state"] != "RUNNING":
|
|
return False
|
|
|
|
if self._data["update_progress"] == 0:
|
|
self._data["update_progress"] = 1
|
|
|
|
return self._data["update_progress"]
|