276 lines
12 KiB
Python
276 lines
12 KiB
Python
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']
|
|
self.error_list = []
|
|
self.log_handle = self.listen_log(self.on_log_error, "ERROR", log="main_log")
|
|
self.error_count_sensor = self.create_entity(
|
|
sensor_name, state=0,
|
|
attributes={"last_error": "", "error_list": []},
|
|
icon="mdi:alert-circle-outline"
|
|
)
|
|
if 'reset_error_count_button' in self.args['error_collector']:
|
|
self.listen_state(self.on_reset_error_count, self.args['error_collector']['reset_error_count_button'])
|
|
else:
|
|
self.log_error("No output_sensor defined for error_collector")
|
|
|
|
if 'min_max_temp_sensors' in self.args:
|
|
for sensor in self.args['min_max_temp_sensors']['temperature_sensors']:
|
|
self.listen_state(self.on_temperature_change, sensor)
|
|
|
|
output_sensors = self.args['min_max_temp_sensors']['output_sensors']
|
|
self.max_temp_sensor = self.create_entity(output_sensors['max_temp'])
|
|
self.max_temp_yesterday_sensor = self.create_entity(output_sensors['max_temp_yesterday'])
|
|
self.min_temp_sensor = self.create_entity(output_sensors['min_temp'])
|
|
self.min_temp_yesterday_sensor = self.create_entity(output_sensors['min_temp_yesterday'])
|
|
|
|
current_min_temp, current_max_temp = self.compute_min_max_temp()
|
|
if not self.max_temp_sensor.exists() or self.max_temp_sensor.get_state() == 'unavailable':
|
|
max_temp = current_max_temp
|
|
try:
|
|
if max_temp is None or self.dataset['max_temp'] > max_temp:
|
|
max_temp = self.dataset['max_temp']
|
|
self.log_info(f"Restoring max_temp from dataset ({max_temp}°C)")
|
|
except KeyError: pass
|
|
self.set_max_temp_sensor(max_temp)
|
|
if not self.min_temp_sensor.exists() or self.min_temp_sensor.get_state() == 'unavailable':
|
|
min_temp = current_min_temp
|
|
try:
|
|
if min_temp is None or self.dataset['min_temp'] < min_temp:
|
|
min_temp = self.dataset['min_temp']
|
|
self.log_info(f"Restoring min_temp from dataset ({min_temp}°C)")
|
|
except KeyError: pass
|
|
self.set_min_temp_sensor(min_temp)
|
|
|
|
self.run_daily(self.run_at_midnight, "00:00:00")
|
|
|
|
if 'group_collectors' in self.args:
|
|
self._group_sensors = {}
|
|
self._group_evaluators = {}
|
|
|
|
for name, config in self.args['group_collectors'].items():
|
|
self._group_sensors[name] = self.create_entity(
|
|
config['output_sensor'], state=0,
|
|
attributes={'active_entities': ''},
|
|
icon='mdi:counter'
|
|
)
|
|
|
|
entities = self._get_group_entities(config)
|
|
|
|
if 'active_condition' in config:
|
|
self._group_evaluators[name] = {
|
|
entity: SmartCondition.Evaluator(
|
|
self,
|
|
config['active_condition'],
|
|
condition_name=f"{name}_{entity}",
|
|
constants={'self': entity},
|
|
log_callback_trigger_reason=False,
|
|
on_change_cb=self._on_group_condition_changed,
|
|
)
|
|
for entity in entities
|
|
}
|
|
else:
|
|
for entity in entities:
|
|
self.listen_state(self.on_group_entity_state_changed, entity)
|
|
|
|
self.update_group_collectors()
|
|
|
|
def set_max_temp_sensor(self,temperature):
|
|
self.dataset['max_temp'] = temperature
|
|
self.max_temp_sensor.set_state(state = temperature, attributes = { 'unit_of_measurement': '°C', 'device_class': 'temperature'})
|
|
|
|
def set_min_temp_sensor(self,temperature):
|
|
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')
|
|
pickle.dump(self.dataset, f)
|
|
f.close()
|
|
|
|
def compute_min_max_temp(self):
|
|
max_temp = None
|
|
min_temp = None
|
|
for sensor in self.args['min_max_temp_sensors']['temperature_sensors']:
|
|
temp = self.get_state(sensor)
|
|
if not max_temp or temp > max_temp: max_temp = temp
|
|
if not min_temp or temp < min_temp: min_temp = temp
|
|
return min_temp,max_temp
|
|
|
|
def on_log_error(self, name, ts, level, type, message, kwargs):
|
|
error_count = int(self.error_count_sensor.get_state()) + 1
|
|
self.error_list.append(message)
|
|
self.error_count_sensor.set_state(state=error_count, attributes={"last_error": message, "error_list": self.error_list})
|
|
|
|
def on_reset_error_count(self, entity, attribute, old, new, kwargs):
|
|
error_log_file = self.get_ad_api()._logging.config['error_log']['filename']
|
|
open(error_log_file, 'w').close()
|
|
self.error_list.clear()
|
|
self.error_count_sensor.set_state(state=0, attributes={"last_error": "", "error_list": self.error_list})
|
|
|
|
def _on_group_condition_changed(self, prev_result,result):
|
|
self.update_group_collectors()
|
|
|
|
def on_group_entity_state_changed(self, entity, attribute, old, new, kwargs):
|
|
if new != old:
|
|
self.log_info(f"updating group collectors ({entity} changed from {old} to {new})")
|
|
self.update_group_collectors()
|
|
|
|
def on_temperature_change(self, entity, attribute, old, new, kwargs):
|
|
if new != old:
|
|
min_temp = self.min_temp_sensor.get_state()
|
|
max_temp = self.max_temp_sensor.get_state()
|
|
if max_temp is None or max_temp == 'unavailable' or new > max_temp: self.set_max_temp_sensor(new)
|
|
if min_temp is None or min_temp == 'unavailable' or new < min_temp: self.set_min_temp_sensor(new)
|
|
|
|
def run_at_midnight(self, kwargs):
|
|
self.max_temp_yesterday_sensor.set_state(state=self.max_temp_sensor.get_state(), attributes={'unit_of_measurement': '°C', 'device_class': 'temperature'})
|
|
self.min_temp_yesterday_sensor.set_state(state=self.min_temp_sensor.get_state(), attributes={'unit_of_measurement': '°C', 'device_class': 'temperature'})
|
|
min_temp, max_temp = self.compute_min_max_temp()
|
|
self.set_max_temp_sensor(max_temp)
|
|
self.set_min_temp_sensor(min_temp)
|
|
|
|
def _get_group_entities(self, config):
|
|
group = config['group']
|
|
if isinstance(group, str):
|
|
return self.get_state(group, attribute='entity_id') or []
|
|
entities = []
|
|
for g in group:
|
|
entities.extend(self.get_state(g, attribute='entity_id') or [])
|
|
return entities
|
|
|
|
def _is_entity_active(self, name, entity):
|
|
if name in self._group_evaluators:
|
|
return self._group_evaluators[name][entity].evaluate(log=False) == SmartCondition.Result.Succeeded
|
|
return self.get_state(entity) == 'on'
|
|
|
|
# def _get_group_entity_name(self, entity):
|
|
# entity_state = self.get_state(entity, attribute='all') or {}
|
|
# attributes = entity_state.get('attributes', {})
|
|
# friendly_name = attributes.get('friendly_name')
|
|
# return friendly_name if isinstance(friendly_name, str) and friendly_name else entity
|
|
|
|
def update_group_collectors(self):
|
|
for name, config in self.args['group_collectors'].items():
|
|
sensor = self._group_sensors[name]
|
|
prev_state = sensor.get_state()
|
|
prev_count = int(prev_state) if prev_state is not None else 0
|
|
|
|
active_entities = []
|
|
for entity in self._get_group_entities(config):
|
|
if self.entity_exists(entity) and self._is_entity_active(name, entity):
|
|
active_entities.append(entity)
|
|
|
|
count = len(active_entities)
|
|
sensor.set_state(state=count, attributes={'active_entities': ', '.join(active_entities)})
|
|
|
|
if prev_count != count:
|
|
self.log_info(f"{name}: {count} active entities (previously {prev_count})")
|
|
|