Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Refactor enphase_envoy to use pyenphase library (#97862)

pull/75251/head^2
J. Nick Koston 2023-08-05 13:33:16 -10:00 committed by GitHub
parent 34013ac3e9
commit 02e546e3ef
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 460 additions and 378 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -302,6 +302,7 @@ omit =
homeassistant/components/enocean/sensor.py
homeassistant/components/enocean/switch.py
homeassistant/components/enphase_envoy/__init__.py
homeassistant/components/enphase_envoy/coordinator.py
homeassistant/components/enphase_envoy/sensor.py
homeassistant/components/entur_public_transport/*
homeassistant/components/environment_canada/__init__.py

90
homeassistant/components/enphase_envoy/__init__.py
View File

@ -1,27 +1,16 @@
"""The Enphase Envoy integration."""
from __future__ import annotations
from datetime import timedelta
import logging
import async_timeout
from envoy_reader.envoy_reader import EnvoyReader
import httpx
from pyenphase import Envoy
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.httpx_client import get_async_client
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import COORDINATOR, DOMAIN, NAME, PLATFORMS
from .sensor import SENSORS
SCAN_INTERVAL = timedelta(seconds=60)
_LOGGER = logging.getLogger(__name__)
from .const import DOMAIN, PLATFORMS
from .coordinator import EnphaseUpdateCoordinator
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
@ -29,64 +18,19 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
config = entry.data
name = config[CONF_NAME]
host = config[CONF_HOST]
username = config[CONF_USERNAME]
password = config[CONF_PASSWORD]
envoy_reader = EnvoyReader(
config[CONF_HOST],
config[CONF_USERNAME],
config[CONF_PASSWORD],
inverters=True,
async_client=get_async_client(hass),
)
envoy = Envoy(host, get_async_client(hass, verify_ssl=False))
async def async_update_data():
"""Fetch data from API endpoint."""
async with async_timeout.timeout(30):
try:
await envoy_reader.getData()
except httpx.HTTPStatusError as err:
raise ConfigEntryAuthFailed from err
except httpx.HTTPError as err:
raise UpdateFailed(f"Error communicating with API: {err}") from err
data = {
description.key: await getattr(envoy_reader, description.key)()
for description in SENSORS
}
data["inverters_production"] = await envoy_reader.inverters_production()
_LOGGER.debug("Retrieved data from API: %s", data)
return data
coordinator = DataUpdateCoordinator(
hass,
_LOGGER,
name=f"envoy {name}",
update_method=async_update_data,
update_interval=SCAN_INTERVAL,
always_update=False,
)
try:
await coordinator.async_config_entry_first_refresh()
except ConfigEntryAuthFailed:
envoy_reader.get_inverters = False
await coordinator.async_config_entry_first_refresh()
coordinator = EnphaseUpdateCoordinator(hass, envoy, name, username, password)
await coordinator.async_config_entry_first_refresh()
if not entry.unique_id:
try:
serial = await envoy_reader.get_full_serial_number()
except httpx.HTTPError as ex:
raise ConfigEntryNotReady(
f"Could not obtain serial number from envoy: {ex}"
) from ex
hass.config_entries.async_update_entry(entry, unique_id=envoy.serial_number)
hass.config_entries.async_update_entry(entry, unique_id=serial)
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = {
COORDINATOR: coordinator,
NAME: name,
}
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
@ -106,13 +50,13 @@ async def async_remove_config_entry_device(
) -> bool:
"""Remove an enphase_envoy config entry from a device."""
dev_ids = {dev_id[1] for dev_id in device_entry.identifiers if dev_id[0] == DOMAIN}
data: dict = hass.data[DOMAIN][config_entry.entry_id]
coordinator: DataUpdateCoordinator = data[COORDINATOR]
envoy_data: dict = coordinator.data
coordinator: EnphaseUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
envoy_data = coordinator.envoy.data
envoy_serial_num = config_entry.unique_id
if envoy_serial_num in dev_ids:
return False
for inverter in envoy_data.get("inverters_production", []):
if str(inverter) in dev_ids:
return False
if envoy_data and envoy_data.inverters:
for inverter in envoy_data.inverters:
if str(inverter) in dev_ids:
return False
return True

146
homeassistant/components/enphase_envoy/config_flow.py
View File

@ -2,12 +2,17 @@
from __future__ import annotations
from collections.abc import Mapping
import contextlib
import logging
from typing import Any
from envoy_reader.envoy_reader import EnvoyReader
import httpx
from awesomeversion import AwesomeVersion
from pyenphase import (
AUTH_TOKEN_MIN_VERSION,
Envoy,
EnvoyAuthenticationError,
EnvoyAuthenticationRequired,
EnvoyError,
)
import voluptuous as vol
from homeassistant import config_entries
@ -15,7 +20,6 @@ from homeassistant.components import zeroconf
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.httpx_client import get_async_client
from homeassistant.util.network import is_ipv4_address
@ -27,25 +31,19 @@ ENVOY = "Envoy"
CONF_SERIAL = "serial"
INVALID_AUTH_ERRORS = (EnvoyAuthenticationError, EnvoyAuthenticationRequired)
async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> EnvoyReader:
INSTALLER_AUTH_USERNAME = "installer"
async def validate_input(
hass: HomeAssistant, host: str, username: str, password: str
) -> Envoy:
"""Validate the user input allows us to connect."""
envoy_reader = EnvoyReader(
data[CONF_HOST],
data[CONF_USERNAME],
data[CONF_PASSWORD],
inverters=False,
async_client=get_async_client(hass),
)
try:
await envoy_reader.getData()
except httpx.HTTPStatusError as err:
raise InvalidAuth from err
except (RuntimeError, httpx.HTTPError) as err:
raise CannotConnect from err
return envoy_reader
envoy = Envoy(host, get_async_client(hass, verify_ssl=False))
await envoy.setup()
await envoy.authenticate(username=username, password=password)
return envoy
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
@ -57,10 +55,11 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Initialize an envoy flow."""
self.ip_address = None
self.username = None
self.protovers: str | None = None
self._reauth_entry = None
@callback
def _async_generate_schema(self):
def _async_generate_schema(self) -> vol.Schema:
"""Generate schema."""
schema = {}
@ -68,15 +67,26 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
schema[vol.Required(CONF_HOST, default=self.ip_address)] = vol.In(
[self.ip_address]
)
else:
elif not self._reauth_entry:
schema[vol.Required(CONF_HOST)] = str
schema[vol.Optional(CONF_USERNAME, default=self.username or "envoy")] = str
default_username = ""
if (
not self.username
and self.protovers
and AwesomeVersion(self.protovers) < AUTH_TOKEN_MIN_VERSION
):
default_username = INSTALLER_AUTH_USERNAME
schema[
vol.Optional(CONF_USERNAME, default=self.username or default_username)
] = str
schema[vol.Optional(CONF_PASSWORD, default="")] = str
return vol.Schema(schema)
@callback
def _async_current_hosts(self):
def _async_current_hosts(self) -> set[str]:
"""Return a set of hosts."""
return {
entry.data[CONF_HOST]
@ -91,6 +101,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
if not is_ipv4_address(discovery_info.host):
return self.async_abort(reason="not_ipv4_address")
serial = discovery_info.properties["serialnum"]
self.protovers = discovery_info.properties.get("protovers")
await self.async_set_unique_id(serial)
self.ip_address = discovery_info.host
self._abort_if_unique_id_configured({CONF_HOST: self.ip_address})
@ -116,81 +127,84 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self._reauth_entry = self.hass.config_entries.async_get_entry(
self.context["entry_id"]
)
assert self._reauth_entry is not None
if unique_id := self._reauth_entry.unique_id:
await self.async_set_unique_id(unique_id, raise_on_progress=False)
return await self.async_step_user()
def _async_envoy_name(self) -> str:
"""Return the name of the envoy."""
if self.unique_id:
return f"{ENVOY} {self.unique_id}"
return ENVOY
async def _async_set_unique_id_from_envoy(self, envoy_reader: EnvoyReader) -> bool:
"""Set the unique id by fetching it from the envoy."""
serial = None
with contextlib.suppress(httpx.HTTPError):
serial = await envoy_reader.get_full_serial_number()
if serial:
await self.async_set_unique_id(serial)
return True
return False
return f"{ENVOY} {self.unique_id}" if self.unique_id else ENVOY
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
errors = {}
errors: dict[str, str] = {}
description_placeholders: dict[str, str] = {}
if self._reauth_entry:
host = self._reauth_entry.data[CONF_HOST]
else:
host = (user_input or {}).get(CONF_HOST) or self.ip_address or ""
if user_input is not None:
if (
not self._reauth_entry
and user_input[CONF_HOST] in self._async_current_hosts()
):
return self.async_abort(reason="already_configured")
if not self._reauth_entry:
if host in self._async_current_hosts():
return self.async_abort(reason="already_configured")
try:
envoy_reader = await validate_input(self.hass, user_input)
except CannotConnect:
errors["base"] = "cannot_connect"
except InvalidAuth:
envoy = await validate_input(
self.hass,
host,
user_input[CONF_USERNAME],
user_input[CONF_PASSWORD],
)
except INVALID_AUTH_ERRORS as e:
errors["base"] = "invalid_auth"
description_placeholders = {"reason": str(e)}
except EnvoyError as e:
errors["base"] = "cannot_connect"
description_placeholders = {"reason": str(e)}
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
data = user_input.copy()
data[CONF_NAME] = self._async_envoy_name()
name = self._async_envoy_name()
if self._reauth_entry:
self.hass.config_entries.async_update_entry(
self._reauth_entry,
data=data,
data=self._reauth_entry.data | user_input,
)
self.hass.async_create_task(
self.hass.config_entries.async_reload(
self._reauth_entry.entry_id
)
)
return self.async_abort(reason="reauth_successful")
if not self.unique_id and await self._async_set_unique_id_from_envoy(
envoy_reader
):
data[CONF_NAME] = self._async_envoy_name()
if not self.unique_id:
await self.async_set_unique_id(envoy.serial_number)
name = self._async_envoy_name()
if self.unique_id:
self._abort_if_unique_id_configured({CONF_HOST: data[CONF_HOST]})
self._abort_if_unique_id_configured({CONF_HOST: host})
return self.async_create_entry(title=data[CONF_NAME], data=data)
# CONF_NAME is still set for legacy backwards compatibility
return self.async_create_entry(
title=name, data={CONF_HOST: host, CONF_NAME: name} | user_input
)
if self.unique_id:
self.context["title_placeholders"] = {
CONF_SERIAL: self.unique_id,
CONF_HOST: self.ip_address,
CONF_HOST: host,
}
return self.async_show_form(
step_id="user",
data_schema=self._async_generate_schema(),
description_placeholders=description_placeholders,
errors=errors,
)
class CannotConnect(HomeAssistantError):
"""Error to indicate we cannot connect."""
class InvalidAuth(HomeAssistantError):
"""Error to indicate there is invalid auth."""

4
homeassistant/components/enphase_envoy/const.py
View File

@ -4,7 +4,3 @@ from homeassistant.const import Platform
DOMAIN = "enphase_envoy"
PLATFORMS = [Platform.SENSOR]
COORDINATOR = "coordinator"
NAME = "name"

76
homeassistant/components/enphase_envoy/coordinator.py Normal file
View File

@ -0,0 +1,76 @@
"""The enphase_envoy component."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any
from pyenphase import (
Envoy,
EnvoyAuthenticationError,
EnvoyAuthenticationRequired,
EnvoyError,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
SCAN_INTERVAL = timedelta(seconds=60)
_LOGGER = logging.getLogger(__name__)
class EnphaseUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
"""DataUpdateCoordinator to gather data from any envoy."""
envoy_serial_number: str
def __init__(
self,
hass: HomeAssistant,
envoy: Envoy,
name: str,
username: str,
password: str,
) -> None:
"""Initialize DataUpdateCoordinator for the envoy."""
self.envoy = envoy
self.username = username
self.password = password
self.name = name
self._setup_complete = False
super().__init__(
hass,
_LOGGER,
name=name,
update_interval=SCAN_INTERVAL,
always_update=False,
)
async def _async_setup_and_authenticate(self) -> None:
"""Set up and authenticate with the envoy."""
envoy = self.envoy
await envoy.setup()
assert envoy.serial_number is not None
self.envoy_serial_number = envoy.serial_number
await envoy.authenticate(username=self.username, password=self.password)
self._setup_complete = True
async def _async_update_data(self) -> dict[str, Any]:
"""Fetch all device and sensor data from api."""
envoy = self.envoy
for tries in range(2):
try:
if not self._setup_complete:
await self._async_setup_and_authenticate()
return (await envoy.update()).raw
except (EnvoyAuthenticationError, EnvoyAuthenticationRequired) as err:
if self._setup_complete and tries == 0:
# token likely expired or firmware changed, try to re-authenticate
self._setup_complete = False
continue
raise ConfigEntryAuthFailed from err
except EnvoyError as err:
raise UpdateFailed(f"Error communicating with API: {err}") from err
raise RuntimeError("Unreachable code in _async_update_data") # pragma: no cover

6
homeassistant/components/enphase_envoy/diagnostics.py
View File

@ -7,9 +7,9 @@ from homeassistant.components.diagnostics import async_redact_data
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_NAME, CONF_PASSWORD, CONF_UNIQUE_ID, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import COORDINATOR, DOMAIN
from .const import DOMAIN
from .coordinator import EnphaseUpdateCoordinator
CONF_TITLE = "title"
@ -27,7 +27,7 @@ async def async_get_config_entry_diagnostics(
hass: HomeAssistant, entry: ConfigEntry
) -> dict[str, Any]:
"""Return diagnostics for a config entry."""
coordinator: DataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][COORDINATOR]
coordinator: EnphaseUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
return async_redact_data(
{

4
homeassistant/components/enphase_envoy/manifest.json
View File

@ -5,8 +5,8 @@
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/enphase_envoy",
"iot_class": "local_polling",
"loggers": ["envoy_reader"],
"requirements": ["envoy-reader==0.20.1"],
"loggers": ["pyenphase"],
"requirements": ["pyenphase==0.8.0"],
"zeroconf": [
{
"type": "_enphase-envoy._tcp.local."

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

@ -5,7 +5,8 @@ from collections.abc import Callable
from dataclasses import dataclass
import datetime
import logging
from typing import cast
from pyenphase import EnvoyInverter, EnvoySystemConsumption, EnvoySystemProduction
from homeassistant.components.sensor import (
SensorDeviceClass,
@ -16,16 +17,15 @@ from homeassistant.components.sensor import (
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import UnitOfEnergy, UnitOfPower
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity import DeviceInfo, Entity
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import UNDEFINED
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from homeassistant.util import dt as dt_util
from .const import COORDINATOR, DOMAIN, NAME
from .const import DOMAIN
from .coordinator import EnphaseUpdateCoordinator
ICON = "mdi:flash"
_LOGGER = logging.getLogger(__name__)
@ -35,100 +35,147 @@ LAST_REPORTED_KEY = "last_reported"
@dataclass
class EnvoyRequiredKeysMixin:
class EnvoyInverterRequiredKeysMixin:
"""Mixin for required keys."""
value_fn: Callable[[tuple[float, str]], datetime.datetime | float | None]
value_fn: Callable[[EnvoyInverter], datetime.datetime | float]
@dataclass
class EnvoySensorEntityDescription(SensorEntityDescription, EnvoyRequiredKeysMixin):
class EnvoyInverterSensorEntityDescription(
SensorEntityDescription, EnvoyInverterRequiredKeysMixin
):
"""Describes an Envoy inverter sensor entity."""
def _inverter_last_report_time(
watt_report_time: tuple[float, str]
) -> datetime.datetime | None:
if (report_time := watt_report_time[1]) is None:
return None
if (last_reported_dt := dt_util.parse_datetime(report_time)) is None:
return None
if last_reported_dt.tzinfo is None:
return last_reported_dt.replace(tzinfo=dt_util.UTC)
return last_reported_dt
INVERTER_SENSORS = (
EnvoySensorEntityDescription(
EnvoyInverterSensorEntityDescription(
key=INVERTERS_KEY,
native_unit_of_measurement=UnitOfPower.WATT,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.POWER,
value_fn=lambda watt_report_time: watt_report_time[0],
value_fn=lambda inverter: inverter.last_report_watts,
),
EnvoySensorEntityDescription(
EnvoyInverterSensorEntityDescription(
key=LAST_REPORTED_KEY,
name="Last Reported",
device_class=SensorDeviceClass.TIMESTAMP,
entity_registry_enabled_default=False,
value_fn=_inverter_last_report_time,
value_fn=lambda inverter: dt_util.utc_from_timestamp(inverter.last_report_date),
),
)
SENSORS = (
SensorEntityDescription(
@dataclass
class EnvoyProductionRequiredKeysMixin:
"""Mixin for required keys."""
value_fn: Callable[[EnvoySystemProduction], int]
@dataclass
class EnvoyProductionSensorEntityDescription(
SensorEntityDescription, EnvoyProductionRequiredKeysMixin
):
"""Describes an Envoy production sensor entity."""
PRODUCTION_SENSORS = (
EnvoyProductionSensorEntityDescription(
key="production",
name="Current Power Production",
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 production: production.watts_now,
),
SensorEntityDescription(
EnvoyProductionSensorEntityDescription(
key="daily_production",
name="Today's Energy Production",
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
state_class=SensorStateClass.TOTAL_INCREASING,
device_class=SensorDeviceClass.ENERGY,
suggested_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
suggested_display_precision=2,
value_fn=lambda production: production.watt_hours_today,
),
SensorEntityDescription(
EnvoyProductionSensorEntityDescription(
key="seven_days_production",
name="Last Seven Days Energy Production",
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
device_class=SensorDeviceClass.ENERGY,
suggested_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
suggested_display_precision=1,
value_fn=lambda production: production.watt_hours_last_7_days,
),
SensorEntityDescription(
EnvoyProductionSensorEntityDescription(
key="lifetime_production",
name="Lifetime Energy 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 production: production.watt_hours_lifetime,
),
SensorEntityDescription(
)
@dataclass
class EnvoyConsumptionRequiredKeysMixin:
"""Mixin for required keys."""
value_fn: Callable[[EnvoySystemConsumption], int]
@dataclass
class EnvoyConsumptionSensorEntityDescription(
SensorEntityDescription, EnvoyConsumptionRequiredKeysMixin
):
"""Describes an Envoy consumption sensor entity."""
CONSUMPTION_SENSORS = (
EnvoyConsumptionSensorEntityDescription(
key="consumption",
name="Current Power 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 consumption: consumption.watts_now,
),
SensorEntityDescription(
EnvoyConsumptionSensorEntityDescription(
key="daily_consumption",
name="Today's Energy Consumption",
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
state_class=SensorStateClass.TOTAL_INCREASING,
device_class=SensorDeviceClass.ENERGY,
suggested_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
suggested_display_precision=2,
value_fn=lambda consumption: consumption.watt_hours_today,
),
SensorEntityDescription(
EnvoyConsumptionSensorEntityDescription(
key="seven_days_consumption",
name="Last Seven Days Energy Consumption",
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
device_class=SensorDeviceClass.ENERGY,
suggested_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
suggested_display_precision=1,
value_fn=lambda consumption: consumption.watt_hours_last_7_days,
),
SensorEntityDescription(
EnvoyConsumptionSensorEntityDescription(
key="lifetime_consumption",
name="Lifetime Energy 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 consumption: consumption.watt_hours_lifetime,
),
)
@ -139,58 +186,47 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up envoy sensor platform."""
data: dict = hass.data[DOMAIN][config_entry.entry_id]
coordinator: DataUpdateCoordinator = data[COORDINATOR]
envoy_data: dict = coordinator.data
envoy_name: str = data[NAME]
coordinator: EnphaseUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
envoy_data = coordinator.envoy.data
assert envoy_data is not None
envoy_serial_num = config_entry.unique_id
assert envoy_serial_num is not None
_LOGGER.debug("Envoy data: %s", envoy_data)
entities: list[Envoy | EnvoyInverter] = []
for description in SENSORS:
sensor_data = envoy_data.get(description.key)
if isinstance(sensor_data, str) and "not available" in sensor_data:
continue
entities.append(
Envoy(
coordinator,
description,
envoy_name,
envoy_serial_num,
)
)
if production := envoy_data.get("inverters_production"):
entities: list[Entity] = [
EnvoyProductionEntity(coordinator, description)
for description in PRODUCTION_SENSORS
]
if envoy_data.system_consumption:
entities.extend(
EnvoyInverter(
coordinator,
description,
envoy_name,
envoy_serial_num,
str(inverter),
)
EnvoyConsumptionEntity(coordinator, description)
for description in CONSUMPTION_SENSORS
)
if envoy_data.inverters:
entities.extend(
EnvoyInverterEntity(coordinator, description, inverter)
for description in INVERTER_SENSORS
for inverter in production
for inverter in envoy_data.inverters
)
async_add_entities(entities)
class Envoy(CoordinatorEntity, SensorEntity):
class EnvoyEntity(CoordinatorEntity[EnphaseUpdateCoordinator], SensorEntity):
"""Envoy inverter entity."""
_attr_icon = ICON
def __init__(
self,
coordinator: DataUpdateCoordinator,
coordinator: EnphaseUpdateCoordinator,
description: SensorEntityDescription,
envoy_name: str,
envoy_serial_num: str,
) -> None:
"""Initialize Envoy entity."""
self.entity_description = description
envoy_name = coordinator.name
envoy_serial_num = coordinator.envoy.serial_number
assert envoy_serial_num is not None
self._attr_name = f"{envoy_name} {description.name}"
self._attr_unique_id = f"{envoy_serial_num}_{description.key}"
self._attr_device_info = DeviceInfo(
@ -198,44 +234,71 @@ class Envoy(CoordinatorEntity, SensorEntity):
manufacturer="Enphase",
model="Envoy",
name=envoy_name,
sw_version=str(coordinator.envoy.firmware),
)
super().__init__(coordinator)
class EnvoyProductionEntity(EnvoyEntity):
"""Envoy production entity."""
entity_description: EnvoyProductionSensorEntityDescription
@property
def native_value(self) -> float | None:
def native_value(self) -> int | None:
"""Return the state of the sensor."""
if (value := self.coordinator.data.get(self.entity_description.key)) is None:
return None
return cast(float, value)
envoy = self.coordinator.envoy
assert envoy.data is not None
assert envoy.data.system_production is not None
return self.entity_description.value_fn(envoy.data.system_production)
class EnvoyInverter(CoordinatorEntity, SensorEntity):
class EnvoyConsumptionEntity(EnvoyEntity):
"""Envoy consumption entity."""
entity_description: EnvoyConsumptionSensorEntityDescription
@property
def native_value(self) -> int | None:
"""Return the state of the sensor."""
envoy = self.coordinator.envoy
assert envoy.data is not None
assert envoy.data.system_consumption is not None
return self.entity_description.value_fn(envoy.data.system_consumption)
class EnvoyInverterEntity(CoordinatorEntity[EnphaseUpdateCoordinator], SensorEntity):
"""Envoy inverter entity."""
_attr_icon = ICON
entity_description: EnvoySensorEntityDescription
entity_description: EnvoyInverterSensorEntityDescription
def __init__(
self,
coordinator: DataUpdateCoordinator,
description: EnvoySensorEntityDescription,
envoy_name: str,
envoy_serial_num: str,
coordinator: EnphaseUpdateCoordinator,
description: EnvoyInverterSensorEntityDescription,
serial_number: str,
) -> None:
"""Initialize Envoy inverter entity."""
self.entity_description = description
envoy_name = coordinator.name
self._serial_number = serial_number
if description.name is not UNDEFINED:
self._attr_name = (
f"{envoy_name} Inverter {serial_number} {description.name}"
)
else:
name = description.name
key = description.key
if key == INVERTERS_KEY:
# Originally there was only one inverter sensor, so we don't want to
# break existing installations by changing the name or unique_id.
self._attr_name = f"{envoy_name} Inverter {serial_number}"
if description.key == INVERTERS_KEY:
self._attr_unique_id = serial_number
else:
self._attr_unique_id = f"{serial_number}_{description.key}"
# Additional sensors have a name and unique_id that includes the
# sensor key.
self._attr_name = f"{envoy_name} Inverter {serial_number} {name}"
self._attr_unique_id = f"{serial_number}_{key}"
envoy_serial_num = coordinator.envoy.serial_number
assert envoy_serial_num is not None
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, serial_number)},
name=f"Inverter {serial_number}",
@ -246,9 +309,10 @@ class EnvoyInverter(CoordinatorEntity, SensorEntity):
super().__init__(coordinator)
@property
def native_value(self) -> datetime.datetime | float | None:
def native_value(self) -> datetime.datetime | float:
"""Return the state of the sensor."""
watt_report_time: tuple[float, str] = self.coordinator.data[
"inverters_production"
][self._serial_number]
return self.entity_description.value_fn(watt_report_time)
envoy = self.coordinator.envoy
assert envoy.data is not None
assert envoy.data.inverters is not None
inverter = envoy.data.inverters[self._serial_number]
return self.entity_description.value_fn(inverter)

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

@ -3,7 +3,7 @@
"flow_title": "{serial} ({host})",
"step": {
"user": {
"description": "For newer models, enter username `envoy` without a password. For older models, enter username `installer` without a password. For all other models, enter a valid username and password.",
"description": "For firmware version 7.0 and later, enter the Enphase cloud credentials, for older models models, enter username `installer` without a password.",
"data": {
"host": "[%key:common::config_flow::data::host%]",
"username": "[%key:common::config_flow::data::username%]",
@ -12,8 +12,8 @@
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"cannot_connect": "Cannot connect: {reason}",
"invalid_auth": "Invalid authentication: {reason}",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {

6
requirements_all.txt
View File

@ -738,9 +738,6 @@ enturclient==0.2.4
# homeassistant.components.environment_canada
env-canada==0.5.36
# homeassistant.components.enphase_envoy
envoy-reader==0.20.1
# homeassistant.components.season
ephem==4.1.2
@ -1664,6 +1661,9 @@ pyedimax==0.2.1
# homeassistant.components.efergy
pyefergy==22.1.1
# homeassistant.components.enphase_envoy
pyenphase==0.8.0
# homeassistant.components.envisalink
pyenvisalink==4.6

6
requirements_test_all.txt
View File

@ -594,9 +594,6 @@ enocean==0.50
# homeassistant.components.environment_canada
env-canada==0.5.36
# homeassistant.components.enphase_envoy
envoy-reader==0.20.1
# homeassistant.components.season
ephem==4.1.2
@ -1231,6 +1228,9 @@ pyeconet==0.1.20
# homeassistant.components.efergy
pyefergy==22.1.1
# homeassistant.components.enphase_envoy
pyenphase==0.8.0
# homeassistant.components.everlights
pyeverlights==0.1.0

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

@ -1,7 +1,13 @@
"""Define test fixtures for Enphase Envoy."""
import json
from unittest.mock import AsyncMock, Mock, patch
from pyenphase import (
Envoy,
EnvoyData,
EnvoyInverter,
EnvoySystemConsumption,
EnvoySystemProduction,
)
import pytest
from homeassistant.components.enphase_envoy import DOMAIN
@ -9,7 +15,7 @@ from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PASSWORD, CONF_USERNA
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry, load_fixture
from tests.common import MockConfigEntry
@pytest.fixture(name="config_entry")
@ -36,66 +42,49 @@ def config_fixture():
}
@pytest.fixture(name="gateway_data", scope="package")
def gateway_data_fixture():
"""Define a fixture to return gateway data."""
return json.loads(load_fixture("data.json", "enphase_envoy"))
@pytest.fixture(name="inverters_production_data", scope="package")
def inverters_production_data_fixture():
"""Define a fixture to return inverter production data."""
return json.loads(load_fixture("inverters_production.json", "enphase_envoy"))
@pytest.fixture(name="mock_envoy_reader")
def mock_envoy_reader_fixture(
gateway_data,
mock_get_data,
mock_get_full_serial_number,
mock_inverters_production,
serial_number,
):
"""Define a mocked EnvoyReader fixture."""
mock_envoy_reader = Mock(
getData=mock_get_data,
get_full_serial_number=mock_get_full_serial_number,
inverters_production=mock_inverters_production,
@pytest.fixture(name="mock_envoy")
def mock_envoy_fixture(serial_number, mock_authenticate, mock_setup):
"""Define a mocked Envoy fixture."""
mock_envoy = Mock(spec=Envoy)
mock_envoy.serial_number = serial_number
mock_envoy.authenticate = mock_authenticate
mock_envoy.setup = mock_setup
mock_envoy.data = EnvoyData(
system_consumption=EnvoySystemConsumption(
watt_hours_last_7_days=1234,
watt_hours_lifetime=1234,
watt_hours_today=1234,
watts_now=1234,
),
system_production=EnvoySystemProduction(
watt_hours_last_7_days=1234,
watt_hours_lifetime=1234,
watt_hours_today=1234,
watts_now=1234,
),
inverters={
"1": EnvoyInverter(
serial_number="1",
last_report_date=1,
last_report_watts=1,
max_report_watts=1,
)
},
raw={"varies_by": "firmware_version"},
)
for key, value in gateway_data.items():
setattr(mock_envoy_reader, key, AsyncMock(return_value=value))
return mock_envoy_reader
@pytest.fixture(name="mock_get_full_serial_number")
def mock_get_full_serial_number_fixture(serial_number):
"""Define a mocked EnvoyReader.get_full_serial_number fixture."""
return AsyncMock(return_value=serial_number)
@pytest.fixture(name="mock_get_data")
def mock_get_data_fixture():
"""Define a mocked EnvoyReader.getData fixture."""
return AsyncMock()
@pytest.fixture(name="mock_inverters_production")
def mock_inverters_production_fixture(inverters_production_data):
"""Define a mocked EnvoyReader.inverters_production fixture."""
return AsyncMock(return_value=inverters_production_data)
mock_envoy.update = AsyncMock(return_value=mock_envoy.data)
return mock_envoy
@pytest.fixture(name="setup_enphase_envoy")
async def setup_enphase_envoy_fixture(hass, config, mock_envoy_reader):
async def setup_enphase_envoy_fixture(hass, config, mock_envoy):
"""Define a fixture to set up Enphase Envoy."""
with patch(
"homeassistant.components.enphase_envoy.config_flow.EnvoyReader",
return_value=mock_envoy_reader,
"homeassistant.components.enphase_envoy.config_flow.Envoy",
return_value=mock_envoy,
), patch(
"homeassistant.components.enphase_envoy.EnvoyReader",
return_value=mock_envoy_reader,
"homeassistant.components.enphase_envoy.Envoy",
return_value=mock_envoy,
), patch(
"homeassistant.components.enphase_envoy.PLATFORMS", []
):
@ -104,6 +93,18 @@ async def setup_enphase_envoy_fixture(hass, config, mock_envoy_reader):
yield
@pytest.fixture(name="mock_authenticate")
def mock_authenticate():
"""Define a mocked Envoy.authenticate fixture."""
return AsyncMock()
@pytest.fixture(name="mock_setup")
def mock_setup():
"""Define a mocked Envoy.setup fixture."""
return AsyncMock()
@pytest.fixture(name="serial_number")
def serial_number_fixture():
"""Define a serial number fixture."""

1
tests/components/enphase_envoy/fixtures/__init__.py
View File

@ -1 +0,0 @@
"""Define data fixtures for Enphase Envoy."""

10
tests/components/enphase_envoy/fixtures/data.json
View File

@ -1,10 +0,0 @@
{
"production": 1840,
"daily_production": 28223,
"seven_days_production": 174482,
"lifetime_production": 5924391,
"consumption": 1840,
"daily_consumption": 5923857,
"seven_days_consumption": 5923857,
"lifetime_consumption": 5923857
}

18
tests/components/enphase_envoy/fixtures/inverters_production.json
View File

@ -1,18 +0,0 @@
{
"202140024014": [136, "2022-10-08 16:43:36"],
"202140023294": [163, "2022-10-08 16:43:41"],
"202140013819": [130, "2022-10-08 16:43:31"],
"202140023794": [139, "2022-10-08 16:43:38"],
"202140023381": [130, "2022-10-08 16:43:47"],
"202140024176": [54, "2022-10-08 16:43:59"],
"202140003284": [132, "2022-10-08 16:43:55"],
"202140019854": [129, "2022-10-08 16:43:58"],
"202140020743": [131, "2022-10-08 16:43:49"],
"202140023531": [28, "2022-10-08 16:43:53"],
"202140024241": [164, "2022-10-08 16:43:33"],
"202140022963": [164, "2022-10-08 16:43:41"],
"202140023149": [118, "2022-10-08 16:43:47"],
"202140024828": [129, "2022-10-08 16:43:36"],
"202140023269": [133, "2022-10-08 16:43:43"],
"202140024157": [112, "2022-10-08 16:43:52"]
}

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

@ -1,7 +1,7 @@
"""Test the Enphase Envoy config flow."""
from unittest.mock import AsyncMock, MagicMock
from unittest.mock import AsyncMock
import httpx
from pyenphase import EnvoyAuthenticationError, EnvoyError
import pytest
from homeassistant import config_entries
@ -65,16 +65,7 @@ async def test_user_no_serial_number(
}
@pytest.mark.parametrize(
"mock_get_full_serial_number",
[
AsyncMock(
side_effect=httpx.HTTPStatusError(
"any", request=MagicMock(), response=MagicMock()
)
)
],
)
@pytest.mark.parametrize("serial_number", [None])
async def test_user_fetching_serial_fails(
hass: HomeAssistant, setup_enphase_envoy
) -> None:
@ -104,13 +95,9 @@ async def test_user_fetching_serial_fails(
@pytest.mark.parametrize(
"mock_get_data",
"mock_authenticate",
[
AsyncMock(
side_effect=httpx.HTTPStatusError(
"any", request=MagicMock(), response=MagicMock()
)
)
AsyncMock(side_effect=EnvoyAuthenticationError("test")),
],
)
async def test_form_invalid_auth(hass: HomeAssistant, setup_enphase_envoy) -> None:
@ -131,7 +118,8 @@ async def test_form_invalid_auth(hass: HomeAssistant, setup_enphase_envoy) -> No
@pytest.mark.parametrize(
"mock_get_data", [AsyncMock(side_effect=httpx.HTTPError("any"))]
"mock_setup",
[AsyncMock(side_effect=EnvoyError)],
)
async def test_form_cannot_connect(hass: HomeAssistant, setup_enphase_envoy) -> None:
"""Test we handle cannot connect error."""
@ -150,7 +138,10 @@ async def test_form_cannot_connect(hass: HomeAssistant, setup_enphase_envoy) ->
assert result2["errors"] == {"base": "cannot_connect"}
@pytest.mark.parametrize("mock_get_data", [AsyncMock(side_effect=ValueError)])
@pytest.mark.parametrize(
"mock_setup",
[AsyncMock(side_effect=ValueError)],
)
async def test_form_unknown_error(hass: HomeAssistant, setup_enphase_envoy) -> None:
"""Test we handle unknown error."""
result = await hass.config_entries.flow.async_init(
@ -168,7 +159,17 @@ async def test_form_unknown_error(hass: HomeAssistant, setup_enphase_envoy) -> N
assert result2["errors"] == {"base": "unknown"}
async def test_zeroconf(hass: HomeAssistant, setup_enphase_envoy) -> None:
def _get_schema_default(schema, key_name):
"""Iterate schema to find a key."""
for schema_key in schema:
if schema_key == key_name:
return schema_key.default()
raise KeyError(f"{key_name} not found in schema")
async def test_zeroconf_pre_token_firmware(
hass: HomeAssistant, setup_enphase_envoy
) -> None:
"""Test we can setup from zeroconf."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
@ -179,13 +180,55 @@ async def test_zeroconf(hass: HomeAssistant, setup_enphase_envoy) -> None:
hostname="mock_hostname",
name="mock_name",
port=None,
properties={"serialnum": "1234"},
properties={"serialnum": "1234", "protovers": "3.0.0"},
type="mock_type",
),
)
assert result["type"] == "form"
assert result["step_id"] == "user"
assert _get_schema_default(result["data_schema"].schema, "username") == "installer"
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "1.1.1.1",
"username": "test-username",
"password": "test-password",
},
)
assert result2["type"] == "create_entry"
assert result2["title"] == "Envoy 1234"
assert result2["result"].unique_id == "1234"
assert result2["data"] == {
"host": "1.1.1.1",
"name": "Envoy 1234",
"username": "test-username",
"password": "test-password",
}
async def test_zeroconf_token_firmware(
hass: HomeAssistant, setup_enphase_envoy
) -> None:
"""Test we can setup from zeroconf."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=zeroconf.ZeroconfServiceInfo(
host="1.1.1.1",
addresses=["1.1.1.1"],
hostname="mock_hostname",
name="mock_name",
port=None,
properties={"serialnum": "1234", "protovers": "7.0.0"},
type="mock_type",
),
)
assert result["type"] == "form"
assert result["step_id"] == "user"
assert _get_schema_default(result["data_schema"].schema, "username") == ""
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
@ -311,7 +354,6 @@ async def test_reauth(hass: HomeAssistant, config_entry, setup_enphase_envoy) ->
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "1.1.1.1",
"username": "test-username",
"password": "test-password",
},

29
tests/components/enphase_envoy/test_diagnostics.py
View File

@ -32,32 +32,5 @@ async def test_entry_diagnostics(
"unique_id": REDACTED,
"disabled_by": None,
},
"data": {
"production": 1840,
"daily_production": 28223,
"seven_days_production": 174482,
"lifetime_production": 5924391,
"consumption": 1840,
"daily_consumption": 5923857,
"seven_days_consumption": 5923857,
"lifetime_consumption": 5923857,
"inverters_production": {
"202140024014": [136, "2022-10-08 16:43:36"],
"202140023294": [163, "2022-10-08 16:43:41"],
"202140013819": [130, "2022-10-08 16:43:31"],
"202140023794": [139, "2022-10-08 16:43:38"],
"202140023381": [130, "2022-10-08 16:43:47"],
"202140024176": [54, "2022-10-08 16:43:59"],
"202140003284": [132, "2022-10-08 16:43:55"],
"202140019854": [129, "2022-10-08 16:43:58"],
"202140020743": [131, "2022-10-08 16:43:49"],
"202140023531": [28, "2022-10-08 16:43:53"],
"202140024241": [164, "2022-10-08 16:43:33"],
"202140022963": [164, "2022-10-08 16:43:41"],
"202140023149": [118, "2022-10-08 16:43:47"],
"202140024828": [129, "2022-10-08 16:43:36"],
"202140023269": [133, "2022-10-08 16:43:43"],
"202140024157": [112, "2022-10-08 16:43:52"],
},
},
"data": {"varies_by": "firmware_version"},
}