evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add individual battery banks as devices (#108339)

Browse source
This commit is contained in:
Josh Pettersen 2024-01-30 21:03:01 -08:00 • committed by GitHub
parent 5fd6028d97
commit b629ad9c3d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 299 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

4
homeassistant/components/powerwall/__init__.py
View file

@ -229,6 +229,7 @@ async def _call_base_info(power_wall: Powerwall, host: str) -> PowerwallBaseInfo
status = tg.create_task(power_wall.get_status())
device_type = tg.create_task(power_wall.get_device_type())
serial_numbers = tg.create_task(power_wall.get_serial_numbers())
batteries = tg.create_task(power_wall.get_batteries())
# Mimic the behavior of asyncio.gather by reraising the first caught exception since
# this is what is expected by the caller of this method
@ -248,6 +249,7 @@ async def _call_base_info(power_wall: Powerwall, host: str) -> PowerwallBaseInfo
device_type=device_type.result(),
serial_numbers=sorted(serial_numbers.result()),
url=f"https://{host}",
batteries={battery.serial_number: battery for battery in batteries.result()},
)
@ -270,6 +272,7 @@ async def _fetch_powerwall_data(power_wall: Powerwall) -> PowerwallData:
meters = tg.create_task(power_wall.get_meters())
grid_services_active = tg.create_task(power_wall.is_grid_services_active())
grid_status = tg.create_task(power_wall.get_grid_status())
batteries = tg.create_task(power_wall.get_batteries())
# Mimic the behavior of asyncio.gather by reraising the first caught exception since
# this is what is expected by the caller of this method
@ -287,6 +290,7 @@ async def _fetch_powerwall_data(power_wall: Powerwall) -> PowerwallData:
grid_services_active=grid_services_active.result(),
grid_status=grid_status.result(),
backup_reserve=backup_reserve.result(),
batteries={battery.serial_number: battery for battery in batteries.result()},
)

35
homeassistant/components/powerwall/entity.py
View file

@ -14,7 +14,7 @@ from .const import (
POWERWALL_BASE_INFO,
POWERWALL_COORDINATOR,
)
from .models import PowerwallData, PowerwallRuntimeData
from .models import BatteryResponse, PowerwallData, PowerwallRuntimeData
class PowerWallEntity(CoordinatorEntity[DataUpdateCoordinator[PowerwallData]]):
@ -43,3 +43,36 @@ class PowerWallEntity(CoordinatorEntity[DataUpdateCoordinator[PowerwallData]]):
def data(self) -> PowerwallData:
"""Return the coordinator data."""
return self.coordinator.data
class BatteryEntity(CoordinatorEntity[DataUpdateCoordinator[PowerwallData]]):
"""Base class for battery entities."""
_attr_has_entity_name = True
def __init__(
self, powerwall_data: PowerwallRuntimeData, battery: BatteryResponse
) -> None:
"""Initialize the entity."""
base_info = powerwall_data[POWERWALL_BASE_INFO]
coordinator = powerwall_data[POWERWALL_COORDINATOR]
assert coordinator is not None
super().__init__(coordinator)
self.serial_number = battery.serial_number
self.power_wall = powerwall_data[POWERWALL_API]
self.base_unique_id = f"{base_info.gateway_din}_{battery.serial_number}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, self.base_unique_id)},
manufacturer=MANUFACTURER,
model=f"{MODEL} ({battery.part_number})",
name=f"{base_info.site_info.site_name} {battery.serial_number}",
sw_version=base_info.status.version,
configuration_url=base_info.url,
via_device=(DOMAIN, base_info.gateway_din),
)
@property
def battery_data(self) -> BatteryResponse:
"""Return the coordinator data."""
return self.coordinator.data.batteries[self.serial_number]

3
homeassistant/components/powerwall/models.py
View file

@ -5,6 +5,7 @@ from dataclasses import dataclass
from typing import TypedDict
from tesla_powerwall import (
BatteryResponse,
DeviceType,
GridStatus,
MetersAggregatesResponse,
@ -27,6 +28,7 @@ class PowerwallBaseInfo:
device_type: DeviceType
serial_numbers: list[str]
url: str
batteries: dict[str, BatteryResponse]
@dataclass
@ -39,6 +41,7 @@ class PowerwallData:
grid_services_active: bool
grid_status: GridStatus
backup_reserve: float | None
batteries: dict[str, BatteryResponse]
class PowerwallRuntimeData(TypedDict):

148
homeassistant/components/powerwall/sensor.py
View file

@ -5,7 +5,7 @@ from collections.abc import Callable
from dataclasses import dataclass
from typing import TYPE_CHECKING, Generic, TypeVar
from tesla_powerwall import MeterResponse, MeterType
from tesla_powerwall import GridState, MeterResponse, MeterType
from homeassistant.components.sensor import (
SensorDeviceClass,
@ -16,6 +16,7 @@ from homeassistant.components.sensor import (
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
PERCENTAGE,
EntityCategory,
UnitOfElectricCurrent,
UnitOfElectricPotential,
UnitOfEnergy,
@ -27,14 +28,14 @@ from homeassistant.helpers.entity import Entity
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN, POWERWALL_COORDINATOR
from .entity import PowerWallEntity
from .models import PowerwallRuntimeData
from .entity import BatteryEntity, PowerWallEntity
from .models import BatteryResponse, PowerwallRuntimeData
_METER_DIRECTION_EXPORT = "export"
_METER_DIRECTION_IMPORT = "import"
_ValueParamT = TypeVar("_ValueParamT")
_ValueT = TypeVar("_ValueT", bound=float)
_ValueT = TypeVar("_ValueT", bound=float | int | str)
@dataclass(frozen=True)
@ -112,6 +113,116 @@ POWERWALL_INSTANT_SENSORS = (
)
def _get_battery_charge(battery_data: BatteryResponse) -> float:
"""Get the current value in %."""
ratio = float(battery_data.energy_remaining) / float(battery_data.capacity)
return round(100 * ratio, 1)
BATTERY_INSTANT_SENSORS: list[PowerwallSensorEntityDescription] = [
PowerwallSensorEntityDescription[BatteryResponse, int](
key="battery_capacity",
translation_key="battery_capacity",
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.ENERGY_STORAGE,
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
suggested_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
suggested_display_precision=1,
value_fn=lambda battery_data: battery_data.capacity,
),
PowerwallSensorEntityDescription[BatteryResponse, float](
key="battery_instant_voltage",
translation_key="battery_instant_voltage",
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.VOLTAGE,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
value_fn=lambda battery_data: round(battery_data.v_out, 1),
),
PowerwallSensorEntityDescription[BatteryResponse, float](
key="instant_frequency",
translation_key="instant_frequency",
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.FREQUENCY,
native_unit_of_measurement=UnitOfFrequency.HERTZ,
entity_registry_enabled_default=False,
value_fn=lambda battery_data: round(battery_data.f_out, 1),
),
PowerwallSensorEntityDescription[BatteryResponse, float](
key="instant_current",
translation_key="instant_current",
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.CURRENT,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
entity_registry_enabled_default=False,
value_fn=lambda battery_data: round(battery_data.i_out, 1),
),
PowerwallSensorEntityDescription[BatteryResponse, int](
key="instant_power",
translation_key="instant_power",
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.POWER,
native_unit_of_measurement=UnitOfPower.WATT,
value_fn=lambda battery_data: battery_data.p_out,
),
PowerwallSensorEntityDescription[BatteryResponse, float](
key="battery_export",
translation_key="battery_export",
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.TOTAL_INCREASING,
device_class=SensorDeviceClass.ENERGY,
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
suggested_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
suggested_display_precision=0,
value_fn=lambda battery_data: battery_data.energy_discharged,
),
PowerwallSensorEntityDescription[BatteryResponse, float](
key="battery_import",
translation_key="battery_import",
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.TOTAL_INCREASING,
device_class=SensorDeviceClass.ENERGY,
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
suggested_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
suggested_display_precision=0,
value_fn=lambda battery_data: battery_data.energy_charged,
),
PowerwallSensorEntityDescription[BatteryResponse, int](
key="battery_remaining",
translation_key="battery_remaining",
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.ENERGY_STORAGE,
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
suggested_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
suggested_display_precision=1,
value_fn=lambda battery_data: battery_data.energy_remaining,
),
PowerwallSensorEntityDescription[BatteryResponse, float](
key="charge",
translation_key="charge",
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.BATTERY,
native_unit_of_measurement=PERCENTAGE,
suggested_display_precision=0,
value_fn=_get_battery_charge,
),
PowerwallSensorEntityDescription[BatteryResponse, str](
key="grid_state",
translation_key="grid_state",
entity_category=EntityCategory.DIAGNOSTIC,
device_class=SensorDeviceClass.ENUM,
options=[state.value.lower() for state in GridState],
value_fn=lambda battery_data: battery_data.grid_state.value.lower(),
),
]
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
@ -137,6 +248,12 @@ async def async_setup_entry(
for description in POWERWALL_INSTANT_SENSORS
)
for battery in data.batteries.values():
entities.extend(
PowerWallBatterySensor(powerwall_data, battery, description)
for description in BATTERY_INSTANT_SENSORS
)
async_add_entities(entities)
@ -281,3 +398,26 @@ class PowerWallImportSensor(PowerWallEnergyDirectionSensor):
if TYPE_CHECKING:
assert meter is not None
return meter.get_energy_imported()
class PowerWallBatterySensor(BatteryEntity, SensorEntity, Generic[_ValueT]):
"""Representation of an Powerwall Battery sensor."""
entity_description: PowerwallSensorEntityDescription[BatteryResponse, _ValueT]
def __init__(
self,
powerwall_data: PowerwallRuntimeData,
battery: BatteryResponse,
description: PowerwallSensorEntityDescription[BatteryResponse, _ValueT],
) -> None:
"""Initialize the sensor."""
self.entity_description = description
super().__init__(powerwall_data, battery)
self._attr_translation_key = description.translation_key
self._attr_unique_id = f"{self.base_unique_id}_{description.key}"
@property
def native_value(self) -> float | int | str:
"""Get the current value."""
return self.entity_description.value_fn(self.battery_data)

14
homeassistant/components/powerwall/strings.json
View file

@ -146,6 +146,20 @@
"battery_export": {
"name": "Battery export"
},
"battery_capacity": {
"name": "Battery capacity"
},
"battery_remaining": {
"name": "Battery remaining"
},
"grid_state": {
"name": "Grid state",
"state": {
"grid_compliant": "Compliant",
"grid_qualifying": "Qualifying",
"grid_uncompliant": "Uncompliant"
}
},
"load_import": {
"name": "Load import"
},

32
tests/components/powerwall/fixtures/batteries.json Normal file
View file

@ -0,0 +1,32 @@
[
{
"PackagePartNumber": "3012170-05-C",
"PackageSerialNumber": "TG0123456789AB",
"energy_charged": 2693355,
"energy_discharged": 2358235,
"nominal_energy_remaining": 14715,
"nominal_full_pack_energy": 14715,
"wobble_detected": false,
"p_out": -100,
"q_out": -1080,
"v_out": 245.70000000000002,
"f_out": 50.037,
"i_out": 0.30000000000000004,
"pinv_grid_state": "Grid_Compliant"
},
{
"PackagePartNumber": "3012170-05-C",
"PackageSerialNumber": "TG9876543210BA",
"energy_charged": 610483,
"energy_discharged": 509907,
"nominal_energy_remaining": 15137,
"nominal_full_pack_energy": 15137,
"wobble_detected": false,
"p_out": -100,
"q_out": -1090,
"v_out": 245.60000000000002,
"f_out": 50.037,
"i_out": 0.1,
"pinv_grid_state": "Grid_Compliant"
}
]

7
tests/components/powerwall/mocks.py
View file

@ -6,6 +6,7 @@ import os
from unittest.mock import MagicMock
from tesla_powerwall import (
BatteryResponse,
DeviceType,
GridStatus,
MetersAggregatesResponse,
@ -29,6 +30,7 @@ async def _mock_powerwall_with_fixtures(hass, empty_meters: bool = False) -> Mag
site_info = tg.create_task(_async_load_json_fixture(hass, "site_info.json"))
status = tg.create_task(_async_load_json_fixture(hass, "status.json"))
device_type = tg.create_task(_async_load_json_fixture(hass, "device_type.json"))
batteries = tg.create_task(_async_load_json_fixture(hass, "batteries.json"))
return await _mock_powerwall_return_value(
site_info=SiteInfoResponse.from_dict(site_info.result()),
@ -41,6 +43,9 @@ async def _mock_powerwall_with_fixtures(hass, empty_meters: bool = False) -> Mag
device_type=DeviceType(device_type.result()["device_type"]),
serial_numbers=["TG0123456789AB", "TG9876543210BA"],
backup_reserve_percentage=15.0,
batteries=[
BatteryResponse.from_dict(battery) for battery in batteries.result()
],
)
@ -55,6 +60,7 @@ async def _mock_powerwall_return_value(
device_type=None,
serial_numbers=None,
backup_reserve_percentage=None,
batteries=None,
):
powerwall_mock = MagicMock(Powerwall)
powerwall_mock.__aenter__.return_value = powerwall_mock
@ -72,6 +78,7 @@ async def _mock_powerwall_return_value(
)
powerwall_mock.is_grid_services_active.return_value = grid_services_active
powerwall_mock.get_gateway_din.return_value = MOCK_GATEWAY_DIN
powerwall_mock.get_batteries.return_value = batteries
return powerwall_mock

61
tests/components/powerwall/test_sensor.py
View file

@ -132,6 +132,67 @@ async def test_sensors(
assert hass.states.get("sensor.mysite_load_frequency").state == STATE_UNKNOWN
assert hass.states.get("sensor.mysite_backup_reserve").state == STATE_UNKNOWN
assert (
float(hass.states.get("sensor.mysite_tg0123456789ab_battery_capacity").state)
== 14.715
)
assert (
float(hass.states.get("sensor.mysite_tg0123456789ab_battery_voltage").state)
== 245.7
)
assert (
float(hass.states.get("sensor.mysite_tg0123456789ab_frequency").state) == 50.0
)
assert float(hass.states.get("sensor.mysite_tg0123456789ab_current").state) == 0.3
assert int(hass.states.get("sensor.mysite_tg0123456789ab_power").state) == -100
assert (
float(hass.states.get("sensor.mysite_tg0123456789ab_battery_export").state)
== 2358.235
)
assert (
float(hass.states.get("sensor.mysite_tg0123456789ab_battery_import").state)
== 2693.355
)
assert (
float(hass.states.get("sensor.mysite_tg0123456789ab_battery_remaining").state)
== 14.715
)
assert float(hass.states.get("sensor.mysite_tg0123456789ab_charge").state) == 100.0
assert (
str(hass.states.get("sensor.mysite_tg0123456789ab_grid_state").state)
== "grid_compliant"
)
assert (
float(hass.states.get("sensor.mysite_tg9876543210ba_battery_capacity").state)
== 15.137
)
assert (
float(hass.states.get("sensor.mysite_tg9876543210ba_battery_voltage").state)
== 245.6
)
assert (
float(hass.states.get("sensor.mysite_tg9876543210ba_frequency").state) == 50.0
)
assert float(hass.states.get("sensor.mysite_tg9876543210ba_current").state) == 0.1
assert int(hass.states.get("sensor.mysite_tg9876543210ba_power").state) == -100
assert (
float(hass.states.get("sensor.mysite_tg9876543210ba_battery_export").state)
== 509.907
)
assert (
float(hass.states.get("sensor.mysite_tg9876543210ba_battery_import").state)
== 610.483
)
assert (
float(hass.states.get("sensor.mysite_tg9876543210ba_battery_remaining").state)
== 15.137
)
assert float(hass.states.get("sensor.mysite_tg9876543210ba_charge").state) == 100.0
assert (
str(hass.states.get("sensor.mysite_tg9876543210ba_grid_state").state)
== "grid_compliant"
)
async def test_sensor_backup_reserve_unavailable(hass: HomeAssistant) -> None:
"""Confirm that backup reserve sensor is not added if data is unavailable from the device."""