Add config flow for Vivotek integration

homeassistant/components/vivotek/__init__.py

@@ -1 +1,91 @@
 """The Vivotek camera component."""
+
+from dataclasses import dataclass
+import logging
+from types import MappingProxyType
+from typing import Any, TypedDict
+
+from libpyvivotek import VivotekCamera
+
+from homeassistant import config_entries
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import (
+    CONF_AUTHENTICATION,
+    CONF_IP_ADDRESS,
+    CONF_PASSWORD,
+    CONF_PORT,
+    CONF_USERNAME,
+    CONF_VERIFY_SSL,
+    HTTP_DIGEST_AUTHENTICATION,
+    Platform,
+)
+from homeassistant.core import HomeAssistant
+
+from .const import CONF_SECURITY_LEVEL
+
+_LOGGER = logging.getLogger(__name__)
+
+PLATFORMS = [Platform.CAMERA]
+
+type VivotekConfigEntry = config_entries.ConfigEntry[VivotekData]
+
+
+class VivotekCameraConf(TypedDict):
+    """Vivotek Camera configuration type."""
+
+    authentication: str
+    ip_address: str
+    password: str
+    port: int
+    security_level: str
+    ssl: bool
+    username: str
+    verify_ssl: bool
+
+
+async def async_build_and_test_cam_client(
+    hass: HomeAssistant,
+    data: dict[str, Any] | MappingProxyType[str, Any] | VivotekCameraConf,
+) -> VivotekCamera:
+    """Build the client and test if the provided configuration is valid."""
+    cam_client = VivotekCamera(
+        host=data[CONF_IP_ADDRESS],
+        port=data[CONF_PORT],
+        verify_ssl=data[CONF_VERIFY_SSL],
+        usr=data[CONF_USERNAME],
+        pwd=data[CONF_PASSWORD],
+        digest_auth=(data[CONF_AUTHENTICATION] == HTTP_DIGEST_AUTHENTICATION),
+        sec_lvl=data[CONF_SECURITY_LEVEL],  # type: ignore[literal-required]
+    )
+    mac = await hass.async_add_executor_job(cam_client.get_mac)
+    assert len(mac) > 0
+    return cam_client
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: VivotekConfigEntry) -> bool:
+    """Set up the Vivotek component from a config entry."""
+
+    try:
+        cam_client = await async_build_and_test_cam_client(hass, entry.data)
+        entry.runtime_data = VivotekData(
+            cam_client=cam_client,
+        )
+    except Exception:
+        _LOGGER.exception("Unexpected exception during setup")
+        return False
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
+
+
+@dataclass
+class VivotekData:
+    """Data for the Vivotek component."""
+
+    cam_client: VivotekCamera

homeassistant/components/vivotek/camera.py

@@ -2,6 +2,10 @@
 
 from __future__ import annotations
 
+import logging
+from types import MappingProxyType
+from typing import Any
+
 from libpyvivotek import VivotekCamera
 import voluptuous as vol
 
@@ -22,20 +26,32 @@
     HTTP_DIGEST_AUTHENTICATION,
 )
 from homeassistant.core import HomeAssistant
-from homeassistant.helpers import config_validation as cv
-from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers import config_validation as cv, device_registry as dr
+from homeassistant.helpers.device_registry import DeviceInfo, format_mac
+from homeassistant.helpers.entity_platform import (
+    AddConfigEntryEntitiesCallback,
+    AddEntitiesCallback,
+)
 from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
 
-CONF_FRAMERATE = "framerate"
-CONF_SECURITY_LEVEL = "security_level"
-CONF_STREAM_PATH = "stream_path"
+from . import VivotekConfigEntry
+from .const import (
+    CONF_FRAMERATE,
+    CONF_SECURITY_LEVEL,
+    CONF_STREAM_PATH,
+    DOMAIN,
+    MANUFACTURER,
+)
+
+_LOGGER = logging.getLogger(__name__)
 
 DEFAULT_CAMERA_BRAND = "VIVOTEK"
 DEFAULT_NAME = "VIVOTEK Camera"
 DEFAULT_EVENT_0_KEY = "event_i0_enable"
 DEFAULT_SECURITY_LEVEL = "admin"
 DEFAULT_STREAM_SOURCE = "live.sdp"
 
+
 PLATFORM_SCHEMA = CAMERA_PLATFORM_SCHEMA.extend(
     {
         vol.Required(CONF_IP_ADDRESS): cv.string,
@@ -54,13 +70,43 @@
 )
 
 
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: VivotekConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up the component from a config entry."""
+    config = entry.data
+    creds = f"{config[CONF_USERNAME]}:{config[CONF_PASSWORD]}"
+    stream_source = (
+        f"rtsp://{creds}@{config[CONF_IP_ADDRESS]}:554/{config[CONF_STREAM_PATH]}"
+    )
+    cam_client = entry.runtime_data.cam_client
+    mac_address = await hass.async_add_executor_job(cam_client.get_mac)
+    unique_id = format_mac(mac_address)
+    device_registry = dr.async_get(hass)
+    device_registry.async_get_or_create(
+        config_entry_id=entry.entry_id,
+        identifiers={(DOMAIN, unique_id)},
+        connections={(dr.CONNECTION_NETWORK_MAC, mac_address)},
+    )
+
+    if not entry.unique_id:
+        hass.config_entries.async_update_entry(
+            entry,
+            unique_id=unique_id,
+            title=str(config[CONF_NAME]),
+        )
+    async_add_entities([VivotekCam(entry.data, cam_client, stream_source, unique_id)])
+
+
 def setup_platform(
     hass: HomeAssistant,
     config: ConfigType,
     add_entities: AddEntitiesCallback,
     discovery_info: DiscoveryInfoType | None = None,
 ) -> None:
-    """Set up a Vivotek IP Camera."""
+    """Set up a Vivotek IP Camera from Yaml."""
     creds = f"{config[CONF_USERNAME]}:{config[CONF_PASSWORD]}"
     cam = VivotekCamera(
         host=config[CONF_IP_ADDRESS],
@@ -74,46 +120,102 @@ def setup_platform(
     stream_source = (
         f"rtsp://{creds}@{config[CONF_IP_ADDRESS]}:554/{config[CONF_STREAM_PATH]}"
     )
-    add_entities([VivotekCam(config, cam, stream_source)], True)
+    unique_id = cam.get_mac()
+    add_entities([VivotekCam(config, cam, stream_source, unique_id)], True)
 
 
 class VivotekCam(Camera):
     """A Vivotek IP camera."""
 
+    _attr_available = False
     _attr_brand = DEFAULT_CAMERA_BRAND
+    _attr_configuration_url: str | None = None
+    _attr_model: str | None = None
+    _attr_serial: str | None = None
     _attr_supported_features = CameraEntityFeature.STREAM
+    _attr_unique_id = ""
 
     def __init__(
-        self, config: ConfigType, cam: VivotekCamera, stream_source: str
+        self,
+        config: ConfigType | MappingProxyType[str, Any],
+        cam_client: VivotekCamera,
+        stream_source: str,
+        unique_id: str,
     ) -> None:
         """Initialize a Vivotek camera."""
         super().__init__()
-
-        self._cam = cam
+        self._cam_client = cam_client
+        self._attr_configuration_url = f"http://{config[CONF_IP_ADDRESS]}"
         self._attr_frame_interval = 1 / config[CONF_FRAMERATE]
         self._attr_name = config[CONF_NAME]
+        self._attr_unique_id = unique_id
         self._stream_source = stream_source
 
     def camera_image(
         self, width: int | None = None, height: int | None = None
     ) -> bytes | None:
         """Return bytes of camera image."""
-        return self._cam.snapshot()
+        return self._cam_client.snapshot()
 
     async def stream_source(self) -> str:
         """Return the source of the stream."""
         return self._stream_source
 
     def disable_motion_detection(self) -> None:
         """Disable motion detection in camera."""
-        response = self._cam.set_param(DEFAULT_EVENT_0_KEY, 0)
+        response = self._cam_client.set_param(DEFAULT_EVENT_0_KEY, 0)
         self._attr_motion_detection_enabled = int(response) == 1
 
     def enable_motion_detection(self) -> None:
         """Enable motion detection in camera."""
-        response = self._cam.set_param(DEFAULT_EVENT_0_KEY, 1)
+        response = self._cam_client.set_param(DEFAULT_EVENT_0_KEY, 1)
         self._attr_motion_detection_enabled = int(response) == 1
 
     def update(self) -> None:
         """Update entity status."""
-        self._attr_model = self._cam.model_name
+        self._attr_model = self._cam_client.model_name
+        self._attr_available = self._attr_model is not None
+        self._attr_serial = self._cam_client.get_serial()
+
+    async def async_update(self) -> None:
+        """Update the entity."""
+        await self.hass.async_add_executor_job(self.update)
+
+    @property
+    def available(self) -> bool:
+        """Return True if entity is available."""
+        return self._attr_available
+
+    @property
+    def brand(self) -> str | None:
+        """Return the camera brand."""
+        return self._attr_brand
+
+    @property
+    def configuration_url(self) -> str | None:
+        """Return the configuration URL for the camera."""
+        return self._attr_configuration_url
+
+    @property
+    def model(self) -> str | None:
+        """Return the camera model."""
+        if self._attr_model is None:
+            self._attr_model = self._cam_client.model_name
+        return self._attr_model
+
+    @property
+    def name(self) -> str | None:
+        """Return the name of this device."""
+        return self._attr_name
+
+    @property
+    def device_info(self) -> DeviceInfo | None:
+        """Return the device info."""
+        return DeviceInfo(
+            configuration_url=self._attr_configuration_url,
+            identifiers={(DOMAIN, self._attr_unique_id)},
+            manufacturer=MANUFACTURER,
+            model=self._attr_model,
+            name=self._attr_name,
+            serial_number=self._attr_unique_id,
+        )