Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
143 changes: 102 additions & 41 deletions homeassistant/components/powerwall/sensor.py
Original file line number Diff line number Diff line change
@@ -1,35 +1,110 @@
"""Support for powerwall sensors."""
from __future__ import annotations

from typing import Any
from collections.abc import Callable
from dataclasses import dataclass

from tesla_powerwall import Meter, MeterType

from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ENERGY_KILO_WATT_HOUR, PERCENTAGE, POWER_KILO_WATT
from homeassistant.const import (
ELECTRIC_CURRENT_AMPERE,
ELECTRIC_POTENTIAL_VOLT,
ENERGY_KILO_WATT_HOUR,
FREQUENCY_HERTZ,
PERCENTAGE,
POWER_KILO_WATT,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback

from .const import (
ATTR_FREQUENCY,
ATTR_INSTANT_AVERAGE_VOLTAGE,
ATTR_INSTANT_TOTAL_CURRENT,
ATTR_IS_ACTIVE,
DOMAIN,
POWERWALL_COORDINATOR,
)
from .const import DOMAIN, POWERWALL_COORDINATOR
from .entity import PowerWallEntity
from .models import PowerwallData, PowerwallRuntimeData

_METER_DIRECTION_EXPORT = "export"
_METER_DIRECTION_IMPORT = "import"


@dataclass
class PowerwallRequiredKeysMixin:
"""Mixin for required keys."""

value_fn: Callable[[Meter], float]


@dataclass
class PowerwallSensorEntityDescription(
SensorEntityDescription, PowerwallRequiredKeysMixin
):
"""Describes Powerwall entity."""


def _get_meter_power(meter: Meter) -> float:
"""Get the current value in kW."""
return meter.get_power(precision=3)


def _get_meter_frequency(meter: Meter) -> float:
"""Get the current value in Hz."""
return round(meter.frequency, 1)


def _get_meter_total_current(meter: Meter) -> float:
"""Get the current value in A."""
return meter.get_instant_total_current()


def _get_meter_average_voltage(meter: Meter) -> float:
"""Get the current value in V."""
return round(meter.average_voltage, 1)


POWERWALL_INSTANT_SENSORS = (
PowerwallSensorEntityDescription(
key="instant_power",
name="Now",
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.POWER,
native_unit_of_measurement=POWER_KILO_WATT,
value_fn=_get_meter_power,
),
PowerwallSensorEntityDescription(
key="instant_frequency",
name="Frequency Now",
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.FREQUENCY,
native_unit_of_measurement=FREQUENCY_HERTZ,
entity_registry_enabled_default=False,
value_fn=_get_meter_frequency,
),
PowerwallSensorEntityDescription(
key="instant_current",
name="Average Current Now",
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.CURRENT,
native_unit_of_measurement=ELECTRIC_CURRENT_AMPERE,
entity_registry_enabled_default=False,
value_fn=_get_meter_total_current,
),
PowerwallSensorEntityDescription(
key="instant_voltage",
name="Average Voltage Now",
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.VOLTAGE,
native_unit_of_measurement=ELECTRIC_POTENTIAL_VOLT,
entity_registry_enabled_default=False,
value_fn=_get_meter_average_voltage,
),
)


async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
Expand All @@ -40,24 +115,17 @@ async def async_setup_entry(
coordinator = powerwall_data[POWERWALL_COORDINATOR]
assert coordinator is not None
data: PowerwallData = coordinator.data
entities: list[
PowerWallEnergySensor
| PowerWallImportSensor
| PowerWallExportSensor
| PowerWallChargeSensor
| PowerWallBackupReserveSensor
] = [
entities: list[PowerWallEntity] = [
PowerWallChargeSensor(powerwall_data),
PowerWallBackupReserveSensor(powerwall_data),
]

for meter in data.meters.meters:
entities.append(PowerWallExportSensor(powerwall_data, meter))
entities.append(PowerWallImportSensor(powerwall_data, meter))
entities.extend(
[
PowerWallEnergySensor(powerwall_data, meter),
PowerWallExportSensor(powerwall_data, meter),
PowerWallImportSensor(powerwall_data, meter),
]
PowerWallEnergySensor(powerwall_data, meter, description)
for description in POWERWALL_INSTANT_SENSORS
)

async_add_entities(entities)
Expand Down Expand Up @@ -85,34 +153,27 @@ def native_value(self) -> int:
class PowerWallEnergySensor(PowerWallEntity, SensorEntity):
"""Representation of an Powerwall Energy sensor."""

_attr_state_class = SensorStateClass.MEASUREMENT
_attr_native_unit_of_measurement = POWER_KILO_WATT
_attr_device_class = SensorDeviceClass.POWER
entity_description: PowerwallSensorEntityDescription

def __init__(self, powerwall_data: PowerwallRuntimeData, meter: MeterType) -> None:
def __init__(
self,
powerwall_data: PowerwallRuntimeData,
meter: MeterType,
description: PowerwallSensorEntityDescription,
) -> None:
"""Initialize the sensor."""
self.entity_description = description
super().__init__(powerwall_data)
self._meter = meter
self._attr_name = f"Powerwall {self._meter.value.title()} Now"
self._attr_name = f"Powerwall {self._meter.value.title()} {description.name}"
self._attr_unique_id = (
f"{self.base_unique_id}_{self._meter.value}_instant_power"
f"{self.base_unique_id}_{self._meter.value}_{description.key}"
)

@property
def native_value(self) -> float:
"""Get the current value in kW."""
return self.data.meters.get_meter(self._meter).get_power(precision=3)

@property
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the device specific state attributes."""
meter = self.data.meters.get_meter(self._meter)
return {
ATTR_FREQUENCY: round(meter.frequency, 1),
ATTR_INSTANT_AVERAGE_VOLTAGE: round(meter.average_voltage, 1),
ATTR_INSTANT_TOTAL_CURRENT: meter.get_instant_total_current(),
ATTR_IS_ACTIVE: meter.is_active(),
}
"""Get the current value."""
return self.entity_description.value_fn(self.data.meters.get_meter(self._meter))


class PowerWallBackupReserveSensor(PowerWallEntity, SensorEntity):
Expand Down
97 changes: 38 additions & 59 deletions tests/components/powerwall/test_sensor.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,15 +2,22 @@
from unittest.mock import patch

from homeassistant.components.powerwall.const import DOMAIN
from homeassistant.const import CONF_IP_ADDRESS, PERCENTAGE
from homeassistant.components.sensor import ATTR_STATE_CLASS
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_FRIENDLY_NAME,
ATTR_UNIT_OF_MEASUREMENT,
CONF_IP_ADDRESS,
PERCENTAGE,
)
from homeassistant.helpers import device_registry as dr

from .mocks import _mock_powerwall_with_fixtures

from tests.common import MockConfigEntry


async def test_sensors(hass):
async def test_sensors(hass, entity_registry_enabled_by_default):
"""Test creation of the sensors."""

mock_powerwall = await _mock_powerwall_with_fixtures(hass)
Expand All @@ -35,77 +42,49 @@ async def test_sensors(hass):
assert reg_device.manufacturer == "Tesla"
assert reg_device.name == "MySite"

state = hass.states.get("sensor.powerwall_site_now")
assert state.state == "0.032"
expected_attributes = {
"frequency": 60,
"instant_average_voltage": 120.7,
"unit_of_measurement": "kW",
"friendly_name": "Powerwall Site Now",
"device_class": "power",
"is_active": False,
}
# Only test for a subset of attributes in case
# HA changes the implementation and a new one appears
for key, value in expected_attributes.items():
assert state.attributes[key] == value

assert float(hass.states.get("sensor.powerwall_site_export").state) == 10429.5
assert float(hass.states.get("sensor.powerwall_site_import").state) == 4824.2

export_attributes = hass.states.get("sensor.powerwall_site_export").attributes
assert export_attributes["unit_of_measurement"] == "kWh"

state = hass.states.get("sensor.powerwall_load_now")
assert state.state == "1.971"
expected_attributes = {
"frequency": 60,
"instant_average_voltage": 120.7,
"unit_of_measurement": "kW",
"friendly_name": "Powerwall Load Now",
"device_class": "power",
"is_active": True,
}
# Only test for a subset of attributes in case
# HA changes the implementation and a new one appears
for key, value in expected_attributes.items():
assert state.attributes[key] == value
attributes = state.attributes
assert attributes[ATTR_DEVICE_CLASS] == "power"
assert attributes[ATTR_UNIT_OF_MEASUREMENT] == "kW"
assert attributes[ATTR_STATE_CLASS] == "measurement"
assert attributes[ATTR_FRIENDLY_NAME] == "Powerwall Load Now"

state = hass.states.get("sensor.powerwall_load_frequency_now")
assert state.state == "60"
attributes = state.attributes
assert attributes[ATTR_DEVICE_CLASS] == "frequency"
assert attributes[ATTR_UNIT_OF_MEASUREMENT] == "Hz"
assert attributes[ATTR_STATE_CLASS] == "measurement"
assert attributes[ATTR_FRIENDLY_NAME] == "Powerwall Load Frequency Now"

state = hass.states.get("sensor.powerwall_load_average_voltage_now")
assert state.state == "120.7"
attributes = state.attributes
assert attributes[ATTR_DEVICE_CLASS] == "voltage"
assert attributes[ATTR_UNIT_OF_MEASUREMENT] == "V"
assert attributes[ATTR_STATE_CLASS] == "measurement"
assert attributes[ATTR_FRIENDLY_NAME] == "Powerwall Load Average Voltage Now"

state = hass.states.get("sensor.powerwall_load_average_current_now")
assert state.state == "0"
attributes = state.attributes
assert attributes[ATTR_DEVICE_CLASS] == "current"
assert attributes[ATTR_UNIT_OF_MEASUREMENT] == "A"
assert attributes[ATTR_STATE_CLASS] == "measurement"
assert attributes[ATTR_FRIENDLY_NAME] == "Powerwall Load Average Current Now"

assert float(hass.states.get("sensor.powerwall_load_export").state) == 1056.8
assert float(hass.states.get("sensor.powerwall_load_import").state) == 4693.0

state = hass.states.get("sensor.powerwall_battery_now")
assert state.state == "-8.55"
expected_attributes = {
"frequency": 60.0,
"instant_average_voltage": 240.6,
"unit_of_measurement": "kW",
"friendly_name": "Powerwall Battery Now",
"device_class": "power",
"is_active": True,
}
# Only test for a subset of attributes in case
# HA changes the implementation and a new one appears
for key, value in expected_attributes.items():
assert state.attributes[key] == value

assert float(hass.states.get("sensor.powerwall_battery_export").state) == 3620.0
assert float(hass.states.get("sensor.powerwall_battery_import").state) == 4216.2

state = hass.states.get("sensor.powerwall_solar_now")
assert state.state == "10.49"
expected_attributes = {
"frequency": 60,
"instant_average_voltage": 120.7,
"unit_of_measurement": "kW",
"friendly_name": "Powerwall Solar Now",
"device_class": "power",
"is_active": True,
}
# Only test for a subset of attributes in case
# HA changes the implementation and a new one appears
for key, value in expected_attributes.items():
assert state.attributes[key] == value

assert float(hass.states.get("sensor.powerwall_solar_export").state) == 9864.2
assert float(hass.states.get("sensor.powerwall_solar_import").state) == 28.2
Expand Down