Merge pull request #133 from catsmanac/GetFirmwareFromInfo

Show FW on device and in diagnostics

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)
 

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...")

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):