Add AppDaemon version sensor to InformationsCollector

apps/apps.yaml

@@ -12,6 +12,8 @@ informations_collector:
 
   mute: True
 
+  appdaemon_version_sensor: sensor.ad_appdaemon_version
+
   error_collector:
     reset_error_count_button: input_button.reset_error_count
     output_sensor: sensor.ad_errors

apps/informationscollector.py

@@ -1,16 +1,41 @@
 import ad_toolbox.smartcondition as SmartCondition
 from ad_toolbox.smartobject import SmartObject
+from importlib import metadata
+import json
 import pickle
 import os
+from urllib.request import urlopen
 
 class InformationsCollector(SmartObject):
 
+    APPDAEMON_PYPI_URL = "https://pypi.org/pypi/appdaemon/json"
+
     def on_initialize_smart_object(self):
         super().on_initialize_smart_object()
 
         if self.dataset is None:
             self.dataset = dict()
 
+        sensor_name = self.args.get('appdaemon_version_sensor')
+        legacy_version_config = self.args.get('appdaemon_version_collector')
+        if sensor_name is None and isinstance(legacy_version_config, dict):
+            sensor_name = legacy_version_config.get('output_sensor')
+
+        if sensor_name is not None:
+            self.appdaemon_version_sensor = self.create_entity(
+                sensor_name,
+                state='unknown',
+                attributes={
+                    'latest_official_version': self.dataset.get('appdaemon_latest_official_version'),
+                    'update_available': None,
+                },
+                icon='mdi:package-up'
+            )
+            self.update_appdaemon_version_sensor()
+            self.run_daily(self.update_appdaemon_version_sensor, "06:00:00")
+        elif legacy_version_config is not None:
+            self.log_error("Invalid appdaemon version sensor configuration")
+
         if 'error_collector' in self.args:
             if 'output_sensor' in self.args['error_collector']:
                 sensor_name = self.args['error_collector']['output_sensor']
@@ -95,6 +120,74 @@ class InformationsCollector(SmartObject):
         self.dataset['min_temp'] = temperature
         self.min_temp_sensor.set_state(state = temperature, attributes = { 'unit_of_measurement': '°C', 'device_class': 'temperature'})
 
+    def update_appdaemon_version_sensor(self, kwargs=None):
+        installed_version = self._get_installed_appdaemon_version()
+        latest_official_version = self._get_latest_official_appdaemon_version()
+
+        if latest_official_version is not None:
+            self.dataset['appdaemon_latest_official_version'] = latest_official_version
+        else:
+            latest_official_version = self.dataset.get('appdaemon_latest_official_version')
+
+        update_available = self._is_appdaemon_update_available(installed_version, latest_official_version)
+
+        self.appdaemon_version_sensor.set_state(
+            state=installed_version or 'unknown',
+            attributes={
+                'latest_official_version': latest_official_version,
+                'update_available': update_available,
+            },
+        )
+
+    def _get_installed_appdaemon_version(self):
+        try:
+            return metadata.version('appdaemon')
+        except metadata.PackageNotFoundError:
+            self.log_warning("AppDaemon package metadata is unavailable")
+        except Exception as err:
+            self.log_warning(f"Unable to read installed AppDaemon version: {err}")
+        return None
+
+    def _get_latest_official_appdaemon_version(self):
+        try:
+            with urlopen(self.APPDAEMON_PYPI_URL, timeout=10) as response:
+                payload = json.load(response)
+            return payload.get('info', {}).get('version')
+        except Exception as err:
+            self.log_warning(f"Unable to fetch latest official AppDaemon version: {err}")
+            return None
+
+    def _is_appdaemon_update_available(self, installed_version, latest_official_version):
+        if installed_version is None or latest_official_version is None:
+            return None
+
+        installed_parts = self._parse_version_parts(installed_version)
+        latest_parts = self._parse_version_parts(latest_official_version)
+
+        if installed_parts is None or latest_parts is None:
+            return latest_official_version != installed_version
+
+        return latest_parts > installed_parts
+
+    def _parse_version_parts(self, version):
+        if not isinstance(version, str):
+            return None
+
+        parts = []
+        for token in version.split('.'):
+            numeric_prefix = []
+            for char in token:
+                if not char.isdigit():
+                    break
+                numeric_prefix.append(char)
+
+            if not numeric_prefix:
+                return None
+
+            parts.append(int(''.join(numeric_prefix)))
+
+        return tuple(parts)
+
     def terminate(self):
         self.log_info("Writing dataset to " + self.get_dataset_name())
         f = open(self.get_dataset_name(), 'wb')