Add return energy and compensation to Opower (#135258)

* Add return energy statistics to Opower coordinator

* Add consumption and return cost statistics to Opower coordinator

* Rename return cost to compensation for clarity and consistency

* Use original cost stats for consumption cost

* Rename return cost to compensation

* Remove comment

* Raise issue for negative consumption/cost values in Opower statistics

* Migrate existing statistics and raise an issue

* Update strings.json

---------

Co-authored-by: tronikos <tronikos@users.noreply.github.com>

homeassistant/components/opower/coordinator.py

@@ -2,7 +2,7 @@
 
 from datetime import datetime, timedelta
 import logging
-from typing import cast
+from typing import Any, cast
 
 from opower import (
     Account,
@@ -30,7 +30,7 @@ from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, UnitOfEnergy, UnitOfVolume
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.exceptions import ConfigEntryAuthFailed
-from homeassistant.helpers import aiohttp_client
+from homeassistant.helpers import aiohttp_client, issue_registry as ir
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 from homeassistant.util import dt as dt_util
 
@@ -123,11 +123,57 @@ class OpowerCoordinator(DataUpdateCoordinator[dict[str, Forecast]]):
                 )
             )
             cost_statistic_id = f"{DOMAIN}:{id_prefix}_energy_cost"
+            compensation_statistic_id = f"{DOMAIN}:{id_prefix}_energy_compensation"
             consumption_statistic_id = f"{DOMAIN}:{id_prefix}_energy_consumption"
+            return_statistic_id = f"{DOMAIN}:{id_prefix}_energy_return"
             _LOGGER.debug(
-                "Updating Statistics for %s and %s",
+                "Updating Statistics for %s, %s, %s, and %s",
                 cost_statistic_id,
+                compensation_statistic_id,
                 consumption_statistic_id,
+                return_statistic_id,
+            )
+
+            name_prefix = (
+                f"Opower {self.api.utility.subdomain()} "
+                f"{account.meter_type.name.lower()} {account.utility_account_id}"
+            )
+            cost_metadata = StatisticMetaData(
+                mean_type=StatisticMeanType.NONE,
+                has_sum=True,
+                name=f"{name_prefix} cost",
+                source=DOMAIN,
+                statistic_id=cost_statistic_id,
+                unit_of_measurement=None,
+            )
+            compensation_metadata = StatisticMetaData(
+                mean_type=StatisticMeanType.NONE,
+                has_sum=True,
+                name=f"{name_prefix} compensation",
+                source=DOMAIN,
+                statistic_id=compensation_statistic_id,
+                unit_of_measurement=None,
+            )
+            consumption_unit = (
+                UnitOfEnergy.KILO_WATT_HOUR
+                if account.meter_type == MeterType.ELEC
+                else UnitOfVolume.CENTUM_CUBIC_FEET
+            )
+            consumption_metadata = StatisticMetaData(
+                mean_type=StatisticMeanType.NONE,
+                has_sum=True,
+                name=f"{name_prefix} consumption",
+                source=DOMAIN,
+                statistic_id=consumption_statistic_id,
+                unit_of_measurement=consumption_unit,
+            )
+            return_metadata = StatisticMetaData(
+                mean_type=StatisticMeanType.NONE,
+                has_sum=True,
+                name=f"{name_prefix} return",
+                source=DOMAIN,
+                statistic_id=return_statistic_id,
+                unit_of_measurement=consumption_unit,
             )
 
             last_stat = await get_instance(self.hass).async_add_executor_job(
@@ -139,9 +185,24 @@ class OpowerCoordinator(DataUpdateCoordinator[dict[str, Forecast]]):
                     account, self.api.utility.timezone()
                 )
                 cost_sum = 0.0
+                compensation_sum = 0.0
                 consumption_sum = 0.0
+                return_sum = 0.0
                 last_stats_time = None
             else:
+                await self._async_maybe_migrate_statistics(
+                    account.utility_account_id,
+                    {
+                        cost_statistic_id: compensation_statistic_id,
+                        consumption_statistic_id: return_statistic_id,
+                    },
+                    {
+                        cost_statistic_id: cost_metadata,
+                        compensation_statistic_id: compensation_metadata,
+                        consumption_statistic_id: consumption_metadata,
+                        return_statistic_id: return_metadata,
+                    },
+                )
                 cost_reads = await self._async_get_cost_reads(
                     account,
                     self.api.utility.timezone(),
@@ -160,7 +221,12 @@ class OpowerCoordinator(DataUpdateCoordinator[dict[str, Forecast]]):
                         self.hass,
                         start,
                         end,
-                        {cost_statistic_id, consumption_statistic_id},
+                        {
+                            cost_statistic_id,
+                            compensation_statistic_id,
+                            consumption_statistic_id,
+                            return_statistic_id,
+                        },
                         "hour",
                         None,
                         {"sum"},
@@ -175,53 +241,56 @@ class OpowerCoordinator(DataUpdateCoordinator[dict[str, Forecast]]):
                 # We are in this code path only if get_last_statistics found a stat
                 # so statistics_during_period should also have found at least one.
                 assert stats
-                cost_sum = cast(float, stats[cost_statistic_id][0]["sum"])
+
-                consumption_sum = cast(float, stats[consumption_statistic_id][0]["sum"])
+                def _safe_get_sum(records: list[Any]) -> float:
+                    if records and "sum" in records[0]:
+                        return float(records[0]["sum"])
+                    return 0.0
+
+                cost_sum = _safe_get_sum(stats.get(cost_statistic_id, []))
+                compensation_sum = _safe_get_sum(
+                    stats.get(compensation_statistic_id, [])
+                )
+                consumption_sum = _safe_get_sum(stats.get(consumption_statistic_id, []))
+                return_sum = _safe_get_sum(stats.get(return_statistic_id, []))
                 last_stats_time = stats[consumption_statistic_id][0]["start"]
 
             cost_statistics = []
+            compensation_statistics = []
             consumption_statistics = []
+            return_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_state = max(0, cost_read.provided_cost)
+                compensation_state = max(0, -cost_read.provided_cost)
+                consumption_state = max(0, cost_read.consumption)
+                return_state = max(0, -cost_read.consumption)
+
+                cost_sum += cost_state
+                compensation_sum += compensation_state
+                consumption_sum += consumption_state
+                return_sum += return_state
 
                 cost_statistics.append(
+                    StatisticData(start=start, state=cost_state, sum=cost_sum)
+                )
+                compensation_statistics.append(
                     StatisticData(
-                        start=start, state=cost_read.provided_cost, sum=cost_sum
+                        start=start, state=compensation_state, sum=compensation_sum
                     )
                 )
                 consumption_statistics.append(
                     StatisticData(
-                        start=start, state=cost_read.consumption, sum=consumption_sum
+                        start=start, state=consumption_state, sum=consumption_sum
                     )
                 )
-
+                return_statistics.append(
-            name_prefix = (
+                    StatisticData(start=start, state=return_state, sum=return_sum)
-                f"Opower {self.api.utility.subdomain()} "
+                )
-                f"{account.meter_type.name.lower()} {account.utility_account_id}"
-            )
-            cost_metadata = StatisticMetaData(
-                mean_type=StatisticMeanType.NONE,
-                has_sum=True,
-                name=f"{name_prefix} cost",
-                source=DOMAIN,
-                statistic_id=cost_statistic_id,
-                unit_of_measurement=None,
-            )
-            consumption_metadata = StatisticMetaData(
-                mean_type=StatisticMeanType.NONE,
-                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,
-            )
 
             _LOGGER.debug(
                 "Adding %s statistics for %s",
@@ -229,6 +298,14 @@ class OpowerCoordinator(DataUpdateCoordinator[dict[str, Forecast]]):
                 cost_statistic_id,
             )
             async_add_external_statistics(self.hass, cost_metadata, cost_statistics)
+            _LOGGER.debug(
+                "Adding %s statistics for %s",
+                len(compensation_statistics),
+                compensation_statistic_id,
+            )
+            async_add_external_statistics(
+                self.hass, compensation_metadata, compensation_statistics
+            )
             _LOGGER.debug(
                 "Adding %s statistics for %s",
                 len(consumption_statistics),
@@ -237,6 +314,133 @@ class OpowerCoordinator(DataUpdateCoordinator[dict[str, Forecast]]):
             async_add_external_statistics(
                 self.hass, consumption_metadata, consumption_statistics
             )
+            _LOGGER.debug(
+                "Adding %s statistics for %s",
+                len(return_statistics),
+                return_statistic_id,
+            )
+            async_add_external_statistics(self.hass, return_metadata, return_statistics)
+
+    async def _async_maybe_migrate_statistics(
+        self,
+        utility_account_id: str,
+        migration_map: dict[str, str],
+        metadata_map: dict[str, StatisticMetaData],
+    ) -> None:
+        """Perform one-time statistics migration based on the provided map.
+
+        Splits negative values from source IDs into target IDs.
+
+        Args:
+            utility_account_id: The account ID (for issue_id).
+            migration_map: Map from source statistic ID to target statistic ID
+                           (e.g., {cost_id: compensation_id}).
+            metadata_map: Map of all statistic IDs (source and target) to their metadata.
+
+        """
+        if not migration_map:
+            return
+
+        need_migration_source_ids = set()
+        for source_id, target_id in migration_map.items():
+            last_target_stat = await get_instance(self.hass).async_add_executor_job(
+                get_last_statistics,
+                self.hass,
+                1,
+                target_id,
+                True,
+                {},
+            )
+            if not last_target_stat:
+                need_migration_source_ids.add(source_id)
+        if not need_migration_source_ids:
+            return
+
+        _LOGGER.info("Starting one-time migration for: %s", need_migration_source_ids)
+
+        processed_stats: dict[str, list[StatisticData]] = {}
+
+        existing_stats = await get_instance(self.hass).async_add_executor_job(
+            statistics_during_period,
+            self.hass,
+            dt_util.utc_from_timestamp(0),
+            None,
+            need_migration_source_ids,
+            "hour",
+            None,
+            {"start", "state", "sum"},
+        )
+        for source_id, source_stats in existing_stats.items():
+            _LOGGER.debug("Found %d statistics for %s", len(source_stats), source_id)
+            if not source_stats:
+                continue
+            target_id = migration_map[source_id]
+
+            updated_source_stats: list[StatisticData] = []
+            new_target_stats: list[StatisticData] = []
+            updated_source_sum = 0.0
+            new_target_sum = 0.0
+            need_migration = False
+
+            prev_sum = 0.0
+            for stat in source_stats:
+                start = dt_util.utc_from_timestamp(stat["start"])
+                curr_sum = cast(float, stat["sum"])
+                state = curr_sum - prev_sum
+                prev_sum = curr_sum
+                if state < 0:
+                    need_migration = True
+
+                updated_source_state = max(0, state)
+                new_target_state = max(0, -state)
+
+                updated_source_sum += updated_source_state
+                new_target_sum += new_target_state
+
+                updated_source_stats.append(
+                    StatisticData(
+                        start=start, state=updated_source_state, sum=updated_source_sum
+                    )
+                )
+                new_target_stats.append(
+                    StatisticData(
+                        start=start, state=new_target_state, sum=new_target_sum
+                    )
+                )
+
+            if need_migration:
+                processed_stats[source_id] = updated_source_stats
+                processed_stats[target_id] = new_target_stats
+            else:
+                need_migration_source_ids.remove(source_id)
+
+        if not need_migration_source_ids:
+            _LOGGER.debug("No migration needed")
+            return
+
+        for stat_id, stats in processed_stats.items():
+            _LOGGER.debug("Applying %d migrated stats for %s", len(stats), stat_id)
+            async_add_external_statistics(self.hass, metadata_map[stat_id], stats)
+
+        ir.async_create_issue(
+            self.hass,
+            DOMAIN,
+            issue_id=f"return_to_grid_migration_{utility_account_id}",
+            is_fixable=False,
+            severity=ir.IssueSeverity.WARNING,
+            translation_key="return_to_grid_migration",
+            translation_placeholders={
+                "utility_account_id": utility_account_id,
+                "energy_settings": "/config/energy",
+                "target_ids": "\n".join(
+                    {
+                        v
+                        for k, v in migration_map.items()
+                        if k in need_migration_source_ids
+                    }
+                ),
+            },
+        )
 
     async def _async_get_cost_reads(
         self, account: Account, time_zone_str: str, start_time: float | None = None

homeassistant/components/opower/strings.json

@@ -31,5 +31,11 @@
       "already_configured": "[%key:common::config_flow::abort::already_configured_service%]",
       "reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
     }
+  },
+  "issues": {
+    "return_to_grid_migration": {
+      "title": "Return to grid statistics for account: {utility_account_id}",
+      "description": "We found negative values in your existing consumption statistics, likely because you have solar. We split those in separate return statistics for a better experience in the Energy dashboard.\n\nPlease visit the [Energy configuration page]({energy_settings}) to add the following statistics in the **Return to grid** section:\n\n{target_ids}"
+    }
   }
 }