Duke Energy Integration

CODEOWNERS

@@ -359,6 +359,8 @@ build.json @home-assistant/supervisor
 /tests/components/dsmr/ @Robbie1221
 /homeassistant/components/dsmr_reader/ @sorted-bits @glodenox @erwindouna
 /tests/components/dsmr_reader/ @sorted-bits @glodenox @erwindouna
+/homeassistant/components/duke_energy/ @hunterjm
+/tests/components/duke_energy/ @hunterjm
 /homeassistant/components/duotecno/ @cereal2nd
 /tests/components/duotecno/ @cereal2nd
 /homeassistant/components/dwd_weather_warnings/ @runningman84 @stephan192 @andarotajo

homeassistant/components/duke_energy/__init__.py

@@ -0,0 +1,33 @@
+"""The Duke Energy 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 DukeEnergyCoordinator
+
+# This integration provides no platforms for now, just inserts statistics
+PLATFORMS: list[Platform] = []
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Duke Energy from a config entry."""
+
+    coordinator = DukeEnergyCoordinator(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/duke_energy/config_flow.py

@@ -0,0 +1,62 @@
+"""Config flow for Duke Energy integration."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from aiodukeenergy import DukeEnergy
+from aiohttp import ClientError, ClientResponseError
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_EMAIL, CONF_PASSWORD, CONF_USERNAME
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_USERNAME): str,
+        vol.Required(CONF_PASSWORD): str,
+    }
+)
+
+
+class DukeEnergyConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Duke Energy."""
+
+    VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the initial step."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            session = async_get_clientsession(self.hass)
+            api = DukeEnergy(
+                user_input[CONF_USERNAME], user_input[CONF_PASSWORD], session
+            )
+            try:
+                auth = await api.authenticate()
+            except ClientResponseError as e:
+                errors["base"] = (
+                    "invalid_auth" if e.status == 404 else "could_not_connect"
+                )
+            except (ClientError, TimeoutError):
+                errors["base"] = "could_not_connect"
+            except Exception:
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+            else:
+                email = auth["email"].lower()
+                data = {CONF_EMAIL: email, **user_input}
+                self._async_abort_entries_match(data)
+                return self.async_create_entry(title=email, data=data)
+
+        return self.async_show_form(
+            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+        )

homeassistant/components/duke_energy/const.py

@@ -0,0 +1,3 @@
+"""Constants for the Duke Energy integration."""
+
+DOMAIN = "duke_energy"

homeassistant/components/duke_energy/coordinator.py

@@ -0,0 +1,218 @@
+"""Coordinator to handle Duke Energy connections."""
+
+from datetime import datetime, timedelta
+import logging
+from types import MappingProxyType
+from typing import Any, cast
+
+from aiodukeenergy import DukeEnergy
+from aiohttp import ClientError
+
+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, callback
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+from homeassistant.util import dt as dt_util
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+_SUPPORTED_METER_TYPES = ("ELECTRIC",)
+
+
+class DukeEnergyCoordinator(DataUpdateCoordinator[None]):
+    """Handle inserting statistics."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        entry_data: MappingProxyType[str, Any],
+    ) -> None:
+        """Initialize the data handler."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name="Duke Energy",
+            # Data is updated daily on Duke Energy.
+            # Refresh every 12h to be at most 12h behind.
+            update_interval=timedelta(hours=12),
+        )
+        self.api = DukeEnergy(
+            entry_data[CONF_USERNAME],
+            entry_data[CONF_PASSWORD],
+            async_get_clientsession(hass),
+        )
+        self._statistic_ids: set = set()
+
+        @callback
+        def _dummy_listener() -> None:
+            pass
+
+        # Force the coordinator to periodically update by registering at least one listener.
+        # Duke Energy does not provide forecast data, so all information is historical.
+        # This makes _async_update_data get periodically called so we can insert statistics.
+        self.async_add_listener(_dummy_listener)
+
+        if self.config_entry:
+            self.config_entry.async_on_unload(self._clear_statistics)
+
+    def _clear_statistics(self) -> None:
+        """Clear statistics."""
+        get_instance(self.hass).async_clear_statistics(list(self._statistic_ids))
+
+    async def _async_update_data(self) -> None:
+        """Insert Duke Energy statistics."""
+        meters: dict[str, dict[str, Any]] = await self.api.get_meters()
+        for serial_number, meter in meters.items():
+            if (
+                not isinstance(meter["serviceType"], str)
+                or meter["serviceType"] not in _SUPPORTED_METER_TYPES
+            ):
+                _LOGGER.debug(
+                    "Skipping unsupported meter type %s", meter["serviceType"]
+                )
+                continue
+
+            id_prefix = f"{meter["serviceType"].lower()}_{serial_number}"
+            consumption_statistic_id = f"{DOMAIN}:{id_prefix}_energy_consumption"
+            self._statistic_ids.add(consumption_statistic_id)
+            _LOGGER.debug(
+                "Updating Statistics for %s",
+                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")
+                usage = await self._async_get_energy_usage(meter)
+                consumption_sum = 0.0
+                last_stats_time = None
+            else:
+                usage = await self._async_get_energy_usage(
+                    meter,
+                    last_stat[consumption_statistic_id][0]["start"],
+                )
+                if not usage:
+                    _LOGGER.debug("No recent usage data. Skipping update")
+                    continue
+                stats = await get_instance(self.hass).async_add_executor_job(
+                    statistics_during_period,
+                    self.hass,
+                    min(usage.keys()),
+                    None,
+                    {consumption_statistic_id},
+                    "hour",
+                    None,
+                    {"sum"},
+                )
+                consumption_sum = cast(float, stats[consumption_statistic_id][0]["sum"])
+                last_stats_time = stats[consumption_statistic_id][0]["start"]
+
+            consumption_statistics = []
+
+            for start, data in usage.items():
+                if last_stats_time is not None and start.timestamp() <= last_stats_time:
+                    continue
+                consumption_sum += data["energy"]
+
+                consumption_statistics.append(
+                    StatisticData(
+                        start=start, state=data["energy"], sum=consumption_sum
+                    )
+                )
+
+            name_prefix = (
+                f"Duke Energy " f"{meter["serviceType"].capitalize()} {serial_number}"
+            )
+            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 meter["serviceType"] == "ELECTRIC"
+                else UnitOfVolume.CENTUM_CUBIC_FEET,
+            )
+
+            _LOGGER.debug(
+                "Adding %s statistics for %s",
+                len(consumption_statistics),
+                consumption_statistic_id,
+            )
+            async_add_external_statistics(
+                self.hass, consumption_metadata, consumption_statistics
+            )
+
+    async def _async_get_energy_usage(
+        self, meter: dict[str, Any], start_time: float | None = None
+    ) -> dict[datetime, dict[str, float | int]]:
+        """Get energy usage.
+
+        If start_time is None, get usage since account activation (or as far back as possible),
+        otherwise since start_time - 30 days to allow corrections in data.
+
+        Duke Energy provides hourly data all the way back to ~3 years.
+        """
+
+        # All of Duke Energy Service Areas are currently in America/New_York timezone
+        # May need to re-think this if that ever changes and determine timezone based
+        # on the service address somehow.
+        tz = await dt_util.async_get_time_zone("America/New_York")
+        lookback = timedelta(days=30)
+        one = timedelta(days=1)
+        if start_time is None:
+            # Max 3 years of data
+            agreement_date = dt_util.parse_datetime(meter["agreementActiveDate"])
+            if agreement_date is None:
+                start = dt_util.now(tz) - timedelta(days=3 * 365)
+            else:
+                start = max(
+                    agreement_date.replace(tzinfo=tz),
+                    dt_util.now(tz) - timedelta(days=3 * 365),
+                )
+        else:
+            start = datetime.fromtimestamp(start_time, tz=tz) - lookback
+
+        start = start.replace(hour=0, minute=0, second=0, microsecond=0)
+        end = dt_util.now(tz).replace(hour=0, minute=0, second=0, microsecond=0) - one
+        _LOGGER.debug("Data lookup range: %s - %s", start, end)
+
+        start_step = end - lookback
+        end_step = end
+        usage: dict[datetime, dict[str, float | int]] = {}
+        while True:
+            _LOGGER.debug("Getting hourly usage: %s - %s", start_step, end_step)
+            try:
+                # Get data
+                results = await self.api.get_energy_usage(
+                    meter["serialNum"], "HOURLY", "DAY", start_step, end_step
+                )
+                usage = {**results["data"], **usage}
+
+                for missing in results["missing"]:
+                    _LOGGER.debug("Missing data: %s", missing)
+
+                # Set next range
+                end_step = start_step - one
+                start_step = max(start_step - lookback, start)
+
+                # Make sure we don't go back too far
+                if end_step < start:
+                    break
+            except (TimeoutError, ClientError):
+                # ClientError is raised when there is no more data for the range
+                break
+
+        _LOGGER.debug("Got %s meter usage reads", len(usage))
+        return usage

homeassistant/components/duke_energy/manifest.json

@@ -0,0 +1,13 @@
+{
+  "domain": "duke_energy",
+  "name": "Duke Energy",
+  "codeowners": ["@hunterjm"],
+  "config_flow": true,
+  "dependencies": ["recorder"],
+  "documentation": "https://www.home-assistant.io/integrations/duke_energy",
+  "homekit": {},
+  "iot_class": "cloud_polling",
+  "requirements": ["aiodukeenergy==0.2.2"],
+  "ssdp": [],
+  "zeroconf": []
+}

homeassistant/components/duke_energy/strings.json

@@ -0,0 +1,20 @@
+{
+  "config": {
+    "step": {
+      "user": {
+        "data": {
+          "username": "[%key:common::config_flow::data::username%]",
+          "password": "[%key:common::config_flow::data::password%]"
+        }
+      }
+    },
+    "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

@@ -139,6 +139,7 @@
         "drop_connect",
         "dsmr",
         "dsmr_reader",
+        "duke_energy",
         "dunehd",
         "duotecno",
         "dwd_weather_warnings",

homeassistant/generated/integrations.json

@@ -1375,6 +1375,12 @@
       "config_flow": false,
       "iot_class": "cloud_polling"
     },
+    "duke_energy": {
+      "name": "Duke Energy",
+      "integration_type": "hub",
+      "config_flow": true,
+      "iot_class": "cloud_polling"
+    },
     "dunehd": {
       "name": "Dune HD",
       "integration_type": "hub",

requirements_all.txt

@@ -221,6 +221,9 @@ aiodiscover==2.1.0
 # homeassistant.components.dnsip
 aiodns==3.2.0
 
+# homeassistant.components.duke_energy
+aiodukeenergy==0.2.2
+
 # homeassistant.components.eafm
 aioeafm==0.1.2