jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-22 20:57:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add Grid import export to Enphase Envoy (#110884)

Browse Source 
* Add Grid import export to enphase Envoy

* Update snapshot for labels dict element in entity registry

* use identity check for enum

* Revert use of identity check, didn't add entities

* Implement review feedback for tests

* ct phase sensors disabled by default

* import PHASENAMES from pyenphase

* Update tests/components/enphase_envoy/test_sensor.py

* Update tests/components/enphase_envoy/test_sensor.py

---------

Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Arie Catsman 2024-02-27 22:43:43 +01:00 committed by GitHub
parent a29d29ad85
commit d8d44069b5
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 3761 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

266
homeassistant/components/enphase_envoy/sensor.py
View File

@ -16,7 +16,14 @@ from pyenphase import (
EnvoySystemConsumption,
EnvoySystemProduction,
)
from pyenphase.const import PHASENAMES, PhaseNames
from pyenphase.const import PHASENAMES
from pyenphase.models.meters import (
CtMeterStatus,
CtState,
CtStatusFlags,
CtType,
EnvoyMeterData,
)
from homeassistant.components.sensor import (
SensorDeviceClass,
@ -28,7 +35,9 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
PERCENTAGE,
UnitOfApparentPower,
UnitOfElectricPotential,
UnitOfEnergy,
UnitOfFrequency,
UnitOfPower,
UnitOfTemperature,
)
@ -87,7 +96,7 @@ class EnvoyProductionRequiredKeysMixin:
"""Mixin for required keys."""
value_fn: Callable[[EnvoySystemProduction], int]
on_phase: PhaseNames | None
on_phase: str | None
@dataclass(frozen=True)
@ -145,7 +154,7 @@ PRODUCTION_SENSORS = (
PRODUCTION_PHASE_SENSORS = {
(on_phase := PhaseNames(PHASENAMES[phase])): [
(on_phase := PHASENAMES[phase]): [
replace(
sensor,
key=f"{sensor.key}_l{phase + 1}",
@ -164,7 +173,7 @@ class EnvoyConsumptionRequiredKeysMixin:
"""Mixin for required keys."""
value_fn: Callable[[EnvoySystemConsumption], int]
on_phase: PhaseNames | None
on_phase: str | None
@dataclass(frozen=True)
@ -222,7 +231,7 @@ CONSUMPTION_SENSORS = (
CONSUMPTION_PHASE_SENSORS = {
(on_phase := PhaseNames(PHASENAMES[phase])): [
(on_phase := PHASENAMES[phase]): [
replace(
sensor,
key=f"{sensor.key}_l{phase + 1}",
@ -236,6 +245,151 @@ CONSUMPTION_PHASE_SENSORS = {
}
@dataclass(frozen=True)
class EnvoyCTRequiredKeysMixin:
"""Mixin for required keys."""
value_fn: Callable[
[EnvoyMeterData],
int | float | str | CtType | CtMeterStatus | CtStatusFlags | CtState | None,
]
on_phase: str | None
@dataclass(frozen=True)
class EnvoyCTSensorEntityDescription(SensorEntityDescription, EnvoyCTRequiredKeysMixin):
"""Describes an Envoy CT sensor entity."""
CT_NET_CONSUMPTION_SENSORS = (
EnvoyCTSensorEntityDescription(
key="lifetime_net_consumption",
translation_key="lifetime_net_consumption",
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
state_class=SensorStateClass.TOTAL_INCREASING,
device_class=SensorDeviceClass.ENERGY,
suggested_unit_of_measurement=UnitOfEnergy.MEGA_WATT_HOUR,
suggested_display_precision=3,
value_fn=lambda ct: ct.energy_delivered,
on_phase=None,
),
EnvoyCTSensorEntityDescription(
key="lifetime_net_production",
translation_key="lifetime_net_production",
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
state_class=SensorStateClass.TOTAL_INCREASING,
device_class=SensorDeviceClass.ENERGY,
suggested_unit_of_measurement=UnitOfEnergy.MEGA_WATT_HOUR,
suggested_display_precision=3,
value_fn=lambda ct: ct.energy_received,
on_phase=None,
),
EnvoyCTSensorEntityDescription(
key="net_consumption",
translation_key="net_consumption",
native_unit_of_measurement=UnitOfPower.WATT,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.POWER,
suggested_unit_of_measurement=UnitOfPower.KILO_WATT,
suggested_display_precision=3,
value_fn=lambda ct: ct.active_power,
on_phase=None,
),
EnvoyCTSensorEntityDescription(
key="frequency",
translation_key="net_ct_frequency",
native_unit_of_measurement=UnitOfFrequency.HERTZ,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.FREQUENCY,
suggested_unit_of_measurement=UnitOfFrequency.HERTZ,
suggested_display_precision=1,
entity_registry_enabled_default=False,
value_fn=lambda ct: ct.frequency,
on_phase=None,
),
EnvoyCTSensorEntityDescription(
key="voltage",
translation_key="net_ct_voltage",
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.VOLTAGE,
suggested_unit_of_measurement=UnitOfElectricPotential.VOLT,
suggested_display_precision=1,
entity_registry_enabled_default=False,
value_fn=lambda ct: ct.voltage,
on_phase=None,
),
EnvoyCTSensorEntityDescription(
key="net_consumption_ct_metering_status",
translation_key="net_ct_metering_status",
device_class=SensorDeviceClass.ENUM,
options=list(CtMeterStatus),
entity_registry_enabled_default=False,
value_fn=lambda ct: ct.metering_status,
on_phase=None,
),
EnvoyCTSensorEntityDescription(
key="net_consumption_ct_status_flags",
translation_key="net_ct_status_flags",
state_class=None,
entity_registry_enabled_default=False,
value_fn=lambda ct: 0 if ct.status_flags is None else len(ct.status_flags),
on_phase=None,
),
)
CT_NET_CONSUMPTION_PHASE_SENSORS = {
(on_phase := PHASENAMES[phase]): [
replace(
sensor,
key=f"{sensor.key}_l{phase + 1}",
translation_key=f"{sensor.translation_key}_phase",
entity_registry_enabled_default=False,
on_phase=on_phase,
translation_placeholders={"phase_name": f"l{phase + 1}"},
)
for sensor in list(CT_NET_CONSUMPTION_SENSORS)
]
for phase in range(0, 3)
}
CT_PRODUCTION_SENSORS = (
EnvoyCTSensorEntityDescription(
key="production_ct_metering_status",
translation_key="production_ct_metering_status",
device_class=SensorDeviceClass.ENUM,
options=list(CtMeterStatus),
entity_registry_enabled_default=False,
value_fn=lambda ct: ct.metering_status,
on_phase=None,
),
EnvoyCTSensorEntityDescription(
key="production_ct_status_flags",
translation_key="production_ct_status_flags",
state_class=None,
entity_registry_enabled_default=False,
value_fn=lambda ct: 0 if ct.status_flags is None else len(ct.status_flags),
on_phase=None,
),
)
CT_PRODUCTION_PHASE_SENSORS = {
(on_phase := PHASENAMES[phase]): [
replace(
sensor,
key=f"{sensor.key}_l{phase + 1}",
translation_key=f"{sensor.translation_key}_phase",
entity_registry_enabled_default=False,
on_phase=on_phase,
translation_placeholders={"phase_name": f"l{phase + 1}"},
)
for sensor in list(CT_PRODUCTION_SENSORS)
]
for phase in range(0, 3)
}
@dataclass(frozen=True)
class EnvoyEnchargeRequiredKeysMixin:
"""Mixin for required keys."""
@ -408,7 +562,7 @@ async def async_setup_entry(
entities.extend(
EnvoyProductionPhaseEntity(coordinator, description)
for use_phase, phase in envoy_data.system_production_phases.items()
for description in PRODUCTION_PHASE_SENSORS[PhaseNames(use_phase)]
for description in PRODUCTION_PHASE_SENSORS[use_phase]
if phase is not None
)
# For each consumption phase reported add consumption entities
@ -416,9 +570,39 @@ async def async_setup_entry(
entities.extend(
EnvoyConsumptionPhaseEntity(coordinator, description)
for use_phase, phase in envoy_data.system_consumption_phases.items()
for description in CONSUMPTION_PHASE_SENSORS[PhaseNames(use_phase)]
for description in CONSUMPTION_PHASE_SENSORS[use_phase]
if phase is not None
)
# Add net consumption CT entities
if ctmeter := envoy_data.ctmeter_consumption:
entities.extend(
EnvoyConsumptionCTEntity(coordinator, description)
for description in CT_NET_CONSUMPTION_SENSORS
if ctmeter.measurement_type == CtType.NET_CONSUMPTION
)
# For each net consumption ct phase reported add net consumption entities
if phase_data := envoy_data.ctmeter_consumption_phases:
entities.extend(
EnvoyConsumptionCTPhaseEntity(coordinator, description)
for use_phase, phase in phase_data.items()
for description in CT_NET_CONSUMPTION_PHASE_SENSORS[use_phase]
if phase.measurement_type == CtType.NET_CONSUMPTION
)
# Add production CT entities
if ctmeter := envoy_data.ctmeter_production:
entities.extend(
EnvoyProductionCTEntity(coordinator, description)
for description in CT_PRODUCTION_SENSORS
if ctmeter.measurement_type == CtType.PRODUCTION
)
# For each production ct phase reported add production ct entities
if phase_data := envoy_data.ctmeter_production_phases:
entities.extend(
EnvoyProductionCTPhaseEntity(coordinator, description)
for use_phase, phase in phase_data.items()
for description in CT_PRODUCTION_PHASE_SENSORS[use_phase]
if phase.measurement_type == CtType.PRODUCTION
)
if envoy_data.inverters:
entities.extend(
@ -549,6 +733,74 @@ class EnvoyConsumptionPhaseEntity(EnvoySystemSensorEntity):
return self.entity_description.value_fn(system_consumption)
class EnvoyConsumptionCTEntity(EnvoySystemSensorEntity):
"""Envoy net consumption CT entity."""
entity_description: EnvoyCTSensorEntityDescription
@property
def native_value(
self,
) -> int | float | str | CtType | CtMeterStatus | CtStatusFlags | None:
"""Return the state of the CT sensor."""
if (ctmeter := self.data.ctmeter_consumption) is None:
return None
return self.entity_description.value_fn(ctmeter)
class EnvoyConsumptionCTPhaseEntity(EnvoySystemSensorEntity):
"""Envoy net consumption CT phase entity."""
entity_description: EnvoyCTSensorEntityDescription
@property
def native_value(
self,
) -> int | float | str | CtType | CtMeterStatus | CtStatusFlags | None:
"""Return the state of the CT phase sensor."""
if TYPE_CHECKING:
assert self.entity_description.on_phase
if (ctmeter := self.data.ctmeter_consumption_phases) is None:
return None
return self.entity_description.value_fn(
ctmeter[self.entity_description.on_phase]
)
class EnvoyProductionCTEntity(EnvoySystemSensorEntity):
"""Envoy net consumption CT entity."""
entity_description: EnvoyCTSensorEntityDescription
@property
def native_value(
self,
) -> int | float | str | CtType | CtMeterStatus | CtStatusFlags | None:
"""Return the state of the CT sensor."""
if (ctmeter := self.data.ctmeter_production) is None:
return None
return self.entity_description.value_fn(ctmeter)
class EnvoyProductionCTPhaseEntity(EnvoySystemSensorEntity):
"""Envoy net consumption CT phase entity."""
entity_description: EnvoyCTSensorEntityDescription
@property
def native_value(
self,
) -> int | float | str | CtType | CtMeterStatus | CtStatusFlags | None:
"""Return the state of the CT phase sensor."""
if TYPE_CHECKING:
assert self.entity_description.on_phase
if (ctmeter := self.data.ctmeter_production_phases) is None:
return None
return self.entity_description.value_fn(
ctmeter[self.entity_description.on_phase]
)
class EnvoyInverterEntity(EnvoySensorBaseEntity):
"""Envoy inverter entity."""

54
homeassistant/components/enphase_envoy/strings.json
View File

@ -143,6 +143,60 @@
"lifetime_consumption_phase": {
"name": "Lifetime energy consumption {phase_name}"
},
"lifetime_net_consumption": {
"name": "Lifetime net energy consumption"
},
"lifetime_net_production": {
"name": "Lifetime net energy production"
},
"net_consumption": {
"name": "Current net power consumption"
},
"net_ct_frequency": {
"name": "Frequency net consumption CT"
},
"net_ct_voltage": {
"name": "Voltage net consumption CT"
},
"net_ct_metering_status": {
"name": "Metering status net consumption CT"
},
"net_ct_status_flags": {
"name": "Meter status flags active net consumption CT"
},
"production_ct_metering_status": {
"name": "Metering status production CT"
},
"production_ct_status_flags": {
"name": "Meter status flags active production CT"
},
"lifetime_net_consumption_phase": {
"name": "Lifetime net energy consumption {phase_name}"
},
"lifetime_net_production_phase": {
"name": "Lifetime net energy production {phase_name}"
},
"net_consumption_phase": {
"name": "Current net power consumption {phase_name}"
},
"net_ct_frequency_phase": {
"name": "Frequency net consumption CT {phase_name}"
},
"net_ct_voltage_phase": {
"name": "Voltage net consumption CT {phase_name}"
},
"net_ct_metering_status_phase": {
"name": "Metering status net consumption CT {phase_name}"
},
"net_ct_status_flags_phase": {
"name": "Meter status flags active net consumption CT {phase_name}"
},
"production_ct_metering_status_phase": {
"name": "Metering status production CT {phase_name}"
},
"production_ct_status_flags_phase": {
"name": "Meter status flags active production CT {phase_name}"
},
"reserve_soc": {
"name": "Reserve battery level"
},

145
tests/components/enphase_envoy/conftest.py
View File

@ -10,7 +10,14 @@ from pyenphase import (
EnvoyTokenAuth,
)
from pyenphase.const import PhaseNames, SupportedFeatures
from pyenphase.models.meters import CtType, EnvoyPhaseMode
from pyenphase.models.meters import (
CtMeterStatus,
CtState,
CtStatusFlags,
CtType,
EnvoyMeterData,
EnvoyPhaseMode,
)
import pytest
from homeassistant.components.enphase_envoy import DOMAIN
@ -52,6 +59,10 @@ def mock_envoy_fixture(serial_number, mock_authenticate, mock_setup, mock_auth):
mock_envoy = Mock(spec=Envoy)
mock_envoy.serial_number = serial_number
mock_envoy.firmware = "7.1.2"
mock_envoy.part_number = "123456789"
mock_envoy.envoy_model = (
"Envoy, phases: 3, phase mode: three, net-consumption CT, production CT"
)
mock_envoy.authenticate = mock_authenticate
mock_envoy.setup = mock_setup
mock_envoy.auth = mock_auth
@ -61,12 +72,14 @@ def mock_envoy_fixture(serial_number, mock_authenticate, mock_setup, mock_auth):
| SupportedFeatures.PRODUCTION
| SupportedFeatures.METERING
| SupportedFeatures.THREEPHASE
| SupportedFeatures.CTMETERS
)
mock_envoy.phase_mode = EnvoyPhaseMode.THREE
mock_envoy.phase_count = 3
mock_envoy.active_phase_count = 3
mock_envoy.ct_meter_count = 2
mock_envoy.consumption_meter_type = CtType.NET_CONSUMPTION
mock_envoy.production_meter_type = CtType.PRODUCTION
mock_envoy.data = EnvoyData(
system_consumption=EnvoySystemConsumption(
watt_hours_last_7_days=1234,
@ -120,6 +133,133 @@ def mock_envoy_fixture(serial_number, mock_authenticate, mock_setup, mock_auth):
watts_now=3234,
),
},
ctmeter_production=EnvoyMeterData(
eid="100000010",
timestamp=1708006110,
energy_delivered=11234,
energy_received=12345,
active_power=100,
power_factor=0.11,
voltage=111,
current=0.2,
frequency=50.1,
state=CtState.ENABLED,
measurement_type=CtType.PRODUCTION,
metering_status=CtMeterStatus.NORMAL,
status_flags=[
CtStatusFlags.PODUCTION_IMBALANCE,
CtStatusFlags.POWER_ON_UNUSED_PHASE,
],
),
ctmeter_consumption=EnvoyMeterData(
eid="100000020",
timestamp=1708006120,
energy_delivered=21234,
energy_received=22345,
active_power=101,
power_factor=0.21,
voltage=112,
current=0.3,
frequency=50.2,
state=CtState.ENABLED,
measurement_type=CtType.NET_CONSUMPTION,
metering_status=CtMeterStatus.NORMAL,
status_flags=[],
),
ctmeter_production_phases={
PhaseNames.PHASE_1: EnvoyMeterData(
eid="100000011",
timestamp=1708006111,
energy_delivered=112341,
energy_received=123451,
active_power=20,
power_factor=0.12,
voltage=111,
current=0.2,
frequency=50.1,
state=CtState.ENABLED,
measurement_type=CtType.PRODUCTION,
metering_status=CtMeterStatus.NORMAL,
status_flags=[CtStatusFlags.PODUCTION_IMBALANCE],
),
PhaseNames.PHASE_2: EnvoyMeterData(
eid="100000012",
timestamp=1708006112,
energy_delivered=112342,
energy_received=123452,
active_power=30,
power_factor=0.13,
voltage=111,
current=0.2,
frequency=50.1,
state=CtState.ENABLED,
measurement_type=CtType.PRODUCTION,
metering_status=CtMeterStatus.NORMAL,
status_flags=[CtStatusFlags.POWER_ON_UNUSED_PHASE],
),
PhaseNames.PHASE_3: EnvoyMeterData(
eid="100000013",
timestamp=1708006113,
energy_delivered=112343,
energy_received=123453,
active_power=50,
power_factor=0.14,
voltage=111,
current=0.2,
frequency=50.1,
state=CtState.ENABLED,
measurement_type=CtType.PRODUCTION,
metering_status=CtMeterStatus.NORMAL,
status_flags=[],
),
},
ctmeter_consumption_phases={
PhaseNames.PHASE_1: EnvoyMeterData(
eid="100000021",
timestamp=1708006121,
energy_delivered=212341,
energy_received=223451,
active_power=21,
power_factor=0.22,
voltage=112,
current=0.3,
frequency=50.2,
state=CtState.ENABLED,
measurement_type=CtType.NET_CONSUMPTION,
metering_status=CtMeterStatus.NORMAL,
status_flags=[],
),
PhaseNames.PHASE_2: EnvoyMeterData(
eid="100000022",
timestamp=1708006122,
energy_delivered=212342,
energy_received=223452,
active_power=31,
power_factor=0.23,
voltage=112,
current=0.3,
frequency=50.2,
state=CtState.ENABLED,
measurement_type=CtType.NET_CONSUMPTION,
metering_status=CtMeterStatus.NORMAL,
status_flags=[],
),
PhaseNames.PHASE_3: EnvoyMeterData(
eid="100000023",
timestamp=1708006123,
energy_delivered=212343,
energy_received=223453,
active_power=51,
power_factor=0.24,
voltage=112,
current=0.3,
frequency=50.2,
state=CtState.ENABLED,
measurement_type=CtType.NET_CONSUMPTION,
metering_status=CtMeterStatus.NORMAL,
status_flags=[],
),
},
inverters={
"1": EnvoyInverter(
serial_number="1",
@ -143,9 +283,6 @@ async def setup_enphase_envoy_fixture(hass, config, mock_envoy):
), patch(
"homeassistant.components.enphase_envoy.Envoy",
return_value=mock_envoy,
), patch(
"homeassistant.components.enphase_envoy.PLATFORMS",
[],
):
assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()

10
tests/components/enphase_envoy/snapshots/test_config_flow.ambr Normal file
View File

@ -0,0 +1,10 @@
# serializer version: 1
# name: test_platforms
list([
<Platform.BINARY_SENSOR: 'binary_sensor'>,
<Platform.NUMBER: 'number'>,
<Platform.SELECT: 'select'>,
<Platform.SENSOR: 'sensor'>,
<Platform.SWITCH: 'switch'>,
])
# ---

3235
tests/components/enphase_envoy/snapshots/test_sensor.ambr Normal file
View File

File diff suppressed because it is too large Load Diff

8
tests/components/enphase_envoy/test_config_flow.py
View File

@ -4,10 +4,11 @@ from unittest.mock import AsyncMock
from pyenphase import EnvoyAuthenticationError, EnvoyError
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant import config_entries
from homeassistant.components import zeroconf
from homeassistant.components.enphase_envoy.const import DOMAIN
from homeassistant.components.enphase_envoy.const import DOMAIN, PLATFORMS
from homeassistant.core import HomeAssistant
@ -402,3 +403,8 @@ async def test_reauth(hass: HomeAssistant, config_entry, setup_enphase_envoy) ->
)
assert result2["type"] == "abort"
assert result2["reason"] == "reauth_successful"
async def test_platforms(snapshot: SnapshotAssertion) -> None:
"""Test if platform list changed and requires more tests."""
assert snapshot == PLATFORMS

55
tests/components/enphase_envoy/test_sensor.py Normal file
View File

@ -0,0 +1,55 @@
"""Test Enphase Envoy sensors."""
from unittest.mock import patch
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.enphase_envoy import DOMAIN
from homeassistant.components.enphase_envoy.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
@pytest.fixture(name="setup_enphase_envoy_sensor")
async def setup_enphase_envoy_sensor_fixture(hass, config, mock_envoy):
"""Define a fixture to set up Enphase Envoy with sensor platform only."""
with patch(
"homeassistant.components.enphase_envoy.config_flow.Envoy",
return_value=mock_envoy,
), patch(
"homeassistant.components.enphase_envoy.Envoy",
return_value=mock_envoy,
), patch(
"homeassistant.components.enphase_envoy.PLATFORMS",
[Platform.SENSOR],
):
assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()
yield
async def test_sensor(
hass: HomeAssistant,
config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
setup_enphase_envoy_sensor,
) -> None:
"""Test enphase_envoy sensor entities."""
entity_registry = er.async_get(hass)
assert entity_registry
# compare registered entities against snapshot of prior run
entity_entries = er.async_entries_for_config_entry(
entity_registry, config_entry.entry_id
)
assert entity_entries
assert entity_entries == snapshot
# Test if all entities still have same state
for entity_entry in entity_entries:
assert hass.states.get(entity_entry.entity_id) == snapshot(
name=f"{entity_entry.entity_id}-state"
)