935 lines
48 KiB
Python
935 lines
48 KiB
Python
import appdaemon.plugins.hass.hassapi as hass
|
|
import smartcondition as SmartCondition
|
|
import smartvalue as SmartValue
|
|
from kwargsparser import kwargsParser
|
|
from expressionparser import ParsingException
|
|
import time
|
|
import datetime
|
|
import re
|
|
from logger_interface import LoggerInterface
|
|
from ad_toolbox.eventhandler import EventHandler
|
|
|
|
class VirtualSensorBase:
|
|
def __init__(self,ad_api,logger_interface, virtual_sensor_name = None, sensor_name = None,super_entity_id = None, yaml_block = None,templates_library = None, constants = None, app_name = None,self_initialize = False):
|
|
self.ad_api = ad_api
|
|
self.logger_interface = logger_interface
|
|
|
|
# sensor_name should be something like 'domain.name' whereas virtual_sensor_name should be only 'name'
|
|
# only one should be provided and the other one will be deduced.
|
|
# the regular use case is giving virtual_sensor_name.
|
|
# Giving a specific sensor_name is usefull when the entity exist before with a different domain
|
|
if virtual_sensor_name:
|
|
assert sensor_name == None
|
|
self.virtual_sensor_name = virtual_sensor_name
|
|
self.sensor_name = self.generate_sensor_name(virtual_sensor_name)
|
|
else:
|
|
assert virtual_sensor_name == None
|
|
self.sensor_name = sensor_name
|
|
self.virtual_sensor_name = sensor_name.split('.')[1]
|
|
self.sensor_domain = self.sensor_name.split(".")[0]
|
|
|
|
if not yaml_block:
|
|
self.log_error(f"{self.sensor_name} have a null Yaml block")
|
|
self.yaml_block = yaml_block
|
|
self.templates_library = templates_library
|
|
|
|
try: self.attributes = self.yaml_block['attributes']
|
|
except (TypeError,KeyError): self.attributes = dict()
|
|
|
|
if app_name:
|
|
self.attributes['ad_app'] = app_name
|
|
|
|
if constants:
|
|
self.constants = dict(constants) # we don't want to modify the parent dict
|
|
self.constants['self'] = self.sensor_name # we need to set self by ourself
|
|
else: self.constants = {'self' : self.sensor_name }
|
|
|
|
if super_entity_id != None:
|
|
self.constants['super'] = super_entity_id
|
|
|
|
self.entity_existed = self.ad_api.entity_exists(self.sensor_name)
|
|
if not self.entity_existed:
|
|
self.ad_api.set_state(self.sensor_name, state='unknown',attributes=self.attributes, check_existence=False)
|
|
self.entity = self.ad_api.get_entity(self.sensor_name)
|
|
|
|
if self_initialize:
|
|
self._initialize()
|
|
|
|
def generate_sensor_name(self,virtual_sensor_name):
|
|
return f"sensor.{virtual_sensor_name}"
|
|
# TODO : restore that code once we rename the sensor
|
|
# splitted_name = virtual_sensor_name.split(".")
|
|
# assert len(splitted_name) == 2,f"Invalid virtual sensor name : {virtual_sensor_name}"
|
|
# return f"sensor.{splitted_name[1]}"
|
|
|
|
# def log(self,message): self.logger_interface.log_info(f"[{self.sensor_name}]{message}")
|
|
|
|
# def log_info(self,message): self.logger_interface.log_info(f"[{self.sensor_name}]{message}")
|
|
|
|
# def log_warning(self,message): self.logger_interface.log_warning(f"[{self.sensor_name}]{message}")
|
|
|
|
# def log_error(self,message): self.logger_interface.log_error(f"[{self.sensor_name}]{message}")
|
|
|
|
def log_info(self,message): self.logger_interface.log_info(message)
|
|
|
|
def log_warning(self,message): self.logger_interface.log_warning(message)
|
|
|
|
def log_error(self,message): self.logger_interface.log_error(message)
|
|
|
|
def generate_dependencies_list(self): return []
|
|
|
|
@staticmethod
|
|
def is_entity_reference(value):
|
|
# Keep plain entity IDs on the legacy path so their raw state type is preserved.
|
|
return isinstance(value, str) and re.fullmatch(r"[a-z_][a-z0-9_]*\.[a-z0-9_]+", value) is not None
|
|
|
|
def generate_attributes_dependencies_list(self):
|
|
dependencies = list()
|
|
if 'sensor_attributes' in self.yaml_block:
|
|
for expression in self.yaml_block['sensor_attributes'].values():
|
|
if self.is_entity_reference(expression):
|
|
dependencies.append(expression)
|
|
else:
|
|
evaluator = SmartValue.Evaluator(
|
|
self.ad_api, expression, constants=self.constants,
|
|
sensor_existence_validation=False, activated=False, log_init=False)
|
|
self.extract_dependencies_from_entities_to_listen(
|
|
evaluator.get_entities_to_listen_as_list(), dependencies)
|
|
return dependencies
|
|
|
|
def _initialize(self):
|
|
self.attribute_expressions = dict()
|
|
if 'sensor_attributes' in self.yaml_block:
|
|
for attribute_name,expression in self.yaml_block['sensor_attributes'].items():
|
|
if self.is_entity_reference(expression):
|
|
if not self.ad_api.entity_exists(expression):
|
|
self.log_error(f"Error in sensor_attribute. {attribute_name}:{expression}. {expression} doesn't exist")
|
|
else:
|
|
self.attributes[attribute_name] = self.ad_api.get_state(expression)
|
|
self.ad_api.listen_state(self.on_attribute_sensor_state_change,expression)
|
|
else:
|
|
try:
|
|
evaluator = SmartValue.Evaluator(
|
|
self.ad_api, expression, logger_interface=self.logger_interface,
|
|
expression_name=f"{self.sensor_name}.{attribute_name}",
|
|
constants=self.constants, sensor_existence_validation=False,
|
|
trigger_callback_on_activation=False,
|
|
on_change_cb=self.on_attribute_expression_change,
|
|
pass_expression_name_to_cb=attribute_name)
|
|
self.attribute_expressions[attribute_name] = evaluator
|
|
self.attributes[attribute_name] = evaluator.evaluate(False)
|
|
except ParsingException as error:
|
|
self.log_error(error)
|
|
|
|
try: self.initialize()
|
|
except ParsingException as e:
|
|
self.log_error(e)
|
|
self.ad_api.set_state(self.sensor_name, attributes = self.attributes)
|
|
|
|
def on_attribute_sensor_state_change(self, entity, attribute, old, new, kwargs):
|
|
attributes_update = False
|
|
for attribute_name,sensor_name in self.yaml_block['sensor_attributes'].items():
|
|
if sensor_name == entity:
|
|
self.attributes[attribute_name] = new
|
|
attributes_update = True
|
|
|
|
if attributes_update:
|
|
self.ad_api.set_state(self.sensor_name, attributes = self.attributes)
|
|
|
|
def on_attribute_expression_change(self, attribute_name, previous_value, value):
|
|
self.attributes[attribute_name] = value
|
|
self.ad_api.set_state(self.sensor_name, attributes=self.attributes)
|
|
|
|
def initialize(self): assert False
|
|
|
|
def extract_dependencies_from_entities_to_listen(self,entities_to_listen,dependencies):
|
|
for entity_with_attribute in entities_to_listen:
|
|
splitted_entity = entity_with_attribute.split('.')
|
|
assert len(splitted_entity) == 3 or len(splitted_entity) == 2
|
|
if len(splitted_entity) == 3:
|
|
entity = f"{splitted_entity[0]}.{splitted_entity[1]}"
|
|
else:
|
|
entity = entity_with_attribute
|
|
|
|
if entity != self.sensor_name and not entity in dependencies:
|
|
dependencies.append(entity)
|
|
|
|
class Averager(VirtualSensorBase):
|
|
def initialize(self):
|
|
self.round = None
|
|
self.confirm_steady_cb_handle = None
|
|
self.confirm_tendency_cb_handle = None
|
|
self.icons_config = None
|
|
self.sensors_list = list()
|
|
self.average_sensor_attributes = dict()
|
|
self.all_sensors_are_mandatory = True
|
|
if isinstance(self.yaml_block,dict):
|
|
if 'all_sensors_are_mandatory' in self.yaml_block: self.all_sensors_are_mandatory = self.yaml_block['all_sensors_are_mandatory']
|
|
if 'round' in self.yaml_block: self.round = self.yaml_block['round']
|
|
if 'sensors' in self.yaml_block:
|
|
self.sensors_list = self.yaml_block['sensors'] if isinstance(self.yaml_block['sensors'],list) else [self.yaml_block['sensors']]
|
|
if 'icons' in self.yaml_block:
|
|
self.icons_config = self.yaml_block['icons']
|
|
if 'attributes' in self.yaml_block:
|
|
self.average_sensor_attributes = self.yaml_block['attributes']
|
|
else:
|
|
self.average_sensor_attributes = { 'tendency' : 'steady' }
|
|
else:
|
|
self.sensors_list = [self.yaml_block]
|
|
|
|
for sensor in self.sensors_list:
|
|
self.ad_api.listen_state(self.on_sensor_state_change,sensor)
|
|
|
|
self.average_sensor = self.ad_api.get_entity(self.sensor_name)
|
|
|
|
if not self.average_sensor.exists():
|
|
average = self.compute_average()
|
|
self.average_sensor.set_state(state = average if average else 'unavailable',attributes = self.average_sensor_attributes)
|
|
|
|
self.update_average()
|
|
|
|
def generate_dependencies_list(self):
|
|
if isinstance(self.yaml_block,dict):
|
|
if 'sensors' in self.yaml_block:
|
|
return self.yaml_block['sensors'] if isinstance(self.yaml_block['sensors'],list) else [self.yaml_block['sensors']]
|
|
else:
|
|
return []
|
|
else:
|
|
return [ self.yaml_block ]
|
|
|
|
def on_sensor_state_change(self, *args):
|
|
self.update_average()
|
|
|
|
def compute_average(self):
|
|
result_count = 0
|
|
average = 0
|
|
for sensor in self.sensors_list:
|
|
try:
|
|
value = float(self.ad_api.get_state(sensor))
|
|
result_count += 1
|
|
average += value
|
|
except (TypeError,ValueError):
|
|
if self.all_sensors_are_mandatory:
|
|
return None
|
|
|
|
if result_count == 0: return None
|
|
average /= result_count
|
|
if self.round != None:
|
|
average = round(average,self.round)
|
|
return average
|
|
|
|
|
|
def update_average(self):
|
|
try: prev_average = float(self.average_sensor.get_state())
|
|
except (TypeError,ValueError): prev_average = None
|
|
new_average = self.compute_average()
|
|
if new_average and new_average != prev_average:
|
|
#if we don't change the value for one hour, let's go back to the default icon
|
|
if self.confirm_steady_cb_handle != None:
|
|
self.ad_api.cancel_timer(self.confirm_steady_cb_handle)
|
|
self.confirm_steady_cb_handle = None
|
|
self.confirm_steady_cb_handle = self.ad_api.run_in(self.on_confirm_steady, 60 * 60)
|
|
|
|
if prev_average:
|
|
if self.confirm_tendency_cb_handle:
|
|
self.ad_api.cancel_timer(self.confirm_tendency_cb_handle)
|
|
self.confirm_tendency_cb_handle = None
|
|
|
|
prev_tendency = self.average_sensor.get_state(attribute = 'tendency')
|
|
if new_average > prev_average:
|
|
if prev_tendency == 'down':
|
|
self.average_sensor_attributes['tendency'] = 'steady'
|
|
self.confirm_tendency_cb_handle = self.ad_api.run_in(self.on_confirm_tendency, 30 * 60,tendency = 'up')
|
|
else:
|
|
self.average_sensor_attributes['tendency'] = 'up'
|
|
else:
|
|
if prev_tendency == 'up':
|
|
self.average_sensor_attributes['tendency'] = 'steady'
|
|
self.confirm_tendency_cb_handle = self.ad_api.run_in(self.on_confirm_tendency, 30 * 60,tendency = 'down')
|
|
else:
|
|
self.average_sensor_attributes['tendency'] = 'down'
|
|
|
|
self.update_icon_attribute()
|
|
|
|
self.average_sensor.set_state(state = new_average,attributes = self.average_sensor_attributes)
|
|
|
|
def update_icon_attribute(self):
|
|
if self.icons_config:
|
|
if self.average_sensor_attributes['tendency'] == 'up' and 'up' in self.icons_config:
|
|
self.average_sensor_attributes['icon'] = self.icons_config['up']
|
|
if self.average_sensor_attributes['tendency'] == 'down' and 'down' in self.icons_config:
|
|
self.average_sensor_attributes['icon'] = self.icons_config['down']
|
|
if self.average_sensor_attributes['tendency'] == 'steady' and 'default' in self.icons_config:
|
|
self.average_sensor_attributes['icon'] = self.icons_config['default']
|
|
|
|
def on_confirm_tendency(self, kwargs):
|
|
tendency = kwargs['tendency']
|
|
self.confirm_tendency_cb_handle = None
|
|
|
|
self.average_sensor_attributes['tendency'] = tendency
|
|
self.update_icon_attribute()
|
|
self.average_sensor.set_state(attributes = self.average_sensor_attributes)
|
|
|
|
def on_confirm_steady(self, kwargs):
|
|
self.confirm_steady_cb_handle = None
|
|
self.average_sensor_attributes['tendency'] = 'steady'
|
|
self.update_icon_attribute()
|
|
self.average_sensor.set_state(attributes = self.average_sensor_attributes)
|
|
|
|
class ValueSelector(VirtualSensorBase):
|
|
class SelectorConfig:
|
|
def __init__(self,name,conditions,value,is_smart_value):
|
|
self.conditions = conditions
|
|
self.value = str(value) #value like "17" are deserialized as int
|
|
self.name = str(name)
|
|
self.is_smart_value = is_smart_value
|
|
|
|
class Selector:
|
|
def __init__(self,config, ad_api = None,logger_interface = None, constants = None, templates_library = None):
|
|
self.config = config
|
|
self.conditions = SmartCondition.Evaluator(ad_api,self.config.conditions, logger_interface = logger_interface, unavaibility_result = SmartCondition.Result.Failed,condition_name = self.config.name, log_callback_trigger_reason = False, constants = constants, templates_library = templates_library)
|
|
if config.is_smart_value:
|
|
self.value = SmartValue.Evaluator(ad_api,self.config.value, logger_interface = logger_interface,expression_name = self.config.name, log_callback_trigger_reason = False, constants = constants)
|
|
else:
|
|
self.value = None
|
|
|
|
def compute_value(self):
|
|
if self.value: return self.value.evaluate(False)
|
|
else: return self.config.value
|
|
|
|
def get_entities_to_listen(self):
|
|
if self.value:
|
|
entities_to_listen = list(self.conditions.get_entities_to_listen_as_list())
|
|
entities_to_listen.extend(self.value.get_entities_to_listen_as_list())
|
|
return entities_to_listen
|
|
else:
|
|
return self.conditions.get_entities_to_listen_as_list()
|
|
|
|
|
|
def initialize(self):
|
|
self.value_selectors = list()
|
|
selectors_config = self.process_config()
|
|
|
|
self.output_sensor = self.ad_api.get_entity(self.sensor_name)
|
|
|
|
if not self.output_sensor.exists():
|
|
default_config = selectors_config[-1] #the last config is the default config
|
|
if default_config.is_smart_value:
|
|
default_value = SmartValue.Evaluator(self.ad_api,default_config.value,expression_name = default_config.name, logger_interface = self.logger_interface, log_callback_trigger_reason = False, constants = self.constants).evaluate()
|
|
else:
|
|
default_value = default_config.value
|
|
self.set_value(default_value)
|
|
|
|
entities_to_listen = set()
|
|
|
|
for config in selectors_config:
|
|
new_selector = self.Selector(config,ad_api = self.ad_api, logger_interface = self.logger_interface, constants = self.constants, templates_library = self.templates_library)
|
|
self.value_selectors.append(new_selector)
|
|
entities_to_listen.update(new_selector.get_entities_to_listen())
|
|
|
|
for entity_to_listen in entities_to_listen:
|
|
slitted_entity = entity_to_listen.split(".")
|
|
if len(slitted_entity) == 3:
|
|
self.ad_api.listen_state(self.on_entities_to_listen_changed,f"{slitted_entity[0]}.{slitted_entity[1]}",attribute = slitted_entity[2])
|
|
else:
|
|
self.ad_api.listen_state(self.on_entities_to_listen_changed,entity_to_listen)
|
|
|
|
self.update_value()
|
|
|
|
def on_entities_to_listen_changed(self, entity, attribute, old, new, kwargs):
|
|
self.update_value()
|
|
|
|
def process_config(self) -> list[SelectorConfig]:
|
|
configs = list()
|
|
#self.log_info(f"yaml_block:{self.yaml_block}")
|
|
if 'values' in self.yaml_block: values_block = self.yaml_block['values']
|
|
else: values_block = self.yaml_block
|
|
|
|
for key,item in values_block.items():
|
|
try: conditions_block = item['conditions']
|
|
except (TypeError,KeyError): conditions_block = item
|
|
|
|
try:
|
|
value = item['value']
|
|
is_smart_value = True
|
|
except (TypeError,KeyError):
|
|
value = key
|
|
is_smart_value = False #SmartValue are not supported for 'inline' value like "17: some_condition"
|
|
|
|
configs.append(self.SelectorConfig(key,conditions_block,value,is_smart_value))
|
|
|
|
return configs
|
|
|
|
def generate_dependencies_list(self):
|
|
dependencies = list()
|
|
|
|
selectors_config = self.process_config()
|
|
|
|
for config in selectors_config:
|
|
#self.log_info(f"config: name = {config.name},conditions = {config.conditions},value = {config.value}")
|
|
conditions = SmartCondition.Evaluator(self.ad_api,config.conditions, logger_interface = self.logger_interface,condition_name = config.name, sensor_existence_validation = False, constants = self.constants, log_init = False)
|
|
self.extract_dependencies_from_entities_to_listen(conditions.get_entities_to_listen_as_list(),dependencies)
|
|
|
|
if config.is_smart_value:
|
|
value = SmartValue.Evaluator(self.ad_api,config.value,expression_name = config.name, sensor_existence_validation = False, constants = self.constants, log_init = False)
|
|
self.extract_dependencies_from_entities_to_listen(value.get_entities_to_listen_as_list(),dependencies)
|
|
|
|
return dependencies
|
|
|
|
def on_value_condition_change(self, *kwargs):
|
|
self.update_value()
|
|
|
|
def set_value(self, value):
|
|
if self.sensor_domain == 'number':
|
|
self.ad_api.call_service("number/set_value", entity_id = self.sensor_name,value = value)
|
|
else:
|
|
self.output_sensor.set_state(state = value)
|
|
|
|
def update_value(self):
|
|
with self.logger_interface.log_group(self.sensor_name):
|
|
previous_value = str(self.output_sensor.get_state())
|
|
|
|
for selector in self.value_selectors:
|
|
if selector.conditions.evaluate(False) == SmartCondition.Result.Succeeded:
|
|
new_value = selector.compute_value()
|
|
if previous_value != str(new_value):
|
|
#TODO: find a way to know wether if it's the condition or the value that trigger the update
|
|
selector.conditions.log_callback_trigger_reason()
|
|
selector.conditions.log_evaluation_result()
|
|
if selector.value:
|
|
selector.value.log_callback_trigger_reason()
|
|
selector.value.log_evaluation_result()
|
|
self.set_value(new_value)
|
|
break
|
|
|
|
self.log_info(f"{self.sensor_name} = {self.output_sensor.get_state()}")
|
|
|
|
class ContinuousCondition(VirtualSensorBase):
|
|
def initialize(self):
|
|
self.cb_handle = None
|
|
self._timer_generation = 0
|
|
self.time = self.yaml_block['time']
|
|
try: self.ignore_unavailable_condition = self.yaml_block['ignore_unavailable_condition']
|
|
except KeyError: self.ignore_unavailable_condition = False
|
|
try: self.invert = self.yaml_block['invert']
|
|
except KeyError: self.invert = False
|
|
self.log_info(f"Initializing continuous condition for {self.sensor_name}")
|
|
self.output_sensor = self.ad_api.get_entity(self.sensor_name)
|
|
|
|
if not self.output_sensor.exists():
|
|
self.log_info(f"{self.sensor_name} does not exist")
|
|
try:
|
|
self.output_sensor.set_state(state = self.yaml_block['default_value'])
|
|
self.log_info(f"adding {self.sensor_name} with default state {self.yaml_block['default_value']}")
|
|
except KeyError:
|
|
self.output_sensor.set_state(state = 'off')
|
|
self.log_info(f"adding {self.sensor_name} with default state off")
|
|
else:
|
|
self.log_info(f"{self.sensor_name} already exist with state = {self.output_sensor.get_state()}")
|
|
|
|
if self.invert:
|
|
succeed_cb = self.on_condition_fail
|
|
fail_cb = self.on_condition_succeed
|
|
else:
|
|
succeed_cb = self.on_condition_succeed
|
|
fail_cb = self.on_condition_fail
|
|
|
|
self.condition = SmartCondition.Evaluator(self.ad_api,self.yaml_block['conditions'], unavaibility_result = SmartCondition.Result.Unavailable if self.ignore_unavailable_condition else None, logger_interface = self.logger_interface,condition_name = self.sensor_name,on_succeed_cb = succeed_cb, on_fail_cb = fail_cb, constants = self.constants, templates_library = self.templates_library)
|
|
# new_value = 'on' if self.condition.evaluate() == SmartCondition.Result.Succeeded else 'off'
|
|
# self.output_sensor.set_state(state = new_value)
|
|
|
|
def generate_sensor_name(self,virtual_sensor_name):
|
|
return f"binary_sensor.{virtual_sensor_name}"
|
|
#TODO : restore that code once we rename the sensor
|
|
splitted_name = virtual_sensor_name.split(".")
|
|
assert len(splitted_name) == 2,f"Invalid virtual sensor name : {virtual_sensor_name}"
|
|
return f"binary_sensor.{splitted_name[1]}"
|
|
|
|
def generate_dependencies_list(self):
|
|
dependencies = list()
|
|
condition = SmartCondition.Evaluator(self.ad_api,self.yaml_block['conditions'], logger_interface = self.logger_interface, constants = self.constants, templates_library = self.templates_library, sensor_existence_validation = False, log_init = False)
|
|
self.extract_dependencies_from_entities_to_listen(condition.get_entities_to_listen_as_list(),dependencies)
|
|
|
|
return dependencies
|
|
|
|
def register_callback(self):
|
|
if self.cb_handle == None:
|
|
self._timer_generation += 1
|
|
self.cb_handle = self.ad_api.run_in(
|
|
self.on_time_elapse, self.time, generation=self._timer_generation)
|
|
|
|
def cancel_callback(self):
|
|
# An expired timer may already have a callback queued in the worker.
|
|
# Invalidate it before cancellation, even if it is no longer scheduled.
|
|
self._timer_generation += 1
|
|
handle = self.cb_handle
|
|
self.cb_handle = None
|
|
if handle is not None:
|
|
self.ad_api.cancel_timer(handle, silent=True)
|
|
|
|
def on_time_elapse(self, kwargs):
|
|
if kwargs.get('generation') != self._timer_generation:
|
|
return
|
|
self.cb_handle = None
|
|
value = 'off' if self.invert else 'on'
|
|
if self.output_sensor.get_state() != value:
|
|
self.output_sensor.set_state(state = value)
|
|
|
|
def on_condition_succeed(self):
|
|
self.register_callback()
|
|
|
|
def on_condition_fail(self):
|
|
self.cancel_callback()
|
|
value = 'on' if self.invert else 'off'
|
|
if self.output_sensor.get_state() != value:
|
|
self.output_sensor.set_state(state = value)
|
|
|
|
class EphemeralCondition(ContinuousCondition):
|
|
"""True during the first `time` seconds of each true condition.
|
|
|
|
condition_started_at is a Unix timestamp retained after expiry, until
|
|
the condition becomes false. Reloads resume this window when the entity
|
|
attributes survive; without a timestamp, a true condition starts anew.
|
|
"""
|
|
|
|
def initialize(self):
|
|
self._condition_active = False
|
|
if self.yaml_block.get('invert', False):
|
|
raise ValueError("ephemeral_condition does not support invert; negate conditions instead")
|
|
started_at = self.ad_api.get_state(
|
|
self.sensor_name, attribute='condition_started_at')
|
|
try:
|
|
started_at = float(started_at)
|
|
if not 0 <= started_at <= time.time():
|
|
started_at = None
|
|
except (TypeError, ValueError):
|
|
started_at = None
|
|
self._started_at = started_at
|
|
self.attributes['condition_started_at'] = started_at
|
|
# Unknown inputs during startup must not erase a restored window.
|
|
self.yaml_block = dict(self.yaml_block)
|
|
self.yaml_block.setdefault('ignore_unavailable_condition', True)
|
|
super().initialize()
|
|
|
|
def on_condition_succeed(self):
|
|
if self._condition_active:
|
|
return
|
|
self._condition_active = True
|
|
if self._started_at is None:
|
|
self._started_at = time.time()
|
|
self.attributes['condition_started_at'] = self._started_at
|
|
remaining = self._started_at + self.time - time.time()
|
|
self.output_sensor.set_state(
|
|
state='on' if remaining > 0 else 'off',
|
|
attributes={'condition_started_at': self._started_at})
|
|
if remaining > 0:
|
|
self._timer_generation += 1
|
|
self.cb_handle = self.ad_api.run_in(
|
|
self.on_time_elapse, remaining, generation=self._timer_generation)
|
|
|
|
def on_condition_fail(self):
|
|
self._condition_active = False
|
|
self._started_at = None
|
|
self.attributes['condition_started_at'] = None
|
|
self.cancel_callback()
|
|
self.output_sensor.set_state(
|
|
state='off', attributes={'condition_started_at': None})
|
|
|
|
def on_time_elapse(self, kwargs):
|
|
if kwargs.get('generation') != self._timer_generation:
|
|
return
|
|
self.cb_handle = None
|
|
self.output_sensor.set_state(state='off')
|
|
|
|
|
|
class RetainCondition(VirtualSensorBase):
|
|
def initialize(self):
|
|
self.cb_handle = None
|
|
self._timer_generation = 0
|
|
self.retain_time = self.yaml_block['retain_time']
|
|
try: self.reset_retain_time_on_new_success = self.yaml_block['reset_retain_time_on_new_success']
|
|
except KeyError: self.reset_retain_time_on_new_success = True
|
|
self.log_info(f"Initializing retain condition for {self.sensor_name}")
|
|
self.output_sensor = self.ad_api.get_entity(self.sensor_name)
|
|
|
|
if not self.output_sensor.exists():
|
|
self.log_info(f"{self.sensor_name} does not exist")
|
|
try:
|
|
self.output_sensor.set_state(state = self.yaml_block['default_value'])
|
|
self.log_info(f"adding {self.sensor_name} with default state {self.yaml_block['default_value']}")
|
|
except KeyError:
|
|
self.output_sensor.set_state(state = 'off')
|
|
self.log_info(f"adding {self.sensor_name} with default state off")
|
|
else:
|
|
self.log_info(f"{self.sensor_name} already exist with state = {self.output_sensor.get_state()}")
|
|
|
|
self.condition = SmartCondition.Evaluator(self.ad_api,self.yaml_block['conditions'], logger_interface = self.logger_interface,condition_name = self.sensor_name,on_succeed_cb = self.on_condition_succeed, on_fail_cb = self.on_condition_fail, constants = self.constants, templates_library = self.templates_library)
|
|
# new_value = 'on' if self.condition.evaluate() == SmartCondition.Result.Succeeded else 'off'
|
|
# self.output_sensor.set_state(state = new_value)
|
|
|
|
def generate_sensor_name(self,virtual_sensor_name):
|
|
return f"binary_sensor.{virtual_sensor_name}"
|
|
|
|
def generate_dependencies_list(self):
|
|
dependencies = list()
|
|
condition = SmartCondition.Evaluator(self.ad_api,self.yaml_block['conditions'], logger_interface = self.logger_interface, constants = self.constants, templates_library = self.templates_library, sensor_existence_validation = False, log_init = False)
|
|
self.extract_dependencies_from_entities_to_listen(condition.get_entities_to_listen_as_list(),dependencies)
|
|
|
|
return dependencies
|
|
|
|
def register_callback(self):
|
|
if self.cb_handle == None:
|
|
self._timer_generation += 1
|
|
self.cb_handle = self.ad_api.run_in(
|
|
self.on_time_elapse, self.retain_time, generation=self._timer_generation)
|
|
|
|
def cancel_callback(self):
|
|
# Invalidate callbacks already queued by an expired timer as well.
|
|
self._timer_generation += 1
|
|
handle = self.cb_handle
|
|
self.cb_handle = None
|
|
if handle is not None:
|
|
self.ad_api.cancel_timer(handle, silent=True)
|
|
|
|
def on_time_elapse(self, kwargs):
|
|
if kwargs.get('generation') != self._timer_generation:
|
|
return
|
|
self.cb_handle = None
|
|
if self.condition.evaluate() == SmartCondition.Result.Failed:
|
|
if self.output_sensor.get_state() != 'off':
|
|
self.output_sensor.set_state(state = 'off')
|
|
|
|
def on_condition_succeed(self):
|
|
if self.output_sensor.get_state() != 'on':
|
|
self.output_sensor.set_state(state = 'on')
|
|
if self.reset_retain_time_on_new_success:
|
|
# A returning signal cancels the pending off transition.
|
|
self.cancel_callback()
|
|
else:
|
|
# Preserve the fixed window when resetting is explicitly disabled.
|
|
self.register_callback()
|
|
|
|
def on_condition_fail(self):
|
|
if self.reset_retain_time_on_new_success:
|
|
# Retain for the full delay after the condition becomes false,
|
|
# even when it has been true for longer than retain_time.
|
|
if self.output_sensor.get_state() == 'on':
|
|
self.register_callback()
|
|
elif self.cb_handle == None:
|
|
if self.output_sensor.get_state() != 'off':
|
|
self.output_sensor.set_state(state = 'off')
|
|
|
|
class TimePeriod(VirtualSensorBase):
|
|
"""Active in [start, end), using two input_datetime entities.
|
|
|
|
Time-only inputs repeat daily and may cross midnight. Dated inputs
|
|
describe a single interval. Equal boundaries describe an empty interval.
|
|
"""
|
|
|
|
def generate_sensor_name(self, virtual_sensor_name):
|
|
return f"binary_sensor.{virtual_sensor_name}"
|
|
|
|
def generate_dependencies_list(self):
|
|
inputs = []
|
|
for key in ('start', 'end'):
|
|
value = self.yaml_block.get(key)
|
|
value = self.constants.get(value, value) if isinstance(value, str) else value
|
|
if not self.is_entity_reference(value) or not value.startswith('input_datetime.'):
|
|
raise ValueError(f"{self.sensor_name}: {key} must reference an input_datetime")
|
|
inputs.append(value)
|
|
return inputs
|
|
|
|
def initialize(self):
|
|
self.start_entity, self.end_entity = self.generate_dependencies_list()
|
|
self.cb_handle = None
|
|
self._timer_generation = 0
|
|
for entity in set((self.start_entity, self.end_entity)):
|
|
self.ad_api.listen_state(self.on_input_change, entity)
|
|
self.update_period()
|
|
|
|
@staticmethod
|
|
def parse_boundary(value):
|
|
for fmt, kind in (('%H:%M:%S', 'time'),
|
|
('%Y-%m-%d %H:%M:%S', 'datetime'),
|
|
('%Y-%m-%d', 'date')):
|
|
try:
|
|
parsed = datetime.datetime.strptime(value, fmt)
|
|
return kind, parsed.time() if kind == 'time' else parsed
|
|
except (TypeError, ValueError):
|
|
continue
|
|
raise ValueError('Unavailable or invalid input_datetime state')
|
|
|
|
def on_input_change(self, entity, attribute, old, new, kwargs):
|
|
self.update_period()
|
|
|
|
def on_time_elapsed(self, kwargs):
|
|
if kwargs.get('generation') != self._timer_generation:
|
|
return
|
|
self.cb_handle = None
|
|
self.update_period()
|
|
|
|
def update_period(self):
|
|
self._timer_generation += 1
|
|
if self.cb_handle is not None:
|
|
self.ad_api.cancel_timer(self.cb_handle, silent=True)
|
|
self.cb_handle = None
|
|
now = self.ad_api.datetime()
|
|
delay = 60.0
|
|
try:
|
|
start_kind, start = self.parse_boundary(self.ad_api.get_state(self.start_entity))
|
|
end_kind, end = self.parse_boundary(self.ad_api.get_state(self.end_entity))
|
|
if start_kind != end_kind:
|
|
raise ValueError('Both inputs must have the same date/time format')
|
|
if start_kind == 'time':
|
|
current = now.time()
|
|
active = (start <= current < end if start <= end
|
|
else current >= start or current < end)
|
|
boundaries = [datetime.datetime.combine(now.date(), value)
|
|
for value in (start, end)]
|
|
boundaries = [value if value > now else value + datetime.timedelta(days=1)
|
|
for value in boundaries]
|
|
else:
|
|
active = start <= now < end
|
|
boundaries = [value for value in (start, end) if value > now]
|
|
state = 'on' if active else 'off'
|
|
if boundaries:
|
|
delay = min(delay, min((value - now).total_seconds() for value in boundaries))
|
|
except ValueError:
|
|
state = 'unavailable'
|
|
if self.entity.get_state() != state:
|
|
self.entity.set_state(state=state)
|
|
# Refresh at boundaries, and at least every minute to follow clock/DST changes.
|
|
self.cb_handle = self.ad_api.run_in(
|
|
self.on_time_elapsed, delay, generation=self._timer_generation)
|
|
|
|
|
|
class BinarySensor(VirtualSensorBase):
|
|
def initialize(self):
|
|
self.output_sensor = self.ad_api.get_entity(self.sensor_name)
|
|
|
|
if not self.output_sensor.exists():
|
|
self.output_sensor.set_state(state = 'off',attributes = self.attributes)
|
|
|
|
#self.ad_api.log(f"Creating Binary Sensor: binary_sensor = {self.virtual_sensor_name}, yaml_block = {self.yaml_block}")
|
|
self.condition = SmartCondition.Evaluator(self.ad_api,self.get_expression_yaml(), logger_interface = self.logger_interface, unavaibility_result = SmartCondition.Result.Unavailable,condition_name = self.virtual_sensor_name,on_change_cb = self.on_sensor_change, constants = self.constants, templates_library = self.templates_library, log_group = self.sensor_name)
|
|
new_value = 'on' if self.condition.evaluate() == SmartCondition.Result.Succeeded else 'off'
|
|
self.output_sensor.set_state(state = new_value)
|
|
|
|
def on_sensor_change(self,prev_result,result):
|
|
target_state = None
|
|
if result == SmartCondition.Result.Unavailable: target_state = 'unavailable'
|
|
elif result == SmartCondition.Result.Succeeded: target_state = 'on'
|
|
elif result == SmartCondition.Result.Failed: target_state = 'off'
|
|
else: target_state = None #Disabled means we keep the previous value
|
|
|
|
if target_state != None:
|
|
self.output_sensor.set_state(state = target_state)
|
|
self.log_info(f"{self.sensor_name} = {target_state}")
|
|
else:
|
|
self.log_info(f"SmartCondition of {self.sensor_name} is disabled")
|
|
|
|
|
|
def generate_dependencies_list(self):
|
|
dependencies = list()
|
|
condition = SmartCondition.Evaluator(self.ad_api,self.get_expression_yaml(), logger_interface = self.logger_interface, condition_name = [self.virtual_sensor_name,"generate_dependencies_list"],constants = self.constants, templates_library = self.templates_library, sensor_existence_validation = False, log_init = False)
|
|
self.extract_dependencies_from_entities_to_listen(condition.get_entities_to_listen_as_list(),dependencies)
|
|
|
|
return dependencies
|
|
|
|
def get_expression_yaml(self):
|
|
try: return self.yaml_block['conditions']
|
|
except (TypeError,KeyError): return self.yaml_block
|
|
|
|
def generate_sensor_name(self,virtual_sensor_name):
|
|
return f"binary_sensor.{virtual_sensor_name}"
|
|
#TODO : restore that code once we rename the sensor
|
|
splitted_name = virtual_sensor_name.split(".")
|
|
assert len(splitted_name) == 2,f"Invalid virtual sensor name : {virtual_sensor_name}"
|
|
return f"binary_sensor.{splitted_name[1]}"
|
|
|
|
# set a binary_sensor to true for x seconds when an event is received
|
|
class EventSensor(VirtualSensorBase):
|
|
def initialize(self):
|
|
self.output_sensor = self.ad_api.get_entity(self.sensor_name)
|
|
|
|
if not self.output_sensor.exists():
|
|
self.output_sensor.set_state(state = 'off',attributes = self.attributes)
|
|
|
|
self.retain_time = self.yaml_block['retain_time']
|
|
self.cb_handle = None
|
|
|
|
self.event_handlers = []
|
|
if "activation_events" in self.yaml_block:
|
|
self.event_handlers.append(EventHandler(self.ad_api, self.yaml_block["activation_events"], self.on_activation_event,logger_interface = self.logger_interface))
|
|
|
|
def on_activation_event(self, event_name, data, kwargs):
|
|
self.output_sensor.set_state(state = 'on')
|
|
if self.cb_handle is not None:
|
|
self.ad_api.cancel_timer(self.cb_handle)
|
|
self.cb_handle = self.ad_api.run_in(self.on_retain_time_elapsed, self.retain_time)
|
|
|
|
def on_retain_time_elapsed(self, kwargs):
|
|
self.output_sensor.set_state(state = 'off')
|
|
self.cb_handle = None
|
|
|
|
def generate_sensor_name(self,virtual_sensor_name):
|
|
return f"binary_sensor.{virtual_sensor_name}"
|
|
|
|
class ValueSensor(VirtualSensorBase):
|
|
def initialize(self):
|
|
# should be done in the base class
|
|
# try: self.sensor_attributes = self.yaml_block['attributes']
|
|
# except TypeError: self.sensor_attributes = {}
|
|
|
|
self.output_sensor = self.ad_api.get_entity(self.sensor_name)
|
|
if not self.output_sensor.exists():
|
|
self.output_sensor.set_state(state = 0, attributes = self.attributes)
|
|
|
|
self.smart_value = SmartValue.Evaluator(self.ad_api,self.get_expression_yaml(), logger_interface = self.logger_interface, expression_name = self.virtual_sensor_name,on_change_cb = self.on_sensor_change, constants = self.constants)
|
|
new_value = self.smart_value.evaluate()
|
|
self.output_sensor.set_state(state = new_value)
|
|
|
|
def on_sensor_change(self,prev_result,result):
|
|
self.output_sensor.set_state(state = result, attributes = self.attributes)
|
|
|
|
def get_expression_yaml(self):
|
|
try: return self.yaml_block['value']
|
|
except (TypeError,KeyError): return self.yaml_block
|
|
|
|
def generate_dependencies_list(self):
|
|
dependencies = list()
|
|
smart_value = SmartValue.Evaluator(self.ad_api,self.get_expression_yaml(), logger_interface = self.logger_interface, expression_name = self.virtual_sensor_name, constants = self.constants, sensor_existence_validation = False, log_init = False)
|
|
self.extract_dependencies_from_entities_to_listen(smart_value.get_entities_to_listen_as_list(),dependencies)
|
|
|
|
return dependencies
|
|
|
|
def generate_sensor_name(self,virtual_sensor_name):
|
|
return f"sensor.{virtual_sensor_name}"
|
|
|
|
class VirtualSensors():
|
|
|
|
def __init__(self, ad_api = None,logger_interface = None, super_entity_id = None, yaml_block = None ,args_to_ignore_in_validation = [], constants = None, templates_library = None, app_name = None):
|
|
assert ad_api
|
|
assert yaml_block
|
|
assert logger_interface
|
|
self.ad_api = ad_api
|
|
self.virtual_sensors = dict()
|
|
self.logger_interface = logger_interface
|
|
|
|
parser = kwargsParser(yaml_block,lambda string: logger_interface.log_error(string),lambda string: logger_interface.log_warning(string))
|
|
|
|
default_values = parser.parse_args('default_values',{})
|
|
for sensor_name,default_value in default_values.items():
|
|
if not self.ad_api.entity_exists(sensor_name):
|
|
self.logger_interface.log_info(f"Creating sensor {sensor_name} = {default_value}")
|
|
self.ad_api.set_state(sensor_name, state = str(default_value),check_existence=False)
|
|
|
|
averagers = parser.parse_args('averagers',{})
|
|
for averager in averagers:
|
|
self.virtual_sensors[f"sensor.{averager}"] = Averager(self.ad_api,self.logger_interface,averager,super_entity_id = super_entity_id,yaml_block = yaml_block['averagers'][averager],constants = constants, templates_library = templates_library,app_name = app_name)
|
|
|
|
continuous_conditions = parser.parse_args('continuous_conditions',{})
|
|
for continuous_condition in continuous_conditions:
|
|
self.virtual_sensors[f"binary_sensor.{continuous_condition}"] = ContinuousCondition(self.ad_api,self.logger_interface,continuous_condition,super_entity_id = super_entity_id,yaml_block = yaml_block['continuous_conditions'][continuous_condition],templates_library = templates_library,app_name = app_name)
|
|
|
|
binary_sensors = parser.parse_args('binary_sensors',{})
|
|
for binary_sensor in binary_sensors:
|
|
#self.logger_interface.log_info(f"Creating Binary Sensor: binary_sensor = {binary_sensor}, super_entity_id = {super_entity_id}, yaml_block = {yaml_block['binary_sensors']}")
|
|
self.virtual_sensors[f"binary_sensor.{binary_sensor}"] = BinarySensor(self.ad_api,self.logger_interface,binary_sensor,super_entity_id = super_entity_id,yaml_block = yaml_block['binary_sensors'][binary_sensor],constants = constants,templates_library = templates_library,app_name = app_name)
|
|
|
|
value_selectors = parser.parse_args('value_selectors',{})
|
|
for value_selector in value_selectors:
|
|
self.virtual_sensors[f"sensor.{value_selector}"] = ValueSelector(self.ad_api,self.logger_interface,value_selector,super_entity_id = super_entity_id,yaml_block = yaml_block['value_selectors'][value_selector],constants = constants,templates_library = templates_library,app_name = app_name)
|
|
|
|
sensors = parser.parse_args('sensors',{})
|
|
for sensor in sensors:
|
|
splitted_sensor = sensor.split('.')
|
|
if len(splitted_sensor) != 2:
|
|
self.logger_interface.log_error(f"Invalid sensor name {splitted_sensor}")
|
|
|
|
if splitted_sensor[0] == 'binary_sensor':
|
|
self.virtual_sensors[f"binary_sensor.{splitted_sensor[1]}"] = BinarySensor(self.ad_api,self.logger_interface,splitted_sensor[1],super_entity_id = super_entity_id,yaml_block = yaml_block['sensors'][sensor],constants = constants,templates_library = templates_library,app_name = app_name)
|
|
elif splitted_sensor[0] == 'time_period':
|
|
self.virtual_sensors[f"binary_sensor.{splitted_sensor[1]}"] = TimePeriod(self.ad_api,self.logger_interface,splitted_sensor[1],super_entity_id = super_entity_id,yaml_block = yaml_block['sensors'][sensor],constants = constants,templates_library = templates_library,app_name = app_name)
|
|
elif splitted_sensor[0] == 'sensor':
|
|
self.virtual_sensors[f"sensor.{splitted_sensor[1]}"] = ValueSensor(self.ad_api,self.logger_interface,splitted_sensor[1],super_entity_id = super_entity_id,yaml_block = yaml_block['sensors'][sensor],constants = constants,templates_library = templates_library,app_name = app_name)
|
|
elif splitted_sensor[0] == 'continuous_condition':
|
|
self.virtual_sensors[f"binary_sensor.{splitted_sensor[1]}"] = ContinuousCondition(self.ad_api,self.logger_interface,splitted_sensor[1],super_entity_id = super_entity_id,yaml_block = yaml_block['sensors'][sensor],constants = constants,templates_library = templates_library,app_name = app_name)
|
|
elif splitted_sensor[0] == 'averager':
|
|
self.virtual_sensors[f"sensor.{splitted_sensor[1]}"] = Averager(self.ad_api,self.logger_interface,splitted_sensor[1],super_entity_id = super_entity_id,yaml_block = yaml_block['sensors'][sensor],constants = constants,templates_library = templates_library,app_name = app_name)
|
|
elif splitted_sensor[0] == 'value_selector':
|
|
self.virtual_sensors[f"sensor.{splitted_sensor[1]}"] = ValueSelector(self.ad_api,self.logger_interface,splitted_sensor[1],super_entity_id = super_entity_id,yaml_block = yaml_block['sensors'][sensor],constants = constants,templates_library = templates_library,app_name = app_name)
|
|
elif splitted_sensor[0] == 'ephemeral_condition':
|
|
self.virtual_sensors[f"binary_sensor.{splitted_sensor[1]}"] = EphemeralCondition(self.ad_api,self.logger_interface,splitted_sensor[1],super_entity_id = super_entity_id,yaml_block = yaml_block['sensors'][sensor],constants = constants,templates_library = templates_library,app_name = app_name)
|
|
elif splitted_sensor[0] == 'retain_condition':
|
|
self.virtual_sensors[f"binary_sensor.{splitted_sensor[1]}"] = RetainCondition(self.ad_api,self.logger_interface,splitted_sensor[1],super_entity_id = super_entity_id,yaml_block = yaml_block['sensors'][sensor],constants = constants,templates_library = templates_library,app_name = app_name)
|
|
elif splitted_sensor[0] == 'event_sensor':
|
|
self.virtual_sensors[f"binary_sensor.{splitted_sensor[1]}"] = EventSensor(self.ad_api,self.logger_interface,splitted_sensor[1],super_entity_id = super_entity_id,yaml_block = yaml_block['sensors'][sensor],constants = constants,templates_library = templates_library,app_name = app_name)
|
|
else:
|
|
self.logger_interface.log_error(f"Invalid sensor prefix {splitted_sensor[0]}")
|
|
|
|
dependencies_graph = dict()
|
|
for sensor_name, virtual_sensor in self.virtual_sensors.items():
|
|
try:
|
|
dependencies_graph[sensor_name] = virtual_sensor.generate_dependencies_list()
|
|
# add the dependencies from the attributes
|
|
dependencies_graph[sensor_name].extend(virtual_sensor.generate_attributes_dependencies_list())
|
|
#self.logger_interface.log_info(f"Dependencies = {dependencies_graph}")
|
|
except ParsingException as e:
|
|
self.logger_interface.log_error(f"[Dependencies][{sensor_name}] {e}")
|
|
|
|
self.remove_obsolete_dependencies(dependencies_graph)
|
|
|
|
while len(dependencies_graph) > 0:
|
|
virtual_sensor_initialized = list()
|
|
#self.logger_interface.log_info(f"Dependencies = {dependencies_graph}")
|
|
|
|
#initialize sensor with no dependencies
|
|
for sensor_name, dependencies_list in dependencies_graph.items():
|
|
if len(dependencies_list) == 0:
|
|
virtual_sensor_initialized.append(sensor_name)
|
|
self.virtual_sensors[sensor_name]._initialize()
|
|
|
|
#removing initialized sensors from dependencies_graph
|
|
for sensor_name in virtual_sensor_initialized:
|
|
del dependencies_graph[sensor_name]
|
|
|
|
#if didn't initialized any sensors, we are in a circular dependencies
|
|
ignore_sensor = None
|
|
if len(virtual_sensor_initialized) == 0:
|
|
foundInDefaultValues = False
|
|
for sensor_name in dependencies_graph:
|
|
if sensor_name in default_values:
|
|
self.logger_interface.log_info(f"Found circular dependencies in the following sensors : {dependencies_graph}. The default values of {sensor_name} will be use to break it.")
|
|
ignore_sensor = sensor_name
|
|
foundInDefaultValues = True
|
|
break
|
|
|
|
if not foundInDefaultValues:
|
|
self.logger_interface.log_error(f"Found circular dependencies in the following sensors : {dependencies_graph}. You can give a default value with 'default_values' to one of them to break it.")
|
|
break
|
|
|
|
#remove all dependencies towards sensors that are already been initialized
|
|
self.remove_obsolete_dependencies(dependencies_graph,ignore_sensor)
|
|
|
|
def remove_obsolete_dependencies(self, dependencies_graph, ignore_sensor = None):
|
|
for sensor_name, dependencies_list in dependencies_graph.items():
|
|
current_dependencies = list()
|
|
for dependency in dependencies_list:
|
|
if dependency in dependencies_graph and dependency != ignore_sensor:
|
|
current_dependencies.append(dependency)
|
|
dependencies_graph[sensor_name] = current_dependencies
|
|
|
|
|
|
def on_sensor_change(self,sensor_name,prev_result,result):
|
|
if result == SmartCondition.Result.Succeeded:
|
|
self.ad_api.set_state(f'binary_sensor.{sensor_name}',state = 'on')
|
|
else :
|
|
self.ad_api.set_state(f'binary_sensor.{sensor_name}',state = 'off')
|
|
|
|
class VirtualSensorsApp(hass.Hass):
|
|
def initialize(self):
|
|
self.logger_interface = LoggerInterface(self.get_ad_api(),default_log = "virtualsensors_log")
|
|
with self.logger_interface.log_group("App Initialization"):
|
|
self.virtual_sensors = VirtualSensors(ad_api = self.get_ad_api(),logger_interface = self.logger_interface,yaml_block = self.args,args_to_ignore_in_validation = ['module','class','global_dependencies','priority','name','config_path'],app_name = self.name)
|