Add Reolink chime play action (#123245)

* Add chime play service

* fix supported_feature

* finalize

* add tests

* Adjust to device service

* fix issue

* Add tests

* actions -> services

* fix styling

* Use conftest fixture for test_chime

* Update tests/components/reolink/test_services.py

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

* use ATTR_RINGTONE and rename chime_play to play_chime

* Add translatable exceptions

* fix styling

* Remove option to use entity_id

* fixes

* Fix translations

* fix

* fix translation key

* remove translation key

* use callback for async_setup_services

* fix styling

* Add test_play_chime_service_unloaded

---------

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
starkillerOG 2024-08-26 21:12:32 +02:00 committed by GitHub
parent d8fe3c5377
commit 7334fb0125
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 315 additions and 32 deletions

View File

@ -3,7 +3,6 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
from dataclasses import dataclass
from datetime import timedelta from datetime import timedelta
import logging import logging
@ -14,13 +13,20 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EVENT_HOMEASSISTANT_STOP, Platform from homeassistant.const import EVENT_HOMEASSISTANT_STOP, Platform
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr, entity_registry as er from homeassistant.helpers import (
config_validation as cv,
device_registry as dr,
entity_registry as er,
)
from homeassistant.helpers.device_registry import format_mac from homeassistant.helpers.device_registry import format_mac
from homeassistant.helpers.typing import ConfigType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN from .const import DOMAIN
from .exceptions import PasswordIncompatible, ReolinkException, UserNotAdmin from .exceptions import PasswordIncompatible, ReolinkException, UserNotAdmin
from .host import ReolinkHost from .host import ReolinkHost
from .services import async_setup_services
from .util import ReolinkData, get_device_uid_and_ch
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -40,14 +46,14 @@ DEVICE_UPDATE_INTERVAL = timedelta(seconds=60)
FIRMWARE_UPDATE_INTERVAL = timedelta(hours=12) FIRMWARE_UPDATE_INTERVAL = timedelta(hours=12)
NUM_CRED_ERRORS = 3 NUM_CRED_ERRORS = 3
CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
@dataclass
class ReolinkData:
"""Data for the Reolink integration."""
host: ReolinkHost async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
device_coordinator: DataUpdateCoordinator[None] """Set up Reolink shared code."""
firmware_coordinator: DataUpdateCoordinator[None]
async_setup_services(hass)
return True
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool: async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
@ -265,28 +271,6 @@ async def async_remove_config_entry_device(
return False return False
def get_device_uid_and_ch(
device: dr.DeviceEntry, host: ReolinkHost
) -> tuple[list[str], int | None, bool]:
"""Get the channel and the split device_uid from a reolink DeviceEntry."""
device_uid = [
dev_id[1].split("_") for dev_id in device.identifiers if dev_id[0] == DOMAIN
][0]
is_chime = False
if len(device_uid) < 2:
# NVR itself
ch = None
elif device_uid[1].startswith("ch") and len(device_uid[1]) <= 5:
ch = int(device_uid[1][2:])
elif device_uid[1].startswith("chime"):
ch = int(device_uid[1][5:])
is_chime = True
else:
ch = host.api.channel_for_uid(device_uid[1])
return (device_uid, ch, is_chime)
def migrate_entity_ids( def migrate_entity_ids(
hass: HomeAssistant, config_entry_id: str, host: ReolinkHost hass: HomeAssistant, config_entry_id: str, host: ReolinkHost
) -> None: ) -> None:

View File

@ -300,6 +300,7 @@
} }
}, },
"services": { "services": {
"ptz_move": "mdi:pan" "ptz_move": "mdi:pan",
"play_chime": "mdi:music"
} }
} }

View File

@ -0,0 +1,80 @@
"""Reolink additional services."""
from __future__ import annotations
from reolink_aio.api import Chime
from reolink_aio.enums import ChimeToneEnum
from reolink_aio.exceptions import InvalidParameterError, ReolinkError
import voluptuous as vol
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import ATTR_DEVICE_ID
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import device_registry as dr
from .const import DOMAIN
from .host import ReolinkHost
from .util import get_device_uid_and_ch
ATTR_RINGTONE = "ringtone"
@callback
def async_setup_services(hass: HomeAssistant) -> None:
"""Set up Reolink services."""
async def async_play_chime(service_call: ServiceCall) -> None:
"""Play a ringtone."""
service_data = service_call.data
device_registry = dr.async_get(hass)
for device_id in service_data[ATTR_DEVICE_ID]:
config_entry = None
device = device_registry.async_get(device_id)
if device is not None:
for entry_id in device.config_entries:
config_entry = hass.config_entries.async_get_entry(entry_id)
if config_entry is not None and config_entry.domain == DOMAIN:
break
if (
config_entry is None
or device is None
or config_entry.state == ConfigEntryState.NOT_LOADED
):
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="service_entry_ex",
translation_placeholders={"service_name": "play_chime"},
)
host: ReolinkHost = hass.data[DOMAIN][config_entry.entry_id].host
(device_uid, chime_id, is_chime) = get_device_uid_and_ch(device, host)
chime: Chime | None = host.api.chime(chime_id)
if not is_chime or chime is None:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="service_not_chime",
translation_placeholders={"device_name": str(device.name)},
)
ringtone = service_data[ATTR_RINGTONE]
try:
await chime.play(ChimeToneEnum[ringtone].value)
except InvalidParameterError as err:
raise ServiceValidationError(err) from err
except ReolinkError as err:
raise HomeAssistantError(err) from err
hass.services.async_register(
DOMAIN,
"play_chime",
async_play_chime,
schema=vol.Schema(
{
vol.Required(ATTR_DEVICE_ID): list[str],
vol.Required(ATTR_RINGTONE): vol.In(
[method.name for method in ChimeToneEnum][1:]
),
}
),
)

View File

@ -16,3 +16,30 @@ ptz_move:
min: 1 min: 1
max: 64 max: 64
step: 1 step: 1
play_chime:
fields:
device_id:
required: true
selector:
device:
multiple: true
filter:
integration: reolink
model: "Reolink Chime"
ringtone:
required: true
selector:
select:
translation_key: ringtone
options:
- citybird
- originaltune
- pianokey
- loop
- attraction
- hophop
- goodday
- operetta
- moonlight
- waybackhome

View File

@ -50,6 +50,14 @@
} }
} }
}, },
"exceptions": {
"service_entry_ex": {
"message": "Reolink {service_name} error: config entry not found or not loaded"
},
"service_not_chime": {
"message": "Reolink play_chime error: {device_name} is not a chime"
}
},
"issues": { "issues": {
"https_webhook": { "https_webhook": {
"title": "Reolink webhook URL uses HTTPS (SSL)", "title": "Reolink webhook URL uses HTTPS (SSL)",
@ -86,6 +94,36 @@
"description": "PTZ move speed." "description": "PTZ move speed."
} }
} }
},
"play_chime": {
"name": "Play chime",
"description": "Play a ringtone on a chime.",
"fields": {
"device_id": {
"name": "Target chime",
"description": "The chime to play the ringtone on."
},
"ringtone": {
"name": "Ringtone",
"description": "Ringtone to play."
}
}
}
},
"selector": {
"ringtone": {
"options": {
"citybird": "[%key:component::reolink::entity::select::motion_tone::state::citybird%]",
"originaltune": "[%key:component::reolink::entity::select::motion_tone::state::originaltune%]",
"pianokey": "[%key:component::reolink::entity::select::motion_tone::state::pianokey%]",
"loop": "[%key:component::reolink::entity::select::motion_tone::state::loop%]",
"attraction": "[%key:component::reolink::entity::select::motion_tone::state::attraction%]",
"hophop": "[%key:component::reolink::entity::select::motion_tone::state::hophop%]",
"goodday": "[%key:component::reolink::entity::select::motion_tone::state::goodday%]",
"operetta": "[%key:component::reolink::entity::select::motion_tone::state::operetta%]",
"moonlight": "[%key:component::reolink::entity::select::motion_tone::state::moonlight%]",
"waybackhome": "[%key:component::reolink::entity::select::motion_tone::state::waybackhome%]"
}
} }
}, },
"entity": { "entity": {

View File

@ -2,11 +2,24 @@
from __future__ import annotations from __future__ import annotations
from dataclasses import dataclass
from homeassistant import config_entries from homeassistant import config_entries
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from . import ReolinkData
from .const import DOMAIN from .const import DOMAIN
from .host import ReolinkHost
@dataclass
class ReolinkData:
"""Data for the Reolink integration."""
host: ReolinkHost
device_coordinator: DataUpdateCoordinator[None]
firmware_coordinator: DataUpdateCoordinator[None]
def is_connected(hass: HomeAssistant, config_entry: config_entries.ConfigEntry) -> bool: def is_connected(hass: HomeAssistant, config_entry: config_entries.ConfigEntry) -> bool:
@ -19,3 +32,25 @@ def is_connected(hass: HomeAssistant, config_entry: config_entries.ConfigEntry)
and config_entry.state == config_entries.ConfigEntryState.LOADED and config_entry.state == config_entries.ConfigEntryState.LOADED
and reolink_data.device_coordinator.last_update_success and reolink_data.device_coordinator.last_update_success
) )
def get_device_uid_and_ch(
device: dr.DeviceEntry, host: ReolinkHost
) -> tuple[list[str], int | None, bool]:
"""Get the channel and the split device_uid from a reolink DeviceEntry."""
device_uid = [
dev_id[1].split("_") for dev_id in device.identifiers if dev_id[0] == DOMAIN
][0]
is_chime = False
if len(device_uid) < 2:
# NVR itself
ch = None
elif device_uid[1].startswith("ch") and len(device_uid[1]) <= 5:
ch = int(device_uid[1][2:])
elif device_uid[1].startswith("chime"):
ch = int(device_uid[1][5:])
is_chime = True
else:
ch = host.api.channel_for_uid(device_uid[1])
return (device_uid, ch, is_chime)

View File

@ -118,6 +118,7 @@ async def test_chime_select(
entity_id = f"{Platform.SELECT}.test_chime_visitor_ringtone" entity_id = f"{Platform.SELECT}.test_chime_visitor_ringtone"
assert hass.states.get(entity_id).state == "pianokey" assert hass.states.get(entity_id).state == "pianokey"
# Test selecting chime ringtone option
test_chime.set_tone = AsyncMock() test_chime.set_tone = AsyncMock()
await hass.services.async_call( await hass.services.async_call(
SELECT_DOMAIN, SELECT_DOMAIN,
@ -145,6 +146,7 @@ async def test_chime_select(
blocking=True, blocking=True,
) )
# Test unavailable
test_chime.event_info = {} test_chime.event_info = {}
freezer.tick(DEVICE_UPDATE_INTERVAL) freezer.tick(DEVICE_UPDATE_INTERVAL)
async_fire_time_changed(hass) async_fire_time_changed(hass)

View File

@ -0,0 +1,116 @@
"""Test the Reolink services."""
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from reolink_aio.api import Chime
from reolink_aio.exceptions import InvalidParameterError, ReolinkError
from homeassistant.components.reolink.const import DOMAIN as REOLINK_DOMAIN
from homeassistant.components.reolink.services import ATTR_RINGTONE
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import ATTR_DEVICE_ID, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry
async def test_play_chime_service_entity(
hass: HomeAssistant,
config_entry: MockConfigEntry,
reolink_connect: MagicMock,
test_chime: Chime,
entity_registry: er.EntityRegistry,
) -> None:
"""Test chime play service."""
with patch("homeassistant.components.reolink.PLATFORMS", [Platform.SELECT]):
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
entity_id = f"{Platform.SELECT}.test_chime_visitor_ringtone"
entity = entity_registry.async_get(entity_id)
assert entity is not None
device_id = entity.device_id
# Test chime play service with device
test_chime.play = AsyncMock()
await hass.services.async_call(
REOLINK_DOMAIN,
"play_chime",
{ATTR_DEVICE_ID: [device_id], ATTR_RINGTONE: "attraction"},
blocking=True,
)
test_chime.play.assert_called_once()
# Test errors
with pytest.raises(ServiceValidationError):
await hass.services.async_call(
REOLINK_DOMAIN,
"play_chime",
{ATTR_DEVICE_ID: ["invalid_id"], ATTR_RINGTONE: "attraction"},
blocking=True,
)
test_chime.play = AsyncMock(side_effect=ReolinkError("Test error"))
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
REOLINK_DOMAIN,
"play_chime",
{ATTR_DEVICE_ID: [device_id], ATTR_RINGTONE: "attraction"},
blocking=True,
)
test_chime.play = AsyncMock(side_effect=InvalidParameterError("Test error"))
with pytest.raises(ServiceValidationError):
await hass.services.async_call(
REOLINK_DOMAIN,
"play_chime",
{ATTR_DEVICE_ID: [device_id], ATTR_RINGTONE: "attraction"},
blocking=True,
)
reolink_connect.chime.return_value = None
with pytest.raises(ServiceValidationError):
await hass.services.async_call(
REOLINK_DOMAIN,
"play_chime",
{ATTR_DEVICE_ID: [device_id], ATTR_RINGTONE: "attraction"},
blocking=True,
)
async def test_play_chime_service_unloaded(
hass: HomeAssistant,
config_entry: MockConfigEntry,
reolink_connect: MagicMock,
test_chime: Chime,
entity_registry: er.EntityRegistry,
) -> None:
"""Test chime play service when config entry is unloaded."""
with patch("homeassistant.components.reolink.PLATFORMS", [Platform.SELECT]):
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
entity_id = f"{Platform.SELECT}.test_chime_visitor_ringtone"
entity = entity_registry.async_get(entity_id)
assert entity is not None
device_id = entity.device_id
# Unload the config entry
assert await hass.config_entries.async_unload(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.NOT_LOADED
# Test chime play service
with pytest.raises(ServiceValidationError):
await hass.services.async_call(
REOLINK_DOMAIN,
"play_chime",
{ATTR_DEVICE_ID: [device_id], ATTR_RINGTONE: "attraction"},
blocking=True,
)