api.py

Go to the documentation of this file.

"""API for Home Connect bound to HASS OAuth."""
 
from asyncio import run_coroutine_threadsafe
import logging
 
import homeconnect
from homeconnect.api import HomeConnectAppliance, HomeConnectError
 
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_entry_oauth2_flow
from homeassistant.helpers.dispatcher import dispatcher_send
 
from .const import ATTR_KEY, ATTR_VALUE, BSH_ACTIVE_PROGRAM, SIGNAL_UPDATE_ENTITIES
 
_LOGGER = logging.getLogger(__name__)
 
 
class ConfigEntryAuth(homeconnect.HomeConnectAPI):
    """Provide Home Connect authentication tied to an OAuth2 based config entry."""
 
    def __init__(
        self,
        hass: HomeAssistant,
        config_entry: ConfigEntry,
        implementation: config_entry_oauth2_flow.AbstractOAuth2Implementation,
    ) -> None:
        """Initialize Home Connect Auth."""
        self.hasshass = hass
        self.config_entryconfig_entry = config_entry
        self.sessionsession = config_entry_oauth2_flow.OAuth2Session(
            hass, config_entry, implementation
        )
        super().__init__(self.sessionsession.token)
        self.devicesdevices: list[HomeConnectDevice] = []
 
    def refresh_tokens(self) -> dict:
        """Refresh and return new Home Connect tokens using Home Assistant OAuth2 session."""
        run_coroutine_threadsafe(
            self.sessionsession.async_ensure_token_valid(), self.hasshass.loop
        ).result()
 
        return self.sessionsession.token
 
    def get_devices(self) -> list[HomeConnectAppliance]:
        """Get a dictionary of devices."""
        appl: list[HomeConnectAppliance] = self.get_appliances()
        self.devicesdevices = [HomeConnectDevice(self.hasshass, app) for app in appl]
        return self.devicesdevices
 
 
class HomeConnectDevice:
    """Generic Home Connect device."""
 
    def __init__(self, hass: HomeAssistant, appliance: HomeConnectAppliance) -> None:
        """Initialize the device class."""
        self.hasshass = hass
        self.applianceappliance = appliance
 
    def initialize(self) -> None:
        """Fetch the info needed to initialize the device."""
        try:
            self.applianceappliance.get_status()
        except (HomeConnectError, ValueError):
            _LOGGER.debug("Unable to fetch appliance status. Probably offline")
        try:
            self.applianceappliance.get_settings()
        except (HomeConnectError, ValueError):
            _LOGGER.debug("Unable to fetch settings. Probably offline")
        try:
            program_active = self.applianceappliance.get_programs_active()
        except (HomeConnectError, ValueError):
            _LOGGER.debug("Unable to fetch active programs. Probably offline")
            program_active = None
        if program_active and ATTR_KEY in program_active:
            self.applianceappliance.status[BSH_ACTIVE_PROGRAM] = {
                ATTR_VALUE: program_active[ATTR_KEY]
            }
        self.applianceappliance.listen_events(callback=self.event_callbackevent_callback)
 
    def event_callback(self, appliance: HomeConnectAppliance) -> None:
        """Handle event."""
        _LOGGER.debug("Update triggered on %s", appliance.name)
        _LOGGER.debug(self.applianceappliance.status)
        dispatcher_send(self.hasshass, SIGNAL_UPDATE_ENTITIES, appliance.haId)