core/homeassistant/components/fyta/image.py

124 lines
3.7 KiB
Python

"""Entity for Fyta plant image."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime
import logging
from typing import Final
from fyta_cli.fyta_models import Plant
from homeassistant.components.image import (
Image,
ImageEntity,
ImageEntityDescription,
valid_image_content_type,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import FytaConfigEntry, FytaCoordinator
from .entity import FytaPlantEntity
_LOGGER = logging.getLogger(__name__)
@dataclass(frozen=True, kw_only=True)
class FytaImageEntityDescription(ImageEntityDescription):
"""Describes Fyta image entity."""
url_fn: Callable[[Plant], str]
name_key: str | None = None
IMAGES: Final[list[FytaImageEntityDescription]] = [
FytaImageEntityDescription(
key="plant_image",
translation_key="plant_image",
url_fn=lambda plant: plant.plant_origin_path,
),
FytaImageEntityDescription(
key="plant_image_user",
translation_key="plant_image_user",
url_fn=lambda plant: plant.user_picture_path,
),
]
async def async_setup_entry(
hass: HomeAssistant,
entry: FytaConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the FYTA plant images."""
coordinator = entry.runtime_data
async_add_entities(
FytaPlantImageEntity(coordinator, entry, description, plant_id)
for plant_id in coordinator.fyta.plant_list
if plant_id in coordinator.data
for description in IMAGES
)
def _async_add_new_device(plant_id: int) -> None:
async_add_entities(
FytaPlantImageEntity(coordinator, entry, description, plant_id)
for description in IMAGES
)
coordinator.new_device_callbacks.append(_async_add_new_device)
class FytaPlantImageEntity(FytaPlantEntity, ImageEntity):
"""Represents a Fyta image."""
entity_description: FytaImageEntityDescription
def __init__(
self,
coordinator: FytaCoordinator,
entry: ConfigEntry,
description: FytaImageEntityDescription,
plant_id: int,
) -> None:
"""Initialize Fyta Image entity."""
super().__init__(coordinator, entry, description, plant_id)
ImageEntity.__init__(self, coordinator.hass)
async def async_image(self) -> bytes | None:
"""Return bytes of image."""
if self.entity_description.key == "plant_image_user":
if self._cached_image is None:
response = await self.coordinator.fyta.get_plant_image(
self.plant.user_picture_path
)
_LOGGER.debug("Response of downloading user image: %s", response)
if response is None:
_LOGGER.debug(
"%s: Error getting new image from %s",
self.entity_id,
self.plant.user_picture_path,
)
return None
content_type, raw_image = response
self._cached_image = Image(
valid_image_content_type(content_type), raw_image
)
return self._cached_image.content
return await ImageEntity.async_image(self)
@property
def image_url(self) -> str:
"""Return the image_url for this plant."""
url = self.entity_description.url_fn(self.plant)
if url != self._attr_image_url:
self._cached_image = None
self._attr_image_last_updated = datetime.now()
return url