Update tests

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

homeassistant/components/energy/data.py

@@ -59,13 +59,38 @@ class FlowToGridSourceType(TypedDict):
     number_energy_price: float | None  # Price for energy ($/kWh)
 
 
-class GridPowerSourceType(TypedDict):
+class PowerConfig(TypedDict, total=False):
+    """Dictionary holding power sensor configuration options.
+
+    Users can configure power sensors in three ways:
+    1. Standard: single sensor (positive=discharge/from_grid, negative=charge/to_grid)
+    2. Inverted: single sensor with opposite polarity (needs to be multiplied by -1)
+    3. Two sensors: separate positive sensors for each direction
+    """
+
+    # Standard: single sensor (positive=discharge/from_grid, negative=charge/to_grid)
+    stat_rate: str
+
+    # Inverted: single sensor with opposite polarity (needs to be multiplied by -1)
+    stat_rate_inverted: str
+
+    # Two sensors: separate positive sensors for each direction
+    # Result = stat_rate_from - stat_rate_to (positive when net outflow)
+    stat_rate_from: str  # Battery: discharge, Grid: consumption
+    stat_rate_to: str  # Battery: charge, Grid: return
+
+
+class GridPowerSourceType(TypedDict, total=False):
     """Dictionary holding the source of grid power consumption."""
 
     # statistic_id of a power meter (kW)
     # negative values indicate grid return
+    # This is either the original sensor or a generated template sensor
     stat_rate: str
 
+    # User's original power sensor configuration
+    power_config: PowerConfig
+
 
 class GridSourceType(TypedDict):
     """Dictionary holding the source of grid energy consumption."""
@@ -97,8 +122,12 @@ class BatterySourceType(TypedDict):
     stat_energy_from: str
     stat_energy_to: str
     # positive when discharging, negative when charging
+    # This is either the original sensor or a generated template sensor
     stat_rate: NotRequired[str]
 
+    # User's original power sensor configuration
+    power_config: NotRequired[PowerConfig]
+
 
 class GasSourceType(TypedDict):
     """Dictionary holding the source of gas consumption."""
@@ -211,10 +240,53 @@ FLOW_TO_GRID_SOURCE_SCHEMA = vol.Schema(
     }
 )
 
-GRID_POWER_SOURCE_SCHEMA = vol.Schema(
-    {
-        vol.Required("stat_rate"): str,
-    }
+
+def _validate_power_config(val: dict[str, Any]) -> dict[str, Any]:
+    """Validate power_config has exactly one configuration method."""
+    if not val:
+        raise vol.Invalid("power_config must have at least one option")
+
+    # Ensure only one configuration method is used
+    has_single = "stat_rate" in val
+    has_inverted = "stat_rate_inverted" in val
+    has_combined = "stat_rate_from" in val or "stat_rate_to" in val
+
+    methods_count = sum([has_single, has_inverted, has_combined])
+    if methods_count > 1:
+        raise vol.Invalid(
+            "power_config must use only one configuration method: "
+            "stat_rate, stat_rate_inverted, or stat_rate_from/stat_rate_to"
+        )
+
+    return val
+
+
+POWER_CONFIG_SCHEMA = vol.All(
+    vol.Schema(
+        {
+            vol.Exclusive("stat_rate", "power_source"): str,
+            vol.Exclusive("stat_rate_inverted", "power_source"): str,
+            # stat_rate_from/stat_rate_to: two sensors for bidirectional power
+            # Battery: from=discharge (out), to=charge (in)
+            # Grid: from=consumption, to=return
+            vol.Inclusive("stat_rate_from", "two_sensors"): str,
+            vol.Inclusive("stat_rate_to", "two_sensors"): str,
+        }
+    ),
+    _validate_power_config,
+)
+
+
+GRID_POWER_SOURCE_SCHEMA = vol.All(
+    vol.Schema(
+        {
+            # stat_rate and power_config are both optional schema keys, but the validator
+            # requires that at least one is provided; power_config takes precedence
+            vol.Optional("stat_rate"): str,
+            vol.Optional("power_config"): POWER_CONFIG_SCHEMA,
+        }
+    ),
+    cv.has_at_least_one_key("stat_rate", "power_config"),
 )
 
 
@@ -225,7 +297,7 @@ def _generate_unique_value_validator(key: str) -> Callable[[list[dict]], list[di
         val: list[dict],
     ) -> list[dict]:
         """Ensure that the user doesn't add duplicate values."""
-        counts = Counter(flow_from[key] for flow_from in val)
+        counts = Counter(item.get(key) for item in val if item.get(key) is not None)
 
         for value, count in counts.items():
             if count > 1:
@@ -267,7 +339,10 @@ BATTERY_SOURCE_SCHEMA = vol.Schema(
         vol.Required("type"): "battery",
         vol.Required("stat_energy_from"): str,
         vol.Required("stat_energy_to"): str,
+        # Both stat_rate and power_config are optional
+        # If power_config is provided, it takes precedence and stat_rate is overwritten
         vol.Optional("stat_rate"): str,
+        vol.Optional("power_config"): POWER_CONFIG_SCHEMA,
     }
 )
 GAS_SOURCE_SCHEMA = vol.Schema(
@@ -387,6 +462,12 @@ class EnergyManager:
             if key in update:
                 data[key] = update[key]
 
+        # Process energy sources and set stat_rate for power configs
+        if "energy_sources" in update:
+            data["energy_sources"] = self._process_energy_sources(
+                data["energy_sources"]
+            )
+
         self.data = data
         self._store.async_delay_save(lambda: data, 60)
 
@@ -395,6 +476,74 @@ class EnergyManager:
 
         await asyncio.gather(*(listener() for listener in self._update_listeners))
 
+    def _process_energy_sources(self, sources: list[SourceType]) -> list[SourceType]:
+        """Process energy sources and set stat_rate for power configs."""
+        from .helpers import generate_power_sensor_entity_id  # noqa: PLC0415
+
+        processed: list[SourceType] = []
+        for source in sources:
+            if source["type"] == "battery":
+                source = self._process_battery_power(
+                    source, generate_power_sensor_entity_id
+                )
+            elif source["type"] == "grid":
+                source = self._process_grid_power(
+                    source, generate_power_sensor_entity_id
+                )
+            processed.append(source)
+        return processed
+
+    def _process_battery_power(
+        self,
+        source: BatterySourceType,
+        generate_entity_id: Callable[[str, PowerConfig], str | None],
+    ) -> BatterySourceType:
+        """Set stat_rate for battery if power_config is specified."""
+        if "power_config" not in source:
+            return source
+
+        config = source["power_config"]
+
+        # If power_config has stat_rate (standard), just use it directly
+        if "stat_rate" in config:
+            return {**source, "stat_rate": config["stat_rate"]}
+
+        # For inverted or two-sensor config, set stat_rate to the generated entity_id
+        entity_id = generate_entity_id("battery", config)
+        if entity_id:
+            return {**source, "stat_rate": entity_id}
+
+        return source
+
+    def _process_grid_power(
+        self,
+        source: GridSourceType,
+        generate_entity_id: Callable[[str, PowerConfig], str | None],
+    ) -> GridSourceType:
+        """Set stat_rate for grid power sources if power_config is specified."""
+        if "power" not in source:
+            return source
+
+        processed_power: list[GridPowerSourceType] = []
+        for power in source["power"]:
+            if "power_config" in power:
+                config = power["power_config"]
+
+                # If power_config has stat_rate (standard), just use it directly
+                if "stat_rate" in config:
+                    processed_power.append({**power, "stat_rate": config["stat_rate"]})
+                    continue
+
+                # For inverted or two-sensor config, set stat_rate to generated entity_id
+                entity_id = generate_entity_id("grid", config)
+                if entity_id:
+                    processed_power.append({**power, "stat_rate": entity_id})
+                    continue
+
+            processed_power.append(power)
+
+        return {**source, "power": processed_power}
+
     @callback
     def async_listen_updates(self, update_listener: Callable[[], Awaitable]) -> None:
         """Listen for data updates."""

homeassistant/components/energy/helpers.py

@@ -0,0 +1,42 @@
+"""Helpers for the Energy integration."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from .data import PowerConfig
+
+
+def generate_power_sensor_unique_id(
+    source_type: str, config: PowerConfig
+) -> str | None:
+    """Generate a unique ID for a power transform sensor."""
+    if "stat_rate_inverted" in config:
+        sensor_id = config["stat_rate_inverted"].replace(".", "_")
+        return f"energy_power_{source_type}_inverted_{sensor_id}"
+    if "stat_rate_from" in config and "stat_rate_to" in config:
+        from_id = config["stat_rate_from"].replace(".", "_")
+        to_id = config["stat_rate_to"].replace(".", "_")
+        return f"energy_power_{source_type}_combined_{from_id}_{to_id}"
+    return None
+
+
+def generate_power_sensor_entity_id(
+    source_type: str, config: PowerConfig
+) -> str | None:
+    """Generate an entity ID for a power transform sensor."""
+    if "stat_rate_inverted" in config:
+        # Use source sensor name with _inverted suffix
+        source = config["stat_rate_inverted"]
+        if source.startswith("sensor."):
+            return f"{source}_inverted"
+        return f"sensor.{source.replace('.', '_')}_inverted"
+    if "stat_rate_from" in config and "stat_rate_to" in config:
+        # Use both sensors in entity ID to ensure uniqueness when multiple
+        # combined configs exist. The entity represents net power (from - to),
+        # e.g., discharge - charge for battery.
+        from_sensor = config["stat_rate_from"].removeprefix("sensor.")
+        to_sensor = config["stat_rate_to"].removeprefix("sensor.")
+        return f"sensor.energy_{source_type}_{from_sensor}_{to_sensor}_net_power"
+    return None

homeassistant/components/energy/sensor.py

@@ -19,7 +19,12 @@ from homeassistant.components.sensor import (
 from homeassistant.components.sensor.recorder import (  # pylint: disable=hass-component-root-import
     reset_detected,
 )
-from homeassistant.const import ATTR_UNIT_OF_MEASUREMENT, UnitOfEnergy, UnitOfVolume
+from homeassistant.const import (
+    ATTR_UNIT_OF_MEASUREMENT,
+    UnitOfEnergy,
+    UnitOfPower,
+    UnitOfVolume,
+)
 from homeassistant.core import (
     HomeAssistant,
     State,
@@ -36,7 +41,8 @@ from homeassistant.util import dt as dt_util, unit_conversion
 from homeassistant.util.unit_system import METRIC_SYSTEM
 
 from .const import DOMAIN
-from .data import EnergyManager, async_get_manager
+from .data import EnergyManager, PowerConfig, async_get_manager
+from .helpers import generate_power_sensor_entity_id, generate_power_sensor_unique_id
 
 SUPPORTED_STATE_CLASSES = {
     SensorStateClass.MEASUREMENT,
@@ -137,6 +143,7 @@ class SensorManager:
         self.manager = manager
         self.async_add_entities = async_add_entities
         self.current_entities: dict[tuple[str, str | None, str], EnergyCostSensor] = {}
+        self.current_power_entities: dict[str, EnergyPowerSensor] = {}
 
     async def async_start(self) -> None:
         """Start."""
@@ -147,8 +154,9 @@ class SensorManager:
 
     async def _process_manager_data(self) -> None:
         """Process manager data."""
-        to_add: list[EnergyCostSensor] = []
+        to_add: list[EnergyCostSensor | EnergyPowerSensor] = []
         to_remove = dict(self.current_entities)
+        power_to_remove = dict(self.current_power_entities)
 
         async def finish() -> None:
             if to_add:
@@ -159,6 +167,10 @@ class SensorManager:
                 self.current_entities.pop(key)
                 await entity.async_remove()
 
+            for power_key, power_entity in power_to_remove.items():
+                self.current_power_entities.pop(power_key)
+                await power_entity.async_remove()
+
         if not self.manager.data:
             await finish()
             return
@@ -185,6 +197,13 @@ class SensorManager:
                         to_remove,
                     )
 
+            # Process power sensors for battery and grid sources
+            self._process_power_sensor_data(
+                energy_source,
+                to_add,
+                power_to_remove,
+            )
+
         await finish()
 
     @callback
@@ -192,7 +211,7 @@ class SensorManager:
         self,
         adapter: SourceAdapter,
         config: Mapping[str, Any],
-        to_add: list[EnergyCostSensor],
+        to_add: list[EnergyCostSensor | EnergyPowerSensor],
         to_remove: dict[tuple[str, str | None, str], EnergyCostSensor],
     ) -> None:
         """Process sensor data."""
@@ -220,6 +239,74 @@ class SensorManager:
         )
         to_add.append(self.current_entities[key])
 
+    @callback
+    def _process_power_sensor_data(
+        self,
+        energy_source: Mapping[str, Any],
+        to_add: list[EnergyCostSensor | EnergyPowerSensor],
+        to_remove: dict[str, EnergyPowerSensor],
+    ) -> None:
+        """Process power sensor data for battery and grid sources."""
+        source_type = energy_source.get("type")
+
+        if source_type == "battery":
+            power_config = energy_source.get("power_config")
+            if power_config and self._needs_power_sensor(power_config):
+                self._create_or_keep_power_sensor(
+                    source_type, power_config, to_add, to_remove
+                )
+
+        elif source_type == "grid":
+            for power in energy_source.get("power", []):
+                power_config = power.get("power_config")
+                if power_config and self._needs_power_sensor(power_config):
+                    self._create_or_keep_power_sensor(
+                        source_type, power_config, to_add, to_remove
+                    )
+
+    @staticmethod
+    def _needs_power_sensor(power_config: PowerConfig) -> bool:
+        """Check if power_config needs a transform sensor."""
+        # Only create sensors for inverted or two-sensor configs
+        # Standard stat_rate configs don't need a transform sensor
+        return "stat_rate_inverted" in power_config or (
+            "stat_rate_from" in power_config and "stat_rate_to" in power_config
+        )
+
+    def _create_or_keep_power_sensor(
+        self,
+        source_type: str,
+        power_config: PowerConfig,
+        to_add: list[EnergyCostSensor | EnergyPowerSensor],
+        to_remove: dict[str, EnergyPowerSensor],
+    ) -> None:
+        """Create a power sensor or keep an existing one."""
+        unique_id = generate_power_sensor_unique_id(source_type, power_config)
+        if not unique_id:
+            return
+
+        # If entity already exists, keep it
+        if unique_id in to_remove:
+            to_remove.pop(unique_id)
+            return
+
+        # If we already have this entity, skip
+        if unique_id in self.current_power_entities:
+            return
+
+        entity_id = generate_power_sensor_entity_id(source_type, power_config)
+        if not entity_id:
+            return
+
+        sensor = EnergyPowerSensor(
+            source_type,
+            power_config,
+            unique_id,
+            entity_id,
+        )
+        self.current_power_entities[unique_id] = sensor
+        to_add.append(sensor)
+
 
 def _set_result_unless_done(future: asyncio.Future[None]) -> None:
     """Set the result of a future unless it is done."""
@@ -495,3 +582,197 @@ class EnergyCostSensor(SensorEntity):
             prefix = self._config[self._adapter.stat_energy_key]
 
         return f"{prefix}_{self._adapter.source_type}_{self._adapter.entity_id_suffix}"
+
+
+class EnergyPowerSensor(SensorEntity):
+    """Transform power sensor values (invert or combine two sensors).
+
+    This sensor handles non-standard power sensor configurations for the energy
+    dashboard by either inverting polarity or combining two positive sensors.
+    """
+
+    _attr_should_poll = False
+    _attr_device_class = SensorDeviceClass.POWER
+    _attr_state_class = SensorStateClass.MEASUREMENT
+    _attr_has_entity_name = True
+
+    def __init__(
+        self,
+        source_type: str,
+        config: PowerConfig,
+        unique_id: str,
+        entity_id: str,
+    ) -> None:
+        """Initialize the sensor."""
+        super().__init__()
+        self._source_type = source_type
+        self._config: PowerConfig = config
+        self._attr_unique_id = unique_id
+        self.entity_id = entity_id
+        self._source_sensors: list[str] = []
+        self._is_inverted = "stat_rate_inverted" in config
+        self._is_combined = "stat_rate_from" in config and "stat_rate_to" in config
+
+        # Determine source sensors
+        if self._is_inverted:
+            self._source_sensors = [config["stat_rate_inverted"]]
+        elif self._is_combined:
+            self._source_sensors = [
+                config["stat_rate_from"],
+                config["stat_rate_to"],
+            ]
+
+        # add_finished is set when either async_added_to_hass or add_to_platform_abort
+        # is called
+        self.add_finished: asyncio.Future[None] = (
+            asyncio.get_running_loop().create_future()
+        )
+
+    @property
+    def available(self) -> bool:
+        """Return if entity is available."""
+        if self._is_inverted:
+            source = self.hass.states.get(self._source_sensors[0])
+            return source is not None and source.state not in (
+                "unknown",
+                "unavailable",
+            )
+        if self._is_combined:
+            discharge = self.hass.states.get(self._source_sensors[0])
+            charge = self.hass.states.get(self._source_sensors[1])
+            return (
+                discharge is not None
+                and charge is not None
+                and discharge.state not in ("unknown", "unavailable")
+                and charge.state not in ("unknown", "unavailable")
+            )
+        return True
+
+    @callback
+    def _update_state(self) -> None:
+        """Update the sensor state based on source sensors."""
+        if self._is_inverted:
+            source_state = self.hass.states.get(self._source_sensors[0])
+            if source_state is None or source_state.state in ("unknown", "unavailable"):
+                self._attr_native_value = None
+                return
+            try:
+                value = float(source_state.state)
+            except ValueError:
+                self._attr_native_value = None
+                return
+
+            self._attr_native_value = value * -1
+
+        elif self._is_combined:
+            discharge_state = self.hass.states.get(self._source_sensors[0])
+            charge_state = self.hass.states.get(self._source_sensors[1])
+
+            if (
+                discharge_state is None
+                or charge_state is None
+                or discharge_state.state in ("unknown", "unavailable")
+                or charge_state.state in ("unknown", "unavailable")
+            ):
+                self._attr_native_value = None
+                return
+
+            try:
+                discharge = float(discharge_state.state)
+                charge = float(charge_state.state)
+            except ValueError:
+                self._attr_native_value = None
+                return
+
+            # Get units from state attributes
+            discharge_unit = discharge_state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
+            charge_unit = charge_state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
+
+            # Convert to Watts if units are present
+            if discharge_unit:
+                discharge = unit_conversion.PowerConverter.convert(
+                    discharge, discharge_unit, UnitOfPower.WATT
+                )
+            if charge_unit:
+                charge = unit_conversion.PowerConverter.convert(
+                    charge, charge_unit, UnitOfPower.WATT
+                )
+
+            self._attr_native_value = discharge - charge
+
+    async def async_added_to_hass(self) -> None:
+        """Register callbacks."""
+        # Set name based on source sensor(s)
+        if self._source_sensors:
+            entity_reg = er.async_get(self.hass)
+            device_id = None
+            source_name = None
+            # Check first sensor
+            if source_entry := entity_reg.async_get(self._source_sensors[0]):
+                device_id = source_entry.device_id
+                # For combined mode, always use Watts because we may have different source units; for inverted mode, copy source unit
+                if self._is_combined:
+                    self._attr_native_unit_of_measurement = UnitOfPower.WATT
+                else:
+                    self._attr_native_unit_of_measurement = (
+                        source_entry.unit_of_measurement
+                    )
+                # Get source name from registry
+                source_name = source_entry.name or source_entry.original_name
+            # Assign power sensor to same device as source sensor(s)
+            # Note: We use manual entity registry update instead of _attr_device_info
+            # because device assignment depends on runtime information from the entity
+            # registry (which source sensor has a device). This information isn't
+            # available during __init__, and the entity is already registered before
+            # async_added_to_hass runs, making the standard _attr_device_info pattern
+            # incompatible with this use case.
+            # If first sensor has no device and we have a second sensor, check it
+            if not device_id and len(self._source_sensors) > 1:
+                if source_entry := entity_reg.async_get(self._source_sensors[1]):
+                    device_id = source_entry.device_id
+            # Update entity registry entry with device_id
+            if device_id and (power_entry := entity_reg.async_get(self.entity_id)):
+                entity_reg.async_update_entity(
+                    power_entry.entity_id, device_id=device_id
+                )
+            else:
+                self._attr_has_entity_name = False
+
+            # Set name for inverted mode
+            if self._is_inverted:
+                if source_name:
+                    self._attr_name = f"{source_name} Inverted"
+                else:
+                    # Fall back to entity_id if no name in registry
+                    sensor_name = split_entity_id(self._source_sensors[0])[1].replace(
+                        "_", " "
+                    )
+                    self._attr_name = f"{sensor_name.title()} Inverted"
+
+        # Set name for combined mode
+        if self._is_combined:
+            self._attr_name = f"{self._source_type.title()} Power"
+
+        self._update_state()
+
+        # Track state changes on all source sensors
+        self.async_on_remove(
+            async_track_state_change_event(
+                self.hass,
+                self._source_sensors,
+                self._async_state_changed_listener,
+            )
+        )
+        _set_result_unless_done(self.add_finished)
+
+    @callback
+    def _async_state_changed_listener(self, *_: Any) -> None:
+        """Handle source sensor state changes."""
+        self._update_state()
+        self.async_write_ha_state()
+
+    @callback
+    def add_to_platform_abort(self) -> None:
+        """Abort adding an entity to a platform."""
+        _set_result_unless_done(self.add_finished)
+        super().add_to_platform_abort()

tests/components/energy/test_data.py

@@ -72,3 +72,186 @@ async def test_energy_preferences_migration_from_old_version(
     assert manager.data is not None
     assert "device_consumption_water" in manager.data
     assert manager.data["device_consumption_water"] == []
+
+
+async def test_battery_power_config_inverted_sets_stat_rate(
+    hass: HomeAssistant,
+) -> None:
+    """Test that battery with inverted power_config sets stat_rate to generated entity_id."""
+    manager = EnergyManager(hass)
+    await manager.async_initialize()
+    manager.data = manager.default_preferences()
+
+    await manager.async_update(
+        {
+            "energy_sources": [
+                {
+                    "type": "battery",
+                    "stat_energy_from": "sensor.battery_energy_from",
+                    "stat_energy_to": "sensor.battery_energy_to",
+                    "power_config": {
+                        "stat_rate_inverted": "sensor.battery_power",
+                    },
+                }
+            ],
+        }
+    )
+
+    # Verify stat_rate was set to the expected entity_id
+    assert manager.data is not None
+    assert len(manager.data["energy_sources"]) == 1
+    source = manager.data["energy_sources"][0]
+    assert source["stat_rate"] == "sensor.battery_power_inverted"
+    # Verify power_config is preserved
+    assert source["power_config"] == {"stat_rate_inverted": "sensor.battery_power"}
+
+
+async def test_battery_power_config_two_sensors_sets_stat_rate(
+    hass: HomeAssistant,
+) -> None:
+    """Test that battery with two-sensor power_config sets stat_rate."""
+    manager = EnergyManager(hass)
+    await manager.async_initialize()
+    manager.data = manager.default_preferences()
+
+    await manager.async_update(
+        {
+            "energy_sources": [
+                {
+                    "type": "battery",
+                    "stat_energy_from": "sensor.battery_energy_from",
+                    "stat_energy_to": "sensor.battery_energy_to",
+                    "power_config": {
+                        "stat_rate_from": "sensor.battery_discharge",
+                        "stat_rate_to": "sensor.battery_charge",
+                    },
+                }
+            ],
+        }
+    )
+
+    assert manager.data is not None
+    source = manager.data["energy_sources"][0]
+    # Entity ID includes discharge sensor name to avoid collisions
+    assert (
+        source["stat_rate"]
+        == "sensor.energy_battery_battery_discharge_battery_charge_net_power"
+    )
+
+
+async def test_grid_power_config_inverted_sets_stat_rate(
+    hass: HomeAssistant,
+) -> None:
+    """Test that grid with inverted power_config sets stat_rate."""
+    manager = EnergyManager(hass)
+    await manager.async_initialize()
+    manager.data = manager.default_preferences()
+
+    await manager.async_update(
+        {
+            "energy_sources": [
+                {
+                    "type": "grid",
+                    "flow_from": [],
+                    "flow_to": [],
+                    "power": [
+                        {
+                            "power_config": {
+                                "stat_rate_inverted": "sensor.grid_power",
+                            },
+                        }
+                    ],
+                    "cost_adjustment_day": 0,
+                }
+            ],
+        }
+    )
+
+    assert manager.data is not None
+    grid_source = manager.data["energy_sources"][0]
+    assert grid_source["power"][0]["stat_rate"] == "sensor.grid_power_inverted"
+
+
+async def test_power_config_standard_uses_stat_rate_directly(
+    hass: HomeAssistant,
+) -> None:
+    """Test that power_config with standard stat_rate uses it directly."""
+    manager = EnergyManager(hass)
+    await manager.async_initialize()
+    manager.data = manager.default_preferences()
+
+    await manager.async_update(
+        {
+            "energy_sources": [
+                {
+                    "type": "battery",
+                    "stat_energy_from": "sensor.battery_energy_from",
+                    "stat_energy_to": "sensor.battery_energy_to",
+                    "power_config": {
+                        "stat_rate": "sensor.battery_power",
+                    },
+                }
+            ],
+        }
+    )
+
+    assert manager.data is not None
+    source = manager.data["energy_sources"][0]
+    # stat_rate should be set directly from power_config.stat_rate
+    assert source["stat_rate"] == "sensor.battery_power"
+
+
+async def test_battery_without_power_config_unchanged(hass: HomeAssistant) -> None:
+    """Test that battery without power_config is unchanged."""
+    manager = EnergyManager(hass)
+    await manager.async_initialize()
+    manager.data = manager.default_preferences()
+
+    await manager.async_update(
+        {
+            "energy_sources": [
+                {
+                    "type": "battery",
+                    "stat_energy_from": "sensor.battery_energy_from",
+                    "stat_energy_to": "sensor.battery_energy_to",
+                    "stat_rate": "sensor.battery_power",
+                }
+            ],
+        }
+    )
+
+    assert manager.data is not None
+    source = manager.data["energy_sources"][0]
+    assert source["stat_rate"] == "sensor.battery_power"
+    assert "power_config" not in source
+
+
+async def test_power_config_takes_precedence_over_stat_rate(
+    hass: HomeAssistant,
+) -> None:
+    """Test that power_config takes precedence when both are provided."""
+    manager = EnergyManager(hass)
+    await manager.async_initialize()
+    manager.data = manager.default_preferences()
+
+    # Frontend sends both stat_rate and power_config
+    await manager.async_update(
+        {
+            "energy_sources": [
+                {
+                    "type": "battery",
+                    "stat_energy_from": "sensor.battery_energy_from",
+                    "stat_energy_to": "sensor.battery_energy_to",
+                    "stat_rate": "sensor.battery_power",  # This should be ignored
+                    "power_config": {
+                        "stat_rate_inverted": "sensor.battery_power",
+                    },
+                }
+            ],
+        }
+    )
+
+    assert manager.data is not None
+    source = manager.data["energy_sources"][0]
+    # stat_rate should be overwritten to point to the generated inverted sensor
+    assert source["stat_rate"] == "sensor.battery_power_inverted"

tests/components/energy/test_sensor.py

@@ -8,7 +8,8 @@ from typing import Any
 from freezegun.api import FrozenDateTimeFactory
 import pytest
 
-from homeassistant.components.energy import data
+from homeassistant.components.energy import async_get_manager, data
+from homeassistant.components.energy.sensor import SensorManager
 from homeassistant.components.recorder.core import Recorder
 from homeassistant.components.recorder.util import session_scope
 from homeassistant.components.sensor import (
@@ -25,15 +26,17 @@ from homeassistant.const import (
     ATTR_UNIT_OF_MEASUREMENT,
     STATE_UNKNOWN,
     UnitOfEnergy,
+    UnitOfPower,
     UnitOfVolume,
 )
 from homeassistant.core import HomeAssistant
-from homeassistant.helpers import entity_registry as er
+from homeassistant.helpers import device_registry as dr, entity_registry as er
 from homeassistant.setup import async_setup_component
 from homeassistant.util import dt as dt_util
 from homeassistant.util.unit_conversion import _WH_TO_CAL, _WH_TO_J
 from homeassistant.util.unit_system import METRIC_SYSTEM, US_CUSTOMARY_SYSTEM
 
+from tests.common import MockConfigEntry
 from tests.components.recorder.common import async_wait_recording_done
 from tests.typing import WebSocketGenerator
 
@@ -1324,3 +1327,610 @@ async def test_inherit_source_unique_id(
     assert entry
     assert entry.unique_id == f"{source_entry.id}_gas_cost"
     assert entry.hidden_by is er.RegistryEntryHider.INTEGRATION
+
+
+async def test_needs_power_sensor_standard(hass: HomeAssistant) -> None:
+    """Test _needs_power_sensor returns False for standard stat_rate."""
+    assert SensorManager._needs_power_sensor({"stat_rate": "sensor.power"}) is False
+
+
+async def test_needs_power_sensor_inverted(hass: HomeAssistant) -> None:
+    """Test _needs_power_sensor returns True for inverted config."""
+    assert (
+        SensorManager._needs_power_sensor({"stat_rate_inverted": "sensor.power"})
+        is True
+    )
+
+
+async def test_needs_power_sensor_combined(hass: HomeAssistant) -> None:
+    """Test _needs_power_sensor returns True for combined config."""
+    assert (
+        SensorManager._needs_power_sensor(
+            {
+                "stat_rate_from": "sensor.discharge",
+                "stat_rate_to": "sensor.charge",
+            }
+        )
+        is True
+    )
+
+
+async def test_needs_power_sensor_partial_combined(hass: HomeAssistant) -> None:
+    """Test _needs_power_sensor returns False for incomplete combined config."""
+    # Only stat_rate_from without stat_rate_to
+    assert (
+        SensorManager._needs_power_sensor({"stat_rate_from": "sensor.discharge"})
+        is False
+    )
+
+
+async def test_power_sensor_manager_creation(
+    recorder_mock: Recorder, hass: HomeAssistant
+) -> None:
+    """Test SensorManager creates power sensors correctly."""
+    assert await async_setup_component(hass, "energy", {"energy": {}})
+    manager = await async_get_manager(hass)
+    manager.data = manager.default_preferences()
+
+    # Set up a source sensor
+    hass.states.async_set(
+        "sensor.battery_power",
+        "100.0",
+        {ATTR_UNIT_OF_MEASUREMENT: UnitOfPower.WATT},
+    )
+    await hass.async_block_till_done()
+
+    # Update with battery that has inverted power_config
+    await manager.async_update(
+        {
+            "energy_sources": [
+                {
+                    "type": "battery",
+                    "stat_energy_from": "sensor.battery_energy_from",
+                    "stat_energy_to": "sensor.battery_energy_to",
+                    "power_config": {
+                        "stat_rate_inverted": "sensor.battery_power",
+                    },
+                }
+            ],
+        }
+    )
+    await hass.async_block_till_done()
+
+    # Verify the power sensor entity was created
+    state = hass.states.get("sensor.battery_power_inverted")
+    assert state is not None
+    assert float(state.state) == -100.0
+
+
+async def test_power_sensor_manager_cleanup(
+    recorder_mock: Recorder, hass: HomeAssistant
+) -> None:
+    """Test SensorManager removes power sensors when config changes."""
+    assert await async_setup_component(hass, "energy", {"energy": {}})
+    manager = await async_get_manager(hass)
+    manager.data = manager.default_preferences()
+
+    # Set up source sensors
+    hass.states.async_set("sensor.battery_power", "100.0")
+    await hass.async_block_till_done()
+
+    # Create with inverted power_config
+    await manager.async_update(
+        {
+            "energy_sources": [
+                {
+                    "type": "battery",
+                    "stat_energy_from": "sensor.battery_energy_from",
+                    "stat_energy_to": "sensor.battery_energy_to",
+                    "power_config": {
+                        "stat_rate_inverted": "sensor.battery_power",
+                    },
+                }
+            ],
+        }
+    )
+    await hass.async_block_till_done()
+
+    # Verify sensor exists and has a valid value
+    state = hass.states.get("sensor.battery_power_inverted")
+    assert state is not None
+    assert state.state == "-100.0"
+
+    # Update to remove power_config (use direct stat_rate)
+    await manager.async_update(
+        {
+            "energy_sources": [
+                {
+                    "type": "battery",
+                    "stat_energy_from": "sensor.battery_energy_from",
+                    "stat_energy_to": "sensor.battery_energy_to",
+                    "stat_rate": "sensor.battery_power",
+                }
+            ],
+        }
+    )
+    await hass.async_block_till_done()
+
+    # Verify sensor was removed (state becomes unavailable when entity is removed)
+    state = hass.states.get("sensor.battery_power_inverted")
+    assert state is None or state.state == "unavailable"
+
+
+async def test_power_sensor_grid_combined(
+    recorder_mock: Recorder, hass: HomeAssistant
+) -> None:
+    """Test power sensor for grid with combined config."""
+    assert await async_setup_component(hass, "energy", {"energy": {}})
+    manager = await async_get_manager(hass)
+    manager.data = manager.default_preferences()
+
+    # Set up source sensors
+    hass.states.async_set(
+        "sensor.grid_import",
+        "500.0",
+        {ATTR_UNIT_OF_MEASUREMENT: UnitOfPower.WATT},
+    )
+    hass.states.async_set(
+        "sensor.grid_export",
+        "200.0",
+        {ATTR_UNIT_OF_MEASUREMENT: UnitOfPower.WATT},
+    )
+    await hass.async_block_till_done()
+
+    # Update with grid that has combined power_config
+    await manager.async_update(
+        {
+            "energy_sources": [
+                {
+                    "type": "grid",
+                    "flow_from": [
+                        {
+                            "stat_energy_from": "sensor.grid_energy_import",
+                        }
+                    ],
+                    "flow_to": [
+                        {
+                            "stat_energy_to": "sensor.grid_energy_export",
+                        }
+                    ],
+                    "power": [
+                        {
+                            "power_config": {
+                                "stat_rate_from": "sensor.grid_import",
+                                "stat_rate_to": "sensor.grid_export",
+                            }
+                        }
+                    ],
+                    "cost_adjustment_day": 0,
+                }
+            ],
+        }
+    )
+    await hass.async_block_till_done()
+
+    # Verify the power sensor entity was created
+    state = hass.states.get("sensor.energy_grid_grid_import_grid_export_net_power")
+    assert state is not None
+    # 500 - 200 = 300 (net import)
+    assert float(state.state) == 300.0
+
+
+async def test_power_sensor_device_assignment(
+    recorder_mock: Recorder,
+    hass: HomeAssistant,
+    entity_registry: er.EntityRegistry,
+    device_registry: dr.DeviceRegistry,
+) -> None:
+    """Test power sensor is assigned to same device as source sensor."""
+    assert await async_setup_component(hass, "energy", {"energy": {}})
+    manager = await async_get_manager(hass)
+    manager.data = manager.default_preferences()
+
+    # Create a config entry for the device
+    config_entry = MockConfigEntry(domain="test")
+    config_entry.add_to_hass(hass)
+
+    # Create a device and register source sensor to it
+    device_entry = device_registry.async_get_or_create(
+        config_entry_id=config_entry.entry_id,
+        identifiers={("test", "battery_device")},
+        name="Battery Device",
+    )
+
+    # Register the source sensor with the device
+    entity_registry.async_get_or_create(
+        "sensor",
+        "test",
+        "battery_power",
+        suggested_object_id="battery_power",
+        device_id=device_entry.id,
+    )
+
+    # Set up source sensor state
+    hass.states.async_set(
+        "sensor.battery_power",
+        "100.0",
+        {ATTR_UNIT_OF_MEASUREMENT: UnitOfPower.WATT},
+    )
+    await hass.async_block_till_done()
+
+    # Update with battery that has inverted power_config
+    await manager.async_update(
+        {
+            "energy_sources": [
+                {
+                    "type": "battery",
+                    "stat_energy_from": "sensor.battery_energy_from",
+                    "stat_energy_to": "sensor.battery_energy_to",
+                    "power_config": {
+                        "stat_rate_inverted": "sensor.battery_power",
+                    },
+                }
+            ],
+        }
+    )
+    await hass.async_block_till_done()
+
+    # Verify the power sensor was created and assigned to same device
+    power_sensor_entry = entity_registry.async_get("sensor.battery_power_inverted")
+    assert power_sensor_entry is not None
+    assert power_sensor_entry.device_id == device_entry.id
+
+
+async def test_power_sensor_device_assignment_combined_second_sensor(
+    recorder_mock: Recorder,
+    hass: HomeAssistant,
+    entity_registry: er.EntityRegistry,
+    device_registry: dr.DeviceRegistry,
+) -> None:
+    """Test power sensor checks second sensor if first has no device."""
+    assert await async_setup_component(hass, "energy", {"energy": {}})
+    manager = await async_get_manager(hass)
+    manager.data = manager.default_preferences()
+
+    # Create a config entry for the device
+    config_entry = MockConfigEntry(domain="test")
+    config_entry.add_to_hass(hass)
+
+    # Create a device and register second sensor to it
+    device_entry = device_registry.async_get_or_create(
+        config_entry_id=config_entry.entry_id,
+        identifiers={("test", "battery_device")},
+        name="Battery Device",
+    )
+
+    # Register first sensor WITHOUT device
+    entity_registry.async_get_or_create(
+        "sensor",
+        "test",
+        "battery_discharge",
+        suggested_object_id="battery_discharge",
+    )
+
+    # Register second sensor WITH device
+    entity_registry.async_get_or_create(
+        "sensor",
+        "test",
+        "battery_charge",
+        suggested_object_id="battery_charge",
+        device_id=device_entry.id,
+    )
+
+    # Set up source sensor states
+    hass.states.async_set(
+        "sensor.battery_discharge",
+        "100.0",
+        {ATTR_UNIT_OF_MEASUREMENT: UnitOfPower.WATT},
+    )
+    hass.states.async_set(
+        "sensor.battery_charge",
+        "50.0",
+        {ATTR_UNIT_OF_MEASUREMENT: UnitOfPower.WATT},
+    )
+    await hass.async_block_till_done()
+
+    # Update with battery that has combined power_config
+    await manager.async_update(
+        {
+            "energy_sources": [
+                {
+                    "type": "battery",
+                    "stat_energy_from": "sensor.battery_energy_from",
+                    "stat_energy_to": "sensor.battery_energy_to",
+                    "power_config": {
+                        "stat_rate_from": "sensor.battery_discharge",
+                        "stat_rate_to": "sensor.battery_charge",
+                    },
+                }
+            ],
+        }
+    )
+    await hass.async_block_till_done()
+
+    # Verify the power sensor was created and assigned to second sensor's device
+    power_sensor_entry = entity_registry.async_get(
+        "sensor.energy_battery_battery_discharge_battery_charge_net_power"
+    )
+    assert power_sensor_entry is not None
+    assert power_sensor_entry.device_id == device_entry.id
+
+
+async def test_power_sensor_inverted_availability(
+    recorder_mock: Recorder, hass: HomeAssistant
+) -> None:
+    """Test inverted power sensor availability follows source sensor."""
+    assert await async_setup_component(hass, "energy", {"energy": {}})
+    manager = await async_get_manager(hass)
+    manager.data = manager.default_preferences()
+
+    # Set up source sensor as available
+    hass.states.async_set("sensor.battery_power", "100.0")
+    await hass.async_block_till_done()
+
+    # Configure battery with inverted power_config
+    await manager.async_update(
+        {
+            "energy_sources": [
+                {
+                    "type": "battery",
+                    "stat_energy_from": "sensor.battery_energy_from",
+                    "stat_energy_to": "sensor.battery_energy_to",
+                    "power_config": {
+                        "stat_rate_inverted": "sensor.battery_power",
+                    },
+                }
+            ],
+        }
+    )
+    await hass.async_block_till_done()
+
+    # Power sensor should be available
+    state = hass.states.get("sensor.battery_power_inverted")
+    assert state
+    assert state.state == "-100.0"
+
+    # Make source unavailable
+    hass.states.async_set("sensor.battery_power", "unavailable")
+    await hass.async_block_till_done()
+
+    # Power sensor should become unavailable
+    state = hass.states.get("sensor.battery_power_inverted")
+    assert state
+    assert state.state == "unavailable"
+
+    # Make source available again
+    hass.states.async_set("sensor.battery_power", "50.0")
+    await hass.async_block_till_done()
+
+    # Power sensor should become available again
+    state = hass.states.get("sensor.battery_power_inverted")
+    assert state
+    assert state.state == "-50.0"
+
+
+async def test_power_sensor_combined_availability(
+    recorder_mock: Recorder, hass: HomeAssistant
+) -> None:
+    """Test combined power sensor availability requires both sources available."""
+    assert await async_setup_component(hass, "energy", {"energy": {}})
+    manager = await async_get_manager(hass)
+    manager.data = manager.default_preferences()
+
+    # Set up both source sensors as available
+    hass.states.async_set("sensor.battery_discharge", "150.0")
+    hass.states.async_set("sensor.battery_charge", "50.0")
+    await hass.async_block_till_done()
+
+    # Configure battery with combined power_config
+    await manager.async_update(
+        {
+            "energy_sources": [
+                {
+                    "type": "battery",
+                    "stat_energy_from": "sensor.battery_energy_from",
+                    "stat_energy_to": "sensor.battery_energy_to",
+                    "power_config": {
+                        "stat_rate_from": "sensor.battery_discharge",
+                        "stat_rate_to": "sensor.battery_charge",
+                    },
+                }
+            ],
+        }
+    )
+    await hass.async_block_till_done()
+
+    # Power sensor should be available and show net power
+    state = hass.states.get(
+        "sensor.energy_battery_battery_discharge_battery_charge_net_power"
+    )
+    assert state
+    assert state.state == "100.0"
+
+    # Make first source unavailable
+    hass.states.async_set("sensor.battery_discharge", "unavailable")
+    await hass.async_block_till_done()
+
+    # Power sensor should become unavailable
+    state = hass.states.get(
+        "sensor.energy_battery_battery_discharge_battery_charge_net_power"
+    )
+    assert state
+    assert state.state == "unavailable"
+
+    # Make first source available again
+    hass.states.async_set("sensor.battery_discharge", "200.0")
+    await hass.async_block_till_done()
+
+    # Power sensor should become available again
+    state = hass.states.get(
+        "sensor.energy_battery_battery_discharge_battery_charge_net_power"
+    )
+    assert state
+    assert state.state == "150.0"
+
+    # Make second source unavailable
+    hass.states.async_set("sensor.battery_charge", "unknown")
+    await hass.async_block_till_done()
+
+    # Power sensor should become unavailable again
+    state = hass.states.get(
+        "sensor.energy_battery_battery_discharge_battery_charge_net_power"
+    )
+    assert state
+    assert state.state == "unavailable"
+
+
+async def test_power_sensor_battery_combined(
+    recorder_mock: Recorder, hass: HomeAssistant
+) -> None:
+    """Test power sensor for battery with combined config."""
+    assert await async_setup_component(hass, "energy", {"energy": {}})
+    manager = await async_get_manager(hass)
+    manager.data = manager.default_preferences()
+
+    # Set up source sensors
+    hass.states.async_set(
+        "sensor.battery_discharge",
+        "150.0",
+        {ATTR_UNIT_OF_MEASUREMENT: UnitOfPower.WATT},
+    )
+    hass.states.async_set(
+        "sensor.battery_charge",
+        "50.0",
+        {ATTR_UNIT_OF_MEASUREMENT: UnitOfPower.WATT},
+    )
+    await hass.async_block_till_done()
+
+    # Update with battery that has combined power_config
+    await manager.async_update(
+        {
+            "energy_sources": [
+                {
+                    "type": "battery",
+                    "stat_energy_from": "sensor.battery_energy_from",
+                    "stat_energy_to": "sensor.battery_energy_to",
+                    "power_config": {
+                        "stat_rate_from": "sensor.battery_discharge",
+                        "stat_rate_to": "sensor.battery_charge",
+                    },
+                }
+            ],
+        }
+    )
+    await hass.async_block_till_done()
+
+    # Verify the power sensor entity was created
+    state = hass.states.get(
+        "sensor.energy_battery_battery_discharge_battery_charge_net_power"
+    )
+    assert state is not None
+    # 150 - 50 = 100 (net discharging)
+    assert float(state.state) == 100.0
+
+    # Test net charging scenario
+    hass.states.async_set("sensor.battery_discharge", "30.0")
+    hass.states.async_set("sensor.battery_charge", "80.0")
+    await hass.async_block_till_done()
+
+    state = hass.states.get(
+        "sensor.energy_battery_battery_discharge_battery_charge_net_power"
+    )
+    assert state is not None
+    # 30 - 80 = -50 (net charging)
+    assert float(state.state) == -50.0
+
+
+async def test_power_sensor_combined_unit_conversion(
+    recorder_mock: Recorder, hass: HomeAssistant
+) -> None:
+    """Test power sensor combined mode with different units."""
+    assert await async_setup_component(hass, "energy", {"energy": {}})
+    manager = await async_get_manager(hass)
+    manager.data = manager.default_preferences()
+
+    # Set up source sensors with different units (kW and W)
+    hass.states.async_set(
+        "sensor.battery_discharge",
+        "1.5",  # 1.5 kW = 1500 W
+        {ATTR_UNIT_OF_MEASUREMENT: UnitOfPower.KILO_WATT},
+    )
+    hass.states.async_set(
+        "sensor.battery_charge",
+        "500.0",  # 500 W
+        {ATTR_UNIT_OF_MEASUREMENT: UnitOfPower.WATT},
+    )
+    await hass.async_block_till_done()
+
+    # Update with battery that has combined power_config
+    await manager.async_update(
+        {
+            "energy_sources": [
+                {
+                    "type": "battery",
+                    "stat_energy_from": "sensor.battery_energy_from",
+                    "stat_energy_to": "sensor.battery_energy_to",
+                    "power_config": {
+                        "stat_rate_from": "sensor.battery_discharge",
+                        "stat_rate_to": "sensor.battery_charge",
+                    },
+                }
+            ],
+        }
+    )
+    await hass.async_block_till_done()
+
+    # Verify the power sensor converts units properly
+    state = hass.states.get(
+        "sensor.energy_battery_battery_discharge_battery_charge_net_power"
+    )
+    assert state is not None
+    # 1500 W - 500 W = 1000 W (units are converted to W internally)
+    assert float(state.state) == 1000.0
+
+
+async def test_power_sensor_inverted_negative_values(
+    recorder_mock: Recorder, hass: HomeAssistant
+) -> None:
+    """Test inverted power sensor with negative source values."""
+    assert await async_setup_component(hass, "energy", {"energy": {}})
+    manager = await async_get_manager(hass)
+    manager.data = manager.default_preferences()
+
+    # Set up source sensor with positive value
+    hass.states.async_set(
+        "sensor.battery_power",
+        "100.0",
+        {ATTR_UNIT_OF_MEASUREMENT: UnitOfPower.WATT},
+    )
+    await hass.async_block_till_done()
+
+    # Update with battery that has inverted power_config
+    await manager.async_update(
+        {
+            "energy_sources": [
+                {
+                    "type": "battery",
+                    "stat_energy_from": "sensor.battery_energy_from",
+                    "stat_energy_to": "sensor.battery_energy_to",
+                    "power_config": {
+                        "stat_rate_inverted": "sensor.battery_power",
+                    },
+                }
+            ],
+        }
+    )
+    await hass.async_block_till_done()
+
+    # Verify inverted value
+    state = hass.states.get("sensor.battery_power_inverted")
+    assert state is not None
+    assert float(state.state) == -100.0
+
+    # Update source to negative value (should become positive)
+    hass.states.async_set("sensor.battery_power", "-50.0")
+    await hass.async_block_till_done()
+
+    state = hass.states.get("sensor.battery_power_inverted")
+    assert state is not None
+    assert float(state.state) == 50.0