camera.py

Go to the documentation of this file.

"""Support for QVR Pro streams."""
 
from __future__ import annotations
 
import logging
 
from pyqvrpro.client import QVRResponseError
 
from homeassistant.components.camera import Camera
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
 
from .const import DOMAIN, SHORT_NAME
 
_LOGGER = logging.getLogger(__name__)
 
 
def setup_platform(
    hass: HomeAssistant,
    config: ConfigType,
    add_entities: AddEntitiesCallback,
    discovery_info: DiscoveryInfoType | None = None,
) -> None:
    """Set up the QVR Pro camera platform."""
    if discovery_info is None:
        return
 
    client = hass.data[DOMAIN]["client"]
 
    entities = []
 
    for channel in hass.data[DOMAIN]["channels"]:
        stream_source = get_stream_source(channel["guid"], client)
        entities.append(
            QVRProCamera(**channel, stream_source=stream_source, client=client)
        )
 
    add_entities(entities)
 
 
def get_stream_source(guid, client):
    """Get channel stream source."""
    try:
        resp = client.get_channel_live_stream(guid, protocol="rtsp")
    except QVRResponseError as ex:
        _LOGGER.error(ex)
        return None
 
    full_url = resp["resourceUris"]
 
    protocol = full_url[:7]
    auth = f"{client.get_auth_string()}@"
    url = full_url[7:]
 
    return f"{protocol}{auth}{url}"
 
 
class QVRProCamera(Camera):
    """Representation of a QVR Pro camera."""
 
    def __init__(self, name, model, brand, channel_index, guid, stream_source, client):
        """Init QVR Pro camera."""
 
        self._name_name = f"{SHORT_NAME} {name}"
        self._model_model = model
        self._brand_brand = brand
        self.indexindex = channel_index
        self.guidguid = guid
        self._client_client = client
        self._stream_source_stream_source = stream_source
 
        super().__init__()
 
    @property
    def name(self):
        """Return the name of the entity."""
        return self._name_name
 
    @property
    def model(self):
        """Return the model of the entity."""
        return self._model_model
 
    @property
    def brand(self):
        """Return the brand of the entity."""
        return self._brand_brand
 
    @property
    def extra_state_attributes(self):
        """Get the state attributes."""
        return {"qvr_guid": self.guidguid}
 
    def camera_image(
        self, width: int | None = None, height: int | None = None
    ) -> bytes | None:
        """Get image bytes from camera."""
        try:
            return self._client_client.get_snapshot(self.guidguid)
 
        except QVRResponseError as ex:
            _LOGGER.error("Error getting image: %s", ex)
            self._client_client.connect()
 
        return self._client_client.get_snapshot(self.guidguid)
 
    async def stream_source(self):
        """Get stream source."""
        return self._stream_source_stream_source