Add System Bridge notifications (#82318)

* System bridge notifications

Add notify platform

Add file to coverage

Restore and fix lint after rebase

Cleanup

Use entity to register notify service

Fix pylint

Update package to 3.6.0 and add audio actions

Update package to fix conflict

Remove addition

* Run pre-commit run --all-files

* Update homeassistant/components/system_bridge/notify.py

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

* Format

* Fix

* Remove duplicate import

---------

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
Aidan Timson 2023-09-10 16:32:52 +01:00 committed by GitHub
parent 6899245020
commit 51899ce5ad
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 106 additions and 3 deletions

View File

@ -1268,6 +1268,7 @@ omit =
homeassistant/components/system_bridge/__init__.py
homeassistant/components/system_bridge/binary_sensor.py
homeassistant/components/system_bridge/coordinator.py
homeassistant/components/system_bridge/notify.py
homeassistant/components/system_bridge/sensor.py
homeassistant/components/systemmonitor/sensor.py
homeassistant/components/tado/__init__.py

View File

@ -20,7 +20,9 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_API_KEY,
CONF_COMMAND,
CONF_ENTITY_ID,
CONF_HOST,
CONF_NAME,
CONF_PATH,
CONF_PORT,
CONF_URL,
@ -28,7 +30,11 @@ from homeassistant.const import (
)
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv, device_registry as dr
from homeassistant.helpers import (
config_validation as cv,
device_registry as dr,
discovery,
)
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
@ -40,6 +46,7 @@ _LOGGER = logging.getLogger(__name__)
PLATFORMS = [
Platform.BINARY_SENSOR,
Platform.NOTIFY,
Platform.SENSOR,
]
@ -142,7 +149,24 @@ async def async_setup_entry(
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][entry.entry_id] = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
# Set up all platforms except notify
await hass.config_entries.async_forward_entry_setups(
entry, [platform for platform in PLATFORMS if platform != Platform.NOTIFY]
)
# Set up notify platform
hass.async_create_task(
discovery.async_load_platform(
hass,
Platform.NOTIFY,
DOMAIN,
{
CONF_NAME: f"{DOMAIN}_{coordinator.data.system.hostname}",
CONF_ENTITY_ID: entry.entry_id,
},
hass.data[DOMAIN][entry.entry_id],
)
)
if hass.services.has_service(DOMAIN, SERVICE_OPEN_URL):
return True
@ -277,7 +301,9 @@ async def async_setup_entry(
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
unload_ok = await hass.config_entries.async_unload_platforms(
entry, [platform for platform in PLATFORMS if platform != Platform.NOTIFY]
)
if unload_ok:
coordinator: SystemBridgeDataUpdateCoordinator = hass.data[DOMAIN][
entry.entry_id

View File

@ -0,0 +1,76 @@
"""Support for System Bridge notification service."""
from __future__ import annotations
import logging
from typing import Any
from systembridgeconnector.models.notification import Notification
from homeassistant.components.notify import (
ATTR_DATA,
ATTR_TITLE,
ATTR_TITLE_DEFAULT,
BaseNotificationService,
)
from homeassistant.const import ATTR_ICON, CONF_ENTITY_ID
from homeassistant.core import HomeAssistant
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from .const import DOMAIN
from .coordinator import SystemBridgeDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
ATTR_ACTIONS = "actions"
ATTR_AUDIO = "audio"
ATTR_IMAGE = "image"
ATTR_TIMEOUT = "timeout"
async def async_get_service(
hass: HomeAssistant,
config: ConfigType,
discovery_info: DiscoveryInfoType | None = None,
) -> SystemBridgeNotificationService | None:
"""Get the System Bridge notification service."""
if discovery_info is None:
return None
coordinator: SystemBridgeDataUpdateCoordinator = hass.data[DOMAIN][
discovery_info[CONF_ENTITY_ID]
]
return SystemBridgeNotificationService(coordinator)
class SystemBridgeNotificationService(BaseNotificationService):
"""Implement the notification service for System Bridge."""
def __init__(
self,
coordinator: SystemBridgeDataUpdateCoordinator,
) -> None:
"""Initialize the service."""
self._coordinator: SystemBridgeDataUpdateCoordinator = coordinator
async def async_send_message(
self,
message: str = "",
**kwargs: Any,
) -> None:
"""Send a message."""
data = kwargs.get(ATTR_DATA, {}) or {}
notification = Notification(
actions=data.get(ATTR_ACTIONS),
audio=data.get(ATTR_AUDIO),
icon=data.get(ATTR_ICON),
image=data.get(ATTR_IMAGE),
message=message,
timeout=data.get(ATTR_TIMEOUT),
title=kwargs.get(ATTR_TITLE, data.get(ATTR_TITLE, ATTR_TITLE_DEFAULT)),
)
_LOGGER.debug("Sending notification: %s", notification.json())
await self._coordinator.websocket_client.send_notification(notification)