Add Mypermobil integration

.coveragerc

@@ -931,6 +931,9 @@ omit =
     homeassistant/components/panasonic_viera/media_player.py
     homeassistant/components/pandora/media_player.py
     homeassistant/components/pencom/switch.py
+    homeassistant/components/permobil/__init__.py
+    homeassistant/components/permobil/coordinator.py
+    homeassistant/components/permobil/sensor.py
     homeassistant/components/philips_js/__init__.py
     homeassistant/components/philips_js/light.py
     homeassistant/components/philips_js/media_player.py

CODEOWNERS

@@ -944,6 +944,8 @@ build.json @home-assistant/supervisor
 /tests/components/peco/ @IceBotYT
 /homeassistant/components/pegel_online/ @mib1185
 /tests/components/pegel_online/ @mib1185
+/homeassistant/components/permobil/ @IsakNyberg
+/tests/components/permobil/ @IsakNyberg
 /homeassistant/components/persistent_notification/ @home-assistant/core
 /tests/components/persistent_notification/ @home-assistant/core
 /homeassistant/components/philips_js/ @elupus

homeassistant/components/permobil/__init__.py

@@ -0,0 +1,63 @@
+"""The MyPermobil integration."""
+from __future__ import annotations
+
+import logging
+
+from mypermobil import MyPermobil, MyPermobilClientException
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import (
+    CONF_CODE,
+    CONF_EMAIL,
+    CONF_REGION,
+    CONF_TOKEN,
+    CONF_TTL,
+    Platform,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryAuthFailed
+
+from .const import APPLICATION, DOMAIN
+from .coordinator import MyPermobilCoordinator
+
+PLATFORMS: list[Platform] = [Platform.SENSOR]
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up MyPermobil from a config entry."""
+
+    # create the API object from the config and save it in hass
+    session = hass.helpers.aiohttp_client.async_get_clientsession()
+    p_api = MyPermobil(
+        application=APPLICATION,
+        session=session,
+        email=entry.data[CONF_EMAIL],
+        region=entry.data[CONF_REGION],
+        code=entry.data[CONF_CODE],
+        token=entry.data[CONF_TOKEN],
+        expiration_date=entry.data[CONF_TTL],
+    )
+    try:
+        p_api.self_authenticate()
+    except MyPermobilClientException as err:
+        _LOGGER.error("Error authenticating  %s", err)
+        raise ConfigEntryAuthFailed(f"Config error for {p_api.email}") from err
+
+    # create the coordinator with the API object
+    coordinator = MyPermobilCoordinator(hass, p_api)
+    await coordinator.async_config_entry_first_refresh()
+
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok

homeassistant/components/permobil/config_flow.py

@@ -0,0 +1,173 @@
+"""Config flow for MyPermobil integration."""
+from __future__ import annotations
+
+from collections.abc import Mapping
+import logging
+from typing import Any
+
+from mypermobil import MyPermobil, MyPermobilAPIException, MyPermobilClientException
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.const import CONF_CODE, CONF_EMAIL, CONF_REGION, CONF_TOKEN, CONF_TTL
+from homeassistant.core import HomeAssistant, async_get_hass
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.helpers import selector
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+import homeassistant.helpers.config_validation as cv
+from homeassistant.helpers.selector import (
+    TextSelector,
+    TextSelectorConfig,
+    TextSelectorType,
+)
+
+from .const import APPLICATION, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+GET_EMAIL_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_EMAIL): TextSelector(
+            TextSelectorConfig(type=TextSelectorType.EMAIL)
+        ),
+    }
+)
+
+GET_TOKEN_SCHEMA = vol.Schema({vol.Required(CONF_CODE): cv.string})
+
+
+class PermobilConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Permobil config flow."""
+
+    VERSION = 1
+    region_names: dict[str, str] = {}
+    data: dict[str, str] = {}
+
+    def __init__(self) -> None:
+        """Initialize flow."""
+        hass: HomeAssistant = async_get_hass()
+        session = async_get_clientsession(hass)
+        self.p_api = MyPermobil(APPLICATION, session=session)
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Invoke when a user initiates a flow via the user interface."""
+        errors: dict[str, str] = {}
+
+        if user_input:
+            try:
+                self.p_api.set_email(user_input[CONF_EMAIL])
+            except MyPermobilClientException:
+                _LOGGER.exception("Error validating email")
+                errors["base"] = "invalid_email"
+
+            self.data.update(user_input)
+
+            await self.async_set_unique_id(self.data[CONF_EMAIL])
+            self._abort_if_unique_id_configured()
+
+        if errors or not user_input:
+            return self.async_show_form(
+                step_id="user", data_schema=GET_EMAIL_SCHEMA, errors=errors
+            )
+        return await self.async_step_region()
+
+    async def async_step_region(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Invoke when a user initiates a flow via the user interface."""
+        errors: dict[str, str] = {}
+        if not user_input:
+            # fetch the list of regions names and urls from the api
+            # for the user to select from.
+            try:
+                self.region_names = await self.p_api.request_region_names()
+                _LOGGER.debug(
+                    "region names %s",
+                    ",".join(list(self.region_names.keys())),
+                )
+            except MyPermobilAPIException:
+                _LOGGER.exception("Error requesting regions")
+                errors["base"] = "region_fetch_error"
+
+        else:
+            region_url = self.region_names[user_input[CONF_REGION]]
+
+            self.data[CONF_REGION] = region_url
+            self.p_api.set_region(region_url)
+            _LOGGER.debug("region %s", self.p_api.region)
+            try:
+                # tell backend to send code to the users email
+                await self.p_api.request_application_code()
+            except MyPermobilAPIException:
+                _LOGGER.exception("Error requesting code")
+                errors["base"] = "code_request_error"
+
+        if errors or not user_input:
+            # the error could either be that the fetch region did not pass
+            # or that the request application code failed
+            schema = vol.Schema(
+                {
+                    vol.Required(CONF_REGION): selector.SelectSelector(
+                        selector.SelectSelectorConfig(
+                            options=list(self.region_names.keys()),
+                            mode=selector.SelectSelectorMode.DROPDOWN,
+                        )
+                    ),
+                }
+            )
+            return self.async_show_form(
+                step_id="region", data_schema=schema, errors=errors
+            )
+
+        return await self.async_step_email_code()
+
+    async def async_step_email_code(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Second step in config flow to enter the email code."""
+        errors: dict[str, str] = {}
+
+        if user_input:
+            try:
+                self.p_api.set_code(user_input[CONF_CODE])
+                self.data.update(user_input)
+                token, ttl = await self.p_api.request_application_token()
+                self.data[CONF_TOKEN] = token
+                self.data[CONF_TTL] = ttl
+            except (MyPermobilAPIException, MyPermobilClientException):
+                # the code did not pass validation by the api client
+                # or the backend returned an error when trying to validate the code
+                _LOGGER.exception("Error verifying code")
+                errors["base"] = "invalid_code"
+
+        if errors or not user_input:
+            return self.async_show_form(
+                step_id="email_code", data_schema=GET_TOKEN_SCHEMA, errors=errors
+            )
+
+        return self.async_create_entry(title=self.data[CONF_EMAIL], data=self.data)
+
+    async def async_step_reauth(self, user_input: Mapping[str, Any]) -> FlowResult:
+        """Perform reauth upon an API authentication error."""
+        reauth_entry = self.hass.config_entries.async_get_entry(
+            self.context["entry_id"]
+        )
+        assert reauth_entry
+
+        try:
+            email: str = reauth_entry.data[CONF_EMAIL]
+            region: str = reauth_entry.data[CONF_REGION]
+            self.p_api.set_email(email)
+            self.p_api.set_region(region)
+            self.data = {
+                CONF_EMAIL: email,
+                CONF_REGION: region,
+            }
+            await self.p_api.request_application_code()
+        except MyPermobilAPIException:
+            _LOGGER.exception("Error requesting code for reauth")
+            return self.async_abort(reason="unknown")
+
+        return await self.async_step_email_code()

homeassistant/components/permobil/const.py

@@ -0,0 +1,11 @@
+"""Constants for the MyPermobil integration."""
+
+DOMAIN = "permobil"
+
+APPLICATION = "Home Assistant"
+
+
+BATTERY_ASSUMED_VOLTAGE = 25.0  # This is the average voltage over all states of charge
+REGIONS = "regions"
+KM = "kilometers"
+MILES = "miles"

homeassistant/components/permobil/coordinator.py

@@ -0,0 +1,57 @@
+"""DataUpdateCoordinator for permobil integration."""
+
+import asyncio
+from dataclasses import dataclass
+from datetime import timedelta
+import logging
+
+from mypermobil import MyPermobil, MyPermobilAPIException
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+_LOGGER = logging.getLogger(__name__)
+
+
+@dataclass
+class MyPermobilData:
+    """MyPermobil data stored in the DataUpdateCoordinator."""
+
+    battery: dict[str, str | float | int | list | dict]
+    daily_usage: dict[str, str | float | int | list | dict]
+    records: dict[str, str | float | int | list | dict]
+
+
+class MyPermobilCoordinator(DataUpdateCoordinator[MyPermobilData]):
+    """MyPermobil coordinator."""
+
+    def __init__(self, hass: HomeAssistant, p_api: MyPermobil) -> None:
+        """Initialize my coordinator."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name="permobil",
+            update_interval=timedelta(minutes=5),
+        )
+        self.p_api = p_api
+
+    async def _async_update_data(self) -> MyPermobilData:
+        """Fetch data from the 3 API endpoints."""
+        try:
+            async with asyncio.timeout(10):
+                battery = await self.p_api.get_battery_info()
+                daily_usage = await self.p_api.get_daily_usage()
+                records = await self.p_api.get_usage_records()
+                return MyPermobilData(
+                    battery=battery,
+                    daily_usage=daily_usage,
+                    records=records,
+                )
+
+        except MyPermobilAPIException as err:
+            _LOGGER.exception(
+                "Error fetching data from MyPermobil API for account %s %s",
+                self.p_api.email,
+                err,
+            )
+            raise UpdateFailed from err

homeassistant/components/permobil/manifest.json

@@ -0,0 +1,9 @@
+{
+  "domain": "permobil",
+  "name": "MyPermobil",
+  "codeowners": ["@IsakNyberg"],
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/permobil",
+  "iot_class": "cloud_polling",
+  "requirements": ["mypermobil==0.1.6"]
+}