Add weenect

custom_components/weenect/__init__.py

@@ -0,0 +1,155 @@
+"""
+Custom integration to integrate weenect with Home Assistant.
+
+For more details about this integration, please refer to
+https://github.com/eifinger/hass-weenect
+"""
+import asyncio
+from datetime import timedelta
+import logging
+from typing import Any
+
+from aioweenect import AioWeenect
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import Config, HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.dispatcher import async_dispatcher_send
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import (
+    CONF_PASSWORD,
+    CONF_UPDATE_RATE,
+    CONF_USERNAME,
+    DEFAULT_UPDATE_RATE,
+    DOMAIN,
+    PLATFORMS,
+    STARTUP_MESSAGE,
+    TRACKER_ADDED,
+)
+from .services import async_setup_services, async_unload_services
+
+_LOGGER: logging.Logger = logging.getLogger(__package__)
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
+    """Set up this integration using UI."""
+    if hass.data.get(DOMAIN) is None:
+        hass.data.setdefault(DOMAIN, {})
+        await async_setup_services(hass)
+        _LOGGER.info(STARTUP_MESSAGE)
+
+    username = entry.data.get(CONF_USERNAME)
+    password = entry.data.get(CONF_PASSWORD)
+
+    session = async_get_clientsession(hass)
+    client = AioWeenect(username=username, password=password, session=session)
+
+    coordinator = WeenectDataUpdateCoordinator(hass, config_entry=entry, client=client)
+    await coordinator.async_refresh()
+
+    if not coordinator.last_update_success:
+        raise ConfigEntryNotReady
+
+    hass.data[DOMAIN][entry.entry_id] = coordinator
+
+    for platform in PLATFORMS:
+        hass.async_add_job(
+            hass.config_entries.async_forward_entry_setup(entry, platform)
+        )
+
+    entry.add_update_listener(async_reload_entry)
+    return True
+
+
+class WeenectDataUpdateCoordinator(DataUpdateCoordinator):
+    """Class to manage fetching data from the API."""
+
+    def __init__(
+        self, hass: HomeAssistant, config_entry: ConfigEntry, client: AioWeenect
+    ) -> None:
+        """Initialize."""
+        if CONF_UPDATE_RATE in config_entry.options:
+            update_interval = timedelta(seconds=config_entry.options[CONF_UPDATE_RATE])
+        else:
+            update_interval = timedelta(seconds=DEFAULT_UPDATE_RATE)
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=DOMAIN,
+            update_interval=update_interval,
+        )
+        self.client = client
+        self.config_entry = config_entry
+        self.unsub_dispatchers = []
+        self.data = {}
+        self.add_options()
+
+    async def _async_update_data(self):
+        """Update data via library."""
+        try:
+            data = await self.client.get_trackers()
+            data = self.transform_data(data)
+            self._detect_added_and_removed_trackers(data)
+            return data
+        except Exception as exception:
+            raise UpdateFailed() from exception
+
+    def _detect_added_and_removed_trackers(self, data: Any):
+        """Detect if trackers were added or removed."""
+        added = set(data.keys()) - set(self.data.keys())
+        async_dispatcher_send(
+            self.hass, f"{self.config_entry.entry_id}_{TRACKER_ADDED}", added
+        )
+
+    @staticmethod
+    def transform_data(data: Any):
+        """Extract trackers from list and put them in a dict by tracker id."""
+        result = {}
+        for tracker in data["items"]:
+            result[tracker["id"]] = tracker
+        return result
+
+    def add_options(self) -> None:
+        """Add options for weenect integration."""
+        if not self.config_entry.options:
+            options = {
+                CONF_UPDATE_RATE: DEFAULT_UPDATE_RATE,
+            }
+            self.hass.config_entries.async_update_entry(
+                self.config_entry, options=options
+            )
+        else:
+            options = dict(self.config_entry.options)
+            if CONF_UPDATE_RATE not in self.config_entry.options:
+                options[CONF_UPDATE_RATE] = DEFAULT_UPDATE_RATE
+            self.hass.config_entries.async_update_entry(
+                self.config_entry, options=options
+            )
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Handle removal of an entry."""
+    for unsub_dispatcher in hass.data[DOMAIN][entry.entry_id].unsub_dispatchers:
+        unsub_dispatcher()
+    unloaded = all(
+        await asyncio.gather(
+            *[
+                hass.config_entries.async_forward_entry_unload(entry, platform)
+                for platform in PLATFORMS
+            ]
+        )
+    )
+    if unloaded:
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    # If there is no instance of this integration registered anymore
+    if not hass.data[DOMAIN]:
+        await async_unload_services(hass)
+    return unloaded
+
+
+async def async_reload_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
+    """Reload config entry."""
+    await async_unload_entry(hass, entry)
+    await async_setup_entry(hass, entry)

custom_components/weenect/binary_sensor.py

@@ -0,0 +1,86 @@
+"""Binary_sensor platform for weenect."""
+import logging
+from typing import Any, Dict, List
+
+from homeassistant.components.binary_sensor import BinarySensorEntity
+from homeassistant.core import callback
+from homeassistant.helpers.dispatcher import async_dispatcher_connect
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .const import BINARY_SENSOR_TYPES, DOMAIN, TRACKER_ADDED
+from .entity import WeenectEntity
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(hass, config_entry, async_add_entities):
+    """Set up the weenect binary_sensors."""
+
+    coordinator = hass.data[DOMAIN][config_entry.entry_id]
+
+    @callback
+    def async_add_binary_sensors(
+        added: List[int],
+    ) -> None:
+        """Add binary_sensors callback."""
+
+        sensors: list = []
+        for tracker_id in added:
+            for sensor_type in BINARY_SENSOR_TYPES:
+                sensors.append(
+                    WeenectBinarySensor(coordinator, tracker_id, sensor_type)
+                )
+
+        async_add_entities(sensors, True)
+
+    unsub_dispatcher = async_dispatcher_connect(
+        hass,
+        f"{config_entry.entry_id}_{TRACKER_ADDED}",
+        async_add_binary_sensors,
+    )
+    coordinator.unsub_dispatchers.append(unsub_dispatcher)
+    if len(coordinator.data) > 0:
+        async_add_binary_sensors(coordinator.data.keys())
+
+
+class WeenectBinarySensor(WeenectEntity, BinarySensorEntity):
+    """weenect binary_sensor."""
+
+    def __init__(
+        self,
+        coordinator: DataUpdateCoordinator,
+        tracker_id: str,
+        sensor_type: Dict[str, Any],
+    ):
+        super().__init__(coordinator, tracker_id)
+        self._device_class = sensor_type["device_class"]
+        self._value_name = sensor_type["value_name"]
+        self._enabled = sensor_type["enabled"]
+        self._name = sensor_type["name"]
+
+    @property
+    def name(self):
+        """Return the name of this tracker."""
+        if self.id in self.coordinator.data:
+            return f"{self.coordinator.data[self.id]['name']} {self._name}"
+
+    @property
+    def unique_id(self):
+        """Return a unique ID to use for this entity."""
+        return f"{self.id}_{self._value_name}"
+
+    @property
+    def is_on(self):
+        """Return True if the binary sensor is on."""
+        if self.id in self.coordinator.data:
+            return self.coordinator.data[self.id]["position"][0][self._value_name]
+
+    @property
+    def device_class(self):
+        """Device class of this entity."""
+        return self._device_class
+
+    @property
+    def entity_registry_enabled_default(self) -> bool:
+        """Return if the entity should be enabled when first added to the entity registry."""
+        return self._enabled

custom_components/weenect/config_flow.py

@@ -0,0 +1,119 @@
+"""Adds config flow for weenect."""
+import logging
+
+from aioweenect import AioWeenect
+from homeassistant import config_entries
+from homeassistant.core import callback
+from homeassistant.helpers.aiohttp_client import async_create_clientsession
+import voluptuous as vol
+
+from .const import (
+    CONF_PASSWORD,
+    CONF_UPDATE_RATE,
+    CONF_USERNAME,
+    DEFAULT_UPDATE_RATE,
+    DOMAIN,
+)
+
+_LOGGER: logging.Logger = logging.getLogger(__package__)
+
+
+class WeenectFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
+    """Config flow for weenect."""
+
+    VERSION = 1
+    CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
+
+    def __init__(self):
+        """Initialize."""
+        self._errors = {}
+
+    async def async_step_user(self, user_input=None):
+        """Handle a flow initialized by the user."""
+        self._errors = {}
+
+        entries = self._async_current_entries()
+        for entry in entries:
+            if (
+                entry.data[CONF_USERNAME] == user_input[CONF_USERNAME]
+                and entry.data[CONF_PASSWORD] == user_input[CONF_PASSWORD]
+            ):
+                return self.async_abort(reason="already_configured")
+
+        if user_input is not None:
+            valid = await self._test_credentials(
+                user_input[CONF_USERNAME], user_input[CONF_PASSWORD]
+            )
+            if valid:
+                return self.async_create_entry(
+                    title=user_input[CONF_USERNAME],
+                    data=user_input,
+                )
+            self._errors["base"] = "auth"
+
+        return await self._show_config_form(user_input)
+
+    @staticmethod
+    @callback
+    def async_get_options_flow(config_entry):
+        return WeenectOptionsFlowHandler(config_entry)
+
+    async def _show_config_form(self, user_input):  # pylint: disable=unused-argument
+        """Show the configuration form."""
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema(
+                {vol.Required(CONF_USERNAME): str, vol.Required(CONF_PASSWORD): str}
+            ),
+            errors=self._errors,
+        )
+
+    async def _test_credentials(self, username, password) -> bool:
+        """Return true if credentials is valid."""
+        try:
+            session = async_create_clientsession(self.hass)
+            client = AioWeenect(username=username, password=password, session=session)
+            await client.login()
+            return True
+        except Exception as exception:  # pylint: disable=broad-except
+            _LOGGER.debug(exception)
+        return False
+
+
+class WeenectOptionsFlowHandler(config_entries.OptionsFlow):
+    """weenect config flow options handler."""
+
+    def __init__(self, config_entry):
+        """Initialize weenect options flow."""
+        self.config_entry = config_entry
+        self.options = dict(config_entry.options)
+
+    async def async_step_init(self, user_input=None):  # pylint: disable=unused-argument
+        """Manage the options."""
+        return await self.async_step_user()
+
+    async def async_step_user(self, user_input=None):
+        """Handle a flow initialized by the user."""
+        if user_input is not None:
+            self.options.update(user_input)
+            return await self._update_options()
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema(
+                {
+                    vol.Optional(
+                        CONF_UPDATE_RATE,
+                        default=self.config_entry.options.get(
+                            CONF_UPDATE_RATE, DEFAULT_UPDATE_RATE
+                        ),
+                    ): int
+                }
+            ),
+        )
+
+    async def _update_options(self):
+        """Update config entry options."""
+        return self.async_create_entry(
+            title=self.config_entry.data.get(CONF_USERNAME), data=self.options
+        )