"""Support for LaMetric time services.""" from __future__ import annotations from collections.abc import Sequence from demetriek import ( AlarmSound, Chart, 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_DATA, CONF_ICON_TYPE, CONF_MESSAGE, CONF_PRIORITY, CONF_SOUND, DOMAIN, SERVICE_CHART, SERVICE_MESSAGE, ) from .coordinator import LaMetricDataUpdateCoordinator from .helpers import async_get_coordinator_by_device_id SERVICE_BASE_SCHEMA = vol.Schema( { vol.Required(CONF_DEVICE_ID): 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) ), } ) SERVICE_MESSAGE_SCHEMA = SERVICE_BASE_SCHEMA.extend( { vol.Required(CONF_MESSAGE): cv.string, vol.Optional(CONF_ICON): cv.string, } ) SERVICE_CHART_SCHEMA = SERVICE_BASE_SCHEMA.extend( { vol.Required(CONF_DATA): vol.All(cv.ensure_list, [vol.Coerce(int)]), } ) @callback def async_setup_services(hass: HomeAssistant) -> None: """Set up services for the LaMetric integration.""" async def _async_service_chart(call: ServiceCall) -> None: """Send a chart to a LaMetric device.""" coordinator = async_get_coordinator_by_device_id( hass, call.data[CONF_DEVICE_ID] ) await async_send_notification( coordinator, call, [Chart(data=call.data[CONF_DATA])] ) async def _async_service_message(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_send_notification( coordinator, call, [ Simple( icon=call.data.get(CONF_ICON), text=call.data[CONF_MESSAGE], ) ], ) hass.services.async_register( DOMAIN, SERVICE_CHART, _async_service_chart, schema=SERVICE_CHART_SCHEMA, ) hass.services.async_register( DOMAIN, SERVICE_MESSAGE, _async_service_message, schema=SERVICE_MESSAGE_SCHEMA, ) async def async_send_notification( coordinator: LaMetricDataUpdateCoordinator, call: ServiceCall, frames: Sequence[Chart | Simple], ) -> None: """Send a notification 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=frames, 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