From dc47a5e6140b51c2f7634470a95c44932daed235 Mon Sep 17 00:00:00 2001 From: Petar Petrov Date: Tue, 14 Oct 2025 14:40:57 +0300 Subject: [PATCH] Separate grid power from energy --- homeassistant/components/energy/data.py | 24 ++++-- homeassistant/components/energy/validate.py | 81 +++++++++++++++++++++ 2 files changed, 100 insertions(+), 5 deletions(-) diff --git a/homeassistant/components/energy/data.py b/homeassistant/components/energy/data.py index 812f43ccb2d..ed5b59174ff 100644 --- a/homeassistant/components/energy/data.py +++ b/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), } ) diff --git a/homeassistant/components/energy/validate.py b/homeassistant/components/energy/validate.py index 6c11c2b068c..081e18bd700 100644 --- a/homeassistant/components/energy/validate.py +++ b/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(