From 7cb4e71959f7a72bea60a78afafe60edc51f2eab Mon Sep 17 00:00:00 2001 From: Pierre Date: Wed, 20 May 2026 22:40:08 +0200 Subject: [PATCH] Add AppDaemon version sensor to InformationsCollector --- apps/apps.yaml | 2 + apps/informationscollector.py | 93 +++++++++++++++++++++++++++++++++++ 2 files changed, 95 insertions(+) diff --git a/apps/apps.yaml b/apps/apps.yaml index 15b4370..26927fb 100644 --- a/apps/apps.yaml +++ b/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 diff --git a/apps/informationscollector.py b/apps/informationscollector.py index b330e99..f92c0e7 100644 --- a/apps/informationscollector.py +++ b/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')