starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-08-24 10:13:52 -05:00
Code Issues Packages Projects Releases Wiki Activity

Migrate grid connections to single objects with import/export/power (#162200)

Browse Source 
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Petar Petrov
2026-02-16 09:55:42 +02:00
committed by GitHub
co-authored by Martin Hjelmare
parent 4386b3d5cc
commit ffbb8c037e
8 changed files with 1014 additions and 467 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/energy/data.py
+247 -40
View File
@@ -15,7 +15,7 @@ from homeassistant.helpers import config_validation as cv, singleton, storage
from .const import DOMAIN
STORAGE_VERSION = 1
STORAGE_MINOR_VERSION = 2
STORAGE_MINOR_VERSION = 3
STORAGE_KEY = DOMAIN
@@ -92,8 +92,11 @@ class GridPowerSourceType(TypedDict, total=False):
power_config: PowerConfig
class GridSourceType(TypedDict):
"""Dictionary holding the source of grid energy consumption."""
class LegacyGridSourceType(TypedDict):
"""Legacy dictionary holding the source of grid energy consumption.
This format is deprecated and will be migrated to GridSourceType.
"""
type: Literal["grid"]
@@ -104,6 +107,40 @@ class GridSourceType(TypedDict):
cost_adjustment_day: float
class GridSourceType(TypedDict):
"""Dictionary holding a unified grid connection (like batteries).
Each grid connection represents a single import/export pair with
optional power tracking. Multiple grid sources are allowed.
"""
type: Literal["grid"]
# Import meter - kWh consumed from grid
# Can be None for export-only or power-only grids migrated from legacy format
stat_energy_from: str | None
# Export meter (optional) - kWh returned to grid (solar/battery export)
stat_energy_to: str | None
# Cost tracking for import
stat_cost: str | None # statistic_id of costs ($) incurred
entity_energy_price: str | None # entity_id providing price ($/kWh)
number_energy_price: float | None # Fixed price ($/kWh)
# Compensation tracking for export
stat_compensation: str | None # statistic_id of compensation ($) received
entity_energy_price_export: str | None # entity_id providing export price ($/kWh)
number_energy_price_export: float | None # Fixed export price ($/kWh)
# Power measurement (optional)
# positive when consuming from grid, negative when exporting
stat_rate: NotRequired[str]
power_config: NotRequired[PowerConfig]
cost_adjustment_day: float
class SolarSourceType(TypedDict):
"""Dictionary holding the source of energy production."""
@@ -308,23 +345,77 @@ def _generate_unique_value_validator(key: str) -> Callable[[list[dict]], list[di
return validate_uniqueness
GRID_SOURCE_SCHEMA = vol.Schema(
{
vol.Required("type"): "grid",
vol.Required("flow_from"): vol.All(
[FLOW_FROM_GRID_SOURCE_SCHEMA],
_generate_unique_value_validator("stat_energy_from"),
),
vol.Required("flow_to"): vol.All(
[FLOW_TO_GRID_SOURCE_SCHEMA],
_generate_unique_value_validator("stat_energy_to"),
),
vol.Optional("power"): vol.All(
[GRID_POWER_SOURCE_SCHEMA],
_generate_unique_value_validator("stat_rate"),
),
vol.Required("cost_adjustment_day"): vol.Coerce(float),
}
def _grid_ensure_single_price_import(
val: dict[str, Any],
) -> dict[str, Any]:
"""Ensure we use a single price source for import."""
if (
val.get("entity_energy_price") is not None
and val.get("number_energy_price") is not None
):
raise vol.Invalid("Define either an entity or a fixed number for import price")
return val
def _grid_ensure_single_price_export(
val: dict[str, Any],
) -> dict[str, Any]:
"""Ensure we use a single price source for export."""
if (
val.get("entity_energy_price_export") is not None
and val.get("number_energy_price_export") is not None
):
raise vol.Invalid("Define either an entity or a fixed number for export price")
return val
def _grid_ensure_at_least_one_stat(
val: dict[str, Any],
) -> dict[str, Any]:
"""Ensure at least one of import, export, or power is configured."""
if (
val.get("stat_energy_from") is None
and val.get("stat_energy_to") is None
and val.get("stat_rate") is None
and val.get("power_config") is None
):
raise vol.Invalid(
"Grid must have at least one of: import meter, export meter, or power sensor"
)
return val
GRID_SOURCE_SCHEMA = vol.All(
vol.Schema(
{
vol.Required("type"): "grid",
# Import meter (can be None for export-only grids from legacy migration)
vol.Optional("stat_energy_from", default=None): vol.Any(str, None),
# Export meter (optional)
vol.Optional("stat_energy_to", default=None): vol.Any(str, None),
# Import cost tracking
vol.Optional("stat_cost", default=None): vol.Any(str, None),
vol.Optional("entity_energy_price", default=None): vol.Any(str, None),
vol.Optional("number_energy_price", default=None): vol.Any(
vol.Coerce(float), None
),
# Export compensation tracking
vol.Optional("stat_compensation", default=None): vol.Any(str, None),
vol.Optional("entity_energy_price_export", default=None): vol.Any(
str, None
),
vol.Optional("number_energy_price_export", default=None): vol.Any(
vol.Coerce(float), None
),
# Power measurement (optional)
vol.Optional("stat_rate"): str,
vol.Optional("power_config"): POWER_CONFIG_SCHEMA,
vol.Required("cost_adjustment_day"): vol.Coerce(float),
}
),
_grid_ensure_single_price_import,
_grid_ensure_single_price_export,
_grid_ensure_at_least_one_stat,
)
SOLAR_SOURCE_SCHEMA = vol.Schema(
{
@@ -369,10 +460,46 @@ WATER_SOURCE_SCHEMA = vol.Schema(
def check_type_limits(value: list[SourceType]) -> list[SourceType]:
"""Validate that we don't have too many of certain types."""
types = Counter([val["type"] for val in value])
# Currently no type limits - multiple grid sources are allowed (like batteries)
return value
if types.get("grid", 0) > 1:
raise vol.Invalid("You cannot have more than 1 grid source")
def _validate_grid_stat_uniqueness(value: list[SourceType]) -> list[SourceType]:
"""Validate that grid statistics are unique across all sources."""
seen_import: set[str] = set()
seen_export: set[str] = set()
seen_rate: set[str] = set()
for source in value:
if source.get("type") != "grid":
continue
# Cast to GridSourceType since we've filtered for grid type
grid_source: GridSourceType = source # type: ignore[assignment]
# Check import meter uniqueness
if (stat_from := grid_source.get("stat_energy_from")) is not None:
if stat_from in seen_import:
raise vol.Invalid(
f"Import meter {stat_from} is used in multiple grid connections"
)
seen_import.add(stat_from)
# Check export meter uniqueness
if (stat_to := grid_source.get("stat_energy_to")) is not None:
if stat_to in seen_export:
raise vol.Invalid(
f"Export meter {stat_to} is used in multiple grid connections"
)
seen_export.add(stat_to)
# Check power stat uniqueness
if (stat_rate := grid_source.get("stat_rate")) is not None:
if stat_rate in seen_rate:
raise vol.Invalid(
f"Power stat {stat_rate} is used in multiple grid connections"
)
seen_rate.add(stat_rate)
return value
@@ -393,6 +520,7 @@ ENERGY_SOURCE_SCHEMA = vol.All(
]
),
check_type_limits,
_validate_grid_stat_uniqueness,
)
DEVICE_CONSUMPTION_SCHEMA = vol.Schema(
@@ -405,6 +533,82 @@ DEVICE_CONSUMPTION_SCHEMA = vol.Schema(
)
def _migrate_legacy_grid_to_unified(
old_grid: dict[str, Any],
) -> list[dict[str, Any]]:
"""Migrate legacy grid format (flow_from/flow_to/power arrays) to unified format.
Each grid connection can have any combination of import, export, and power -
all are optional as long as at least one is configured.
Migration pairs arrays by index position:
- flow_from[i], flow_to[i], and power[i] combine into grid connection i
- If arrays have different lengths, missing entries get None for that field
- The number of grid connections equals max(len(flow_from), len(flow_to), len(power))
"""
flow_from = old_grid.get("flow_from", [])
flow_to = old_grid.get("flow_to", [])
power_list = old_grid.get("power", [])
cost_adj = old_grid.get("cost_adjustment_day", 0.0)
new_sources: list[dict[str, Any]] = []
# Number of grid connections = max length across all three arrays
# If all arrays are empty, don't create any grid sources
max_len = max(len(flow_from), len(flow_to), len(power_list))
if max_len == 0:
return []
for i in range(max_len):
source: dict[str, Any] = {
"type": "grid",
"cost_adjustment_day": cost_adj,
}
# Import fields from flow_from
if i < len(flow_from):
ff = flow_from[i]
source["stat_energy_from"] = ff.get("stat_energy_from") or None
source["stat_cost"] = ff.get("stat_cost")
source["entity_energy_price"] = ff.get("entity_energy_price")
source["number_energy_price"] = ff.get("number_energy_price")
else:
# Export-only entry - set import to None (validation will flag this)
source["stat_energy_from"] = None
source["stat_cost"] = None
source["entity_energy_price"] = None
source["number_energy_price"] = None
# Export fields from flow_to
if i < len(flow_to):
ft = flow_to[i]
source["stat_energy_to"] = ft.get("stat_energy_to")
source["stat_compensation"] = ft.get("stat_compensation")
source["entity_energy_price_export"] = ft.get("entity_energy_price")
source["number_energy_price_export"] = ft.get("number_energy_price")
else:
source["stat_energy_to"] = None
source["stat_compensation"] = None
source["entity_energy_price_export"] = None
source["number_energy_price_export"] = None
# Power config at index i goes to grid connection at index i
if i < len(power_list):
power = power_list[i]
if "power_config" in power:
source["power_config"] = power["power_config"]
if "stat_rate" in power:
source["stat_rate"] = power["stat_rate"]
new_sources.append(source)
return new_sources
def _is_legacy_grid_format(source: dict[str, Any]) -> bool:
"""Check if a grid source is in the legacy format."""
return source.get("type") == "grid" and "flow_from" in source
class _EnergyPreferencesStore(storage.Store[EnergyPreferences]):
"""Energy preferences store with migration support."""
@@ -419,6 +623,18 @@ class _EnergyPreferencesStore(storage.Store[EnergyPreferences]):
if old_major_version == 1 and old_minor_version < 2:
# Add device_consumption_water field if it doesn't exist
data.setdefault("device_consumption_water", [])
if old_major_version == 1 and old_minor_version < 3:
# Migrate legacy grid format to unified format
new_sources: list[dict[str, Any]] = []
for source in data.get("energy_sources", []):
if _is_legacy_grid_format(source):
# Convert legacy grid to multiple unified grid sources
new_sources.extend(_migrate_legacy_grid_to_unified(source))
else:
new_sources.append(source)
data["energy_sources"] = new_sources
return data
@@ -516,27 +732,18 @@ class EnergyManager:
source: GridSourceType,
generate_entity_id: Callable[[str, PowerConfig], str],
) -> GridSourceType:
"""Set stat_rate for grid power sources if power_config is specified."""
if "power" not in source:
"""Set stat_rate for grid if power_config is specified."""
if "power_config" not in source:
return source
processed_power: list[GridPowerSourceType] = []
for power in source["power"]:
if "power_config" in power:
config = power["power_config"]
config = source["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"]})
else:
# For inverted or two-sensor config, set stat_rate to generated entity_id
processed_power.append(
{**power, "stat_rate": generate_entity_id("grid", config)}
)
else:
processed_power.append(power)
# If power_config has stat_rate (standard), just use it directly
if "stat_rate" in config:
return {**source, "stat_rate": config["stat_rate"]}
return {**source, "power": processed_power}
# For inverted or two-sensor config, set stat_rate to the generated entity_id
return {**source, "stat_rate": generate_entity_id("grid", config)}
@callback
def async_listen_updates(self, update_listener: Callable[[], Awaitable]) -> None:
homeassistant/components/energy/sensor.py
+95 -35
View File
@@ -94,22 +94,15 @@ class SourceAdapter:
SOURCE_ADAPTERS: Final = (
# Grid import cost (unified format)
SourceAdapter(
"grid",
"flow_from",
None, # No flow_type - unified format
"stat_energy_from",
"stat_cost",
"Cost",
"cost",
),
SourceAdapter(
"grid",
"flow_to",
"stat_energy_to",
"stat_compensation",
"Compensation",
"compensation",
),
SourceAdapter(
"gas",
None,
@@ -128,6 +121,16 @@ SOURCE_ADAPTERS: Final = (
),
)
# Separate adapter for grid export compensation (needs different price field)
GRID_EXPORT_ADAPTER: Final = SourceAdapter(
"grid",
None, # No flow_type - unified format
"stat_energy_to",
"stat_compensation",
"Compensation",
"compensation",
)
class EntityNotFoundError(HomeAssistantError):
"""When a referenced entity was not found."""
@@ -183,22 +186,20 @@ class SensorManager:
if adapter.source_type != energy_source["type"]:
continue
if adapter.flow_type is None:
self._process_sensor_data(
adapter,
energy_source,
to_add,
to_remove,
)
continue
self._process_sensor_data(
adapter,
energy_source,
to_add,
to_remove,
)
for flow in energy_source[adapter.flow_type]: # type: ignore[typeddict-item]
self._process_sensor_data(
adapter,
flow,
to_add,
to_remove,
)
# Handle grid export compensation (unified format uses different price fields)
if energy_source["type"] == "grid":
self._process_grid_export_sensor(
energy_source,
to_add,
to_remove,
)
# Process power sensors for battery and grid sources
self._process_power_sensor_data(
@@ -222,11 +223,16 @@ class SensorManager:
if config.get(adapter.total_money_key) is not None:
return
key = (adapter.source_type, adapter.flow_type, config[adapter.stat_energy_key])
# Skip if the energy stat is not configured (e.g., export-only or power-only grids)
stat_energy = config.get(adapter.stat_energy_key)
if not stat_energy:
return
key = (adapter.source_type, adapter.flow_type, stat_energy)
# Make sure the right data is there
# If the entity existed, we don't pop it from to_remove so it's removed
if not valid_entity_id(config[adapter.stat_energy_key]) or (
if not valid_entity_id(stat_energy) or (
config.get("entity_energy_price") is None
and config.get("number_energy_price") is None
):
@@ -242,6 +248,56 @@ class SensorManager:
)
to_add.append(self.current_entities[key])
@callback
def _process_grid_export_sensor(
self,
config: Mapping[str, Any],
to_add: list[EnergyCostSensor | EnergyPowerSensor],
to_remove: dict[tuple[str, str | None, str], EnergyCostSensor],
) -> None:
"""Process grid export compensation sensor (unified format).
The unified grid format uses different field names for export pricing:
- entity_energy_price_export instead of entity_energy_price
- number_energy_price_export instead of number_energy_price
"""
# No export meter configured
stat_energy_to = config.get("stat_energy_to")
if stat_energy_to is None:
return
# Already have a compensation stat
if config.get("stat_compensation") is not None:
return
key = ("grid", None, stat_energy_to)
# Check for export pricing fields (different names in unified format)
if not valid_entity_id(stat_energy_to) or (
config.get("entity_energy_price_export") is None
and config.get("number_energy_price_export") is None
):
return
# Create a config wrapper that maps the sell price fields to standard names
# so EnergyCostSensor can use them
export_config: dict[str, Any] = {
"stat_energy_to": stat_energy_to,
"stat_compensation": config.get("stat_compensation"),
"entity_energy_price": config.get("entity_energy_price_export"),
"number_energy_price": config.get("number_energy_price_export"),
}
if current_entity := to_remove.pop(key, None):
current_entity.update_config(export_config)
return
self.current_entities[key] = EnergyCostSensor(
GRID_EXPORT_ADAPTER,
export_config,
)
to_add.append(self.current_entities[key])
@callback
def _process_power_sensor_data(
self,
@@ -252,21 +308,14 @@ class SensorManager:
"""Process power sensor data for battery and grid sources."""
source_type = energy_source.get("type")
if source_type == "battery":
if source_type in ("battery", "grid"):
# Both battery and grid now use unified format with power_config at top level
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."""
@@ -312,6 +361,17 @@ class EnergyCostSensor(SensorEntity):
This is intended as a fallback for when no specific cost sensor is available for the
utility.
Expected config fields (from adapter or export_config wrapper):
- stat_energy_key (via adapter): Key to get the energy statistic ID
- total_money_key (via adapter): Key to get the existing cost/compensation stat
- entity_energy_price: Entity ID providing price per unit (e.g., $/kWh)
- number_energy_price: Fixed price per unit
Note: For grid export compensation, the unified format uses different field names
(entity_energy_price_export, number_energy_price_export). The _process_grid_export_sensor
method in SensorManager creates a wrapper config that maps these to the standard
field names (entity_energy_price, number_energy_price) so this class can use them.
"""
_attr_entity_registry_visible_default = False
homeassistant/components/energy/validate.py
+32 -23
View File
@@ -401,16 +401,20 @@ def _validate_grid_source(
source_result: ValidationIssues,
validate_calls: list[functools.partial[None]],
) -> None:
"""Validate grid energy source."""
flow_from: data.FlowFromGridSourceType
for flow_from in source["flow_from"]:
wanted_statistics_metadata.add(flow_from["stat_energy_from"])
"""Validate grid energy source (unified format)."""
stat_energy_from = source.get("stat_energy_from")
stat_energy_to = source.get("stat_energy_to")
stat_rate = source.get("stat_rate")
# Validate import meter (optional)
if stat_energy_from:
wanted_statistics_metadata.add(stat_energy_from)
validate_calls.append(
functools.partial(
_async_validate_usage_stat,
hass,
statistics_metadata,
flow_from["stat_energy_from"],
stat_energy_from,
ENERGY_USAGE_DEVICE_CLASSES,
ENERGY_USAGE_UNITS,
ENERGY_UNIT_ERROR,
@@ -418,7 +422,8 @@ def _validate_grid_source(
)
)
if (stat_cost := flow_from.get("stat_cost")) is not None:
# Validate import cost tracking (only if import meter exists)
if (stat_cost := source.get("stat_cost")) is not None:
wanted_statistics_metadata.add(stat_cost)
validate_calls.append(
functools.partial(
@@ -429,7 +434,7 @@ def _validate_grid_source(
source_result,
)
)
elif (entity_energy_price := flow_from.get("entity_energy_price")) is not None:
elif (entity_energy_price := source.get("entity_energy_price")) is not None:
validate_calls.append(
functools.partial(
_async_validate_price_entity,
@@ -442,27 +447,27 @@ def _validate_grid_source(
)
if (
flow_from.get("entity_energy_price") is not None
or flow_from.get("number_energy_price") is not None
source.get("entity_energy_price") is not None
or source.get("number_energy_price") is not None
):
validate_calls.append(
functools.partial(
_async_validate_auto_generated_cost_entity,
hass,
flow_from["stat_energy_from"],
stat_energy_from,
source_result,
)
)
flow_to: data.FlowToGridSourceType
for flow_to in source["flow_to"]:
wanted_statistics_metadata.add(flow_to["stat_energy_to"])
# Validate export meter (optional)
if stat_energy_to:
wanted_statistics_metadata.add(stat_energy_to)
validate_calls.append(
functools.partial(
_async_validate_usage_stat,
hass,
statistics_metadata,
flow_to["stat_energy_to"],
stat_energy_to,
ENERGY_USAGE_DEVICE_CLASSES,
ENERGY_USAGE_UNITS,
ENERGY_UNIT_ERROR,
@@ -470,7 +475,8 @@ def _validate_grid_source(
)
)
if (stat_compensation := flow_to.get("stat_compensation")) is not None:
# Validate export compensation tracking
if (stat_compensation := source.get("stat_compensation")) is not None:
wanted_statistics_metadata.add(stat_compensation)
validate_calls.append(
functools.partial(
@@ -481,12 +487,14 @@ def _validate_grid_source(
source_result,
)
)
elif (entity_energy_price := flow_to.get("entity_energy_price")) is not None:
elif (
entity_price_export := source.get("entity_energy_price_export")
) is not None:
validate_calls.append(
functools.partial(
_async_validate_price_entity,
hass,
entity_energy_price,
entity_price_export,
source_result,
ENERGY_PRICE_UNITS,
ENERGY_PRICE_UNIT_ERROR,
@@ -494,26 +502,27 @@ def _validate_grid_source(
)
if (
flow_to.get("entity_energy_price") is not None
or flow_to.get("number_energy_price") is not None
source.get("entity_energy_price_export") is not None
or source.get("number_energy_price_export") is not None
):
validate_calls.append(
functools.partial(
_async_validate_auto_generated_cost_entity,
hass,
flow_to["stat_energy_to"],
stat_energy_to,
source_result,
)
)
for power_stat in source.get("power", []):
wanted_statistics_metadata.add(power_stat["stat_rate"])
# Validate power sensor (optional)
if stat_rate:
wanted_statistics_metadata.add(stat_rate)
validate_calls.append(
functools.partial(
_async_validate_power_stat,
hass,
statistics_metadata,
power_stat["stat_rate"],
stat_rate,
POWER_USAGE_DEVICE_CLASSES,
POWER_USAGE_UNITS,
POWER_UNIT_ERROR,
tests/components/energy/test_data.py
+450 -80
View File
@@ -160,15 +160,17 @@ async def test_grid_power_config_inverted_sets_stat_rate(
"energy_sources": [
{
"type": "grid",
"flow_from": [],
"flow_to": [],
"power": [
{
"power_config": {
"stat_rate_inverted": "sensor.grid_power",
},
}
],
"stat_energy_from": "sensor.grid_import",
"stat_energy_to": None,
"stat_cost": None,
"stat_compensation": None,
"entity_energy_price": None,
"number_energy_price": None,
"entity_energy_price_export": None,
"number_energy_price_export": None,
"power_config": {
"stat_rate_inverted": "sensor.grid_power",
},
"cost_adjustment_day": 0,
}
],
@@ -177,7 +179,7 @@ async def test_grid_power_config_inverted_sets_stat_rate(
assert manager.data is not None
grid_source = manager.data["energy_sources"][0]
assert grid_source["power"][0]["stat_rate"] == "sensor.grid_power_inverted"
assert grid_source["stat_rate"] == "sensor.grid_power_inverted"
async def test_power_config_standard_uses_stat_rate_directly(
@@ -319,25 +321,27 @@ async def test_flow_from_validation_multiple_prices() -> None:
async def test_energy_sources_validation_multiple_grids() -> None:
"""Test that multiple grid sources are rejected."""
# Multiple grid sources should fail validation
with pytest.raises(vol.Invalid, match="You cannot have more than 1 grid source"):
ENERGY_SOURCE_SCHEMA(
[
{
"type": "grid",
"flow_from": [],
"flow_to": [],
"cost_adjustment_day": 0,
},
{
"type": "grid",
"flow_from": [],
"flow_to": [],
"cost_adjustment_day": 0,
},
]
)
"""Test that multiple grid sources are allowed (like batteries)."""
# Multiple grid sources should now pass validation
result = ENERGY_SOURCE_SCHEMA(
[
{
"type": "grid",
"stat_energy_from": "sensor.grid1_import",
"stat_energy_to": "sensor.grid1_export",
"cost_adjustment_day": 0,
},
{
"type": "grid",
"stat_energy_from": "sensor.grid2_import",
"stat_energy_to": None,
"cost_adjustment_day": 0,
},
]
)
assert len(result) == 2
assert result[0]["stat_energy_from"] == "sensor.grid1_import"
assert result[1]["stat_energy_from"] == "sensor.grid2_import"
async def test_power_config_validation_passes() -> None:
@@ -371,15 +375,17 @@ async def test_grid_power_config_standard_stat_rate(hass: HomeAssistant) -> None
"energy_sources": [
{
"type": "grid",
"flow_from": [],
"flow_to": [],
"power": [
{
"power_config": {
"stat_rate": "sensor.grid_power",
},
}
],
"stat_energy_from": "sensor.grid_import",
"stat_energy_to": None,
"stat_cost": None,
"stat_compensation": None,
"entity_energy_price": None,
"number_energy_price": None,
"entity_energy_price_export": None,
"number_energy_price_export": None,
"power_config": {
"stat_rate": "sensor.grid_power",
},
"cost_adjustment_day": 0,
}
],
@@ -389,37 +395,44 @@ async def test_grid_power_config_standard_stat_rate(hass: HomeAssistant) -> None
assert manager.data is not None
grid_source = manager.data["energy_sources"][0]
# stat_rate should be set directly from power_config.stat_rate
assert grid_source["power"][0]["stat_rate"] == "sensor.grid_power"
assert grid_source["stat_rate"] == "sensor.grid_power"
async def test_flow_from_duplicate_stat_energy_from() -> None:
"""Test that duplicate stat_energy_from values are rejected."""
with pytest.raises(
vol.Invalid, match="Cannot specify sensor.energy more than once"
):
ENERGY_SOURCE_SCHEMA(
[
{
"type": "grid",
"flow_from": [
{
"stat_energy_from": "sensor.energy",
"stat_cost": None,
"entity_energy_price": None,
"number_energy_price": 0.15,
},
{
"stat_energy_from": "sensor.energy", # Duplicate
"stat_cost": None,
"entity_energy_price": None,
"number_energy_price": 0.20,
},
],
"flow_to": [],
"cost_adjustment_day": 0,
},
]
)
async def test_grid_new_format_validates_correctly() -> None:
"""Test that new unified grid format validates correctly."""
# Valid grid source with import and export
result = ENERGY_SOURCE_SCHEMA(
[
{
"type": "grid",
"stat_energy_from": "sensor.energy_import",
"stat_energy_to": "sensor.energy_export",
"stat_cost": None,
"stat_compensation": None,
"entity_energy_price": None,
"number_energy_price": 0.15,
"entity_energy_price_export": None,
"number_energy_price_export": 0.08,
"cost_adjustment_day": 0,
},
]
)
assert len(result) == 1
assert result[0]["stat_energy_from"] == "sensor.energy_import"
assert result[0]["stat_energy_to"] == "sensor.energy_export"
# Valid grid source with import only (no export)
result = ENERGY_SOURCE_SCHEMA(
[
{
"type": "grid",
"stat_energy_from": "sensor.energy_import",
"stat_energy_to": None,
"cost_adjustment_day": 0,
},
]
)
assert result[0]["stat_energy_to"] is None
async def test_async_update_when_data_is_none(hass: HomeAssistant) -> None:
@@ -453,7 +466,7 @@ async def test_async_update_when_data_is_none(hass: HomeAssistant) -> None:
async def test_grid_power_without_power_config(hass: HomeAssistant) -> None:
"""Test that grid power entry without power_config is preserved unchanged."""
"""Test that grid without power_config is preserved unchanged."""
manager = EnergyManager(hass)
await manager.async_initialize()
manager.data = manager.default_preferences()
@@ -463,14 +476,16 @@ async def test_grid_power_without_power_config(hass: HomeAssistant) -> None:
"energy_sources": [
{
"type": "grid",
"flow_from": [],
"flow_to": [],
"power": [
{
# No power_config, just stat_rate directly
"stat_rate": "sensor.grid_power",
}
],
"stat_energy_from": "sensor.grid_import",
"stat_energy_to": None,
"stat_cost": None,
"stat_compensation": None,
"entity_energy_price": None,
"number_energy_price": None,
"entity_energy_price_export": None,
"number_energy_price_export": None,
# No power_config, just stat_rate directly
"stat_rate": "sensor.grid_power",
"cost_adjustment_day": 0,
}
],
@@ -479,7 +494,362 @@ async def test_grid_power_without_power_config(hass: HomeAssistant) -> None:
assert manager.data is not None
grid_source = manager.data["energy_sources"][0]
# Power entry should be preserved unchanged
assert len(grid_source["power"]) == 1
assert grid_source["power"][0]["stat_rate"] == "sensor.grid_power"
assert "power_config" not in grid_source["power"][0]
# stat_rate should be preserved unchanged
assert grid_source["stat_rate"] == "sensor.grid_power"
assert "power_config" not in grid_source
async def test_grid_migration_single_import_export(hass: HomeAssistant) -> None:
"""Test migration from legacy format with 1 import + 1 export creates 1 grid."""
# Create legacy format data (v1.2) with flow_from/flow_to arrays
old_data = {
"energy_sources": [
{
"type": "grid",
"flow_from": [
{
"stat_energy_from": "sensor.grid_import",
"stat_cost": "sensor.grid_cost",
"entity_energy_price": None,
"number_energy_price": None,
}
],
"flow_to": [
{
"stat_energy_to": "sensor.grid_export",
"stat_compensation": None,
"entity_energy_price": "sensor.sell_price",
"number_energy_price": None,
}
],
"cost_adjustment_day": 0.5,
}
],
"device_consumption": [],
"device_consumption_water": [],
}
# Save with old version (1.2) - migration will run to upgrade to 1.3
old_store = storage.Store(hass, 1, "energy", minor_version=2)
await old_store.async_save(old_data)
# Load with manager - should trigger migration
manager = EnergyManager(hass)
await manager.async_initialize()
# Verify migration created unified grid source
assert manager.data is not None
assert len(manager.data["energy_sources"]) == 1
grid = manager.data["energy_sources"][0]
assert grid["type"] == "grid"
assert grid["stat_energy_from"] == "sensor.grid_import"
assert grid["stat_energy_to"] == "sensor.grid_export"
assert grid["stat_cost"] == "sensor.grid_cost"
assert grid["stat_compensation"] is None
assert grid["entity_energy_price"] is None
assert grid["entity_energy_price_export"] == "sensor.sell_price"
assert grid["cost_adjustment_day"] == 0.5
# Should not have legacy fields
assert "flow_from" not in grid
assert "flow_to" not in grid
async def test_grid_migration_multiple_imports_exports_paired(
hass: HomeAssistant,
) -> None:
"""Test migration with 2 imports + 2 exports creates 2 paired grids."""
old_data = {
"energy_sources": [
{
"type": "grid",
"flow_from": [
{
"stat_energy_from": "sensor.grid_import_1",
"stat_cost": None,
"entity_energy_price": None,
"number_energy_price": 0.15,
},
{
"stat_energy_from": "sensor.grid_import_2",
"stat_cost": None,
"entity_energy_price": None,
"number_energy_price": 0.20,
},
],
"flow_to": [
{
"stat_energy_to": "sensor.grid_export_1",
"stat_compensation": None,
"entity_energy_price": None,
"number_energy_price": 0.08,
},
{
"stat_energy_to": "sensor.grid_export_2",
"stat_compensation": None,
"entity_energy_price": None,
"number_energy_price": 0.05,
},
],
"cost_adjustment_day": 0,
}
],
"device_consumption": [],
"device_consumption_water": [],
}
old_store = storage.Store(hass, 1, "energy", minor_version=2)
await old_store.async_save(old_data)
manager = EnergyManager(hass)
await manager.async_initialize()
assert manager.data is not None
assert len(manager.data["energy_sources"]) == 2
# First grid: paired import_1 with export_1
grid1 = manager.data["energy_sources"][0]
assert grid1["stat_energy_from"] == "sensor.grid_import_1"
assert grid1["stat_energy_to"] == "sensor.grid_export_1"
assert grid1["number_energy_price"] == 0.15
assert grid1["number_energy_price_export"] == 0.08
# Second grid: paired import_2 with export_2
grid2 = manager.data["energy_sources"][1]
assert grid2["stat_energy_from"] == "sensor.grid_import_2"
assert grid2["stat_energy_to"] == "sensor.grid_export_2"
assert grid2["number_energy_price"] == 0.20
assert grid2["number_energy_price_export"] == 0.05
async def test_grid_migration_more_imports_than_exports(hass: HomeAssistant) -> None:
"""Test migration with 3 imports + 1 export creates 3 grids (first has export)."""
old_data = {
"energy_sources": [
{
"type": "grid",
"flow_from": [
{"stat_energy_from": "sensor.import_1"},
{"stat_energy_from": "sensor.import_2"},
{"stat_energy_from": "sensor.import_3"},
],
"flow_to": [
{"stat_energy_to": "sensor.export_1"},
],
"cost_adjustment_day": 0,
}
],
"device_consumption": [],
"device_consumption_water": [],
}
old_store = storage.Store(hass, 1, "energy", minor_version=2)
await old_store.async_save(old_data)
manager = EnergyManager(hass)
await manager.async_initialize()
assert manager.data is not None
assert len(manager.data["energy_sources"]) == 3
# First grid: has both import and export
grid1 = manager.data["energy_sources"][0]
assert grid1["stat_energy_from"] == "sensor.import_1"
assert grid1["stat_energy_to"] == "sensor.export_1"
# Second and third grids: import only
grid2 = manager.data["energy_sources"][1]
assert grid2["stat_energy_from"] == "sensor.import_2"
assert grid2["stat_energy_to"] is None
grid3 = manager.data["energy_sources"][2]
assert grid3["stat_energy_from"] == "sensor.import_3"
assert grid3["stat_energy_to"] is None
async def test_grid_migration_with_power(hass: HomeAssistant) -> None:
"""Test migration preserves power config and stat_rate from first grid.
Note: Migration preserves the original stat_rate value from the legacy power array.
The stat_rate regeneration from power_config only happens during async_update()
for new data submissions, not during storage migration.
"""
old_data = {
"energy_sources": [
{
"type": "grid",
"flow_from": [
{"stat_energy_from": "sensor.grid_import"},
],
"flow_to": [
{"stat_energy_to": "sensor.grid_export"},
],
"power": [
{
"stat_rate": "sensor.grid_power",
"power_config": {"stat_rate_inverted": "sensor.grid_power"},
}
],
"cost_adjustment_day": 0,
}
],
"device_consumption": [],
"device_consumption_water": [],
}
old_store = storage.Store(hass, 1, "energy", minor_version=2)
await old_store.async_save(old_data)
manager = EnergyManager(hass)
await manager.async_initialize()
assert manager.data is not None
grid = manager.data["energy_sources"][0]
# Verify power_config is preserved
assert grid["power_config"] == {"stat_rate_inverted": "sensor.grid_power"}
# Migration preserves the original stat_rate value from the legacy power array
# (stat_rate regeneration from power_config only happens in async_update)
assert grid["stat_rate"] == "sensor.grid_power"
async def test_grid_migration_import_only(hass: HomeAssistant) -> None:
"""Test migration with imports but no exports creates import-only grids."""
old_data = {
"energy_sources": [
{
"type": "grid",
"flow_from": [
{"stat_energy_from": "sensor.grid_import"},
],
"flow_to": [],
"cost_adjustment_day": 0,
}
],
"device_consumption": [],
"device_consumption_water": [],
}
old_store = storage.Store(hass, 1, "energy", minor_version=2)
await old_store.async_save(old_data)
manager = EnergyManager(hass)
await manager.async_initialize()
assert manager.data is not None
assert len(manager.data["energy_sources"]) == 1
grid = manager.data["energy_sources"][0]
assert grid["stat_energy_from"] == "sensor.grid_import"
assert grid["stat_energy_to"] is None
async def test_grid_migration_power_only(hass: HomeAssistant) -> None:
"""Test migration with only power configured (no import/export meters)."""
old_data = {
"energy_sources": [
{
"type": "grid",
"flow_from": [],
"flow_to": [],
"power": [
{"stat_rate": "sensor.grid_power"},
],
"cost_adjustment_day": 0.5,
}
],
"device_consumption": [],
"device_consumption_water": [],
}
old_store = storage.Store(hass, 1, "energy", minor_version=2)
await old_store.async_save(old_data)
manager = EnergyManager(hass)
await manager.async_initialize()
assert manager.data is not None
assert len(manager.data["energy_sources"]) == 1
grid = manager.data["energy_sources"][0]
assert grid["type"] == "grid"
# No import or export meters
assert grid["stat_energy_from"] is None
assert grid["stat_energy_to"] is None
# Power is preserved
assert grid["stat_rate"] == "sensor.grid_power"
assert grid["cost_adjustment_day"] == 0.5
async def test_grid_new_format_no_migration_needed(hass: HomeAssistant) -> None:
"""Test that new format data doesn't get migrated."""
new_data = {
"energy_sources": [
{
"type": "grid",
"stat_energy_from": "sensor.grid_import",
"stat_energy_to": "sensor.grid_export",
"stat_cost": None,
"stat_compensation": None,
"entity_energy_price": None,
"number_energy_price": 0.15,
"entity_energy_price_export": None,
"number_energy_price_export": 0.08,
"cost_adjustment_day": 0,
}
],
"device_consumption": [],
"device_consumption_water": [],
}
# Save with current version (1.3)
old_store = storage.Store(hass, 1, "energy", minor_version=3)
await old_store.async_save(new_data)
manager = EnergyManager(hass)
await manager.async_initialize()
assert manager.data is not None
assert len(manager.data["energy_sources"]) == 1
grid = manager.data["energy_sources"][0]
assert grid["stat_energy_from"] == "sensor.grid_import"
assert grid["stat_energy_to"] == "sensor.grid_export"
async def test_grid_validation_single_import_price() -> None:
"""Test that grid validation rejects both entity and number import price."""
with pytest.raises(
vol.Invalid, match="Define either an entity or a fixed number for import price"
):
ENERGY_SOURCE_SCHEMA(
[
{
"type": "grid",
"stat_energy_from": "sensor.grid_import",
"entity_energy_price": "sensor.price",
"number_energy_price": 0.15,
"cost_adjustment_day": 0,
}
]
)
async def test_grid_validation_single_export_price() -> None:
"""Test that grid validation rejects both entity and number export price."""
with pytest.raises(
vol.Invalid, match="Define either an entity or a fixed number for export price"
):
ENERGY_SOURCE_SCHEMA(
[
{
"type": "grid",
"stat_energy_from": "sensor.grid_import",
"stat_energy_to": "sensor.grid_export",
"entity_energy_price_export": "sensor.sell_price",
"number_energy_price_export": 0.08,
"cost_adjustment_day": 0,
}
]
)
tests/components/energy/test_sensor.py
+114 -105
View File
@@ -149,13 +149,35 @@ async def test_cost_sensor_attributes(
("price_entity", "fixed_price"), [("sensor.energy_price", None), (None, 1)]
)
@pytest.mark.parametrize(
("usage_sensor_entity_id", "cost_sensor_entity_id", "flow_type"),
(
"usage_sensor_entity_id",
"cost_sensor_entity_id",
"flow_type",
"energy_source_data",
"price_update_key",
),
[
("sensor.energy_consumption", "sensor.energy_consumption_cost", "flow_from"),
(
"sensor.energy_consumption",
"sensor.energy_consumption_cost",
"flow_from",
{
"type": "grid",
"stat_energy_from": "sensor.energy_consumption",
"cost_adjustment_day": 0,
},
"number_energy_price",
),
(
"sensor.energy_production",
"sensor.energy_production_compensation",
"flow_to",
{
"type": "grid",
"stat_energy_to": "sensor.energy_production",
"cost_adjustment_day": 0,
},
"number_energy_price_export",
),
],
)
@@ -173,6 +195,8 @@ async def test_cost_sensor_price_entity_total_increasing(
usage_sensor_entity_id,
cost_sensor_entity_id,
flow_type,
energy_source_data: dict[str, Any],
price_update_key: str,
) -> None:
"""Test energy cost price from total_increasing type sensor entity."""
@@ -188,32 +212,15 @@ async def test_cost_sensor_price_entity_total_increasing(
}
energy_data = data.EnergyManager.default_preferences()
energy_data["energy_sources"].append(
{
"type": "grid",
"flow_from": [
{
"stat_energy_from": "sensor.energy_consumption",
"stat_cost": None,
"entity_energy_price": price_entity,
"number_energy_price": fixed_price,
}
]
if flow_type == "flow_from"
else [],
"flow_to": [
{
"stat_energy_to": "sensor.energy_production",
"stat_compensation": None,
"entity_energy_price": price_entity,
"number_energy_price": fixed_price,
}
]
if flow_type == "flow_to"
else [],
"cost_adjustment_day": 0,
}
)
# Build energy source from test parameter data, adding price fields
energy_source = copy.deepcopy(energy_source_data)
if flow_type == "flow_from":
energy_source["entity_energy_price"] = price_entity
energy_source["number_energy_price"] = fixed_price
else:
energy_source["entity_energy_price_export"] = price_entity
energy_source["number_energy_price_export"] = fixed_price
energy_data["energy_sources"].append(energy_source)
hass_storage[data.STORAGE_KEY] = {
"version": 1,
@@ -282,7 +289,7 @@ async def test_cost_sensor_price_entity_total_increasing(
await hass.async_block_till_done()
else:
energy_data = copy.deepcopy(energy_data)
energy_data["energy_sources"][0][flow_type][0]["number_energy_price"] = 2
energy_data["energy_sources"][0][price_update_key] = 2
client = await hass_ws_client(hass)
await client.send_json({"id": 5, "type": "energy/save_prefs", **energy_data})
msg = await client.receive_json()
@@ -360,13 +367,35 @@ async def test_cost_sensor_price_entity_total_increasing(
("price_entity", "fixed_price"), [("sensor.energy_price", None), (None, 1)]
)
@pytest.mark.parametrize(
("usage_sensor_entity_id", "cost_sensor_entity_id", "flow_type"),
(
"usage_sensor_entity_id",
"cost_sensor_entity_id",
"flow_type",
"energy_source_data",
"price_update_key",
),
[
("sensor.energy_consumption", "sensor.energy_consumption_cost", "flow_from"),
(
"sensor.energy_consumption",
"sensor.energy_consumption_cost",
"flow_from",
{
"type": "grid",
"stat_energy_from": "sensor.energy_consumption",
"cost_adjustment_day": 0,
},
"number_energy_price",
),
(
"sensor.energy_production",
"sensor.energy_production_compensation",
"flow_to",
{
"type": "grid",
"stat_energy_to": "sensor.energy_production",
"cost_adjustment_day": 0,
},
"number_energy_price_export",
),
],
)
@@ -385,6 +414,8 @@ async def test_cost_sensor_price_entity_total(
usage_sensor_entity_id,
cost_sensor_entity_id,
flow_type,
energy_source_data: dict[str, Any],
price_update_key: str,
energy_state_class,
) -> None:
"""Test energy cost price from total type sensor entity."""
@@ -401,32 +432,15 @@ async def test_cost_sensor_price_entity_total(
}
energy_data = data.EnergyManager.default_preferences()
energy_data["energy_sources"].append(
{
"type": "grid",
"flow_from": [
{
"stat_energy_from": "sensor.energy_consumption",
"stat_cost": None,
"entity_energy_price": price_entity,
"number_energy_price": fixed_price,
}
]
if flow_type == "flow_from"
else [],
"flow_to": [
{
"stat_energy_to": "sensor.energy_production",
"stat_compensation": None,
"entity_energy_price": price_entity,
"number_energy_price": fixed_price,
}
]
if flow_type == "flow_to"
else [],
"cost_adjustment_day": 0,
}
)
# Build energy source from test parameter data, adding price fields
energy_source = copy.deepcopy(energy_source_data)
if flow_type == "flow_from":
energy_source["entity_energy_price"] = price_entity
energy_source["number_energy_price"] = fixed_price
else:
energy_source["entity_energy_price_export"] = price_entity
energy_source["number_energy_price_export"] = fixed_price
energy_data["energy_sources"].append(energy_source)
hass_storage[data.STORAGE_KEY] = {
"version": 1,
@@ -496,7 +510,7 @@ async def test_cost_sensor_price_entity_total(
await hass.async_block_till_done()
else:
energy_data = copy.deepcopy(energy_data)
energy_data["energy_sources"][0][flow_type][0]["number_energy_price"] = 2
energy_data["energy_sources"][0][price_update_key] = 2
client = await hass_ws_client(hass)
await client.send_json({"id": 5, "type": "energy/save_prefs", **energy_data})
msg = await client.receive_json()
@@ -575,13 +589,35 @@ async def test_cost_sensor_price_entity_total(
("price_entity", "fixed_price"), [("sensor.energy_price", None), (None, 1)]
)
@pytest.mark.parametrize(
("usage_sensor_entity_id", "cost_sensor_entity_id", "flow_type"),
(
"usage_sensor_entity_id",
"cost_sensor_entity_id",
"flow_type",
"energy_source_data",
"price_update_key",
),
[
("sensor.energy_consumption", "sensor.energy_consumption_cost", "flow_from"),
(
"sensor.energy_consumption",
"sensor.energy_consumption_cost",
"flow_from",
{
"type": "grid",
"stat_energy_from": "sensor.energy_consumption",
"cost_adjustment_day": 0,
},
"number_energy_price",
),
(
"sensor.energy_production",
"sensor.energy_production_compensation",
"flow_to",
{
"type": "grid",
"stat_energy_to": "sensor.energy_production",
"cost_adjustment_day": 0,
},
"number_energy_price_export",
),
],
)
@@ -600,6 +636,8 @@ async def test_cost_sensor_price_entity_total_no_reset(
usage_sensor_entity_id,
cost_sensor_entity_id,
flow_type,
energy_source_data: dict[str, Any],
price_update_key: str,
energy_state_class,
) -> None:
"""Test energy cost price from total type sensor entity with no last_reset."""
@@ -616,32 +654,15 @@ async def test_cost_sensor_price_entity_total_no_reset(
}
energy_data = data.EnergyManager.default_preferences()
energy_data["energy_sources"].append(
{
"type": "grid",
"flow_from": [
{
"stat_energy_from": "sensor.energy_consumption",
"stat_cost": None,
"entity_energy_price": price_entity,
"number_energy_price": fixed_price,
}
]
if flow_type == "flow_from"
else [],
"flow_to": [
{
"stat_energy_to": "sensor.energy_production",
"stat_compensation": None,
"entity_energy_price": price_entity,
"number_energy_price": fixed_price,
}
]
if flow_type == "flow_to"
else [],
"cost_adjustment_day": 0,
}
)
# Build energy source from test parameter data, adding price fields
energy_source = copy.deepcopy(energy_source_data)
if flow_type == "flow_from":
energy_source["entity_energy_price"] = price_entity
energy_source["number_energy_price"] = fixed_price
else:
energy_source["entity_energy_price_export"] = price_entity
energy_source["number_energy_price_export"] = fixed_price
energy_data["energy_sources"].append(energy_source)
hass_storage[data.STORAGE_KEY] = {
"version": 1,
@@ -710,7 +731,7 @@ async def test_cost_sensor_price_entity_total_no_reset(
await hass.async_block_till_done()
else:
energy_data = copy.deepcopy(energy_data)
energy_data["energy_sources"][0][flow_type][0]["number_energy_price"] = 2
energy_data["energy_sources"][0][price_update_key] = 2
client = await hass_ws_client(hass)
await client.send_json({"id": 5, "type": "energy/save_prefs", **energy_data})
msg = await client.receive_json()
@@ -1490,24 +1511,12 @@ async def test_power_sensor_grid_combined(
"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",
}
}
],
"stat_energy_from": "sensor.grid_energy_import",
"stat_energy_to": "sensor.grid_energy_export",
"power_config": {
"stat_rate_from": "sensor.grid_import",
"stat_rate_to": "sensor.grid_export",
},
"cost_adjustment_day": 0,
}
],
tests/components/energy/test_validate.py
+18 -49
View File
@@ -371,19 +371,11 @@ async def test_validation_grid(
"energy_sources": [
{
"type": "grid",
"flow_from": [
{
"stat_energy_from": "sensor.grid_consumption_1",
"stat_cost": "sensor.grid_cost_1",
}
],
"flow_to": [
{
"stat_energy_to": "sensor.grid_production_1",
"stat_compensation": "sensor.grid_compensation_1",
}
],
"power": [],
"stat_energy_from": "sensor.grid_consumption_1",
"stat_energy_to": "sensor.grid_production_1",
"stat_cost": "sensor.grid_cost_1",
"stat_compensation": "sensor.grid_compensation_1",
"cost_adjustment_day": 0.0,
}
]
}
@@ -464,19 +456,11 @@ async def test_validation_grid_external_cost_compensation(
"energy_sources": [
{
"type": "grid",
"flow_from": [
{
"stat_energy_from": "sensor.grid_consumption_1",
"stat_cost": "external:grid_cost_1",
}
],
"flow_to": [
{
"stat_energy_to": "sensor.grid_production_1",
"stat_compensation": "external:grid_compensation_1",
}
],
"power": [],
"stat_energy_from": "sensor.grid_consumption_1",
"stat_energy_to": "sensor.grid_production_1",
"stat_cost": "external:grid_cost_1",
"stat_compensation": "external:grid_compensation_1",
"cost_adjustment_day": 0.0,
}
]
}
@@ -559,20 +543,11 @@ async def test_validation_grid_price_not_exist(
"energy_sources": [
{
"type": "grid",
"flow_from": [
{
"stat_energy_from": "sensor.grid_consumption_1",
"entity_energy_price": "sensor.grid_price_1",
"number_energy_price": None,
}
],
"flow_to": [
{
"stat_energy_to": "sensor.grid_production_1",
"entity_energy_price": None,
"number_energy_price": 0.10,
}
],
"stat_energy_from": "sensor.grid_consumption_1",
"stat_energy_to": "sensor.grid_production_1",
"entity_energy_price": "sensor.grid_price_1",
"number_energy_price_export": 0.10,
"cost_adjustment_day": 0.0,
}
]
}
@@ -710,15 +685,9 @@ async def test_validation_grid_price_errors(
"energy_sources": [
{
"type": "grid",
"flow_from": [
{
"stat_energy_from": "sensor.grid_consumption_1",
"entity_energy_price": "sensor.grid_price_1",
"number_energy_price": None,
}
],
"flow_to": [],
"power": [],
"stat_energy_from": "sensor.grid_consumption_1",
"entity_energy_price": "sensor.grid_price_1",
"cost_adjustment_day": 0.0,
}
]
}
tests/components/energy/test_validate_power.py
+18 -75
View File
@@ -27,13 +27,7 @@ async def test_validation_grid_power_valid(
"energy_sources": [
{
"type": "grid",
"flow_from": [],
"flow_to": [],
"power": [
{
"stat_rate": "sensor.grid_power",
}
],
"stat_rate": "sensor.grid_power",
"cost_adjustment_day": 0.0,
}
]
@@ -66,13 +60,7 @@ async def test_validation_grid_power_wrong_unit(
"energy_sources": [
{
"type": "grid",
"flow_from": [],
"flow_to": [],
"power": [
{
"stat_rate": "sensor.grid_power",
}
],
"stat_rate": "sensor.grid_power",
"cost_adjustment_day": 0.0,
}
]
@@ -113,13 +101,7 @@ async def test_validation_grid_power_wrong_state_class(
"energy_sources": [
{
"type": "grid",
"flow_from": [],
"flow_to": [],
"power": [
{
"stat_rate": "sensor.grid_power",
}
],
"stat_rate": "sensor.grid_power",
"cost_adjustment_day": 0.0,
}
]
@@ -160,13 +142,7 @@ async def test_validation_grid_power_entity_missing(
"energy_sources": [
{
"type": "grid",
"flow_from": [],
"flow_to": [],
"power": [
{
"stat_rate": "sensor.missing_power",
}
],
"stat_rate": "sensor.missing_power",
"cost_adjustment_day": 0.0,
}
]
@@ -203,13 +179,7 @@ async def test_validation_grid_power_entity_unavailable(
"energy_sources": [
{
"type": "grid",
"flow_from": [],
"flow_to": [],
"power": [
{
"stat_rate": "sensor.unavailable_power",
}
],
"stat_rate": "sensor.unavailable_power",
"cost_adjustment_day": 0.0,
}
]
@@ -242,13 +212,7 @@ async def test_validation_grid_power_entity_non_numeric(
"energy_sources": [
{
"type": "grid",
"flow_from": [],
"flow_to": [],
"power": [
{
"stat_rate": "sensor.non_numeric_power",
}
],
"stat_rate": "sensor.non_numeric_power",
"cost_adjustment_day": 0.0,
}
]
@@ -289,13 +253,7 @@ async def test_validation_grid_power_wrong_device_class(
"energy_sources": [
{
"type": "grid",
"flow_from": [],
"flow_to": [],
"power": [
{
"stat_rate": "sensor.wrong_device_class_power",
}
],
"stat_rate": "sensor.wrong_device_class_power",
"cost_adjustment_day": 0.0,
}
]
@@ -333,23 +291,20 @@ async def test_validation_grid_power_different_units(
hass: HomeAssistant, mock_energy_manager, mock_get_metadata
) -> None:
"""Test validating grid with power sensors using different valid units."""
# With unified format, each grid has one power sensor, so we use two grids
await mock_energy_manager.async_update(
{
"energy_sources": [
{
"type": "grid",
"flow_from": [],
"flow_to": [],
"power": [
{
"stat_rate": "sensor.power_watt",
},
{
"stat_rate": "sensor.power_milliwatt",
},
],
"stat_rate": "sensor.power_watt",
"cost_adjustment_day": 0.0,
}
},
{
"type": "grid",
"stat_rate": "sensor.power_milliwatt",
"cost_adjustment_day": 0.0,
},
]
}
)
@@ -374,7 +329,7 @@ async def test_validation_grid_power_different_units(
result = await validate.async_validate(hass)
assert result.as_dict() == {
"energy_sources": [[]],
"energy_sources": [[], []],
"device_consumption": [],
"device_consumption_water": [],
}
@@ -391,13 +346,7 @@ async def test_validation_grid_power_external_statistics(
"energy_sources": [
{
"type": "grid",
"flow_from": [],
"flow_to": [],
"power": [
{
"stat_rate": "external:power_stat",
}
],
"stat_rate": "external:power_stat",
"cost_adjustment_day": 0.0,
}
]
@@ -431,13 +380,7 @@ async def test_validation_grid_power_recorder_untracked(
"energy_sources": [
{
"type": "grid",
"flow_from": [],
"flow_to": [],
"power": [
{
"stat_rate": "sensor.untracked_power",
}
],
"stat_rate": "sensor.untracked_power",
"cost_adjustment_day": 0.0,
}
]
tests/components/energy/test_websocket_api.py
+40 -60
View File
@@ -107,41 +107,31 @@ async def test_save_preferences(
new_prefs = {
"energy_sources": [
# Grid 1: heat_pump_meter paired with return_to_grid_peak + power
{
"type": "grid",
"flow_from": [
{
"stat_energy_from": "sensor.heat_pump_meter",
"stat_cost": "heat_pump_kwh_cost",
"entity_energy_price": None,
"number_energy_price": None,
},
{
"stat_energy_from": "sensor.heat_pump_meter_2",
"stat_cost": None,
"entity_energy_price": None,
"number_energy_price": 0.20,
},
],
"flow_to": [
{
"stat_energy_to": "sensor.return_to_grid_peak",
"stat_compensation": None,
"entity_energy_price": None,
"number_energy_price": None,
},
{
"stat_energy_to": "sensor.return_to_grid_offpeak",
"stat_compensation": None,
"entity_energy_price": None,
"number_energy_price": 0.20,
},
],
"power": [
{
"stat_rate": "sensor.grid_power",
}
],
"stat_energy_from": "sensor.heat_pump_meter",
"stat_energy_to": "sensor.return_to_grid_peak",
"stat_cost": "heat_pump_kwh_cost",
"stat_compensation": None,
"entity_energy_price": None,
"number_energy_price": None,
"entity_energy_price_export": None,
"number_energy_price_export": None,
"stat_rate": "sensor.grid_power",
"cost_adjustment_day": 1.2,
},
# Grid 2: heat_pump_meter_2 paired with return_to_grid_offpeak
{
"type": "grid",
"stat_energy_from": "sensor.heat_pump_meter_2",
"stat_energy_to": "sensor.return_to_grid_offpeak",
"stat_cost": None,
"stat_compensation": None,
"entity_energy_price": None,
"number_energy_price": 0.20,
"entity_energy_price_export": None,
"number_energy_price_export": 0.20,
"cost_adjustment_day": 1.2,
},
{
@@ -206,20 +196,19 @@ async def test_save_preferences(
"solar_forecast_domains": ["some_domain"],
}
# Prefs with limited options
# Prefs with limited options (defaults will be applied by schema)
new_prefs_2 = {
"energy_sources": [
{
"type": "grid",
"flow_from": [
{
"stat_energy_from": "sensor.heat_pump_meter",
"stat_cost": None,
"entity_energy_price": None,
"number_energy_price": None,
}
],
"flow_to": [],
"stat_energy_from": "sensor.heat_pump_meter",
"stat_energy_to": None,
"stat_cost": None,
"stat_compensation": None,
"entity_energy_price": None,
"number_energy_price": None,
"entity_energy_price_export": None,
"number_energy_price_export": None,
"cost_adjustment_day": 1.2,
},
{
@@ -242,9 +231,10 @@ async def test_save_preferences(
async def test_handle_duplicate_from_stat(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
"""Test we handle duplicate from stats."""
"""Test we handle duplicate from stats across multiple grid sources."""
client = await hass_ws_client(hass)
# Try to create two grids with the same import meter
await client.send_json(
{
"id": 5,
@@ -252,22 +242,12 @@ async def test_handle_duplicate_from_stat(
"energy_sources": [
{
"type": "grid",
"flow_from": [
{
"stat_energy_from": "sensor.heat_pump_meter",
"stat_cost": None,
"entity_energy_price": None,
"number_energy_price": None,
},
{
"stat_energy_from": "sensor.heat_pump_meter",
"stat_cost": None,
"entity_energy_price": None,
"number_energy_price": None,
},
],
"flow_to": [],
"power": [],
"stat_energy_from": "sensor.heat_pump_meter",
"cost_adjustment_day": 0,
},
{
"type": "grid",
"stat_energy_from": "sensor.heat_pump_meter",
"cost_adjustment_day": 0,
},
],