Use entity description for Reolink cameras (#104139)

* Use entity description for cams

* expend for loops
This commit is contained in:
starkillerOG 2023-11-20 18:30:39 +01:00 committed by GitHub
parent cd5595a130
commit ce497dd7ed
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 156 additions and 33 deletions

View File

@ -1,11 +1,17 @@
"""Component providing support for Reolink IP cameras.""" """Component providing support for Reolink IP cameras."""
from __future__ import annotations from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
import logging import logging
from reolink_aio.api import DUAL_LENS_MODELS from reolink_aio.api import DUAL_LENS_MODELS, Host
from homeassistant.components.camera import Camera, CameraEntityFeature from homeassistant.components.camera import (
Camera,
CameraEntityDescription,
CameraEntityFeature,
)
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
@ -17,6 +23,70 @@ from .entity import ReolinkChannelCoordinatorEntity
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@dataclass(kw_only=True)
class ReolinkCameraEntityDescription(
CameraEntityDescription,
):
"""A class that describes camera entities for a camera channel."""
stream: str
supported: Callable[[Host, int], bool] = lambda api, ch: True
CAMERA_ENTITIES = (
ReolinkCameraEntityDescription(
key="sub",
stream="sub",
translation_key="sub",
),
ReolinkCameraEntityDescription(
key="main",
stream="main",
translation_key="main",
entity_registry_enabled_default=False,
),
ReolinkCameraEntityDescription(
key="snapshots_sub",
stream="snapshots_sub",
translation_key="snapshots_sub",
entity_registry_enabled_default=False,
),
ReolinkCameraEntityDescription(
key="snapshots",
stream="snapshots_main",
translation_key="snapshots_main",
entity_registry_enabled_default=False,
),
ReolinkCameraEntityDescription(
key="ext",
stream="ext",
translation_key="ext",
supported=lambda api, ch: api.protocol in ["rtmp", "flv"],
entity_registry_enabled_default=False,
),
ReolinkCameraEntityDescription(
key="autotrack_sub",
stream="autotrack_sub",
translation_key="autotrack_sub",
supported=lambda api, ch: api.supported(ch, "autotrack_stream"),
),
ReolinkCameraEntityDescription(
key="autotrack_snapshots_sub",
stream="autotrack_snapshots_sub",
translation_key="autotrack_snapshots_sub",
supported=lambda api, ch: api.supported(ch, "autotrack_stream"),
entity_registry_enabled_default=False,
),
ReolinkCameraEntityDescription(
key="autotrack_snapshots_main",
stream="autotrack_snapshots_main",
translation_key="autotrack_snapshots_main",
supported=lambda api, ch: api.supported(ch, "autotrack_stream"),
entity_registry_enabled_default=False,
),
)
async def async_setup_entry( async def async_setup_entry(
hass: HomeAssistant, hass: HomeAssistant,
config_entry: ConfigEntry, config_entry: ConfigEntry,
@ -24,62 +94,59 @@ async def async_setup_entry(
) -> None: ) -> None:
"""Set up a Reolink IP Camera.""" """Set up a Reolink IP Camera."""
reolink_data: ReolinkData = hass.data[DOMAIN][config_entry.entry_id] reolink_data: ReolinkData = hass.data[DOMAIN][config_entry.entry_id]
host = reolink_data.host
cameras = [] entities: list[ReolinkCamera] = []
for channel in host.api.stream_channels: for entity_description in CAMERA_ENTITIES:
streams = ["sub", "main", "snapshots_sub", "snapshots_main"] for channel in reolink_data.host.api.stream_channels:
if host.api.protocol in ["rtmp", "flv"]: if not entity_description.supported(reolink_data.host.api, channel):
streams.append("ext") continue
stream_url = await reolink_data.host.api.get_stream_source(
if host.api.supported(channel, "autotrack_stream"): channel, entity_description.stream
streams.extend( )
["autotrack_sub", "autotrack_snapshots_sub", "autotrack_snapshots_main"] if stream_url is None and "snapshots" not in entity_description.stream:
)
for stream in streams:
stream_url = await host.api.get_stream_source(channel, stream)
if stream_url is None and "snapshots" not in stream:
continue continue
cameras.append(ReolinkCamera(reolink_data, channel, stream))
async_add_entities(cameras) entities.append(ReolinkCamera(reolink_data, channel, entity_description))
async_add_entities(entities)
class ReolinkCamera(ReolinkChannelCoordinatorEntity, Camera): class ReolinkCamera(ReolinkChannelCoordinatorEntity, Camera):
"""An implementation of a Reolink IP camera.""" """An implementation of a Reolink IP camera."""
_attr_supported_features: CameraEntityFeature = CameraEntityFeature.STREAM _attr_supported_features: CameraEntityFeature = CameraEntityFeature.STREAM
entity_description: ReolinkCameraEntityDescription
def __init__( def __init__(
self, self,
reolink_data: ReolinkData, reolink_data: ReolinkData,
channel: int, channel: int,
stream: str, entity_description: ReolinkCameraEntityDescription,
) -> None: ) -> None:
"""Initialize Reolink camera stream.""" """Initialize Reolink camera stream."""
self.entity_description = entity_description
ReolinkChannelCoordinatorEntity.__init__(self, reolink_data, channel) ReolinkChannelCoordinatorEntity.__init__(self, reolink_data, channel)
Camera.__init__(self) Camera.__init__(self)
self._stream = stream
stream_name = self._stream.replace("_", " ")
if self._host.api.model in DUAL_LENS_MODELS: if self._host.api.model in DUAL_LENS_MODELS:
self._attr_name = f"{stream_name} lens {self._channel}" self._attr_translation_key = (
else: f"{entity_description.translation_key}_lens_{self._channel}"
self._attr_name = stream_name )
stream_id = self._stream
if stream_id == "snapshots_main": self._attr_unique_id = (
stream_id = "snapshots" f"{self._host.unique_id}_{channel}_{entity_description.key}"
self._attr_unique_id = f"{self._host.unique_id}_{self._channel}_{stream_id}" )
self._attr_entity_registry_enabled_default = stream in ["sub", "autotrack_sub"]
async def stream_source(self) -> str | None: async def stream_source(self) -> str | None:
"""Return the source of the stream.""" """Return the source of the stream."""
return await self._host.api.get_stream_source(self._channel, self._stream) return await self._host.api.get_stream_source(
self._channel, self.entity_description.stream
)
async def async_camera_image( async def async_camera_image(
self, width: int | None = None, height: int | None = None self, width: int | None = None, height: int | None = None
) -> bytes | None: ) -> bytes | None:
"""Return a still image response from the camera.""" """Return a still image response from the camera."""
return await self._host.api.get_snapshot(self._channel, self._stream) return await self._host.api.get_snapshot(
self._channel, self.entity_description.stream
)

View File

@ -147,6 +147,62 @@
"name": "Guard set current position" "name": "Guard set current position"
} }
}, },
"camera": {
"sub": {
"name": "Fluent"
},
"main": {
"name": "Clear"
},
"snapshots_sub": {
"name": "Snapshots fluent"
},
"snapshots_main": {
"name": "Snapshots clear"
},
"ext": {
"name": "Balanced"
},
"sub_lens_0": {
"name": "Fluent lens 0"
},
"main_lens_0": {
"name": "Clear lens 0"
},
"snapshots_sub_lens_0": {
"name": "Snapshots fluent lens 0"
},
"snapshots_main_lens_0": {
"name": "Snapshots clear lens 0"
},
"ext_lens_0": {
"name": "Balanced lens 0"
},
"sub_lens_1": {
"name": "Fluent lens 1"
},
"main_lens_1": {
"name": "Clear lens 1"
},
"snapshots_sub_lens_1": {
"name": "Snapshots fluent lens 1"
},
"snapshots_main_lens_1": {
"name": "Snapshots clear lens 1"
},
"ext_lens_1": {
"name": "Balanced lens 1"
},
"autotrack_sub": {
"name": "Autotrack fluent"
},
"autotrack_snapshots_sub": {
"name": "Autotrack snapshots fluent"
},
"autotrack_snapshots_main": {
"name": "Autotrack snapshots clear"
}
},
"light": { "light": {
"floodlight": { "floodlight": {
"name": "Floodlight" "name": "Floodlight"