Add Opower integration for getting electricity/gas usage and cost for many utilities

.coveragerc

@@ -859,6 +859,9 @@ omit =
     homeassistant/components/openweathermap/sensor.py
     homeassistant/components/openweathermap/weather_update_coordinator.py
     homeassistant/components/opnsense/__init__.py
+    homeassistant/components/opower/__init__.py
+    homeassistant/components/opower/coordinator.py
+    homeassistant/components/opower/sensor.py
     homeassistant/components/opnsense/device_tracker.py
     homeassistant/components/opple/light.py
     homeassistant/components/oru/*

CODEOWNERS

@@ -897,6 +897,8 @@ build.json @home-assistant/supervisor
 /tests/components/openweathermap/ @fabaff @freekode @nzapponi
 /homeassistant/components/opnsense/ @mtreinish
 /tests/components/opnsense/ @mtreinish
+/homeassistant/components/opower/ @tronikos
+/tests/components/opower/ @tronikos
 /homeassistant/components/oralb/ @bdraco @Lash-L
 /tests/components/oralb/ @bdraco @Lash-L
 /homeassistant/components/oru/ @bvlaicu

homeassistant/components/opower/__init__.py

@@ -0,0 +1,31 @@
+"""The Opower integration."""
+from __future__ import annotations
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+
+from .const import DOMAIN
+from .coordinator import OpowerCoordinator
+
+PLATFORMS: list[Platform] = [Platform.SENSOR]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Opower from a config entry."""
+
+    coordinator = OpowerCoordinator(hass, entry.data)
+    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/opower/config_flow.py

@@ -0,0 +1,112 @@
+"""Config flow for Opower integration."""
+from __future__ import annotations
+
+from collections.abc import Mapping
+import logging
+from typing import Any
+
+from opower import CannotConnect, InvalidAuth, Opower, get_supported_utility_names
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.helpers.aiohttp_client import async_create_clientsession
+
+from .const import CONF_UTILITY, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_UTILITY): vol.In(get_supported_utility_names()),
+        vol.Required(CONF_USERNAME): str,
+        vol.Required(CONF_PASSWORD): str,
+    }
+)
+
+
+async def _validate_login(
+    hass: HomeAssistant, login_data: dict[str, str]
+) -> dict[str, str]:
+    """Validate login data and return any errors."""
+    api = Opower(
+        async_create_clientsession(hass),
+        login_data[CONF_UTILITY],
+        login_data[CONF_USERNAME],
+        login_data[CONF_PASSWORD],
+    )
+    errors: dict[str, str] = {}
+    try:
+        await api.async_login()
+    except InvalidAuth:
+        errors["base"] = "invalid_auth"
+    except CannotConnect:
+        errors["base"] = "cannot_connect"
+    return errors
+
+
+class OpowerConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Opower."""
+
+    VERSION = 1
+
+    def __init__(self) -> None:
+        """Initialize a new OpowerConfigFlow."""
+        self.reauth_entry: config_entries.ConfigEntry | None = None
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the initial step."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            self._async_abort_entries_match(
+                {
+                    CONF_UTILITY: user_input[CONF_UTILITY],
+                    CONF_USERNAME: user_input[CONF_USERNAME],
+                }
+            )
+            errors = await _validate_login(self.hass, user_input)
+            if not errors:
+                return self.async_create_entry(
+                    title=f"{user_input[CONF_UTILITY]} ({user_input[CONF_USERNAME]})",
+                    data=user_input,
+                )
+        return self.async_show_form(
+            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+        )
+
+    async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> FlowResult:
+        """Handle configuration by re-auth."""
+        self.reauth_entry = self.hass.config_entries.async_get_entry(
+            self.context["entry_id"]
+        )
+        return await self.async_step_reauth_confirm()
+
+    async def async_step_reauth_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Dialog that informs the user that reauth is required."""
+        assert self.reauth_entry
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            data = {**self.reauth_entry.data, **user_input}
+            errors = await _validate_login(self.hass, data)
+            if not errors:
+                self.hass.config_entries.async_update_entry(
+                    self.reauth_entry, data=data
+                )
+                await self.hass.config_entries.async_reload(self.reauth_entry.entry_id)
+                return self.async_abort(reason="reauth_successful")
+        return self.async_show_form(
+            step_id="reauth_confirm",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_USERNAME): self.reauth_entry.data[CONF_USERNAME],
+                    vol.Required(CONF_PASSWORD): str,
+                }
+            ),
+            errors=errors,
+        )

homeassistant/components/opower/const.py

@@ -0,0 +1,5 @@
+"""Constants for the Opower integration."""
+
+DOMAIN = "opower"
+
+CONF_UTILITY = "utility"

homeassistant/components/opower/coordinator.py

@@ -0,0 +1,220 @@
+"""Coordinator to handle Opower connections."""
+from datetime import datetime, timedelta
+import logging
+from types import MappingProxyType
+from typing import Any, cast
+
+from opower import (
+    Account,
+    AggregateType,
+    CostRead,
+    Forecast,
+    InvalidAuth,
+    MeterType,
+    Opower,
+)
+
+from homeassistant.components.recorder import get_instance
+from homeassistant.components.recorder.models import StatisticData, StatisticMetaData
+from homeassistant.components.recorder.statistics import (
+    async_add_external_statistics,
+    get_last_statistics,
+    statistics_during_period,
+)
+from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, UnitOfEnergy, UnitOfVolume
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryAuthFailed
+from homeassistant.helpers import aiohttp_client
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .const import CONF_UTILITY, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class OpowerCoordinator(DataUpdateCoordinator):
+    """Handle fetching Opower data, updating sensors and inserting statistics."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        entry_data: MappingProxyType[str, Any],
+    ) -> None:
+        """Initialize the data handler."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name="Opower",
+            # Data is updated daily on Opower.
+            # Refresh every 12h to be at most 12h behind.
+            update_interval=timedelta(hours=12),
+        )
+        self.api = Opower(
+            aiohttp_client.async_get_clientsession(hass),
+            entry_data[CONF_UTILITY],
+            entry_data[CONF_USERNAME],
+            entry_data[CONF_PASSWORD],
+        )
+
+    async def _async_update_data(
+        self,
+    ) -> dict[str, Forecast]:
+        """Fetch data from API endpoint."""
+        try:
+            # Login expires after a few minutes.
+            # Given the infrequent updating (every 12h)
+            # assume previous session has expired and re-login.
+            await self.api.async_login()
+        except InvalidAuth as err:
+            raise ConfigEntryAuthFailed from err
+        forecasts: list[Forecast] = await self.api.async_get_forecast()
+        _LOGGER.debug("Updating sensor data with: %s", forecasts)
+        await self._insert_statistics([forecast.account for forecast in forecasts])
+        return {forecast.account.utility_account_id: forecast for forecast in forecasts}
+
+    async def _insert_statistics(self, accounts: list[Account]) -> None:
+        """Insert Opower statistics."""
+        for account in accounts:
+            id_prefix = "_".join(
+                (
+                    self.api.utility.subdomain(),
+                    account.meter_type.name.lower(),
+                    account.utility_account_id,
+                )
+            )
+            cost_statistic_id = f"{DOMAIN}:{id_prefix}_energy_cost"
+            consumption_statistic_id = f"{DOMAIN}:{id_prefix}_energy_consumption"
+            _LOGGER.debug(
+                "Updating Statistics for %s and %s",
+                cost_statistic_id,
+                consumption_statistic_id,
+            )
+
+            last_stat = await get_instance(self.hass).async_add_executor_job(
+                get_last_statistics, self.hass, 1, consumption_statistic_id, True, set()
+            )
+            if not last_stat:
+                _LOGGER.debug("Updating statistic for the first time")
+                cost_reads = await self._async_get_all_cost_reads(account)
+                cost_sum = 0.0
+                consumption_sum = 0.0
+                last_stats_time = None
+            else:
+                cost_reads = await self._async_get_recent_cost_reads(
+                    account, last_stat[consumption_statistic_id][0]["start"]
+                )
+                if not cost_reads:
+                    _LOGGER.debug("No recent usage/cost data. Skipping update")
+                    continue
+                stats = await get_instance(self.hass).async_add_executor_job(
+                    statistics_during_period,
+                    self.hass,
+                    cost_reads[0].start_time,
+                    None,
+                    {cost_statistic_id, consumption_statistic_id},
+                    "hour" if account.meter_type == MeterType.ELEC else "day",
+                    None,
+                    {"sum"},
+                )
+                cost_sum = cast(float, stats[cost_statistic_id][0]["sum"])
+                consumption_sum = cast(float, stats[consumption_statistic_id][0]["sum"])
+                last_stats_time = stats[cost_statistic_id][0]["start"]
+
+            cost_statistics = []
+            consumption_statistics = []
+
+            for cost_read in cost_reads:
+                start = cost_read.start_time
+                if last_stats_time is not None and start.timestamp() <= last_stats_time:
+                    continue
+                cost_sum += cost_read.provided_cost
+                consumption_sum += cost_read.consumption
+
+                cost_statistics.append(
+                    StatisticData(
+                        start=start, state=cost_read.provided_cost, sum=cost_sum
+                    )
+                )
+                consumption_statistics.append(
+                    StatisticData(
+                        start=start, state=cost_read.consumption, sum=consumption_sum
+                    )
+                )
+
+            name_prefix = " ".join(
+                (
+                    "Opower",
+                    self.api.utility.subdomain(),
+                    account.meter_type.name.lower(),
+                    account.utility_account_id,
+                )
+            )
+            cost_metadata = StatisticMetaData(
+                has_mean=False,
+                has_sum=True,
+                name=f"{name_prefix} cost",
+                source=DOMAIN,
+                statistic_id=cost_statistic_id,
+                unit_of_measurement=None,
+            )
+            consumption_metadata = StatisticMetaData(
+                has_mean=False,
+                has_sum=True,
+                name=f"{name_prefix} consumption",
+                source=DOMAIN,
+                statistic_id=consumption_statistic_id,
+                unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR
+                if account.meter_type == MeterType.ELEC
+                else UnitOfVolume.CENTUM_CUBIC_FEET,
+            )
+
+            async_add_external_statistics(self.hass, cost_metadata, cost_statistics)
+            async_add_external_statistics(
+                self.hass, consumption_metadata, consumption_statistics
+            )
+
+    async def _async_get_all_cost_reads(self, account: Account) -> list[CostRead]:
+        """Get all cost reads since account activation but at different resolutions depending on age.
+
+        - month resolution for all years (since account activation)
+        - day resolution for past 3 years
+        - hour resolution for past 2 months, only for electricity, not gas
+        """
+        cost_reads = []
+        start = None
+        end = datetime.now() - timedelta(days=3 * 365)
+        cost_reads += await self.api.async_get_cost_reads(
+            account, AggregateType.BILL, start, end
+        )
+        start = end if not cost_reads else cost_reads[-1].end_time
+        end = (
+            datetime.now() - timedelta(days=2 * 30)
+            if account.meter_type == MeterType.ELEC
+            else datetime.now()
+        )
+        cost_reads += await self.api.async_get_cost_reads(
+            account, AggregateType.DAY, start, end
+        )
+        if account.meter_type == MeterType.ELEC:
+            start = end if not cost_reads else cost_reads[-1].end_time
+            end = datetime.now()
+            cost_reads += await self.api.async_get_cost_reads(
+                account, AggregateType.HOUR, start, end
+            )
+        return cost_reads
+
+    async def _async_get_recent_cost_reads(
+        self, account: Account, last_stat_time: float
+    ) -> list[CostRead]:
+        """Get cost reads within the past 30 days to allow corrections in data from utilities.
+
+        Hourly for electricity, daily for gas.
+        """
+        return await self.api.async_get_cost_reads(
+            account,
+            AggregateType.HOUR
+            if account.meter_type == MeterType.ELEC
+            else AggregateType.DAY,
+            datetime.fromtimestamp(last_stat_time) - timedelta(days=30),
+            datetime.now(),
+        )