Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add update entity for car diffuser #46

Merged
merged 1 commit into from
Feb 29, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions custom_components/pura/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
Platform.SELECT,
Platform.SENSOR,
Platform.SWITCH,
Platform.UPDATE,
]


Expand Down
31 changes: 31 additions & 0 deletions custom_components/pura/coordinator.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,3 +63,34 @@ async def _async_update_data(self):
)
raise UpdateFailed(err) from err
return self.devices


class PuraCarFirmwareDataUpdateCoordinator(DataUpdateCoordinator):
"""Class to manage fetching data from the API."""

def __init__(self, hass: HomeAssistant, client: Pura) -> None:
"""Initialize."""
self.api = client

super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=timedelta(seconds=UPDATE_INTERVAL),
)

async def _async_update_data(self):
"""Update data via library, refresh token if necessary."""
try:
details: str = await self.hass.async_add_executor_job(
self.api.get_latest_firmware_details, "car", "v1"
)
return {
(part := line.split("=", 1))[0].lower(): part[1]
for line in details.split("\r\n")
}
except Exception as err: # pylint: disable=broad-except
_LOGGER.error(
"Unknown exception while updating Pura data: %s", err, exc_info=1
)
raise UpdateFailed(err) from err
95 changes: 95 additions & 0 deletions custom_components/pura/update.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,95 @@
"""Support for Pura update."""
from __future__ import annotations

from dataclasses import dataclass
import logging

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

from .const import DOMAIN
from .coordinator import PuraDataUpdateCoordinator
from .entity import PuraEntity
from .helpers import get_device_id

_LOGGER = logging.getLogger(__name__)


@dataclass
class RequiredKeysMixin:
"""Required keys mixin."""

lookup_key: str


@dataclass
class PuraUpdateEntityDescription(UpdateEntityDescription, RequiredKeysMixin):
"""Pura update entity description."""


UPDATE = PuraUpdateEntityDescription(key="firmware", lookup_key="fw_version")


async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up Pura updates using config entry."""
coordinator: PuraDataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]

entities = [
PuraUpdateEntity(
coordinator=coordinator,
config_entry=config_entry,
description=UPDATE,
device_type=device_type,
device_id=get_device_id(device),
)
for device_type, devices in coordinator.devices.items()
if device_type == "car"
for device in devices
]

if not entities:
return

async_add_entities(entities, True)


class PuraUpdateEntity(PuraEntity, UpdateEntity):
"""Pura update."""

entity_description: PuraUpdateEntityDescription
_attr_device_class = UpdateDeviceClass.FIRMWARE
_attr_should_poll = True
_attr_release_summary = (
"https://help.pura.com/en/car_diffuser/Update-Pura-Car-Firmware"
)

@property
def installed_version(self) -> str | None:
"""Version installed and in use."""
return self.get_device().get(self.entity_description.lookup_key)

async def async_update(self) -> None:
"""Update the entity."""
try:
details: str = await self.hass.async_add_executor_job(
self.coordinator.api.get_latest_firmware_details, "car", "v1"
)
firmware = {
(part := line.split("=", 1))[0].lower(): part[1]
for line in details.split("\r\n")
}
self._attr_latest_version = ".".join(
firmware[key] for key in ("major", "minor", "patch")
)
except Exception as ex: # pylint: disable=broad-except
_LOGGER.exception(ex)
Loading