Add message service to LaMetric (#80448)

This commit is contained in:
Franck Nijhof 2022-10-17 16:14:20 +02:00 committed by GitHub
parent f730f96024
commit 326a1b21ca
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 462 additions and 1 deletions

View File

@ -12,6 +12,7 @@ from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN, PLATFORMS
from .coordinator import LaMetricDataUpdateCoordinator
from .services import async_setup_services
CONFIG_SCHEMA = vol.Schema(
vol.All(
@ -31,6 +32,7 @@ CONFIG_SCHEMA = vol.Schema(
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the LaMetric integration."""
async_setup_services(hass)
hass.data[DOMAIN] = {"hass_config": config}
if DOMAIN in config:
async_create_issue(
@ -51,7 +53,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
coordinator = LaMetricDataUpdateCoordinator(hass, entry)
await coordinator.async_config_entry_first_refresh()
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
hass.data[DOMAIN][entry.entry_id] = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
# Set up notify platform, no entry support for notify component yet,

View File

@ -23,3 +23,6 @@ CONF_ICON_TYPE: Final = "icon_type"
CONF_LIFETIME: Final = "lifetime"
CONF_PRIORITY: Final = "priority"
CONF_SOUND: Final = "sound"
CONF_MESSAGE: Final = "message"
SERVICE_MESSAGE: Final = "message"

View File

@ -7,8 +7,12 @@ from typing import Any, TypeVar
from demetriek import LaMetricConnectionError, LaMetricError
from typing_extensions import Concatenate, ParamSpec
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr
from .const import DOMAIN
from .coordinator import LaMetricDataUpdateCoordinator
from .entity import LaMetricEntity
_LaMetricEntityT = TypeVar("_LaMetricEntityT", bound=LaMetricEntity)
@ -44,3 +48,27 @@ def lametric_exception_handler(
) from error
return handler
@callback
def async_get_coordinator_by_device_id(
hass: HomeAssistant, device_id: str
) -> LaMetricDataUpdateCoordinator:
"""Get the LaMetric coordinator for this device ID."""
device_registry = dr.async_get(hass)
if (device_entry := device_registry.async_get(device_id)) is None:
raise ValueError(f"Unknown LaMetric device ID: {device_id}")
for entry_id in device_entry.config_entries:
if (
(entry := hass.config_entries.async_get_entry(entry_id))
and entry.domain == DOMAIN
and entry.entry_id in hass.data[DOMAIN]
):
coordinator: LaMetricDataUpdateCoordinator = hass.data[DOMAIN][
entry.entry_id
]
return coordinator
raise ValueError(f"No coordinator for device ID: {device_id}")

View File

@ -0,0 +1,96 @@
"""Support for LaMetric time services."""
from demetriek import (
AlarmSound,
LaMetricError,
Model,
Notification,
NotificationIconType,
NotificationPriority,
NotificationSound,
Simple,
Sound,
)
import voluptuous as vol
from homeassistant.const import CONF_DEVICE_ID, CONF_ICON
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_validation as cv
from .const import (
CONF_CYCLES,
CONF_ICON_TYPE,
CONF_MESSAGE,
CONF_PRIORITY,
CONF_SOUND,
DOMAIN,
SERVICE_MESSAGE,
)
from .coordinator import LaMetricDataUpdateCoordinator
from .helpers import async_get_coordinator_by_device_id
SERVICE_MESSAGE_SCHEMA = vol.Schema(
{
vol.Required(CONF_DEVICE_ID): cv.string,
vol.Required(CONF_MESSAGE): cv.string,
vol.Optional(CONF_CYCLES, default=1): cv.positive_int,
vol.Optional(CONF_ICON_TYPE, default=NotificationIconType.NONE): vol.Coerce(
NotificationIconType
),
vol.Optional(CONF_PRIORITY, default=NotificationPriority.INFO): vol.Coerce(
NotificationPriority
),
vol.Optional(CONF_SOUND): vol.Any(
vol.Coerce(AlarmSound), vol.Coerce(NotificationSound)
),
vol.Optional(CONF_ICON): cv.string,
}
)
@callback
def async_setup_services(hass: HomeAssistant) -> None:
"""Set up services for the LaMetric integration."""
async def _async_service_text(call: ServiceCall) -> None:
"""Send a message to a LaMetric device."""
coordinator = async_get_coordinator_by_device_id(
hass, call.data[CONF_DEVICE_ID]
)
await async_service_text(coordinator, call)
hass.services.async_register(
DOMAIN,
SERVICE_MESSAGE,
_async_service_text,
schema=SERVICE_MESSAGE_SCHEMA,
)
async def async_service_text(
coordinator: LaMetricDataUpdateCoordinator, call: ServiceCall
) -> None:
"""Send a message to an LaMetric device."""
sound = None
if CONF_SOUND in call.data:
sound = Sound(id=call.data[CONF_SOUND], category=None)
notification = Notification(
icon_type=NotificationIconType(call.data[CONF_ICON_TYPE]),
priority=NotificationPriority(call.data.get(CONF_PRIORITY)),
model=Model(
frames=[
Simple(
icon=call.data.get(CONF_ICON),
text=call.data[CONF_MESSAGE],
)
],
cycles=call.data[CONF_CYCLES],
sound=sound,
),
)
try:
await coordinator.lametric.notify(notification=notification)
except LaMetricError as ex:
raise HomeAssistantError("Could not send LaMetric notification") from ex

View File

@ -0,0 +1,174 @@
message:
name: Display a message
description: Display a message with an optional icon on a LaMetric device.
fields:
device_id:
name: Device
description: The LaMetric device to display the message on.
required: true
selector:
device:
integration: lametric
message:
name: Message
description: The message to display.
required: true
selector:
text:
icon:
name: Icon
description: >-
The ID number of the icon or animation to display. List of all icons
and their IDs can be found at: https://developer.lametric.com/icons
required: false
selector:
text:
sound:
name: Sound
description: The notification sound to play.
required: false
selector:
select:
options:
- label: "Alarm 1"
value: "alarm1"
- label: "Alarm 2"
value: "alarm2"
- label: "Alarm 3"
value: "alarm3"
- label: "Alarm 4"
value: "alarm4"
- label: "Alarm 5"
value: "alarm5"
- label: "Alarm 6"
value: "alarm6"
- label: "Alarm 7"
value: "alarm7"
- label: "Alarm 8"
value: "alarm8"
- label: "Alarm 9"
value: "alarm9"
- label: "Alarm 10"
value: "alarm10"
- label: "Alarm 11"
value: "alarm11"
- label: "Alarm 12"
value: "alarm12"
- label: "Alarm 13"
value: "alarm13"
- label: "Bicycle"
value: "bicycle"
- label: "Car"
value: "car"
- label: "Cash"
value: "cash"
- label: "Cat"
value: "cat"
- label: "Dog 1"
value: "dog"
- label: "Dog 2"
value: "dog2"
- label: "Energy"
value: "energy"
- label: "Knock knock"
value: "knock-knock"
- label: "Letter email"
value: "letter_email"
- label: "Lose 1"
value: "lose1"
- label: "Lose 2"
value: "lose2"
- label: "Negative 1"
value: "negative1"
- label: "Negative 2"
value: "negative2"
- label: "Negative 3"
value: "negative3"
- label: "Negative 4"
value: "negative4"
- label: "Negative 5"
value: "negative5"
- label: "Notification 1"
value: "notification"
- label: "Notification 2"
value: "notification2"
- label: "Notification 3"
value: "notification3"
- label: "Notification 4"
value: "notification4"
- label: "Open door"
value: "open_door"
- label: "Positive 1"
value: "positive1"
- label: "Positive 2"
value: "positive2"
- label: "Positive 3"
value: "positive3"
- label: "Positive 4"
value: "positive4"
- label: "Positive 5"
value: "positive5"
- label: "Positive 6"
value: "positive6"
- label: "Static"
value: "static"
- label: "Thunder"
value: "thunder"
- label: "Water 1"
value: "water1"
- label: "Water 2"
value: "water2"
- label: "Win 1"
value: "win"
- label: "Win 2"
value: "win2"
- label: "Wind"
value: "wind"
- label: "Wind short"
value: "wind_short"
cycles:
name: Cycles
description: >-
The number of times to display the message. When set to 0, the message
will be displayed until dismissed.
required: false
default: 1
selector:
number:
min: 0
max: 10
mode: slider
icon_type:
name: Icon type
description: >-
The type of icon to display, indicating the nature of the notification.
required: false
default: "none"
selector:
select:
mode: dropdown
options:
- label: "None"
value: "none"
- label: "Info"
value: "info"
- label: "Alert"
value: "alert"
priority:
name: Priority
description: >-
The priority of the notification. When the device is running in
screensaver or kiosk mode, only critical priority notifications
will be accepted.
required: false
default: "info"
selector:
select:
mode: dropdown
options:
- label: "Info"
value: "info"
- label: "Warning"
value: "warning"
- label: "Critical"
value: "critical"

View File

@ -0,0 +1,38 @@
"""Tests for the LaMetric helpers."""
from unittest.mock import MagicMock
import pytest
from homeassistant.components.lametric.helpers import async_get_coordinator_by_device_id
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry
async def test_get_coordinator_by_device_id(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_lametric: MagicMock,
) -> None:
"""Test get LaMetric coordinator by device ID ."""
entity_registry = er.async_get(hass)
with pytest.raises(ValueError, match="Unknown LaMetric device ID: bla"):
async_get_coordinator_by_device_id(hass, "bla")
entry = entity_registry.async_get("button.frenck_s_lametric_next_app")
assert entry
assert entry.device_id
coordinator = async_get_coordinator_by_device_id(hass, entry.device_id)
assert coordinator.data == mock_lametric.device.return_value
# Unload entry
await hass.config_entries.async_unload(init_integration.entry_id)
await hass.async_block_till_done()
with pytest.raises(
ValueError, match=f"No coordinator for device ID: {entry.device_id}"
):
async_get_coordinator_by_device_id(hass, entry.device_id)

View File

@ -0,0 +1,120 @@
"""Tests for the LaMetric services."""
from unittest.mock import MagicMock
from demetriek import (
LaMetricError,
Notification,
NotificationIconType,
NotificationPriority,
NotificationSound,
NotificationSoundCategory,
Simple,
)
import pytest
from homeassistant.components.lametric.const import (
CONF_CYCLES,
CONF_ICON_TYPE,
CONF_MESSAGE,
CONF_PRIORITY,
CONF_SOUND,
DOMAIN,
SERVICE_MESSAGE,
)
from homeassistant.const import CONF_DEVICE_ID, CONF_ICON
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry
async def test_service_message(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_lametric: MagicMock,
) -> None:
"""Test the LaMetric text service."""
entity_registry = er.async_get(hass)
entry = entity_registry.async_get("button.frenck_s_lametric_next_app")
assert entry
assert entry.device_id
await hass.services.async_call(
DOMAIN,
SERVICE_MESSAGE,
{
CONF_DEVICE_ID: entry.device_id,
CONF_MESSAGE: "Hi!",
},
blocking=True,
)
assert len(mock_lametric.notify.mock_calls) == 1
notification: Notification = mock_lametric.notify.mock_calls[0][2]["notification"]
assert notification.icon_type is NotificationIconType.NONE
assert notification.life_time is None
assert notification.model.cycles == 1
assert notification.model.sound is None
assert notification.notification_id is None
assert notification.notification_type is None
assert notification.priority is NotificationPriority.INFO
assert len(notification.model.frames) == 1
frame = notification.model.frames[0]
assert type(frame) is Simple
assert frame.icon is None
assert frame.text == "Hi!"
await hass.services.async_call(
DOMAIN,
SERVICE_MESSAGE,
{
CONF_DEVICE_ID: entry.device_id,
CONF_MESSAGE: "Meow!",
CONF_CYCLES: 3,
CONF_ICON_TYPE: "info",
CONF_PRIORITY: "critical",
CONF_SOUND: "cat",
CONF_ICON: "6916",
},
blocking=True,
)
assert len(mock_lametric.notify.mock_calls) == 2
notification: Notification = mock_lametric.notify.mock_calls[1][2]["notification"]
assert notification.icon_type is NotificationIconType.INFO
assert notification.life_time is None
assert notification.model.cycles == 3
assert notification.model.sound is not None
assert notification.model.sound.category is NotificationSoundCategory.NOTIFICATIONS
assert notification.model.sound.sound is NotificationSound.CAT
assert notification.model.sound.repeat == 1
assert notification.notification_id is None
assert notification.notification_type is None
assert notification.priority is NotificationPriority.CRITICAL
assert len(notification.model.frames) == 1
frame = notification.model.frames[0]
assert type(frame) is Simple
assert frame.icon == 6916
assert frame.text == "Meow!"
mock_lametric.notify.side_effect = LaMetricError
with pytest.raises(
HomeAssistantError, match="Could not send LaMetric notification"
):
await hass.services.async_call(
DOMAIN,
SERVICE_MESSAGE,
{
CONF_DEVICE_ID: entry.device_id,
CONF_MESSAGE: "Epic failure!",
},
blocking=True,
)
assert len(mock_lametric.notify.mock_calls) == 3