Add UniFi Protect service to remove privacy zones (#111292)

This commit is contained in:
Christopher Bailey 2024-03-14 13:34:45 -04:00 committed by GitHub
parent 02521c9da3
commit 2bf6170a6b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 142 additions and 5 deletions

View File

@ -8,15 +8,15 @@ from typing import Any, cast
from pydantic import ValidationError
from pyunifiprotect.api import ProtectApiClient
from pyunifiprotect.data import Chime
from pyunifiprotect.data import Camera, Chime
from pyunifiprotect.exceptions import ClientError
import voluptuous as vol
from homeassistant.components.binary_sensor import BinarySensorDeviceClass
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import ATTR_DEVICE_ID, Platform
from homeassistant.const import ATTR_DEVICE_ID, ATTR_NAME, Platform
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import (
config_validation as cv,
device_registry as dr,
@ -30,6 +30,8 @@ from .data import async_ufp_instance_for_config_entry_ids
SERVICE_ADD_DOORBELL_TEXT = "add_doorbell_text"
SERVICE_REMOVE_DOORBELL_TEXT = "remove_doorbell_text"
SERVICE_SET_PRIVACY_ZONE = "set_privacy_zone"
SERVICE_REMOVE_PRIVACY_ZONE = "remove_privacy_zone"
SERVICE_SET_DEFAULT_DOORBELL_TEXT = "set_default_doorbell_text"
SERVICE_SET_CHIME_PAIRED = "set_chime_paired_doorbells"
@ -38,6 +40,7 @@ ALL_GLOBAL_SERIVCES = [
SERVICE_REMOVE_DOORBELL_TEXT,
SERVICE_SET_DEFAULT_DOORBELL_TEXT,
SERVICE_SET_CHIME_PAIRED,
SERVICE_REMOVE_PRIVACY_ZONE,
]
DOORBELL_TEXT_SCHEMA = vol.All(
@ -60,6 +63,16 @@ CHIME_PAIRED_SCHEMA = vol.All(
cv.has_at_least_one_key(ATTR_DEVICE_ID),
)
REMOVE_PRIVACY_ZONE_SCHEMA = vol.All(
vol.Schema(
{
**cv.ENTITY_SERVICE_FIELDS,
vol.Required(ATTR_NAME): cv.string,
},
),
cv.has_at_least_one_key(ATTR_DEVICE_ID),
)
@callback
def _async_get_ufp_instance(hass: HomeAssistant, device_id: str) -> ProtectApiClient:
@ -77,6 +90,21 @@ def _async_get_ufp_instance(hass: HomeAssistant, device_id: str) -> ProtectApiCl
raise HomeAssistantError(f"No device found for device id: {device_id}")
@callback
def _async_get_ufp_camera(hass: HomeAssistant, call: ServiceCall) -> Camera:
ref = async_extract_referenced_entity_ids(hass, call)
entity_registry = er.async_get(hass)
entity_id = ref.indirectly_referenced.pop()
camera_entity = entity_registry.async_get(entity_id)
assert camera_entity is not None
assert camera_entity.device_id is not None
camera_mac = _async_unique_id_to_mac(camera_entity.unique_id)
instance = _async_get_ufp_instance(hass, camera_entity.device_id)
return cast(Camera, instance.bootstrap.get_device_from_mac(camera_mac))
@callback
def _async_get_protect_from_call(
hass: HomeAssistant, call: ServiceCall
@ -123,6 +151,29 @@ async def set_default_doorbell_text(hass: HomeAssistant, call: ServiceCall) -> N
await _async_service_call_nvr(hass, call, "set_default_doorbell_message", message)
async def remove_privacy_zone(hass: HomeAssistant, call: ServiceCall) -> None:
"""Remove privacy zone from camera."""
name: str = call.data[ATTR_NAME]
camera = _async_get_ufp_camera(hass, call)
remove_index: int | None = None
for index, zone in enumerate(camera.privacy_zones):
if zone.name == name:
remove_index = index
break
if remove_index is None:
raise ServiceValidationError(
f"Could not find privacy zone with name {name} on camera {camera.display_name}."
)
def remove_zone() -> None:
camera.privacy_zones.pop(remove_index)
await camera.queue_update(remove_zone)
@callback
def _async_unique_id_to_mac(unique_id: str) -> str:
"""Extract the MAC address from the registry entry unique id."""
@ -190,6 +241,11 @@ def async_setup_services(hass: HomeAssistant) -> None:
functools.partial(set_chime_paired_doorbells, hass),
CHIME_PAIRED_SCHEMA,
),
(
SERVICE_REMOVE_PRIVACY_ZONE,
functools.partial(remove_privacy_zone, hass),
REMOVE_PRIVACY_ZONE_SCHEMA,
),
]
for name, method, schema in services:
if hass.services.has_service(DOMAIN, name):

View File

@ -52,3 +52,16 @@ set_chime_paired_doorbells:
integration: unifiprotect
domain: binary_sensor
device_class: occupancy
remove_privacy_zone:
fields:
device_id:
required: true
selector:
device:
integration: unifiprotect
entity:
domain: camera
name:
required: true
selector:
text:

View File

@ -157,6 +157,20 @@
"description": "The doorbells to link to the chime."
}
}
},
"remove_privacy_zone": {
"name": "Remove camera privacy zone",
"description": "Use to remove a privacy zone from a camera.",
"fields": {
"device_id": {
"name": "Camera",
"description": "Camera you want to remove privacy zone from."
},
"name": {
"name": "Privacy Zone Name",
"description": "The name of the zone to remove."
}
}
}
}
}

View File

@ -5,17 +5,19 @@ from __future__ import annotations
from unittest.mock import AsyncMock, Mock
import pytest
from pyunifiprotect.data import Camera, Chime, Light, ModelType
from pyunifiprotect.data import Camera, Chime, Color, Light, ModelType
from pyunifiprotect.data.devices import CameraZone
from pyunifiprotect.exceptions import BadRequest
from homeassistant.components.unifiprotect.const import ATTR_MESSAGE, DOMAIN
from homeassistant.components.unifiprotect.services import (
SERVICE_ADD_DOORBELL_TEXT,
SERVICE_REMOVE_DOORBELL_TEXT,
SERVICE_REMOVE_PRIVACY_ZONE,
SERVICE_SET_CHIME_PAIRED,
SERVICE_SET_DEFAULT_DOORBELL_TEXT,
)
from homeassistant.const import ATTR_DEVICE_ID, ATTR_ENTITY_ID
from homeassistant.const import ATTR_DEVICE_ID, ATTR_ENTITY_ID, ATTR_NAME
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr, entity_registry as er
@ -177,3 +179,55 @@ async def test_set_chime_paired_doorbells(
ufp.api.update_device.assert_called_once_with(
ModelType.CHIME, chime.id, {"cameraIds": sorted([camera1.id, camera2.id])}
)
async def test_remove_privacy_zone_no_zone(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
) -> None:
"""Test remove_privacy_zone service."""
ufp.api.update_device = AsyncMock()
doorbell.privacy_zones = []
await init_entry(hass, ufp, [doorbell])
registry = er.async_get(hass)
camera_entry = registry.async_get("binary_sensor.test_camera_doorbell")
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
DOMAIN,
SERVICE_REMOVE_PRIVACY_ZONE,
{ATTR_DEVICE_ID: camera_entry.device_id, ATTR_NAME: "Testing"},
blocking=True,
)
ufp.api.update_device.assert_not_called()
async def test_remove_privacy_zone(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
) -> None:
"""Test remove_privacy_zone service."""
ufp.api.update_device = AsyncMock()
doorbell.privacy_zones = [
CameraZone(id=0, name="Testing", color=Color("red"), points=[(0, 0), (1, 1)])
]
await init_entry(hass, ufp, [doorbell])
registry = er.async_get(hass)
camera_entry = registry.async_get("binary_sensor.test_camera_doorbell")
await hass.services.async_call(
DOMAIN,
SERVICE_REMOVE_PRIVACY_ZONE,
{ATTR_DEVICE_ID: camera_entry.device_id, ATTR_NAME: "Testing"},
blocking=True,
)
ufp.api.update_device.assert_called()
assert not len(doorbell.privacy_zones)