Separate grid power from energy

homeassistant/components/energy/data.py

@@ -32,10 +32,6 @@ class FlowFromGridSourceType(TypedDict):
     # statistic_id of an energy meter (kWh)
     stat_energy_from: str
 
-    # statistic_id of an power meter (kW)
-    # negative values indicate grid return
-    stat_power: str | None
-
     # statistic_id of costs ($) incurred from the energy meter
     # If set to None and entity_energy_price or number_energy_price are configured,
     # an EnergyCostSensor will be automatically created
@@ -62,6 +58,14 @@ class FlowToGridSourceType(TypedDict):
     number_energy_price: float | None  # Price for energy ($/kWh)
 
 
+class GridPowerSourceType(TypedDict):
+    """Dictionary holding the source of grid power consumption."""
+
+    # statistic_id of an power meter (kW)
+    # negative values indicate grid return
+    stat_power: str
+
+
 class GridSourceType(TypedDict):
     """Dictionary holding the source of grid energy consumption."""
 
@@ -69,6 +73,7 @@ class GridSourceType(TypedDict):
 
     flow_from: list[FlowFromGridSourceType]
     flow_to: list[FlowToGridSourceType]
+    power: list[GridPowerSourceType]
 
     cost_adjustment_day: float
 
@@ -182,7 +187,6 @@ FLOW_FROM_GRID_SOURCE_SCHEMA = vol.All(
     vol.Schema(
         {
             vol.Required("stat_energy_from"): str,
-            vol.Optional("stat_power"): str,
             vol.Optional("stat_cost"): vol.Any(str, None),
             # entity_energy_from was removed in HA Core 2022.10
             vol.Remove("entity_energy_from"): vol.Any(str, None),
@@ -205,6 +209,12 @@ FLOW_TO_GRID_SOURCE_SCHEMA = vol.Schema(
     }
 )
 
+GRID_POWER_SOURCE_SCHEMA = vol.Schema(
+    {
+        vol.Required("stat_power"): str,
+    }
+)
+
 
 def _generate_unique_value_validator(key: str) -> Callable[[list[dict]], list[dict]]:
     """Generate a validator that ensures a value is only used once."""
@@ -235,6 +245,10 @@ GRID_SOURCE_SCHEMA = vol.Schema(
             [FLOW_TO_GRID_SOURCE_SCHEMA],
             _generate_unique_value_validator("stat_energy_to"),
         ),
+        vol.Required("power"): vol.All(
+            [GRID_POWER_SOURCE_SCHEMA],
+            _generate_unique_value_validator("stat_power"),
+        ),
         vol.Required("cost_adjustment_day"): vol.Coerce(float),
     }
 )

homeassistant/components/energy/validate.py

@@ -12,6 +12,7 @@ from homeassistant.const import (
     STATE_UNAVAILABLE,
     STATE_UNKNOWN,
     UnitOfEnergy,
+    UnitOfPower,
     UnitOfVolume,
 )
 from homeassistant.core import HomeAssistant, callback, valid_entity_id
@@ -23,12 +24,17 @@ ENERGY_USAGE_DEVICE_CLASSES = (sensor.SensorDeviceClass.ENERGY,)
 ENERGY_USAGE_UNITS: dict[str, tuple[UnitOfEnergy, ...]] = {
     sensor.SensorDeviceClass.ENERGY: tuple(UnitOfEnergy)
 }
+POWER_USAGE_DEVICE_CLASSES = (sensor.SensorDeviceClass.POWER,)
+POWER_USAGE_UNITS: dict[str, tuple[UnitOfPower, ...]] = {
+    sensor.SensorDeviceClass.POWER: tuple(UnitOfPower)
+}
 
 ENERGY_PRICE_UNITS = tuple(
     f"/{unit}" for units in ENERGY_USAGE_UNITS.values() for unit in units
 )
 ENERGY_UNIT_ERROR = "entity_unexpected_unit_energy"
 ENERGY_PRICE_UNIT_ERROR = "entity_unexpected_unit_energy_price"
+POWER_UNIT_ERROR = "entity_unexpected_unit_power"
 GAS_USAGE_DEVICE_CLASSES = (
     sensor.SensorDeviceClass.ENERGY,
     sensor.SensorDeviceClass.GAS,
@@ -82,6 +88,10 @@ def _get_placeholders(hass: HomeAssistant, issue_type: str) -> dict[str, str] |
                 f"{currency}{unit}" for unit in ENERGY_PRICE_UNITS
             ),
         }
+    if issue_type == POWER_UNIT_ERROR:
+        return {
+            "power_units": ", ".join(POWER_USAGE_UNITS[sensor.SensorDeviceClass.POWER]),
+        }
     if issue_type == GAS_UNIT_ERROR:
         return {
             "energy_units": ", ".join(GAS_USAGE_UNITS[sensor.SensorDeviceClass.ENERGY]),
@@ -255,6 +265,62 @@ def _async_validate_price_entity(
         issues.add_issue(hass, unit_error, entity_id, unit)
 
 
+@callback
+def _async_validate_power_stat(
+    hass: HomeAssistant,
+    metadata: dict[str, tuple[int, recorder.models.StatisticMetaData]],
+    stat_id: str,
+    allowed_device_classes: Sequence[str],
+    allowed_units: Mapping[str, Sequence[str]],
+    unit_error: str,
+    issues: ValidationIssues,
+) -> None:
+    """Validate a power statistic."""
+    if stat_id not in metadata:
+        issues.add_issue(hass, "statistics_not_defined", stat_id)
+
+    has_entity_source = valid_entity_id(stat_id)
+
+    if not has_entity_source:
+        return
+
+    entity_id = stat_id
+
+    if not recorder.is_entity_recorded(hass, entity_id):
+        issues.add_issue(hass, "recorder_untracked", entity_id)
+        return
+
+    if (state := hass.states.get(entity_id)) is None:
+        issues.add_issue(hass, "entity_not_defined", entity_id)
+        return
+
+    if state.state in (STATE_UNAVAILABLE, STATE_UNKNOWN):
+        issues.add_issue(hass, "entity_unavailable", entity_id, state.state)
+        return
+
+    try:
+        float(state.state)
+    except ValueError:
+        issues.add_issue(hass, "entity_state_non_numeric", entity_id, state.state)
+        return
+
+    device_class = state.attributes.get(ATTR_DEVICE_CLASS)
+    if device_class not in allowed_device_classes:
+        issues.add_issue(
+            hass, "entity_unexpected_device_class", entity_id, device_class
+        )
+    else:
+        unit = state.attributes.get("unit_of_measurement")
+
+        if device_class and unit not in allowed_units.get(device_class, []):
+            issues.add_issue(hass, unit_error, entity_id, unit)
+
+    state_class = state.attributes.get(sensor.ATTR_STATE_CLASS)
+
+    if state_class != sensor.SensorStateClass.MEASUREMENT:
+        issues.add_issue(hass, "entity_unexpected_state_class", entity_id, state_class)
+
+
 @callback
 def _async_validate_cost_stat(
     hass: HomeAssistant,
@@ -434,6 +500,21 @@ async def async_validate(hass: HomeAssistant) -> EnergyPreferencesValidation:
                         )
                     )
 
+            for power_stat in source["power"]:
+                wanted_statistics_metadata.add(power_stat["stat_power"])
+                validate_calls.append(
+                    functools.partial(
+                        _async_validate_power_stat,
+                        hass,
+                        statistics_metadata,
+                        power_stat["stat_power"],
+                        POWER_USAGE_DEVICE_CLASSES,
+                        POWER_USAGE_UNITS,
+                        POWER_UNIT_ERROR,
+                        source_result,
+                    )
+                )
+
         elif source["type"] == "gas":
             wanted_statistics_metadata.add(source["stat_energy_from"])
             validate_calls.append(