core/homeassistant/components/playstation_network/notify.py

"""Notify platform for PlayStation Network."""

from __future__ import annotations

from enum import StrEnum
from typing import TYPE_CHECKING

from psnawp_api.core.psnawp_exceptions import (
    PSNAWPClientError,
    PSNAWPForbiddenError,
    PSNAWPNotFoundError,
    PSNAWPServerError,
)
from psnawp_api.models.group.group import Group

from homeassistant.components.notify import (
    DOMAIN as NOTIFY_DOMAIN,
    NotifyEntity,
    NotifyEntityDescription,
)
from homeassistant.config_entries import ConfigSubentry
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback

from .const import DOMAIN
from .coordinator import (
    PlaystationNetworkConfigEntry,
    PlaystationNetworkFriendDataCoordinator,
    PlaystationNetworkGroupsUpdateCoordinator,
)
from .entity import PlaystationNetworkServiceEntity

PARALLEL_UPDATES = 20


class PlaystationNetworkNotify(StrEnum):
    """PlayStation Network sensors."""

    GROUP_MESSAGE = "group_message"
    DIRECT_MESSAGE = "direct_message"


async def async_setup_entry(
    hass: HomeAssistant,
    config_entry: PlaystationNetworkConfigEntry,
    async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
    """Set up the notify entity platform."""

    coordinator = config_entry.runtime_data.groups

    groups_added: set[str] = set()
    entity_registry = er.async_get(hass)

    @callback
    def add_entities() -> None:
        nonlocal groups_added

        new_groups = set(coordinator.data.keys()) - groups_added
        if new_groups:
            async_add_entities(
                PlaystationNetworkNotifyEntity(coordinator, group_id)
                for group_id in new_groups
            )
            groups_added |= new_groups

        deleted_groups = groups_added - set(coordinator.data.keys())
        for group_id in deleted_groups:
            if entity_id := entity_registry.async_get_entity_id(
                NOTIFY_DOMAIN,
                DOMAIN,
                f"{coordinator.config_entry.unique_id}_{group_id}",
            ):
                entity_registry.async_remove(entity_id)

    coordinator.async_add_listener(add_entities)
    add_entities()

    for subentry_id, friend_coordinator in config_entry.runtime_data.friends.items():
        async_add_entities(
            [
                PlaystationNetworkDirectMessageNotifyEntity(
                    friend_coordinator,
                    config_entry.subentries[subentry_id],
                )
            ],
            config_subentry_id=subentry_id,
        )


class PlaystationNetworkNotifyBaseEntity(PlaystationNetworkServiceEntity, NotifyEntity):
    """Base class of PlayStation Network notify entity."""

    group: Group | None = None

    def send_message(self, message: str, title: str | None = None) -> None:
        """Send a message."""
        if TYPE_CHECKING:
            assert self.group
        try:
            self.group.send_message(message)
        except PSNAWPNotFoundError as e:
            raise HomeAssistantError(
                translation_domain=DOMAIN,
                translation_key="group_invalid",
                translation_placeholders=dict(self.translation_placeholders),
            ) from e
        except PSNAWPForbiddenError as e:
            raise HomeAssistantError(
                translation_domain=DOMAIN,
                translation_key="send_message_forbidden",
                translation_placeholders=dict(self.translation_placeholders),
            ) from e
        except (PSNAWPServerError, PSNAWPClientError) as e:
            raise HomeAssistantError(
                translation_domain=DOMAIN,
                translation_key="send_message_failed",
                translation_placeholders=dict(self.translation_placeholders),
            ) from e


class PlaystationNetworkNotifyEntity(PlaystationNetworkNotifyBaseEntity):
    """Representation of a PlayStation Network notify entity."""

    coordinator: PlaystationNetworkGroupsUpdateCoordinator

    def __init__(
        self,
        coordinator: PlaystationNetworkGroupsUpdateCoordinator,
        group_id: str,
    ) -> None:
        """Initialize a notification entity."""
        self.group = coordinator.psn.psn.group(group_id=group_id)
        group_details = coordinator.data[group_id]
        self.entity_description = NotifyEntityDescription(
            key=group_id,
            translation_key=PlaystationNetworkNotify.GROUP_MESSAGE,
            translation_placeholders={
                "group_name": group_details["groupName"]["value"]
                or ", ".join(
                    member["onlineId"]
                    for member in group_details["members"]
                    if member["accountId"] != coordinator.psn.user.account_id
                )
            },
        )

        super().__init__(coordinator, self.entity_description)


class PlaystationNetworkDirectMessageNotifyEntity(PlaystationNetworkNotifyBaseEntity):
    """Representation of a PlayStation Network notify entity for sending direct messages."""

    coordinator: PlaystationNetworkFriendDataCoordinator

    def __init__(
        self,
        coordinator: PlaystationNetworkFriendDataCoordinator,
        subentry: ConfigSubentry,
    ) -> None:
        """Initialize a notification entity."""

        self.entity_description = NotifyEntityDescription(
            key=PlaystationNetworkNotify.DIRECT_MESSAGE,
            translation_key=PlaystationNetworkNotify.DIRECT_MESSAGE,
        )

        super().__init__(coordinator, self.entity_description, subentry)

    def send_message(self, message: str, title: str | None = None) -> None:
        """Send a message."""

        if not self.group:
            self.group = self.coordinator.psn.psn.group(
                users_list=[self.coordinator.user]
            )
        super().send_message(message, title)