mirror of https://github.com/home-assistant/core
91 lines
3.0 KiB
Python
91 lines
3.0 KiB
Python
"""Support for August doorbell camera."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from aiohttp import ClientSession
|
|
from yalexs.activity import ActivityType
|
|
from yalexs.doorbell import Doorbell
|
|
from yalexs.util import update_doorbell_image_from_activity
|
|
|
|
from homeassistant.components.camera import Camera
|
|
from homeassistant.core import HomeAssistant, callback
|
|
from homeassistant.helpers import aiohttp_client
|
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
|
|
|
from . import AugustConfigEntry, AugustData
|
|
from .const import DEFAULT_NAME, DEFAULT_TIMEOUT
|
|
from .entity import AugustEntity
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
async def async_setup_entry(
|
|
hass: HomeAssistant,
|
|
config_entry: AugustConfigEntry,
|
|
async_add_entities: AddEntitiesCallback,
|
|
) -> None:
|
|
"""Set up August cameras."""
|
|
data = config_entry.runtime_data
|
|
# Create an aiohttp session instead of using the default one since the
|
|
# default one is likely to trigger august's WAF if another integration
|
|
# is also using Cloudflare
|
|
session = aiohttp_client.async_create_clientsession(hass)
|
|
async_add_entities(
|
|
AugustCamera(data, doorbell, session, DEFAULT_TIMEOUT)
|
|
for doorbell in data.doorbells
|
|
)
|
|
|
|
|
|
class AugustCamera(AugustEntity, Camera):
|
|
"""An implementation of an August security camera."""
|
|
|
|
_attr_translation_key = "camera"
|
|
_attr_motion_detection_enabled = True
|
|
_attr_brand = DEFAULT_NAME
|
|
_image_url: str | None = None
|
|
_image_content: bytes | None = None
|
|
|
|
def __init__(
|
|
self, data: AugustData, device: Doorbell, session: ClientSession, timeout: int
|
|
) -> None:
|
|
"""Initialize an August security camera."""
|
|
super().__init__(data, device, "camera")
|
|
self._timeout = timeout
|
|
self._session = session
|
|
self._attr_model = self._detail.model
|
|
|
|
@property
|
|
def is_recording(self) -> bool:
|
|
"""Return true if the device is recording."""
|
|
return self._device.has_subscription
|
|
|
|
async def _async_update(self):
|
|
"""Update device."""
|
|
_LOGGER.debug("async_update called %s", self._detail.device_name)
|
|
await self._data.refresh_camera_by_id(self._device_id)
|
|
self._update_from_data()
|
|
|
|
@callback
|
|
def _update_from_data(self) -> None:
|
|
"""Get the latest state of the sensor."""
|
|
if doorbell_activity := self._get_latest(
|
|
{ActivityType.DOORBELL_MOTION, ActivityType.DOORBELL_IMAGE_CAPTURE}
|
|
):
|
|
update_doorbell_image_from_activity(self._detail, doorbell_activity)
|
|
|
|
async def async_camera_image(
|
|
self, width: int | None = None, height: int | None = None
|
|
) -> bytes | None:
|
|
"""Return bytes of camera image."""
|
|
self._update_from_data()
|
|
|
|
if self._image_url is not self._detail.image_url:
|
|
self._image_content = await self._data.async_get_doorbell_image(
|
|
self._device_id, self._session, timeout=self._timeout
|
|
)
|
|
self._image_url = self._detail.image_url
|
|
|
|
return self._image_content
|