camera.py

Go to the documentation of this file.

"""Support for Agent camera streaming."""
 
from datetime import timedelta
import logging
 
from agent import AgentError
 
from homeassistant.components.camera import CameraEntityFeature
from homeassistant.components.mjpeg import MjpegCamera, filter_urllib3_logging
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import (
    AddEntitiesCallback,
    async_get_current_platform,
)
 
from . import AgentDVRConfigEntry
from .const import ATTRIBUTION, CAMERA_SCAN_INTERVAL_SECS, DOMAIN as AGENT_DOMAIN
 
SCAN_INTERVAL = timedelta(seconds=CAMERA_SCAN_INTERVAL_SECS)
 
_LOGGER = logging.getLogger(__name__)
 
_DEV_EN_ALT = "enable_alerts"
_DEV_DS_ALT = "disable_alerts"
_DEV_EN_REC = "start_recording"
_DEV_DS_REC = "stop_recording"
_DEV_SNAP = "snapshot"
 
CAMERA_SERVICES = {
    _DEV_EN_ALT: "async_enable_alerts",
    _DEV_DS_ALT: "async_disable_alerts",
    _DEV_EN_REC: "async_start_recording",
    _DEV_DS_REC: "async_stop_recording",
    _DEV_SNAP: "async_snapshot",
}
 
 
async def async_setup_entry(
    hass: HomeAssistant,
    config_entry: AgentDVRConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    """Set up the Agent cameras."""
    filter_urllib3_logging()
    cameras = []
 
    server = config_entry.runtime_data
    if not server.devices:
        _LOGGER.warning("Could not fetch cameras from Agent server")
        return
 
    for device in server.devices:
        if device.typeID == 2:
            camera = AgentCamera(device)
            cameras.append(camera)
 
    async_add_entities(cameras)
 
    platform = async_get_current_platform()
    for service, method in CAMERA_SERVICES.items():
        platform.async_register_entity_service(service, None, method)
 
 
class AgentCamera(MjpegCamera):
    """Representation of an Agent Device Stream."""
 
    _attr_attribution = ATTRIBUTION
    _attr_should_poll = True  # Cameras default to False
    _attr_supported_features = CameraEntityFeature.ON_OFF
    _attr_has_entity_name = True
    _attr_name = None
 
    def __init__(self, device):
        """Initialize as a subclass of MjpegCamera."""
        self.devicedevice = device
        self._removed_removed = False
        self._attr_unique_id_attr_unique_id = f"{device.client.unique}_{device.typeID}_{device.id}"
        super().__init__(
            name=device.name,
            mjpeg_url=f"{device.client._server_url}{device.mjpeg_image_url}&size={device.mjpegStreamWidth}x{device.mjpegStreamHeight}",  # noqa: SLF001
            still_image_url=f"{device.client._server_url}{device.still_image_url}&size={device.mjpegStreamWidth}x{device.mjpegStreamHeight}",  # noqa: SLF001
        )
        self._attr_device_info_attr_device_info = DeviceInfo(
            identifiers={(AGENT_DOMAIN, self.unique_id)},
            manufacturer="Agent",
            model="Camera",
            name=f"{device.client.name} {device.name}",
            sw_version=device.client.version,
        )
 
    async def async_update(self) -> None:
        """Update our state from the Agent API."""
        try:
            await self.devicedevice.update()
            if self._removed_removed:
                _LOGGER.debug("%s reacquired", self.name)
            self._removed_removed = False
        except AgentError:
            # server still available - camera error
            if self.devicedevice.client.is_available and not self._removed_removed:
                _LOGGER.error("%s lost", self.name)
                self._removed_removed = True
        self._attr_icon_attr_icon = "mdi:camcorder-off"
        if self.is_onis_on:
            self._attr_icon_attr_icon = "mdi:camcorder"
        self._attr_available_attr_available = self.devicedevice.client.is_available
        self._attr_extra_state_attributes_attr_extra_state_attributes = {
            "editable": False,
            "enabled": self.is_onis_on,
            "connected": self.connectedconnected,
            "detected": self.is_detectedis_detected,
            "alerted": self.is_alertedis_alerted,
            "has_ptz": self.devicedevice.has_ptz,
            "alerts_enabled": self.devicedevice.alerts_active,
        }
 
    @property
    def is_recording(self) -> bool:
        """Return whether the monitor is recording."""
        return self.devicedevice.recording
 
    @property
    def is_alerted(self) -> bool:
        """Return whether the monitor has alerted."""
        return self.devicedevice.alerted
 
    @property
    def is_detected(self) -> bool:
        """Return whether the monitor has alerted."""
        return self.devicedevice.detected
 
    @property
    def connected(self) -> bool:
        """Return True if entity is connected."""
        return self.devicedevice.connected
 
    @property
    def is_on(self) -> bool:
        """Return true if on."""
        return self.devicedevice.online
 
    @property
    def motion_detection_enabled(self) -> bool:
        """Return the camera motion detection status."""
        return self.devicedevice.detector_active
 
    async def async_enable_alerts(self):
        """Enable alerts."""
        await self.devicedevice.alerts_on()
 
    async def async_disable_alerts(self):
        """Disable alerts."""
        await self.devicedevice.alerts_off()
 
    async def async_enable_motion_detection(self) -> None:
        """Enable motion detection."""
        await self.devicedevice.detector_on()
 
    async def async_disable_motion_detection(self) -> None:
        """Disable motion detection."""
        await self.devicedevice.detector_off()
 
    async def async_start_recording(self):
        """Start recording."""
        await self.devicedevice.record()
 
    async def async_stop_recording(self):
        """Stop recording."""
        await self.devicedevice.record_stop()
 
    async def async_turn_on(self) -> None:
        """Enable the camera."""
        await self.devicedevice.enable()
 
    async def async_snapshot(self):
        """Take a snapshot."""
        await self.devicedevice.snapshot()
 
    async def async_turn_off(self) -> None:
        """Disable the camera."""
        await self.devicedevice.disable()