243 lines
8.9 KiB
Python
243 lines
8.9 KiB
Python
from datetime import datetime, timedelta
|
|
from typing import Any
|
|
|
|
from homeassistant.components.image import ImageEntity
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.core import HomeAssistant, callback
|
|
from homeassistant.helpers.device_registry import DeviceInfo
|
|
from homeassistant.helpers.event import async_track_time_interval
|
|
from homeassistant.helpers.entity import EntityCategory
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from . import CONF_CAMERA_IMAGE_REFRESH_INTERVAL, DOMAIN, RTKeyCamerasApi, _LOGGER
|
|
|
|
|
|
async def async_setup_entry(
|
|
hass: HomeAssistant,
|
|
config_entry: ConfigEntry,
|
|
async_add_entities: Any,
|
|
) -> None:
|
|
cameras_api: RTKeyCamerasApi = hass.data[DOMAIN][config_entry.entry_id]["cameras_api"]
|
|
cameras_info = await cameras_api.get_cameras_info()
|
|
|
|
# Проверяем, включены ли обычные скриншоты
|
|
interval = config_entry.options.get(CONF_CAMERA_IMAGE_REFRESH_INTERVAL, 60)
|
|
|
|
entities = []
|
|
|
|
# Обычные скриншоты (если interval > 0)
|
|
if interval > 0:
|
|
entities.extend([
|
|
RTKeyCameraImageEntity(hass, config_entry, cameras_api, camera_info)
|
|
for camera_info in cameras_info["data"]["items"]
|
|
])
|
|
else:
|
|
_LOGGER.info("Обычные скриншоты отключены (interval=0)")
|
|
|
|
# Скриншоты событий
|
|
if cameras_api.event_screenshot_enabled:
|
|
intercoms_info = await cameras_api.get_intercoms_info()
|
|
for intercom_info in intercoms_info.get("data", {}).get("devices", []):
|
|
entities.append(
|
|
RTKeyEventImage(hass, config_entry, cameras_api, intercom_info)
|
|
)
|
|
else:
|
|
_LOGGER.info("Скриншоты событий отключены")
|
|
|
|
async_add_entities(entities)
|
|
|
|
|
|
class RTKeyCameraImageEntity(ImageEntity):
|
|
"""Периодический скриншот камеры."""
|
|
|
|
def __init__(
|
|
self,
|
|
hass: HomeAssistant,
|
|
config_entry: ConfigEntry,
|
|
cameras_api: RTKeyCamerasApi,
|
|
camera_info: dict,
|
|
) -> None:
|
|
super().__init__(hass)
|
|
|
|
self.hass = hass
|
|
self.cameras_api = cameras_api
|
|
self.camera_id = camera_info["id"]
|
|
|
|
self.device_name = cameras_api.build_device_name(camera_info.get("title", "Устройство"))
|
|
|
|
self._device_model = camera_info.get("model", "")
|
|
self._device_vendor = camera_info.get("vendor", "")
|
|
self._device_serial = camera_info.get("serial_number", "")
|
|
self._device_status_title = camera_info.get("status_title", "Неизвестно")
|
|
|
|
self._attr_unique_id = f"image_{self.camera_id}"
|
|
self._attr_name = "Скриншот"
|
|
|
|
self.camera_image_refresh_interval = config_entry.options.get(
|
|
CONF_CAMERA_IMAGE_REFRESH_INTERVAL, 60
|
|
)
|
|
|
|
self._attr_image_last_updated = datetime.now()
|
|
self._cached_image: bytes | None = None
|
|
self._unsub_timer: Any = None
|
|
|
|
async def async_added_to_hass(self) -> None:
|
|
await super().async_added_to_hass()
|
|
|
|
self._unsub_timer = async_track_time_interval(
|
|
self.hass,
|
|
self._async_update_image,
|
|
timedelta(seconds=self.camera_image_refresh_interval),
|
|
)
|
|
await self._async_update_image()
|
|
|
|
async def async_will_remove_from_hass(self) -> None:
|
|
if self._unsub_timer:
|
|
self._unsub_timer()
|
|
await super().async_will_remove_from_hass()
|
|
|
|
async def _async_update_image(self, now: datetime | None = None) -> None:
|
|
"""Продолжаем опрашивать API даже если камера недоступна,
|
|
чтобы узнать, когда она снова появится."""
|
|
try:
|
|
image = await self.cameras_api.get_camera_image(self.camera_id)
|
|
if image:
|
|
self._cached_image = image
|
|
self._attr_image_last_updated = datetime.now()
|
|
self.async_write_ha_state()
|
|
_LOGGER.debug(f"Обновлено изображение: {self.device_name}")
|
|
except Exception as e:
|
|
_LOGGER.error(f"Ошибка обновления изображения {self.device_name}: {e}")
|
|
|
|
async def async_image(self) -> bytes | None:
|
|
return self._cached_image
|
|
|
|
@property
|
|
def available(self) -> bool:
|
|
"""Сущность доступна, если камера не в списке недоступных."""
|
|
return self.cameras_api.is_camera_available(self.camera_id)
|
|
|
|
@property
|
|
def extra_state_attributes(self) -> dict[str, Any]:
|
|
"""Атрибуты с динамической проверкой доступности камеры."""
|
|
if self.cameras_api.is_camera_available(self.camera_id):
|
|
status = self._device_status_title
|
|
else:
|
|
status = "❌ Нет связи"
|
|
|
|
attrs = {"Статус": status}
|
|
if self._device_model:
|
|
attrs["Модель"] = self._device_model
|
|
if self._device_serial:
|
|
attrs["Серийный номер"] = self._device_serial
|
|
return attrs
|
|
|
|
@property
|
|
def device_info(self) -> DeviceInfo:
|
|
info = {
|
|
"identifiers": {(DOMAIN, self.camera_id)},
|
|
"name": self.device_name,
|
|
"manufacturer": self._device_vendor or "RT Key",
|
|
}
|
|
if self._device_model:
|
|
info["model"] = self._device_model
|
|
if self._device_serial:
|
|
info["serial_number"] = self._device_serial
|
|
return info
|
|
|
|
|
|
class RTKeyEventImage(ImageEntity):
|
|
"""Скриншот на момент последнего события домофона."""
|
|
|
|
_attr_should_poll = False
|
|
_attr_entity_category = EntityCategory.DIAGNOSTIC
|
|
|
|
def __init__(
|
|
self,
|
|
hass: HomeAssistant,
|
|
config_entry: ConfigEntry,
|
|
cameras_api: RTKeyCamerasApi,
|
|
intercom_info: dict,
|
|
) -> None:
|
|
super().__init__(hass)
|
|
|
|
self.hass = hass
|
|
self.cameras_api = cameras_api
|
|
self.intercom_id = str(intercom_info["id"])
|
|
self.camera_id = intercom_info.get("camera_id")
|
|
|
|
self.device_name = cameras_api.build_device_name(
|
|
intercom_info.get("name_by_user")
|
|
or intercom_info.get("description")
|
|
or intercom_info.get("name_by_company", "Домофон")
|
|
)
|
|
|
|
self._attr_unique_id = f"event_image_{self.intercom_id}"
|
|
self._attr_name = "Скриншот события"
|
|
|
|
self._attr_image_last_updated = None
|
|
self._cached_image: bytes | None = None
|
|
self._event_description = None
|
|
self._event_time = None
|
|
|
|
async def async_added_to_hass(self) -> None:
|
|
await super().async_added_to_hass()
|
|
self.cameras_api.register_event_listener(self._handle_event_update)
|
|
|
|
if self.intercom_id in self.cameras_api.last_events:
|
|
self._update_from_event()
|
|
|
|
async def async_will_remove_from_hass(self) -> None:
|
|
self.cameras_api.unregister_event_listener(self._handle_event_update)
|
|
await super().async_will_remove_from_hass()
|
|
|
|
@callback
|
|
def _handle_event_update(self, intercom_id: str):
|
|
if intercom_id == self.intercom_id:
|
|
self._update_from_event()
|
|
|
|
def _update_from_event(self):
|
|
event_info = self.cameras_api.last_events.get(self.intercom_id)
|
|
if not event_info:
|
|
return
|
|
|
|
screenshot_data = event_info.get("screenshot_data")
|
|
if screenshot_data:
|
|
self._cached_image = screenshot_data
|
|
self._event_description = event_info.get("description")
|
|
self._event_time = event_info.get("local_time")
|
|
|
|
# ✅ Используем время события, а не текущее время
|
|
event_timestamp = event_info.get("timestamp")
|
|
if event_timestamp:
|
|
self._attr_image_last_updated = datetime.fromtimestamp(
|
|
event_timestamp,
|
|
tz=ZoneInfo(self.hass.config.time_zone),
|
|
)
|
|
else:
|
|
self._attr_image_last_updated = datetime.now()
|
|
|
|
self.async_write_ha_state()
|
|
_LOGGER.debug(f"Обновлён скриншот события для {self.device_name}")
|
|
|
|
async def async_image(self) -> bytes | None:
|
|
return self._cached_image
|
|
|
|
@property
|
|
def extra_state_attributes(self) -> dict[str, Any]:
|
|
attrs = {}
|
|
if self._event_description:
|
|
attrs["Событие"] = self._event_description
|
|
if self._event_time:
|
|
attrs["Время"] = self._event_time
|
|
return attrs
|
|
|
|
@property
|
|
def device_info(self) -> DeviceInfo:
|
|
device_id = self.camera_id or self.intercom_id
|
|
return {
|
|
"identifiers": {(DOMAIN, device_id)},
|
|
"name": self.device_name,
|
|
"manufacturer": "RT Key",
|
|
"model": "Домофон",
|
|
} |