mirror of https://github.com/home-assistant/core
86 lines
3.2 KiB
Python
86 lines
3.2 KiB
Python
"""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.hass = hass
|
|
self.config_entry = config_entry
|
|
self.session = config_entry_oauth2_flow.OAuth2Session(
|
|
hass, config_entry, implementation
|
|
)
|
|
super().__init__(self.session.token)
|
|
self.devices: list[HomeConnectDevice] = []
|
|
|
|
def refresh_tokens(self) -> dict:
|
|
"""Refresh and return new Home Connect tokens using Home Assistant OAuth2 session."""
|
|
run_coroutine_threadsafe(
|
|
self.session.async_ensure_token_valid(), self.hass.loop
|
|
).result()
|
|
|
|
return self.session.token
|
|
|
|
def get_devices(self) -> list[HomeConnectAppliance]:
|
|
"""Get a dictionary of devices."""
|
|
appl: list[HomeConnectAppliance] = self.get_appliances()
|
|
self.devices = [HomeConnectDevice(self.hass, app) for app in appl]
|
|
return self.devices
|
|
|
|
|
|
class HomeConnectDevice:
|
|
"""Generic Home Connect device."""
|
|
|
|
def __init__(self, hass: HomeAssistant, appliance: HomeConnectAppliance) -> None:
|
|
"""Initialize the device class."""
|
|
self.hass = hass
|
|
self.appliance = appliance
|
|
|
|
def initialize(self) -> None:
|
|
"""Fetch the info needed to initialize the device."""
|
|
try:
|
|
self.appliance.get_status()
|
|
except (HomeConnectError, ValueError):
|
|
_LOGGER.debug("Unable to fetch appliance status. Probably offline")
|
|
try:
|
|
self.appliance.get_settings()
|
|
except (HomeConnectError, ValueError):
|
|
_LOGGER.debug("Unable to fetch settings. Probably offline")
|
|
try:
|
|
program_active = self.appliance.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.appliance.status[BSH_ACTIVE_PROGRAM] = {
|
|
ATTR_VALUE: program_active[ATTR_KEY]
|
|
}
|
|
self.appliance.listen_events(callback=self.event_callback)
|
|
|
|
def event_callback(self, appliance: HomeConnectAppliance) -> None:
|
|
"""Handle event."""
|
|
_LOGGER.debug("Update triggered on %s", appliance.name)
|
|
_LOGGER.debug(self.appliance.status)
|
|
dispatcher_send(self.hass, SIGNAL_UPDATE_ENTITIES, appliance.haId)
|