Загрузить файлы в «custom_components/rtkey»
This commit is contained in:
@@ -0,0 +1,243 @@
|
||||
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": "Домофон",
|
||||
}
|
||||
Reference in New Issue
Block a user