mirror of https://github.com/home-assistant/core
178 lines
6.2 KiB
Python
178 lines
6.2 KiB
Python
"""Diagnostics support for Enphase Envoy."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from attr import asdict
|
|
from pyenphase.envoy import Envoy
|
|
from pyenphase.exceptions import EnvoyError
|
|
|
|
from homeassistant.components.diagnostics import async_redact_data
|
|
from homeassistant.const import (
|
|
CONF_NAME,
|
|
CONF_PASSWORD,
|
|
CONF_TOKEN,
|
|
CONF_UNIQUE_ID,
|
|
CONF_USERNAME,
|
|
)
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers import device_registry as dr, entity_registry as er
|
|
from homeassistant.helpers.json import json_dumps
|
|
from homeassistant.util.json import json_loads
|
|
|
|
from .const import OPTION_DIAGNOSTICS_INCLUDE_FIXTURES
|
|
from .coordinator import EnphaseConfigEntry
|
|
|
|
CONF_TITLE = "title"
|
|
CLEAN_TEXT = "<<envoyserial>>"
|
|
|
|
TO_REDACT = {
|
|
CONF_NAME,
|
|
CONF_PASSWORD,
|
|
# Config entry title and unique ID may contain sensitive data:
|
|
CONF_TITLE,
|
|
CONF_UNIQUE_ID,
|
|
CONF_USERNAME,
|
|
CONF_TOKEN,
|
|
}
|
|
|
|
|
|
async def _get_fixture_collection(envoy: Envoy, serial: str) -> dict[str, Any]:
|
|
"""Collect Envoy endpoints to use for test fixture set."""
|
|
fixture_data: dict[str, Any] = {}
|
|
end_points = [
|
|
"/info",
|
|
"/api/v1/production",
|
|
"/api/v1/production/inverters",
|
|
"/production.json",
|
|
"/production.json?details=1",
|
|
"/production",
|
|
"/ivp/ensemble/power",
|
|
"/ivp/ensemble/inventory",
|
|
"/ivp/ensemble/dry_contacts",
|
|
"/ivp/ensemble/status",
|
|
"/ivp/ensemble/secctrl",
|
|
"/ivp/ss/dry_contact_settings",
|
|
"/admin/lib/tariff",
|
|
"/ivp/ss/gen_config",
|
|
"/ivp/ss/gen_schedule",
|
|
"/ivp/sc/pvlimit",
|
|
"/ivp/ss/pel_settings",
|
|
"/ivp/ensemble/generator",
|
|
"/ivp/meters",
|
|
"/ivp/meters/readings",
|
|
]
|
|
|
|
for end_point in end_points:
|
|
response = await envoy.request(end_point)
|
|
fixture_data[end_point] = response.text.replace("\n", "").replace(
|
|
serial, CLEAN_TEXT
|
|
)
|
|
fixture_data[f"{end_point}_log"] = json_dumps(
|
|
{
|
|
"headers": dict(response.headers.items()),
|
|
"code": response.status_code,
|
|
}
|
|
)
|
|
return fixture_data
|
|
|
|
|
|
async def async_get_config_entry_diagnostics(
|
|
hass: HomeAssistant, entry: EnphaseConfigEntry
|
|
) -> dict[str, Any]:
|
|
"""Return diagnostics for a config entry."""
|
|
coordinator = entry.runtime_data
|
|
|
|
if TYPE_CHECKING:
|
|
assert coordinator.envoy.data
|
|
envoy_data = coordinator.envoy.data
|
|
envoy = coordinator.envoy
|
|
|
|
device_registry = dr.async_get(hass)
|
|
entity_registry = er.async_get(hass)
|
|
|
|
device_entities = []
|
|
# for each device associated with the envoy get entity and state information
|
|
for device in dr.async_entries_for_config_entry(device_registry, entry.entry_id):
|
|
entities = []
|
|
for entity in er.async_entries_for_device(
|
|
entity_registry, device_id=device.id, include_disabled_entities=True
|
|
):
|
|
state_dict = None
|
|
if state := hass.states.get(entity.entity_id):
|
|
state_dict = dict(state.as_dict())
|
|
state_dict.pop("context", None)
|
|
entity_dict = asdict(entity)
|
|
entity_dict.pop("_cache", None)
|
|
entities.append({"entity": entity_dict, "state": state_dict})
|
|
device_dict = asdict(device)
|
|
device_dict.pop("_cache", None)
|
|
device_entities.append({"device": device_dict, "entities": entities})
|
|
|
|
# remove envoy serial
|
|
old_serial = coordinator.envoy_serial_number
|
|
|
|
coordinator_data = copy.deepcopy(coordinator.data)
|
|
coordinator_data_cleaned = json_dumps(coordinator_data).replace(
|
|
old_serial, CLEAN_TEXT
|
|
)
|
|
|
|
device_entities_cleaned = json_dumps(device_entities).replace(
|
|
old_serial, CLEAN_TEXT
|
|
)
|
|
|
|
envoy_model: dict[str, Any] = {
|
|
"encharge_inventory": envoy_data.encharge_inventory,
|
|
"encharge_power": envoy_data.encharge_power,
|
|
"encharge_aggregate": envoy_data.encharge_aggregate,
|
|
"enpower": envoy_data.enpower,
|
|
"system_consumption": envoy_data.system_consumption,
|
|
"system_production": envoy_data.system_production,
|
|
"system_consumption_phases": envoy_data.system_consumption_phases,
|
|
"system_production_phases": envoy_data.system_production_phases,
|
|
"ctmeter_production": envoy_data.ctmeter_production,
|
|
"ctmeter_consumption": envoy_data.ctmeter_consumption,
|
|
"ctmeter_storage": envoy_data.ctmeter_storage,
|
|
"ctmeter_production_phases": envoy_data.ctmeter_production_phases,
|
|
"ctmeter_consumption_phases": envoy_data.ctmeter_consumption_phases,
|
|
"ctmeter_storage_phases": envoy_data.ctmeter_storage_phases,
|
|
"dry_contact_status": envoy_data.dry_contact_status,
|
|
"dry_contact_settings": envoy_data.dry_contact_settings,
|
|
"inverters": envoy_data.inverters,
|
|
"tariff": envoy_data.tariff,
|
|
}
|
|
|
|
envoy_properties: dict[str, Any] = {
|
|
"envoy_firmware": envoy.firmware,
|
|
"part_number": envoy.part_number,
|
|
"envoy_model": envoy.envoy_model,
|
|
"supported_features": [feature.name for feature in envoy.supported_features],
|
|
"phase_mode": envoy.phase_mode,
|
|
"phase_count": envoy.phase_count,
|
|
"active_phasecount": envoy.active_phase_count,
|
|
"ct_count": envoy.ct_meter_count,
|
|
"ct_consumption_meter": envoy.consumption_meter_type,
|
|
"ct_production_meter": envoy.production_meter_type,
|
|
"ct_storage_meter": envoy.storage_meter_type,
|
|
}
|
|
|
|
fixture_data: dict[str, Any] = {}
|
|
if entry.options.get(OPTION_DIAGNOSTICS_INCLUDE_FIXTURES, False):
|
|
try:
|
|
fixture_data = await _get_fixture_collection(envoy=envoy, serial=old_serial)
|
|
except EnvoyError as err:
|
|
fixture_data["Error"] = repr(err)
|
|
|
|
diagnostic_data: dict[str, Any] = {
|
|
"config_entry": async_redact_data(entry.as_dict(), TO_REDACT),
|
|
"envoy_properties": envoy_properties,
|
|
"raw_data": json_loads(coordinator_data_cleaned),
|
|
"envoy_model_data": envoy_model,
|
|
"envoy_entities_by_device": json_loads(device_entities_cleaned),
|
|
"fixtures": fixture_data,
|
|
}
|
|
|
|
return diagnostic_data
|