Add Peblar Rocksolid EV Chargers integration (#133501)

* Add Peblar Rocksolid EV Chargers integration

* Process review comments

.strict-typing

@@ -363,6 +363,7 @@ homeassistant.components.otbr.*
 homeassistant.components.overkiz.*
 homeassistant.components.p1_monitor.*
 homeassistant.components.panel_custom.*
+homeassistant.components.peblar.*
 homeassistant.components.peco.*
 homeassistant.components.persistent_notification.*
 homeassistant.components.pi_hole.*

CODEOWNERS

@@ -1113,6 +1113,8 @@ build.json @home-assistant/supervisor
 /tests/components/palazzetti/ @dotvav
 /homeassistant/components/panel_custom/ @home-assistant/frontend
 /tests/components/panel_custom/ @home-assistant/frontend
+/homeassistant/components/peblar/ @frenck
+/tests/components/peblar/ @frenck
 /homeassistant/components/peco/ @IceBotYT
 /tests/components/peco/ @IceBotYT
 /homeassistant/components/pegel_online/ @mib1185

homeassistant/components/peblar/__init__.py

@@ -0,0 +1,54 @@
+"""Integration for Peblar EV chargers."""
+
+from __future__ import annotations
+
+from aiohttp import CookieJar
+from peblar import (
+    AccessMode,
+    Peblar,
+    PeblarAuthenticationError,
+    PeblarConnectionError,
+    PeblarError,
+)
+
+from homeassistant.const import CONF_HOST, CONF_PASSWORD, Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady
+from homeassistant.helpers.aiohttp_client import async_create_clientsession
+
+from .coordinator import PeblarConfigEntry, PeblarMeterDataUpdateCoordinator
+
+PLATFORMS = [Platform.SENSOR]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: PeblarConfigEntry) -> bool:
+    """Set up Peblar from a config entry."""
+
+    peblar = Peblar(
+        host=entry.data[CONF_HOST],
+        session=async_create_clientsession(hass, cookie_jar=CookieJar(unsafe=True)),
+    )
+    try:
+        await peblar.login(password=entry.data[CONF_PASSWORD])
+        api = await peblar.rest_api(enable=True, access_mode=AccessMode.READ_WRITE)
+    except PeblarConnectionError as err:
+        raise ConfigEntryNotReady("Could not connect to Peblar charger") from err
+    except PeblarAuthenticationError as err:
+        raise ConfigEntryError("Could not login to Peblar charger") from err
+    except PeblarError as err:
+        raise ConfigEntryNotReady(
+            "Unknown error occurred while connecting to Peblar charger"
+        ) from err
+
+    coordinator = PeblarMeterDataUpdateCoordinator(hass, entry, api)
+    await coordinator.async_config_entry_first_refresh()
+
+    entry.runtime_data = coordinator
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: PeblarConfigEntry) -> bool:
+    """Unload Peblar config entry."""
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

homeassistant/components/peblar/config_flow.py

@@ -0,0 +1,71 @@
+"""Config flow to configure the Peblar integration."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from aiohttp import CookieJar
+from peblar import Peblar, PeblarAuthenticationError, PeblarConnectionError
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_HOST, CONF_PASSWORD
+from homeassistant.helpers.aiohttp_client import async_create_clientsession
+from homeassistant.helpers.selector import (
+    TextSelector,
+    TextSelectorConfig,
+    TextSelectorType,
+)
+
+from .const import DOMAIN, LOGGER
+
+
+class PeblarFlowHandler(ConfigFlow, domain=DOMAIN):
+    """Handle a Peblar config flow."""
+
+    VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle a flow initiated by the user."""
+        errors = {}
+
+        if user_input is not None:
+            peblar = Peblar(
+                host=user_input[CONF_HOST],
+                session=async_create_clientsession(
+                    self.hass, cookie_jar=CookieJar(unsafe=True)
+                ),
+            )
+            try:
+                await peblar.login(password=user_input[CONF_PASSWORD])
+                info = await peblar.system_information()
+            except PeblarAuthenticationError:
+                errors[CONF_PASSWORD] = "invalid_auth"
+            except PeblarConnectionError:
+                errors[CONF_HOST] = "cannot_connect"
+            except Exception:  # noqa: BLE001
+                LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+            else:
+                await self.async_set_unique_id(info.product_serial_number)
+                self._abort_if_unique_id_configured()
+                return self.async_create_entry(title="Peblar", data=user_input)
+        else:
+            user_input = {}
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(
+                        CONF_HOST, default=user_input.get(CONF_HOST)
+                    ): TextSelector(TextSelectorConfig(autocomplete="off")),
+                    vol.Required(CONF_PASSWORD): TextSelector(
+                        TextSelectorConfig(type=TextSelectorType.PASSWORD)
+                    ),
+                }
+            ),
+            errors=errors,
+        )

homeassistant/components/peblar/const.py

@@ -0,0 +1,10 @@
+"""Constants for the Peblar integration."""
+
+from __future__ import annotations
+
+import logging
+from typing import Final
+
+DOMAIN: Final = "peblar"
+
+LOGGER = logging.getLogger(__package__)

homeassistant/components/peblar/coordinator.py

@@ -0,0 +1,37 @@
+"""Data update coordinator for Peblar EV chargers."""
+
+from datetime import timedelta
+
+from peblar import PeblarApi, PeblarError, PeblarMeter
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import LOGGER
+
+type PeblarConfigEntry = ConfigEntry[PeblarMeterDataUpdateCoordinator]
+
+
+class PeblarMeterDataUpdateCoordinator(DataUpdateCoordinator[PeblarMeter]):
+    """Class to manage fetching Peblar meter data."""
+
+    def __init__(
+        self, hass: HomeAssistant, entry: PeblarConfigEntry, api: PeblarApi
+    ) -> None:
+        """Initialize the coordinator."""
+        self.api = api
+        super().__init__(
+            hass,
+            LOGGER,
+            config_entry=entry,
+            name=f"Peblar {entry.title} meter",
+            update_interval=timedelta(seconds=10),
+        )
+
+    async def _async_update_data(self) -> PeblarMeter:
+        """Fetch data from the Peblar device."""
+        try:
+            return await self.api.meter()
+        except PeblarError as err:
+            raise UpdateFailed(err) from err

homeassistant/components/peblar/entity.py

@@ -0,0 +1,26 @@
+"""Base entity for the Peblar integration."""
+
+from __future__ import annotations
+
+from homeassistant.const import CONF_HOST
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DOMAIN
+from .coordinator import PeblarConfigEntry, PeblarMeterDataUpdateCoordinator
+
+
+class PeblarEntity(CoordinatorEntity[PeblarMeterDataUpdateCoordinator]):
+    """Defines a Peblar entity."""
+
+    _attr_has_entity_name = True
+
+    def __init__(self, entry: PeblarConfigEntry) -> None:
+        """Initialize the Peblar entity."""
+        super().__init__(coordinator=entry.runtime_data)
+        self._attr_device_info = DeviceInfo(
+            configuration_url=f"http://{entry.data[CONF_HOST]}",
+            identifiers={(DOMAIN, str(entry.unique_id))},
+            manufacturer="Peblar",
+            name="Peblar EV charger",
+        )

homeassistant/components/peblar/manifest.json

@@ -0,0 +1,11 @@
+{
+  "domain": "peblar",
+  "name": "Peblar",
+  "codeowners": ["@frenck"],
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/peblar",
+  "integration_type": "device",
+  "iot_class": "local_polling",
+  "quality_scale": "bronze",
+  "requirements": ["peblar==0.2.1"]
+}

homeassistant/components/peblar/quality_scale.yaml

@@ -0,0 +1,79 @@
+rules:
+  # Bronze
+  action-setup:
+    status: exempt
+    comment: Integration does not register custom actions.
+  appropriate-polling: done
+  brands: done
+  common-modules: done
+  config-flow-test-coverage: done
+  config-flow: done
+  dependency-transparency: done
+  docs-actions:
+    status: exempt
+    comment: |
+      This integration does not have any custom actions.
+  docs-high-level-description: done
+  docs-installation-instructions: done
+  docs-removal-instructions: done
+  entity-event-setup:
+    status: exempt
+    comment: |
+      Entities of this integration does not explicitly subscribe to events.
+  entity-unique-id: done
+  has-entity-name: done
+  runtime-data: done
+  test-before-configure: done
+  test-before-setup: done
+  unique-config-entry: done
+
+  # Silver
+  action-exceptions: todo
+  config-entry-unloading: done
+  docs-configuration-parameters: todo
+  docs-installation-parameters: todo
+  entity-unavailable: done
+  integration-owner: done
+  log-when-unavailable: done
+  parallel-updates: todo
+  reauthentication-flow: todo
+  test-coverage: todo
+  # Gold
+  devices: todo
+  diagnostics: todo
+  discovery-update-info: todo
+  discovery: todo
+  docs-data-update: todo
+  docs-examples: todo
+  docs-known-limitations: todo
+  docs-supported-devices: todo
+  docs-supported-functions: todo
+  docs-troubleshooting: todo
+  docs-use-cases: todo
+  dynamic-devices:
+    status: exempt
+    comment: |
+      This integration connects to a single device.
+  entity-category: todo
+  entity-device-class: todo
+  entity-disabled-by-default: todo
+  entity-translations: todo
+  exception-translations:
+    status: exempt
+    comment: |
+      The coordinator needs translation when the update failed.
+  icon-translations: todo
+  reconfiguration-flow: todo
+  repair-issues:
+    status: exempt
+    comment: |
+      This integration does not raise any repairable issues.
+  stale-devices:
+    status: exempt
+    comment: |
+      This integration connects to a single device.
+
+  # Platinum
+  async-dependency: done
+  inject-websession: done
+  strict-typing: done

homeassistant/components/peblar/sensor.py

@@ -0,0 +1,73 @@
+"""Support for Peblar sensors."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+
+from peblar import PeblarMeter
+
+from homeassistant.components.sensor import (
+    SensorDeviceClass,
+    SensorEntity,
+    SensorEntityDescription,
+    SensorStateClass,
+)
+from homeassistant.const import UnitOfEnergy
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from .coordinator import PeblarConfigEntry
+from .entity import PeblarEntity
+
+
+@dataclass(frozen=True, kw_only=True)
+class PeblarSensorDescription(SensorEntityDescription):
+    """Describe an Peblar sensor."""
+
+    value_fn: Callable[[PeblarMeter], int | None]
+
+
+SENSORS: tuple[PeblarSensorDescription, ...] = (
+    PeblarSensorDescription(
+        key="energy_total",
+        device_class=SensorDeviceClass.ENERGY,
+        native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
+        state_class=SensorStateClass.TOTAL_INCREASING,
+        suggested_display_precision=2,
+        suggested_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
+        value_fn=lambda x: x.energy_total,
+    ),
+)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: PeblarConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up Peblar sensors based on a config entry."""
+    async_add_entities(
+        PeblarSensorEntity(entry, description) for description in SENSORS
+    )
+
+
+class PeblarSensorEntity(PeblarEntity, SensorEntity):
+    """Defines a Peblar sensor."""
+
+    entity_description: PeblarSensorDescription
+
+    def __init__(
+        self,
+        entry: PeblarConfigEntry,
+        description: PeblarSensorDescription,
+    ) -> None:
+        """Initialize the Peblar entity."""
+        super().__init__(entry)
+        self.entity_description = description
+        self._attr_unique_id = f"{entry.unique_id}_{description.key}"
+
+    @property
+    def native_value(self) -> int | None:
+        """Return the state of the sensor."""
+        return self.entity_description.value_fn(self.coordinator.data)

homeassistant/components/peblar/strings.json

@@ -0,0 +1,25 @@
+{
+  "config": {
+    "step": {
+      "user": {
+        "description": "Set up your Peblar EV charger to integrate with Home Assistant.\n\nTo do so, you will need to get the IP address of your Peblar charger and the password you use to log into the Peblar device' web interface.\n\nHome Assistant will automatically configure your Peblar charger for use with Home Assistant.",
+        "data": {
+          "host": "[%key:common::config_flow::data::host%]",
+          "password": "[%key:common::config_flow::data::password%]"
+        },
+        "data_description": {
+          "host": "The hostname or IP address of your Peblar charger on your home network.",
+          "password": "The same password as you use to log in to the Peblar device' local web interface."
+        }
+      }
+    },
+    "error": {
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
+      "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
+      "unknown": "[%key:common::config_flow::error::unknown%]"
+    },
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
+    }
+  }
+}

homeassistant/generated/config_flows.py

@@ -452,6 +452,7 @@ FLOWS = {
         "p1_monitor",
         "palazzetti",
         "panasonic_viera",
+        "peblar",
         "peco",
         "pegel_online",
         "permobil",

homeassistant/generated/integrations.json

@@ -4618,6 +4618,12 @@
       "integration_type": "virtual",
       "supported_by": "upb"
     },
+    "peblar": {
+      "name": "Peblar",
+      "integration_type": "device",
+      "config_flow": true,
+      "iot_class": "local_polling"
+    },
     "peco": {
       "name": "PECO Outage Counter",
       "integration_type": "hub",

mypy.ini

@@ -3386,6 +3386,16 @@ disallow_untyped_defs = true
 warn_return_any = true
 warn_unreachable = true
 
+[mypy-homeassistant.components.peblar.*]
+check_untyped_defs = true
+disallow_incomplete_defs = true
+disallow_subclassing_any = true
+disallow_untyped_calls = true
+disallow_untyped_decorators = true
+disallow_untyped_defs = true
+warn_return_any = true
+warn_unreachable = true
+
 [mypy-homeassistant.components.peco.*]
 check_untyped_defs = true
 disallow_incomplete_defs = true

requirements_all.txt

@@ -1599,6 +1599,9 @@ panasonic-viera==0.4.2
 # homeassistant.components.dunehd
 pdunehd==1.3.2
 
+# homeassistant.components.peblar
+peblar==0.2.1
+
 # homeassistant.components.peco
 peco==0.0.30
 

requirements_test_all.txt

@@ -1326,6 +1326,9 @@ panasonic-viera==0.4.2
 # homeassistant.components.dunehd
 pdunehd==1.3.2
 
+# homeassistant.components.peblar
+peblar==0.2.1
+
 # homeassistant.components.peco
 peco==0.0.30
 

tests/components/peblar/__init__.py

@@ -0,0 +1 @@
+"""Integration tests for the Peblar integration."""

tests/components/peblar/conftest.py

@@ -0,0 +1,48 @@
+"""Fixtures for the Peblar integration tests."""
+
+from __future__ import annotations
+
+from collections.abc import Generator
+from unittest.mock import MagicMock, patch
+
+from peblar.models import PeblarSystemInformation
+import pytest
+
+from homeassistant.components.peblar.const import DOMAIN
+from homeassistant.const import CONF_HOST, CONF_PASSWORD
+
+from tests.common import MockConfigEntry, load_fixture
+
+
+@pytest.fixture
+def mock_config_entry() -> MockConfigEntry:
+    """Return the default mocked config entry."""
+    return MockConfigEntry(
+        title="Peblar",
+        domain=DOMAIN,
+        data={
+            CONF_HOST: "127.0.0.127",
+            CONF_PASSWORD: "OMGSPIDERS",
+        },
+        unique_id="23-45-A4O-MOF",
+    )
+
+
+@pytest.fixture
+def mock_setup_entry() -> Generator[None]:
+    """Mock setting up a config entry."""
+    with patch("homeassistant.components.peblar.async_setup_entry", return_value=True):
+        yield
+
+
+@pytest.fixture
+def mock_peblar() -> Generator[MagicMock]:
+    """Return a mocked Peblar client."""
+    with patch(
+        "homeassistant.components.peblar.config_flow.Peblar", autospec=True
+    ) as peblar_mock:
+        peblar = peblar_mock.return_value
+        peblar.system_information.return_value = PeblarSystemInformation.from_json(
+            load_fixture("system_information.json", DOMAIN)
+        )
+        yield peblar

tests/components/peblar/fixtures/system_information.json

@@ -0,0 +1,57 @@
+{
+  "BopCalIGainA": 264625,
+  "BopCalIGainB": 267139,
+  "BopCalIGainC": 239155,
+  "CanChangeChargingPhases": false,
+  "CanChargeSinglePhase": true,
+  "CanChargeThreePhases": false,
+  "CustomerId": "PBLR-0000645",
+  "CustomerUpdatePackagePubKey": "-----BEGIN PUBLIC KEY-----\nlorem ipsum\n-----END PUBLIC KEY-----\n",
+  "EthMacAddr": "00:0F:11:58:86:97",
+  "FwIdent": "1.6.1+1+WL-1",
+  "Hostname": "PBLR-0000645",
+  "HwFixedCableRating": 20,
+  "HwFwCompat": "wlac-2",
+  "HwHas4pRelay": false,
+  "HwHasBop": true,
+  "HwHasBuzzer": true,
+  "HwHasDualSocket": false,
+  "HwHasEichrechtLaserMarking": false,
+  "HwHasEthernet": true,
+  "HwHasLed": true,
+  "HwHasLte": false,
+  "HwHasMeter": true,
+  "HwHasMeterDisplay": true,
+  "HwHasPlc": false,
+  "HwHasRfid": true,
+  "HwHasRs485": true,
+  "HwHasShutter": false,
+  "HwHasSocket": false,
+  "HwHasTpm": false,
+  "HwHasWlan": true,
+  "HwMaxCurrent": 16,
+  "HwOneOrThreePhase": 3,
+  "HwUKCompliant": false,
+  "MainboardPn": "6004-2300-7600",
+  "MainboardSn": "23-38-A4E-2MC",
+  "MeterCalIGainA": 267369,
+  "MeterCalIGainB": 228286,
+  "MeterCalIGainC": 246455,
+  "MeterCalIRmsOffsetA": 15573,
+  "MeterCalIRmsOffsetB": 268422963,
+  "MeterCalIRmsOffsetC": 9082,
+  "MeterCalPhaseA": 250,
+  "MeterCalPhaseB": 271,
+  "MeterCalPhaseC": 271,
+  "MeterCalVGainA": 250551,
+  "MeterCalVGainB": 246074,
+  "MeterCalVGainC": 230191,
+  "MeterFwIdent": "b9cbcd",
+  "NorFlash": true,
+  "ProductModelName": "WLAC1-H11R0WE0ICR00",
+  "ProductPn": "6004-2300-8002",
+  "ProductSn": "23-45-A4O-MOF",
+  "ProductVendorName": "Peblar",
+  "WlanApMacAddr": "00:0F:11:58:86:98",
+  "WlanStaMacAddr": "00:0F:11:58:86:99"
+}

tests/components/peblar/test_config_flow.py

@@ -0,0 +1,115 @@
+"""Configuration flow tests for the Peblar integration."""
+
+from unittest.mock import MagicMock
+
+from peblar import PeblarAuthenticationError, PeblarConnectionError
+import pytest
+
+from homeassistant.components.peblar.const import DOMAIN
+from homeassistant.config_entries import SOURCE_USER
+from homeassistant.const import CONF_HOST, CONF_PASSWORD
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResultType
+
+from tests.common import MockConfigEntry
+
+pytestmark = pytest.mark.usefixtures("mock_setup_entry")
+
+
+@pytest.mark.usefixtures("mock_peblar")
+async def test_user_flow(hass: HomeAssistant) -> None:
+    """Test the full happy path user flow from start to finish."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": SOURCE_USER},
+    )
+
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "user"
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        user_input={
+            CONF_HOST: "127.0.0.1",
+            CONF_PASSWORD: "OMGPUPPIES",
+        },
+    )
+
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+
+    config_entry = result["result"]
+    assert config_entry.unique_id == "23-45-A4O-MOF"
+    assert config_entry.data == {
+        CONF_HOST: "127.0.0.1",
+        CONF_PASSWORD: "OMGPUPPIES",
+    }
+    assert not config_entry.options
+
+
+@pytest.mark.parametrize(
+    ("side_effect", "expected_error"),
+    [
+        (PeblarConnectionError, {CONF_HOST: "cannot_connect"}),
+        (PeblarAuthenticationError, {CONF_PASSWORD: "invalid_auth"}),
+        (Exception, {"base": "unknown"}),
+    ],
+)
+async def test_user_flow_errors(
+    hass: HomeAssistant,
+    mock_peblar: MagicMock,
+    side_effect: Exception,
+    expected_error: dict[str, str],
+) -> None:
+    """Test we show user form on a connection error."""
+    mock_peblar.login.side_effect = side_effect
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": SOURCE_USER},
+        data={
+            CONF_HOST: "127.0.0.1",
+            CONF_PASSWORD: "OMGCATS!",
+        },
+    )
+
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "user"
+    assert result["errors"] == expected_error
+
+    mock_peblar.login.side_effect = None
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        user_input={
+            CONF_HOST: "127.0.0.2",
+            CONF_PASSWORD: "OMGPUPPIES!",
+        },
+    )
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+
+    config_entry = result["result"]
+    assert config_entry.unique_id == "23-45-A4O-MOF"
+    assert config_entry.data == {
+        CONF_HOST: "127.0.0.2",
+        CONF_PASSWORD: "OMGPUPPIES!",
+    }
+    assert not config_entry.options
+
+
+@pytest.mark.usefixtures("mock_peblar")
+async def test_user_flow_already_configured(
+    hass: HomeAssistant, mock_config_entry: MockConfigEntry
+) -> None:
+    """Test configuration flow aborts when the device is already configured."""
+    mock_config_entry.add_to_hass(hass)
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": SOURCE_USER},
+        data={
+            CONF_HOST: "127.0.0.1",
+            CONF_PASSWORD: "OMGSPIDERS",
+        },
+    )
+
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "already_configured"