arian/home-assistant-core
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Split SolarEdge power-flow attributes into sensor entities (#170457)

Browse Source 
Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Mika
2026-05-19 15:42:36 +02:00
committed by GitHub
parent de70d9ed82
commit 7cf3cba27b
5 changed files with 182 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/solaredge/coordinator.py
+5 -3
View File
@@ -321,14 +321,16 @@ class SolarEdgePowerFlowDataService(SolarEdgeDataService):
export = key.lower() in power_to
if self.data[key]:
self.data[key] *= -1 if export else 1
self.attributes[key]["flow"] = "export" if export else "import"
self.data["grid_flow_direction"] = "export" if export else "import"
if key == "STORAGE":
charge = key.lower() in power_to
if self.data[key]:
self.data[key] *= -1 if charge else 1
self.attributes[key]["flow"] = "charge" if charge else "discharge"
self.attributes[key]["soc"] = value["chargeLevel"]
self.data["storage_flow_direction"] = (
"charge" if charge else "discharge"
)
self.data["storage_level"] = value["chargeLevel"]
LOGGER.debug("Updated SolarEdge power flow: %s, %s", self.data, self.attributes)
homeassistant/components/solaredge/icons.json
+6
View File
@@ -13,6 +13,9 @@
"energy_today": {
"default": "mdi:solar-power"
},
"grid_flow_direction": {
"default": "mdi:transmission-tower"
},
"grid_power": {
"default": "mdi:power-plug"
},
@@ -25,6 +28,9 @@
"solar_power": {
"default": "mdi:solar-power"
},
"storage_flow_direction": {
"default": "mdi:battery-charging"
},
"storage_power": {
"default": "mdi:car-battery"
}
homeassistant/components/solaredge/sensor.py
+19 -17
View File
@@ -152,6 +152,22 @@ SENSOR_TYPES = [
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.POWER,
),
SolarEdgeSensorEntityDescription(
key="grid_flow_direction",
json_key="grid_flow_direction",
translation_key="grid_flow_direction",
entity_registry_enabled_default=False,
device_class=SensorDeviceClass.ENUM,
options=["export", "import"],
),
SolarEdgeSensorEntityDescription(
key="storage_flow_direction",
json_key="storage_flow_direction",
translation_key="storage_flow_direction",
entity_registry_enabled_default=False,
device_class=SensorDeviceClass.ENUM,
options=["charge", "discharge"],
),
SolarEdgeSensorEntityDescription(
key="purchased_energy",
json_key="Purchased",
@@ -199,7 +215,7 @@ SENSOR_TYPES = [
),
SolarEdgeSensorEntityDescription(
key="storage_level",
json_key="STORAGE",
json_key="storage_level",
translation_key="storage_level",
entity_registry_enabled_default=False,
state_class=SensorStateClass.MEASUREMENT,
@@ -371,8 +387,8 @@ class SolarEdgeSensorFactory:
for key in ("power_consumption", "solar_power", "grid_power", "storage_power"):
self.services[key] = (SolarEdgePowerFlowSensor, flow)
for key in ("storage_level",):
self.services[key] = (SolarEdgeStorageLevelSensor, flow)
for key in ("storage_level", "grid_flow_direction", "storage_flow_direction"):
self.services[key] = (SolarEdgeOverviewSensor, flow)
for key in (
"purchased_energy",
@@ -560,20 +576,6 @@ class SolarEdgePowerFlowSensor(SolarEdgeSensorEntity):
return self.data_service.data.get(self.entity_description.json_key)
class SolarEdgeStorageLevelSensor(SolarEdgeSensorEntity):
"""Representation of an SolarEdge Monitoring API storage level sensor."""
_attr_device_class = SensorDeviceClass.BATTERY
@property
def native_value(self) -> str | None:
"""Return the state of the sensor."""
attr = self.data_service.attributes.get(self.entity_description.json_key)
if attr and "soc" in attr:
return attr["soc"]
return None
class SolarEdgeStorageDataSensor(SolarEdgeSensorEntity):
"""Representation of an SolarEdge aggregate storage data sensor."""
homeassistant/components/solaredge/strings.json
+14
View File
@@ -118,6 +118,13 @@
"gateways": {
"name": "Gateways"
},
"grid_flow_direction": {
"name": "Grid flow direction",
"state": {
"export": "Export",
"import": "Import"
}
},
"grid_power": {
"name": "Grid power"
},
@@ -157,6 +164,13 @@
"storage_discharge_energy": {
"name": "Storage discharge energy today"
},
"storage_flow_direction": {
"name": "Storage flow direction",
"state": {
"charge": "Charge",
"discharge": "Discharge"
}
},
"storage_level": {
"name": "Storage level"
},
tests/components/solaredge/test_sensor.py
+138
View File
@@ -699,3 +699,141 @@ async def test_storage_service_not_retried_after_recovery_with_no_batteries(
"sensor", DOMAIN, f"{SITE_ID}_storage_charge_energy"
)
assert charge_entry is None
def _power_flow_payload(
*,
grid_export: bool,
storage_charging: bool,
charge_level: int = 60,
) -> dict[str, dict]:
"""Build a siteCurrentPowerFlow payload for the given grid/storage flows."""
if grid_export:
grid_connection = {"from": "LOAD", "to": "Grid"}
else:
grid_connection = {"from": "GRID", "to": "Load"}
if storage_charging:
storage_connection = {"from": "PV", "to": "Storage"}
else:
storage_connection = {"from": "STORAGE", "to": "Load"}
return {
"siteCurrentPowerFlow": {
"unit": "W",
"connections": [grid_connection, storage_connection],
"GRID": {"status": "Active", "currentPower": 1200.0},
"LOAD": {"status": "Active", "currentPower": 800.0},
"PV": {"status": "Active", "currentPower": 2000.0},
"STORAGE": {
"status": "Charging" if storage_charging else "Discharging",
"currentPower": 300.0,
"chargeLevel": charge_level,
},
}
}
@pytest.mark.parametrize(
("grid_export", "storage_charging", "expected_grid", "expected_storage"),
[
(True, False, "export", "discharge"),
(False, True, "import", "charge"),
],
)
@patch("homeassistant.components.solaredge.SolarEdge")
async def test_power_flow_direction_sensors(
mock_solaredge: MagicMock,
recorder_mock: Recorder,
hass: HomeAssistant,
mock_solaredge_api: AsyncMock,
mock_config_entry: MockConfigEntry,
grid_export: bool,
storage_charging: bool,
expected_grid: str,
expected_storage: str,
) -> None:
"""Test that grid and storage flow direction ENUM sensors are populated."""
mock_solaredge_api.get_current_power_flow = AsyncMock(
return_value=_power_flow_payload(
grid_export=grid_export, storage_charging=storage_charging
)
)
mock_solaredge.return_value = mock_solaredge_api
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
grid_dir = hass.states.get("sensor.solaredge_grid_flow_direction")
assert grid_dir is not None
assert grid_dir.state == expected_grid
assert grid_dir.attributes["options"] == ["export", "import"]
assert grid_dir.attributes["device_class"] == "enum"
storage_dir = hass.states.get("sensor.solaredge_storage_flow_direction")
assert storage_dir is not None
assert storage_dir.state == expected_storage
assert storage_dir.attributes["options"] == ["charge", "discharge"]
assert storage_dir.attributes["device_class"] == "enum"
# Power sensors must no longer expose flow/soc attributes
grid_power = hass.states.get("sensor.solaredge_grid_power")
assert grid_power is not None
assert "flow" not in grid_power.attributes
storage_power = hass.states.get("sensor.solaredge_storage_power")
assert storage_power is not None
assert "flow" not in storage_power.attributes
assert "soc" not in storage_power.attributes
@patch("homeassistant.components.solaredge.SolarEdge")
async def test_storage_level_from_power_flow(
mock_solaredge: MagicMock,
recorder_mock: Recorder,
hass: HomeAssistant,
mock_solaredge_api: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test that the storage level sensor reads chargeLevel from power-flow data."""
mock_solaredge_api.get_current_power_flow = AsyncMock(
return_value=_power_flow_payload(
grid_export=False, storage_charging=True, charge_level=42
)
)
mock_solaredge.return_value = mock_solaredge_api
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get("sensor.solaredge_storage_level")
assert state is not None
assert state.state == "42"
@patch("homeassistant.components.solaredge.SolarEdge")
async def test_power_flow_direction_sensors_missing_connections(
mock_solaredge: MagicMock,
recorder_mock: Recorder,
hass: HomeAssistant,
mock_solaredge_api: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test direction sensors stay unknown when power flow has no connections."""
mock_solaredge.return_value = mock_solaredge_api
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
grid_dir = hass.states.get("sensor.solaredge_grid_flow_direction")
assert grid_dir is not None
assert grid_dir.state == "unknown"
storage_dir = hass.states.get("sensor.solaredge_storage_flow_direction")
assert storage_dir is not None
assert storage_dir.state == "unknown"
storage_level = hass.states.get("sensor.solaredge_storage_level")
assert storage_level is not None
assert storage_level.state == "unknown"