Duke Energy Integration (#125489)

* Duke Energy Integration

* add recorder mock fixture to all tests

* address PR comments

* update tests

* add basic coordinator tests

* PR comments round 2

* Fix

---------

Co-authored-by: Joostlek <joostlek@outlook.com>

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,22 @@
+"""The Duke Energy integration."""
+
+from __future__ import annotations
+
+from homeassistant.core import HomeAssistant
+
+from .coordinator import DukeEnergyConfigEntry, DukeEnergyCoordinator
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: DukeEnergyConfigEntry) -> bool:
+    """Set up Duke Energy from a config entry."""
+
+    coordinator = DukeEnergyCoordinator(hass, entry.data)
+    await coordinator.async_config_entry_first_refresh()
+    entry.runtime_data = coordinator
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: DukeEnergyConfigEntry) -> bool:
+    """Unload a config entry."""
+    return True

homeassistant/components/duke_energy/config_flow.py

@@ -0,0 +1,67 @@
+"""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 "cannot_connect"
+            except (ClientError, TimeoutError):
+                errors["base"] = "cannot_connect"
+            except Exception:
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+            else:
+                username = auth["cdp_internal_user_id"].lower()
+                await self.async_set_unique_id(username)
+                self._abort_if_unique_id_configured()
+                email = auth["email"].lower()
+                data = {
+                    CONF_EMAIL: email,
+                    CONF_USERNAME: username,
+                    CONF_PASSWORD: user_input[CONF_PASSWORD],
+                }
+                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,222 @@
+"""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.config_entries import ConfigEntry
+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",)
+
+type DukeEnergyConfigEntry = ConfigEntry[DukeEnergyCoordinator]
+
+
+class DukeEnergyCoordinator(DataUpdateCoordinator[None]):
+    """Handle inserting statistics."""
+
+    config_entry: DukeEnergyConfigEntry
+
+    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)
+
+        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,10 @@
+{
+  "domain": "duke_energy",
+  "name": "Duke Energy",
+  "codeowners": ["@hunterjm"],
+  "config_flow": true,
+  "dependencies": ["recorder"],
+  "documentation": "https://www.home-assistant.io/integrations/duke_energy",
+  "iot_class": "cloud_polling",
+  "requirements": ["aiodukeenergy==0.2.2"]
+}

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 @@ FLOWS = {
         "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
 

requirements_test_all.txt

@@ -209,6 +209,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
 

tests/components/duke_energy/__init__.py

@@ -0,0 +1 @@
+"""Tests for the Duke Energy integration."""

tests/components/duke_energy/conftest.py

@@ -0,0 +1,90 @@
+"""Common fixtures for the Duke Energy tests."""
+
+from collections.abc import Generator
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from homeassistant.components.duke_energy.const import DOMAIN
+from homeassistant.const import CONF_EMAIL, CONF_PASSWORD, CONF_USERNAME
+from homeassistant.core import HomeAssistant
+from homeassistant.util import dt as dt_util
+
+from tests.common import MockConfigEntry
+from tests.typing import RecorderInstanceGenerator
+
+
+@pytest.fixture
+async def mock_recorder_before_hass(
+    async_test_recorder: RecorderInstanceGenerator,
+) -> None:
+    """Set up recorder."""
+
+
+@pytest.fixture
+def mock_setup_entry() -> Generator[AsyncMock]:
+    """Override async_setup_entry."""
+    with patch(
+        "homeassistant.components.duke_energy.async_setup_entry",
+        return_value=True,
+    ) as mock_setup_entry:
+        yield mock_setup_entry
+
+
+@pytest.fixture
+def mock_config_entry(hass: HomeAssistant) -> Generator[AsyncMock]:
+    """Return the default mocked config entry."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={
+            CONF_EMAIL: "test@example.com",
+            CONF_USERNAME: "test-username",
+            CONF_PASSWORD: "test-password",
+        },
+    )
+    config_entry.add_to_hass(hass)
+    return config_entry
+
+
+@pytest.fixture
+def mock_api() -> Generator[AsyncMock]:
+    """Mock a successful Duke Energy API."""
+    with (
+        patch(
+            "homeassistant.components.duke_energy.config_flow.DukeEnergy",
+            autospec=True,
+        ) as mock_api,
+        patch(
+            "homeassistant.components.duke_energy.coordinator.DukeEnergy",
+            new=mock_api,
+        ),
+    ):
+        api = mock_api.return_value
+        api.authenticate.return_value = {
+            "email": "TEST@EXAMPLE.COM",
+            "cdp_internal_user_id": "test-username",
+        }
+        api.get_meters.return_value = {}
+        yield api
+
+
+@pytest.fixture
+def mock_api_with_meters(mock_api: AsyncMock) -> AsyncMock:
+    """Mock a successful Duke Energy API with meters."""
+    mock_api.get_meters.return_value = {
+        "123": {
+            "serialNum": "123",
+            "serviceType": "ELECTRIC",
+            "agreementActiveDate": "2000-01-01",
+        },
+    }
+    mock_api.get_energy_usage.return_value = {
+        "data": {
+            dt_util.now(): {
+                "energy": 1.3,
+                "temperature": 70,
+            }
+        },
+        "missing": [],
+    }
+    return mock_api

tests/components/duke_energy/test_config_flow.py

@@ -0,0 +1,118 @@
+"""Test the Duke Energy config flow."""
+
+from unittest.mock import AsyncMock, Mock
+
+from aiohttp import ClientError, ClientResponseError
+import pytest
+
+from homeassistant import config_entries
+from homeassistant.components.duke_energy.const import DOMAIN
+from homeassistant.components.recorder import Recorder
+from homeassistant.const import CONF_EMAIL, CONF_PASSWORD, CONF_USERNAME
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResultType
+
+
+async def test_user(
+    hass: HomeAssistant,
+    recorder_mock: Recorder,
+    mock_api: AsyncMock,
+    mock_setup_entry: AsyncMock,
+) -> None:
+    """Test user config."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result.get("type") is FlowResultType.FORM
+    assert result.get("step_id") == "user"
+
+    # test with all provided
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {CONF_USERNAME: "test-username", CONF_PASSWORD: "test-password"},
+    )
+    assert result.get("type") is FlowResultType.CREATE_ENTRY
+    assert result.get("title") == "test@example.com"
+
+    data = result.get("data")
+    assert data
+    assert data[CONF_USERNAME] == "test-username"
+    assert data[CONF_PASSWORD] == "test-password"
+    assert data[CONF_EMAIL] == "test@example.com"
+
+
+async def test_abort_if_already_setup(
+    hass: HomeAssistant,
+    recorder_mock: Recorder,
+    mock_api: AsyncMock,
+    mock_config_entry: AsyncMock,
+) -> None:
+    """Test we abort if the email is already setup."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_USER},
+        data={
+            CONF_USERNAME: "test-username",
+            CONF_PASSWORD: "test-password",
+        },
+    )
+    assert result
+    assert result.get("type") is FlowResultType.ABORT
+    assert result.get("reason") == "already_configured"
+
+
+async def test_abort_if_already_setup_alternate_username(
+    hass: HomeAssistant,
+    recorder_mock: Recorder,
+    mock_api: AsyncMock,
+    mock_config_entry: AsyncMock,
+) -> None:
+    """Test we abort if the email is already setup."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_USER},
+        data={
+            CONF_USERNAME: "test@example.com",
+            CONF_PASSWORD: "test-password",
+        },
+    )
+    assert result
+    assert result.get("type") is FlowResultType.ABORT
+    assert result.get("reason") == "already_configured"
+
+
+@pytest.mark.parametrize(
+    ("side_effect", "expected_error"),
+    [
+        (ClientResponseError(None, None, status=404), "invalid_auth"),
+        (ClientResponseError(None, None, status=500), "cannot_connect"),
+        (TimeoutError(), "cannot_connect"),
+        (ClientError(), "cannot_connect"),
+        (Exception(), "unknown"),
+    ],
+)
+async def test_api_errors(
+    hass: HomeAssistant,
+    recorder_mock: Recorder,
+    mock_api: Mock,
+    side_effect,
+    expected_error,
+) -> None:
+    """Test the failure scenarios."""
+    mock_api.authenticate.side_effect = side_effect
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_USER},
+        data={CONF_USERNAME: "test-username", CONF_PASSWORD: "test-password"},
+    )
+    assert result.get("type") is FlowResultType.FORM
+    assert result.get("errors") == {"base": expected_error}
+
+    mock_api.authenticate.side_effect = None
+
+    # test with all provided
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {CONF_USERNAME: "test-username", CONF_PASSWORD: "test-password"},
+    )
+    assert result.get("type") is FlowResultType.CREATE_ENTRY

tests/components/duke_energy/test_coordinator.py

@@ -0,0 +1,44 @@
+"""Tests for the SolarEdge coordinator services."""
+
+from datetime import timedelta
+from unittest.mock import Mock, patch
+
+from freezegun.api import FrozenDateTimeFactory
+
+from homeassistant.components.recorder import Recorder
+from homeassistant.core import HomeAssistant
+from homeassistant.util import dt as dt_util
+
+from tests.common import MockConfigEntry, async_fire_time_changed
+
+
+async def test_update(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_api_with_meters: Mock,
+    freezer: FrozenDateTimeFactory,
+    recorder_mock: Recorder,
+) -> None:
+    """Test Coordinator."""
+    mock_config_entry.add_to_hass(hass)
+    await hass.config_entries.async_setup(mock_config_entry.entry_id)
+
+    assert mock_api_with_meters.get_meters.call_count == 1
+    # 3 years of data
+    assert mock_api_with_meters.get_energy_usage.call_count == 37
+
+    with patch(
+        "homeassistant.components.duke_energy.coordinator.get_last_statistics",
+        return_value={
+            "duke_energy:electric_123_energy_consumption": [
+                {"start": dt_util.now().timestamp()}
+            ]
+        },
+    ):
+        freezer.tick(timedelta(hours=12))
+        async_fire_time_changed(hass)
+        await hass.async_block_till_done(wait_background_tasks=True)
+
+        assert mock_api_with_meters.get_meters.call_count == 2
+        # Now have stats, so only one call
+        assert mock_api_with_meters.get_energy_usage.call_count == 38