subscriber.py

Go to the documentation of this file.

"""Mixin class for handling harmony callback subscriptions."""
 
from __future__ import annotations
 
import asyncio
import logging
from typing import Any, NamedTuple
 
from homeassistant.core import CALLBACK_TYPE, HassJob, HomeAssistant, callback
 
_LOGGER = logging.getLogger(__name__)
 
type NoParamCallback = HassJob[[], Any] | None
type ActivityCallback = HassJob[[tuple], Any] | None
 
 
class HarmonyCallback(NamedTuple):
    """Callback type for Harmony Hub notifications."""
 
    connected: NoParamCallback
    disconnected: NoParamCallback
    config_updated: NoParamCallback
    activity_starting: ActivityCallback
    activity_started: ActivityCallback
 
 
class HarmonySubscriberMixin:
    """Base implementation for a subscriber."""
 
    def __init__(self, hass: HomeAssistant) -> None:
        """Initialize an subscriber."""
        super().__init__()
        self._hass_hass = hass
        self._subscriptions: list[HarmonyCallback] = []
        self._activity_lock_activity_lock = asyncio.Lock()
 
    async def async_lock_start_activity(self) -> None:
        """Acquire the lock."""
        await self._activity_lock_activity_lock.acquire()
 
    @callback
    def async_unlock_start_activity(self) -> None:
        """Release the lock."""
        if self._activity_lock_activity_lock.locked():
            self._activity_lock_activity_lock.release()
 
    @callback
    def async_subscribe(self, update_callbacks: HarmonyCallback) -> CALLBACK_TYPE:
        """Add a callback subscriber."""
        self._subscriptions.append(update_callbacks)
 
        def _unsubscribe() -> None:
            self.async_unsubscribeasync_unsubscribe(update_callbacks)
 
        return _unsubscribe
 
    @callback
    def async_unsubscribe(self, update_callback: HarmonyCallback) -> None:
        """Remove a callback subscriber."""
        self._subscriptions.remove(update_callback)
 
    def _config_updated(self, _: dict | None = None) -> None:
        _LOGGER.debug("config_updated")
        self._call_callbacks_call_callbacks("config_updated")
 
    def _connected(self, _: str | None = None) -> None:
        _LOGGER.debug("connected")
        self.async_unlock_start_activityasync_unlock_start_activity()
        self._available_available = True
        self._call_callbacks_call_callbacks("connected")
 
    def _disconnected(self, _: str | None = None) -> None:
        _LOGGER.debug("disconnected")
        self.async_unlock_start_activityasync_unlock_start_activity()
        self._available_available = False
        self._call_callbacks_call_callbacks("disconnected")
 
    def _activity_starting(self, activity_info: tuple) -> None:
        _LOGGER.debug("activity %s starting", activity_info)
        self._call_callbacks_call_callbacks("activity_starting", activity_info)
 
    def _activity_started(self, activity_info: tuple) -> None:
        _LOGGER.debug("activity %s started", activity_info)
        self.async_unlock_start_activityasync_unlock_start_activity()
        self._call_callbacks_call_callbacks("activity_started", activity_info)
 
    def _call_callbacks(
        self, callback_func_name: str, argument: tuple | None = None
    ) -> None:
        for subscription in self._subscriptions:
            if current_callback_job := getattr(subscription, callback_func_name):
                if argument:
                    self._hass_hass.async_run_hass_job(current_callback_job, argument)
                else:
                    self._hass_hass.async_run_hass_job(current_callback_job)