diff --git a/custom_components/enphase_envoy_custom/__init__.py b/custom_components/enphase_envoy_custom/__init__.py index 6021f1a..b53c48d 100644 --- a/custom_components/enphase_envoy_custom/__init__.py +++ b/custom_components/enphase_envoy_custom/__init__.py @@ -108,6 +108,7 @@ async def async_update_data(): ] = await envoy_reader.lifetime_consumption_phase(description.key) data["grid_status"] = await envoy_reader.grid_status() + data["envoy_info"] = await envoy_reader.envoy_info() _LOGGER.debug("Retrieved data from API: %s", data) diff --git a/custom_components/enphase_envoy_custom/envoy_reader.py b/custom_components/enphase_envoy_custom/envoy_reader.py index fd9fe81..174e64b 100644 --- a/custom_components/enphase_envoy_custom/envoy_reader.py +++ b/custom_components/enphase_envoy_custom/envoy_reader.py @@ -12,6 +12,7 @@ import xmltodict from envoy_utils.envoy_utils import EnvoyUtils from homeassistant.util.network import is_ipv6_address +import xmltodict # # Legacy parser is only used on ancient firmwares @@ -33,6 +34,7 @@ ENDPOINT_URL_CHECK_JWT = "https://{}/auth/check_jwt" ENDPOINT_URL_ENSEMBLE_INVENTORY = "http{}://{}/ivp/ensemble/inventory" ENDPOINT_URL_HOME_JSON = "http{}://{}/home.json" +ENDPOINT_URL_INFO_XML = "http{}://{}/info" # pylint: disable=pointless-string-statement @@ -99,6 +101,7 @@ def __init__( # pylint: disable=too-many-arguments use_enlighten_owner_token=False, token_refresh_buffer_seconds=0, store=None, + info_refresh_buffer_seconds=3600, ): """Init the EnvoyReader.""" self.host = host.lower() @@ -129,6 +132,9 @@ def __init__( # pylint: disable=too-many-arguments self.https_flag = https_flag self.use_enlighten_owner_token = use_enlighten_owner_token self.token_refresh_buffer_seconds = token_refresh_buffer_seconds + self.endpoint_info_results = None + self.info_refresh_buffer_seconds = info_refresh_buffer_seconds + self.info_next_refresh_time = datetime.datetime.now() self._store = store self._store_data = {} self._store_update_pending = False @@ -167,6 +173,8 @@ async def _update(self): await self._update_from_p_endpoint() if self.endpoint_type == ENVOY_MODEL_LEGACY: await self._update_from_p0_endpoint() + + await self._update_info_endpoint() async def _update_from_pc_endpoint(self): """Update from PC endpoint.""" @@ -193,6 +201,25 @@ async def _update_from_p0_endpoint(self): "endpoint_production_results", ENDPOINT_URL_PRODUCTION ) + async def _update_info_endpoint(self): + """Update from info endpoint if next time expried.""" + if self.info_next_refresh_time <= datetime.datetime.now(): + await self._update_endpoint("endpoint_info_results", ENDPOINT_URL_INFO_XML) + self.info_next_refresh_time = datetime.datetime.now() + datetime.timedelta( + seconds=self.info_refresh_buffer_seconds + ) + _LOGGER.debug( + "Info endpoint updated, set next update time: %s using interval: %s", + self.info_next_refresh_time, + self.info_refresh_buffer_seconds, + ) + else: + _LOGGER.debug( + "Info endpoint next update time is: %s using interval: %s", + self.info_next_refresh_time, + self.info_refresh_buffer_seconds, + ) + async def _update_endpoint(self, attr, url): """Update a property from an endpoint.""" formatted_url = url.format(self.https_flag, self.host) @@ -436,6 +463,8 @@ async def detect_model(self): + "Please enter in the needed Enlighten credentials during setup." ) + await self._update_info_endpoint() + if ( self.endpoint_production_json_results and self.endpoint_production_json_results.status_code == 200 @@ -842,6 +871,29 @@ async def grid_status(self): self.has_grid_status = False return None + async def envoy_info(self): + """Return information reported by Envoy info.xml.""" + device_data = {} + + if self.endpoint_info_results: + try: + data = xmltodict.parse(self.endpoint_info_results.text) + device_data["software"] = data["envoy_info"]["device"]["software"] + device_data["pn"] = data["envoy_info"]["device"]["pn"] + device_data["metered"] = data["envoy_info"]["device"]["imeter"] + except Exception: # pylint: disable=broad-except + pass + # add internal key information for envoy class + device_data["Using-model"] = self.endpoint_type + device_data["Using-httpsflag"] = self.https_flag + device_data["Using-MeteringEnabled"] = self.isMeteringEnabled + device_data["Using-GetInverters"] = self.get_inverters + device_data["Using-UseEnligthen"] = self.use_enlighten_owner_token + device_data["Using-InfoUpdateInterval"] = self.info_refresh_buffer_seconds + device_data["Using-hasgridstatus"] = self.has_grid_status + + return device_data + def run_in_console(self): """If running this module directly, print all the values in the console.""" print("Reading...") diff --git a/custom_components/enphase_envoy_custom/sensor.py b/custom_components/enphase_envoy_custom/sensor.py index e20bc0e..ca90fea 100644 --- a/custom_components/enphase_envoy_custom/sensor.py +++ b/custom_components/enphase_envoy_custom/sensor.py @@ -189,11 +189,21 @@ def device_info(self) -> DeviceInfo | None: """Return the device_info of the device.""" if not self._device_serial_number: return None + + sw_version = None + hw_version = None + + if self.coordinator.data.get("envoy_info"): + sw_version = self.coordinator.data.get("envoy_info").get("software", None) + hw_version = self.coordinator.data.get("envoy_info").get("pn", None) + return DeviceInfo( identifiers={(DOMAIN, str(self._device_serial_number))}, manufacturer="Enphase", model="Envoy", name=self._device_name, + sw_version=sw_version, + hw_version=hw_version, ) class CoordinatedEnvoyEntity(EnvoyEntity, CoordinatorEntity):