Add WattTime integration (home-assistant#56149)

.coveragerc

@@ -1178,6 +1178,8 @@ omit =
     homeassistant/components/waterfurnace/*
     homeassistant/components/watson_iot/*
     homeassistant/components/watson_tts/tts.py
+    homeassistant/components/watttime/__init__.py
+    homeassistant/components/watttime/sensor.py
     homeassistant/components/waze_travel_time/__init__.py
     homeassistant/components/waze_travel_time/helpers.py
     homeassistant/components/waze_travel_time/sensor.py

CODEOWNERS

@@ -574,6 +574,7 @@ homeassistant/components/wake_on_lan/* @ntilley905
 homeassistant/components/wallbox/* @hesselonline
 homeassistant/components/waqi/* @andrey-git
 homeassistant/components/watson_tts/* @rutkai
+homeassistant/components/watttime/* @bachya
 homeassistant/components/weather/* @fabaff
 homeassistant/components/webostv/* @bendavid @thecode
 homeassistant/components/websocket_api/* @home-assistant/core

homeassistant/components/watttime/__init__.py

@@ -0,0 +1,78 @@
+"""The WattTime integration."""
+from __future__ import annotations
+
+from datetime import timedelta
+
+from aiowatttime import Client
+from aiowatttime.emissions import RealTimeEmissionsResponseType
+from aiowatttime.errors import WattTimeError
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import (
+    CONF_LATITUDE,
+    CONF_LONGITUDE,
+    CONF_PASSWORD,
+    CONF_USERNAME,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import aiohttp_client
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import DATA_COORDINATOR, DOMAIN, LOGGER
+
+DEFAULT_UPDATE_INTERVAL = timedelta(minutes=5)
+
+PLATFORMS: list[str] = ["sensor"]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up WattTime from a config entry."""
+    hass.data.setdefault(DOMAIN, {DATA_COORDINATOR: {}})
+
+    session = aiohttp_client.async_get_clientsession(hass)
+
+    try:
+        client = await Client.async_login(
+            entry.data[CONF_USERNAME],
+            entry.data[CONF_PASSWORD],
+            session=session,
+            logger=LOGGER,
+        )
+    except WattTimeError as err:
+        LOGGER.error("Error while authenticating with WattTime: %s", err)
+        return False
+
+    async def async_update_data() -> RealTimeEmissionsResponseType:
+        """Get the latest realtime emissions data."""
+        try:
+            return await client.emissions.async_get_realtime_emissions(
+                entry.data[CONF_LATITUDE], entry.data[CONF_LONGITUDE]
+            )
+        except WattTimeError as err:
+            raise UpdateFailed(
+                f"Error while requesting data from WattTime: {err}"
+            ) from err
+
+    coordinator = DataUpdateCoordinator(
+        hass,
+        LOGGER,
+        name=entry.title,
+        update_interval=DEFAULT_UPDATE_INTERVAL,
+        update_method=async_update_data,
+    )
+
+    await coordinator.async_config_entry_first_refresh()
+    hass.data[DOMAIN][DATA_COORDINATOR][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][DATA_COORDINATOR].pop(entry.entry_id)
+
+    return unload_ok

homeassistant/components/watttime/config_flow.py

@@ -0,0 +1,165 @@
+"""Config flow for WattTime integration."""
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+from aiowatttime import Client
+from aiowatttime.errors import CoordinatesNotFoundError, InvalidCredentialsError
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.const import (
+    CONF_LATITUDE,
+    CONF_LONGITUDE,
+    CONF_PASSWORD,
+    CONF_USERNAME,
+)
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.helpers import aiohttp_client, config_validation as cv
+
+from .const import (
+    CONF_BALANCING_AUTHORITY,
+    CONF_BALANCING_AUTHORITY_ABBREV,
+    DOMAIN,
+    LOGGER,
+)
+
+CONF_LOCATION_TYPE = "location_type"
+
+LOCATION_TYPE_COORDINATES = "Specify coordinates"
+LOCATION_TYPE_HOME = "Use home location"
+
+STEP_COORDINATES_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_LATITUDE): cv.latitude,
+        vol.Required(CONF_LONGITUDE): cv.longitude,
+    }
+)
+
+STEP_LOCATION_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_LOCATION_TYPE): vol.In(
+            [LOCATION_TYPE_HOME, LOCATION_TYPE_COORDINATES]
+        ),
+    }
+)
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_USERNAME): str,
+        vol.Required(CONF_PASSWORD): str,
+    }
+)
+
+
+class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for WattTime."""
+
+    VERSION = 1
+
+    def __init__(self) -> None:
+        """Initialize."""
+        self._client: Client | None = None
+        self._password: str | None = None
+        self._username: str | None = None
+
+    async def async_step_coordinates(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the coordinates step."""
+        if not user_input:
+            return self.async_show_form(
+                step_id="coordinates", data_schema=STEP_COORDINATES_DATA_SCHEMA
+            )
+
+        if TYPE_CHECKING:
+            assert self._client
+
+        unique_id = f"{user_input[CONF_LATITUDE]}, {user_input[CONF_LONGITUDE]}"
+        await self.async_set_unique_id(unique_id)
+        self._abort_if_unique_id_configured()
+
+        try:
+            grid_region = await self._client.emissions.async_get_grid_region(
+                user_input[CONF_LATITUDE], user_input[CONF_LONGITUDE]
+            )
+        except CoordinatesNotFoundError:
+            return self.async_show_form(
+                step_id="coordinates",
+                data_schema=STEP_COORDINATES_DATA_SCHEMA,
+                errors={CONF_LATITUDE: "unknown_coordinates"},
+            )
+        except Exception as err:  # pylint: disable=broad-except
+            LOGGER.exception("Unexpected exception while getting region: %s", err)
+            return self.async_show_form(
+                step_id="coordinates",
+                data_schema=STEP_COORDINATES_DATA_SCHEMA,
+                errors={"base": "unknown"},
+            )
+
+        return self.async_create_entry(
+            title=unique_id,
+            data={
+                CONF_USERNAME: self._username,
+                CONF_PASSWORD: self._password,
+                CONF_LATITUDE: user_input[CONF_LATITUDE],
+                CONF_LONGITUDE: user_input[CONF_LONGITUDE],
+                CONF_BALANCING_AUTHORITY: grid_region["name"],
+                CONF_BALANCING_AUTHORITY_ABBREV: grid_region["abbrev"],
+            },
+        )
+
+    async def async_step_location(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the "pick a location" step."""
+        if not user_input:
+            return self.async_show_form(
+                step_id="location", data_schema=STEP_LOCATION_DATA_SCHEMA
+            )
+
+        if user_input[CONF_LOCATION_TYPE] == LOCATION_TYPE_COORDINATES:
+            return self.async_show_form(
+                step_id="coordinates", data_schema=STEP_COORDINATES_DATA_SCHEMA
+            )
+        return await self.async_step_coordinates(
+            {
+                CONF_LATITUDE: self.hass.config.latitude,
+                CONF_LONGITUDE: self.hass.config.longitude,
+            }
+        )
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the initial step."""
+        if not user_input:
+            return self.async_show_form(
+                step_id="user", data_schema=STEP_USER_DATA_SCHEMA
+            )
+
+        session = aiohttp_client.async_get_clientsession(self.hass)
+
+        try:
+            self._client = await Client.async_login(
+                user_input[CONF_USERNAME],
+                user_input[CONF_PASSWORD],
+                session=session,
+            )
+        except InvalidCredentialsError:
+            return self.async_show_form(
+                step_id="user",
+                data_schema=STEP_USER_DATA_SCHEMA,
+                errors={CONF_USERNAME: "invalid_auth"},
+            )
+        except Exception as err:  # pylint: disable=broad-except
+            LOGGER.exception("Unexpected exception while logging in: %s", err)
+            return self.async_show_form(
+                step_id="user",
+                data_schema=STEP_USER_DATA_SCHEMA,
+                errors={"base": "unknown"},
+            )
+
+        self._username = user_input[CONF_USERNAME]
+        self._password = user_input[CONF_PASSWORD]
+        return await self.async_step_location()

homeassistant/components/watttime/const.py

@@ -0,0 +1,11 @@
+"""Constants for the WattTime integration."""
+import logging
+
+DOMAIN = "watttime"
+
+LOGGER = logging.getLogger(__package__)
+
+CONF_BALANCING_AUTHORITY = "balancing_authority"
+CONF_BALANCING_AUTHORITY_ABBREV = "balancing_authority_abbreviation"
+
+DATA_COORDINATOR = "coordinator"

homeassistant/components/watttime/manifest.json

@@ -0,0 +1,17 @@
+{
+  "domain": "watttime",
+  "name": "WattTime",
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/watttime",
+  "requirements": [
+    "aiowatttime==0.1.1"
+  ],
+  "ssdp": [],
+  "zeroconf": [],
+  "homekit": {},
+  "dependencies": [],
+  "codeowners": [
+    "@bachya"
+  ],
+  "iot_class": "cloud_polling"
+}