Add initial version for the YouLess integration (#41942)

Co-authored-by: Franck Nijhof <frenck@frenck.nl>
Co-authored-by: Franck Nijhof <git@frenck.dev>

.coveragerc

@@ -1224,6 +1224,9 @@ omit =
     homeassistant/components/yandex_transport/*
     homeassistant/components/yeelightsunflower/light.py
     homeassistant/components/yi/camera.py
+    homeassistant/components/youless/__init__.py
+    homeassistant/components/youless/const.py
+    homeassistant/components/youless/sensor.py
     homeassistant/components/zabbix/*
     homeassistant/components/zamg/sensor.py
     homeassistant/components/zamg/weather.py

CODEOWNERS

@@ -585,6 +585,7 @@ homeassistant/components/yandex_transport/* @rishatik92 @devbis
 homeassistant/components/yeelight/* @rytilahti @zewelor @shenxn
 homeassistant/components/yeelightsunflower/* @lindsaymarkward
 homeassistant/components/yi/* @bachya
+homeassistant/components/youless/* @gjong
 homeassistant/components/zeroconf/* @bdraco
 homeassistant/components/zerproc/* @emlove
 homeassistant/components/zha/* @dmulcahey @adminiuga

homeassistant/components/youless/__init__.py

@@ -0,0 +1,58 @@
+"""The youless integration."""
+from datetime import timedelta
+import logging
+from urllib.error import URLError
+
+from youless_api import YoulessAPI
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_HOST
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .const import DOMAIN
+
+PLATFORMS = ["sensor"]
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up youless from a config entry."""
+    api = YoulessAPI(entry.data[CONF_HOST])
+
+    try:
+        await hass.async_add_executor_job(api.initialize)
+    except URLError as exception:
+        raise ConfigEntryNotReady from exception
+
+    async def async_update_data():
+        """Fetch data from the API."""
+        await hass.async_add_executor_job(api.update)
+        return api
+
+    coordinator = DataUpdateCoordinator(
+        hass,
+        _LOGGER,
+        name="youless_gateway",
+        update_method=async_update_data,
+        update_interval=timedelta(seconds=2),
+    )
+
+    await coordinator.async_config_entry_first_refresh()
+
+    hass.data.setdefault(DOMAIN, {})
+    hass.data[DOMAIN][entry.entry_id] = coordinator
+    hass.config_entries.async_setup_platforms(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
+    if unload_ok:
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok

homeassistant/components/youless/config_flow.py

@@ -0,0 +1,50 @@
+"""Config flow for youless integration."""
+from __future__ import annotations
+
+import logging
+from typing import Any
+from urllib.error import HTTPError, URLError
+
+import voluptuous as vol
+from youless_api import YoulessAPI
+
+from homeassistant.config_entries import ConfigFlow
+from homeassistant.const import CONF_DEVICE, CONF_HOST
+from homeassistant.data_entry_flow import FlowResult
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+DATA_SCHEMA = vol.Schema({vol.Required(CONF_HOST): str})
+
+
+class YoulessConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for youless."""
+
+    VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the initial step."""
+        errors = {}
+        if user_input is not None:
+            try:
+                api = YoulessAPI(user_input[CONF_HOST])
+                await self.hass.async_add_executor_job(api.initialize)
+            except (HTTPError, URLError):
+                _LOGGER.exception("Cannot connect to host")
+                errors["base"] = "cannot_connect"
+            else:
+                return self.async_create_entry(
+                    title=user_input[CONF_HOST],
+                    data={
+                        CONF_HOST: user_input[CONF_HOST],
+                        CONF_DEVICE: api.mac_address,
+                    },
+                )
+
+        return self.async_show_form(
+            step_id="user", data_schema=DATA_SCHEMA, errors=errors
+        )

homeassistant/components/youless/const.py

@@ -0,0 +1,3 @@
+"""Constants for the youless integration."""
+
+DOMAIN = "youless"

homeassistant/components/youless/manifest.json

@@ -0,0 +1,9 @@
+{
+  "domain": "youless",
+  "name": "YouLess",
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/youless",
+  "requirements": ["youless-api==0.10"],
+  "codeowners": ["@gjong"],
+  "iot_class": "local_polling"
+}

homeassistant/components/youless/sensor.py

@@ -0,0 +1,197 @@
+"""The sensor entity for the Youless integration."""
+from __future__ import annotations
+
+from youless_api.youless_sensor import YoulessSensor
+
+from homeassistant.components.youless import DOMAIN
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_DEVICE, DEVICE_CLASS_POWER
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity import Entity
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.typing import StateType
+from homeassistant.helpers.update_coordinator import (
+    CoordinatorEntity,
+    DataUpdateCoordinator,
+)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
+) -> None:
+    """Initialize the integration."""
+    coordinator = hass.data[DOMAIN][entry.entry_id]
+    device = entry.data[CONF_DEVICE]
+    if device is None:
+        device = entry.entry_id
+
+    async_add_entities(
+        [
+            GasSensor(coordinator, device),
+            PowerMeterSensor(coordinator, device, "low"),
+            PowerMeterSensor(coordinator, device, "high"),
+            PowerMeterSensor(coordinator, device, "total"),
+            CurrentPowerSensor(coordinator, device),
+            DeliveryMeterSensor(coordinator, device, "low"),
+            DeliveryMeterSensor(coordinator, device, "high"),
+            ExtraMeterSensor(coordinator, device, "total"),
+            ExtraMeterSensor(coordinator, device, "usage"),
+        ]
+    )
+
+
+class YoulessBaseSensor(CoordinatorEntity, Entity):
+    """The base sensor for Youless."""
+
+    def __init__(
+        self,
+        coordinator: DataUpdateCoordinator,
+        device: str,
+        device_group: str,
+        friendly_name: str,
+        sensor_id: str,
+    ) -> None:
+        """Create the sensor."""
+        super().__init__(coordinator)
+        self._device = device
+        self._device_group = device_group
+        self._sensor_id = sensor_id
+
+        self._attr_unique_id = f"{DOMAIN}_{device}_{sensor_id}"
+        self._attr_device_info = {
+            "identifiers": {(DOMAIN, f"{device}_{device_group}")},
+            "name": friendly_name,
+            "manufacturer": "YouLess",
+            "model": self.coordinator.data.model,
+        }
+
+    @property
+    def get_sensor(self) -> YoulessSensor | None:
+        """Property to get the underlying sensor object."""
+        return None
+
+    @property
+    def unit_of_measurement(self) -> str | None:
+        """Return the unit of measurement for the sensor."""
+        if self.get_sensor is None:
+            return None
+
+        return self.get_sensor.unit_of_measurement
+
+    @property
+    def state(self) -> StateType:
+        """Determine the state value, only if a sensor is initialized."""
+        if self.get_sensor is None:
+            return None
+
+        return self.get_sensor.value
+
+    @property
+    def available(self) -> bool:
+        """Return a flag to indicate the sensor not being available."""
+        return super().available and self.get_sensor is not None
+
+
+class GasSensor(YoulessBaseSensor):
+    """The Youless gas sensor."""
+
+    def __init__(self, coordinator: DataUpdateCoordinator, device: str) -> None:
+        """Instantiate a gas sensor."""
+        super().__init__(coordinator, device, "gas", "Gas meter", "gas")
+        self._attr_name = "Gas usage"
+        self._attr_icon = "mdi:fire"
+
+    @property
+    def get_sensor(self) -> YoulessSensor | None:
+        """Get the sensor for providing the value."""
+        return self.coordinator.data.gas_meter
+
+
+class CurrentPowerSensor(YoulessBaseSensor):
+    """The current power usage sensor."""
+
+    _attr_device_class = DEVICE_CLASS_POWER
+
+    def __init__(self, coordinator: DataUpdateCoordinator, device: str) -> None:
+        """Instantiate the usage meter."""
+        super().__init__(coordinator, device, "power", "Power usage", "usage")
+        self._device = device
+        self._attr_name = "Power Usage"
+
+    @property
+    def get_sensor(self) -> YoulessSensor | None:
+        """Get the sensor for providing the value."""
+        return self.coordinator.data.current_power_usage
+
+
+class DeliveryMeterSensor(YoulessBaseSensor):
+    """The Youless delivery meter value sensor."""
+
+    _attr_device_class = DEVICE_CLASS_POWER
+
+    def __init__(
+        self, coordinator: DataUpdateCoordinator, device: str, dev_type: str
+    ) -> None:
+        """Instantiate a delivery meter sensor."""
+        super().__init__(
+            coordinator, device, "delivery", "Power delivery", f"delivery_{dev_type}"
+        )
+        self._type = dev_type
+        self._attr_name = f"Power delivery {dev_type}"
+
+    @property
+    def get_sensor(self) -> YoulessSensor | None:
+        """Get the sensor for providing the value."""
+        if self.coordinator.data.delivery_meter is None:
+            return None
+
+        return getattr(self.coordinator.data.delivery_meter, f"_{self._type}", None)
+
+
+class PowerMeterSensor(YoulessBaseSensor):
+    """The Youless low meter value sensor."""
+
+    _attr_device_class = DEVICE_CLASS_POWER
+
+    def __init__(
+        self, coordinator: DataUpdateCoordinator, device: str, dev_type: str
+    ) -> None:
+        """Instantiate a power meter sensor."""
+        super().__init__(
+            coordinator, device, "power", "Power usage", f"power_{dev_type}"
+        )
+        self._device = device
+        self._type = dev_type
+        self._attr_name = f"Power {dev_type}"
+
+    @property
+    def get_sensor(self) -> YoulessSensor | None:
+        """Get the sensor for providing the value."""
+        if self.coordinator.data.power_meter is None:
+            return None
+
+        return getattr(self.coordinator.data.power_meter, f"_{self._type}", None)
+
+
+class ExtraMeterSensor(YoulessBaseSensor):
+    """The Youless extra meter value sensor (s0)."""
+
+    _attr_device_class = DEVICE_CLASS_POWER
+
+    def __init__(
+        self, coordinator: DataUpdateCoordinator, device: str, dev_type: str
+    ) -> None:
+        """Instantiate an extra meter sensor."""
+        super().__init__(
+            coordinator, device, "extra", "Extra meter", f"extra_{dev_type}"
+        )
+        self._type = dev_type
+        self._attr_name = f"Extra {dev_type}"
+
+    @property
+    def get_sensor(self) -> YoulessSensor | None:
+        """Get the sensor for providing the value."""
+        if self.coordinator.data.extra_meter is None:
+            return None
+
+        return getattr(self.coordinator.data.extra_meter, f"_{self._type}", None)

homeassistant/components/youless/strings.json

@@ -0,0 +1,15 @@
+{
+  "config": {
+    "step": {
+      "user": {
+        "data": {
+          "name": "[%key:common::config_flow::data::name%]",
+          "host": "[%key:common::config_flow::data::host%]"
+        }
+      }
+    },
+    "error": {
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
+    }
+  }
+}

homeassistant/components/youless/translations/en.json

@@ -0,0 +1,21 @@
+{
+    "config": {
+        "abort": {
+            "already_configured": "Device is already configured"
+        },
+        "error": {
+            "cannot_connect": "Failed to connect",
+            "invalid_auth": "Invalid authentication",
+            "unknown": "Unexpected error"
+        },
+        "step": {
+            "user": {
+                "data": {
+                    "host": "Host",
+                    "password": "Password",
+                    "username": "Username"
+                }
+            }
+        }
+    }
+}

homeassistant/generated/config_flows.py

@@ -303,6 +303,7 @@ FLOWS = [
     "yale_smart_alarm",
     "yamaha_musiccast",
     "yeelight",
+    "youless",
     "zerproc",
     "zha",
     "zwave",

requirements_all.txt

@@ -2422,6 +2422,9 @@ yeelight==0.6.3
 # homeassistant.components.yeelightsunflower
 yeelightsunflower==0.0.10
 
+# homeassistant.components.youless
+youless-api==0.10
+
 # homeassistant.components.media_extractor
 youtube_dl==2021.04.26
 

requirements_test_all.txt

@@ -1334,6 +1334,9 @@ yalexs==1.1.12
 # homeassistant.components.yeelight
 yeelight==0.6.3
 
+# homeassistant.components.youless
+youless-api==0.10
+
 # homeassistant.components.onvif
 zeep[async]==4.0.0
 

tests/components/youless/__init__.py

@@ -0,0 +1 @@
+"""Tests for the youless component."""

tests/components/youless/test_config_flows.py

@@ -0,0 +1,72 @@
+"""Test the youless config flow."""
+from unittest.mock import MagicMock, patch
+from urllib.error import URLError
+
+from homeassistant.components.youless import DOMAIN
+from homeassistant.config_entries import SOURCE_USER
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import RESULT_TYPE_CREATE_ENTRY, RESULT_TYPE_FORM
+
+
+def _get_mock_youless_api(initialize=None):
+    mock_youless = MagicMock()
+    if isinstance(initialize, Exception):
+        type(mock_youless).initialize = MagicMock(side_effect=initialize)
+    else:
+        type(mock_youless).initialize = MagicMock(return_value=initialize)
+
+    type(mock_youless).mac_address = None
+    return mock_youless
+
+
+async def test_full_flow(hass: HomeAssistant) -> None:
+    """Check setup."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_USER}
+    )
+
+    assert result.get("type") == RESULT_TYPE_FORM
+    assert result.get("errors") == {}
+    assert result.get("step_id") == SOURCE_USER
+    assert "flow_id" in result
+
+    mock_youless = _get_mock_youless_api(
+        initialize={"homes": [{"id": 1, "name": "myhome"}]}
+    )
+    with patch(
+        "homeassistant.components.youless.config_flow.YoulessAPI",
+        return_value=mock_youless,
+    ) as mocked_youless:
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"host": "localhost"},
+        )
+
+    assert result2.get("type") == RESULT_TYPE_CREATE_ENTRY
+    assert result2.get("title") == "localhost"
+    assert len(mocked_youless.mock_calls) == 1
+
+
+async def test_not_found(hass: HomeAssistant) -> None:
+    """Check setup."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_USER}
+    )
+
+    assert result.get("type") == RESULT_TYPE_FORM
+    assert result.get("errors") == {}
+    assert result.get("step_id") == SOURCE_USER
+    assert "flow_id" in result
+
+    mock_youless = _get_mock_youless_api(initialize=URLError(""))
+    with patch(
+        "homeassistant.components.youless.config_flow.YoulessAPI",
+        return_value=mock_youless,
+    ) as mocked_youless:
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"host": "localhost"},
+        )
+
+    assert result2.get("type") == RESULT_TYPE_FORM
+    assert len(mocked_youless.mock_calls) == 1