feat: add EventSensor class for temporary binary sensor activation

adexpressioncontext.py

@@ -1,5 +1,5 @@
 import pyparsing as pp
-from expressionparser import BaseExpressionContext
+from expressionparser import BaseExpressionContext, ExpressionParser
 
 class ADExpressionContext(BaseExpressionContext):
     def __init__(self,appdaemon_api, sensor_existence_validation = True):
@@ -13,6 +13,7 @@ class ADExpressionContext(BaseExpressionContext):
         self.sensors_default_value = dict()
         self.sensor_states = dict()
         self.sensor_existence_validation = sensor_existence_validation
+        self.local_expressions = {}
 
         self.declare_default_constants()
 
@@ -131,7 +132,22 @@ class ADExpressionContext(BaseExpressionContext):
     def get_entities_to_listen(self):
         return self.entities_to_listen
 
-    def get_constant_value(self,name): return self.variables[name]
+    def get_constant_value(self, name):
+        if name in self.local_expressions:
+            return self.local_expressions[name].eval()
+        return self.variables[name]
+
+    def declare_local_expression(self, name: str, expression_string: str):
+        assert name not in self.variables, f"'{name}' is already declared"
+        assert name.replace('_', '').isalnum() and name.isascii(), f"Invalid local expression name: {name}"
+        # Register as a constant placeholder so the variable parser recognises it
+        self.variables[name] = None
+        self.constants.append(name)
+        self.parser = None  # reset variable parser cache
+        # Parse the expression; set_context registers its sensor dependencies
+        # Store only the parsed tree so evaluation skips the top-level reset()
+        full_parser = ExpressionParser(expression_string, self)
+        self.local_expressions[name] = full_parser.parsed_data
 
     def declare_constant(self,name : str,value):
         def is_a_sensor(string):

eventhandler.py

@@ -42,6 +42,9 @@ import json
 #         <field>: <value>
 #       reset_data:                         # optional — latch reset filter
 #         <field>: <value>
+#       debounce_seconds: <float>           # optional — ignore re-triggers within
+#                                           #   this delay (seconds).  Useful for
+#                                           #   buttons that fire duplicate events.
 #
 # Usage:
 #   handler = EventHandler(ad_api, self.args['events_to_listen'], my_callback)
@@ -57,8 +60,10 @@ import json
 # =============================================================================
 
 class EventDispatcher:
-    def __init__(self,ad_api,event_name,callback,event_data,reset_data,event_context):
+    def __init__(self,ad_api,logger_interface,event_name,callback,event_data,reset_data,event_context,debounce_seconds=None):
         self.ad_api = ad_api        
+        self.logger_interface = logger_interface
+
         event_data = self._resolve_zha_device_name(event_data)
         reset_data = self._resolve_zha_device_name(reset_data)
         self.event_name = event_name
@@ -67,6 +72,8 @@ class EventDispatcher:
         self.reset_data = reset_data
         self.waiting_for_reset = False
         self.event_context = event_context
+        self.debounce_seconds = debounce_seconds
+        self._last_trigger_time = None
         if event_data == None:
             self.ad_api.listen_event(self.on_event,event_name)
         else:
@@ -121,7 +128,7 @@ class EventDispatcher:
                 result = resp.read().decode('utf-8').strip()
                 return result if result else None
         except Exception as e:
-            self.ad_api.log_error(f"[EventDispatcher] Failed to resolve ZHA device name '{device_name}': {e}")
+            self.logger_interface.log_error(f"[EventDispatcher] Failed to resolve ZHA device name '{device_name}': {e}")
             return None
 
     def _resolve_zha_device_name(self, data):
@@ -134,10 +141,10 @@ class EventDispatcher:
         ieee = self._lookup_zha_ieee(name)
         resolved = {k: v for k, v in data.items() if k != 'device_name'}
         if ieee is not None:
-            self.ad_api.log_info(f"[EventDispatcher] Resolved ZHA device name '{name}' to IEEE '{ieee}'")
+            self.logger_interface.log_info(f"[EventDispatcher] Resolved ZHA device name '{name}' to IEEE '{ieee}'")
             resolved['device_ieee'] = ieee
         else:
-            self.ad_api.log_error(f"[EventDispatcher] ZHA device '{name}' not found — 'device_name' filter ignored")
+            self.logger_interface.log_error(f"[EventDispatcher] ZHA device '{name}' not found — 'device_name' filter ignored")
         return resolved
 
     def process_event(self,data):
@@ -159,6 +166,13 @@ class EventDispatcher:
 
         #special threatment for event
         if match_event_data and not self.waiting_for_reset:
+            if self.debounce_seconds is not None:
+                import time
+                now = time.monotonic()
+                if self._last_trigger_time is not None and (now - self._last_trigger_time) < self.debounce_seconds:
+                    return False
+                self._last_trigger_time = now
+
             if self.reset_data != None: 
                 self.waiting_for_reset = True
             
@@ -171,11 +185,15 @@ class EventDispatcher:
         return False
 
 class EventHandler:
-    def __init__(self, ad_api,events_block,callback,event_context = None): #event_context is passed back to the CB has a context
+    def __init__(self, ad_api,events_block,callback,event_context = None, logger_interface = None): #event_context is passed back to the CB has a context
         def register_event_with_params(event_block,callback,event_context):            
-            self.add_dispatcher(event_block['event_name'],callback,event_block['event_data'] if 'event_data' in event_block else None, event_block['reset_data'] if 'reset_data' in event_block else None,event_context)
+            self.add_dispatcher(event_block['event_name'],callback,event_block['event_data'] if 'event_data' in event_block else None, event_block['reset_data'] if 'reset_data' in event_block else None,event_context, event_block.get('debounce_seconds'))
 
         self.__ad_api = ad_api
+        if logger_interface:
+            self.__logger_interface = logger_interface
+        else:
+            self.__logger_interface = ad_api
         self.event_dispatchers = []
 
         if isinstance(events_block, list):
@@ -185,7 +203,8 @@ class EventHandler:
             for event_block in events_block.values():
                 register_event_with_params(event_block,callback,event_context)
 
-    def add_dispatcher(self,event_name,callback,event_data = None,reset_data = None ,event_context = None):
-        self.__ad_api.log_info(f'Registering dispatcher {callback.__name__} for event "{event_name}" ({event_data})')
-        dispatcher = EventDispatcher(self.__ad_api,event_name,callback,event_data,reset_data,event_context)
+    def add_dispatcher(self,event_name,callback,event_data = None,reset_data = None ,event_context = None, debounce_seconds = None):
+        if self.__logger_interface:
+            self.__logger_interface.log_info(f'Registering dispatcher {callback.__name__} for event "{event_name}" ({event_data})')
+        dispatcher = EventDispatcher(self.__ad_api,self.__logger_interface,event_name,callback,event_data,reset_data,event_context,debounce_seconds)
         self.event_dispatchers.append(dispatcher)

expressionparser.py

@@ -71,6 +71,7 @@ class EvaluateBase(object):
         if value == 'false': return False
         if value == False: return False
         if value == None: return False
+        if value == '': return False
         # this should be handle explicitely by the condition
         # edit: this is actually usefull if don't want disable to be evaluated as true
         if value == 'unavailable': return False

smartcondition.py

@@ -31,6 +31,7 @@ class Evaluator():
     __LOG_INIT                          = "log_init"
     __EXTRA_ENTITIES_TO_LISTEN          = "extra_entities_to_listen"
     __CONSTANTS                         = "constants"
+    __LOCAL_EXPRESSIONS_STRING          = "local_expressions"
     __UNAVAIBILITY_RESULT               = "unavaibility_result"
     
     def __init__(self, appdaemon_api, conditions_block,**kwargs):
@@ -313,6 +314,12 @@ class Evaluator():
                                     if disable_conditions_args:
                                         add_conditions_to_conditions_block(conditions_block,self.__DISABLE_CONDITIONS_STRING,disable_conditions_args)
 
+                                    local_expressions_args = param_parser.parse_args(self.__LOCAL_EXPRESSIONS_STRING, None)
+                                    if local_expressions_args:
+                                        if self.__LOCAL_EXPRESSIONS_STRING not in conditions_block:
+                                            conditions_block[self.__LOCAL_EXPRESSIONS_STRING] = {}
+                                        conditions_block[self.__LOCAL_EXPRESSIONS_STRING].update(local_expressions_args)
+
                                     if param_parser.parse_args(self.__TEMPLATE_CONDITIONS_STRING, None):
                                         self.__log(f"Template '{template}' contain recursive template",level = "ERROR")                                                            
                                     
@@ -378,6 +385,18 @@ class Evaluator():
                         self.__log(f"{constant_name} = {value}")
                     self.expression_context.declare_constant(constant_name,value)
 
+            local_expressions = param_parser.parse_args(self.__LOCAL_EXPRESSIONS_STRING, None)
+            if local_expressions:
+                if self.__log_init:
+                    self.__log(f"declaring local expressions:")
+                for expr_name, expr_string in local_expressions.items():
+                    if self.__log_init:
+                        self.__log(f"{expr_name} = {expr_string}")
+                    self.expression_context.declare_local_expression(expr_name, str(expr_string))
+                # Ensure sensors referenced only in local_expressions are registered as
+                # listeners even if no condition string references them directly.
+                self.__add_to_entities_to_listen(self.expression_context.get_entities_to_listen())
+
 
             trigger_conditions_args = param_parser.parse_args(self.__TRIGGER_CONDITIONS_STRING,None)
             if trigger_conditions_args:

smartobject.py

@@ -135,6 +135,22 @@ class SmartObject(hass.Hass,LoggerInterface):
     # Called at the end of initialize() after all YAML args are processed.
     def on_initialize_smart_object(self): pass
 
+    # Presence simulation extension point.
+    # HomeAlarm can call this method by app name.
+    def start_presence_simulation(self, source_app=None, slot_name=None, payload=None):
+        self.log_error(
+            f"{self.name} does not implement start_presence_simulation "
+            f"(source_app={source_app}, slot_name={slot_name})"
+        )
+
+    # Presence simulation extension point.
+    # HomeAlarm can call this method by app name.
+    def stop_presence_simulation(self, source_app=None, slot_name=None, payload=None):
+        self.log_error(
+            f"{self.name} does not implement stop_presence_simulation "
+            f"(source_app={source_app}, slot_name={slot_name})"
+        )
+
     # ------------------------------------------------------------------
     # Dataset persistence helpers
     # ------------------------------------------------------------------

unit_tests/test_template.yaml

@@ -23,6 +23,10 @@ templates_library:
     trigger_conditions: binary_sensor.off == False
     blocking_conditions:
       - binary_sensor.on
+  T4:
+    local_expressions:
+      local_threshold: "binary_sensor.true ? 170 : 200"
+    trigger_conditions: sensor.int170 == local_threshold
 
 tests:
   test_1: 
@@ -44,3 +48,25 @@ tests:
     conditions: 
       template_conditions: <T2>,<T3>
       trigger_conditions: binary_sensor.true and binary_sensor.false
+
+  # local_expressions in a template
+  test_5_local_expr_in_template_true:
+    expected_result: Succeeded
+    conditions:
+      template_conditions: <T4>
+
+  # local_expressions declared directly in conditions block
+  test_6_local_expr_inline_true:
+    expected_result: Succeeded
+    conditions:
+      local_expressions:
+        my_threshold: "binary_sensor.true ? 17 : 200"
+      trigger_conditions: sensor.int17 == my_threshold
+
+  # local_expressions inline, condition not met
+  test_7_local_expr_inline_false:
+    expected_result: Failed
+    conditions:
+      local_expressions:
+        my_threshold: "binary_sensor.false ? 17 : 200"
+      trigger_conditions: sensor.int17 == my_threshold

virtualsensors.py

@@ -5,6 +5,7 @@ from kwargsparser import kwargsParser
 from expressionparser import ParsingException
 import time
 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):        
@@ -25,7 +26,8 @@ class VirtualSensorBase:
             self.virtual_sensor_name = sensor_name.split('.')[1]
         self.sensor_domain = self.sensor_name.split(".")[0]
 
-        assert yaml_block ,f"{self.sensor_name} have a null Yaml block"
+        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
 
@@ -524,6 +526,34 @@ class BinarySensor(VirtualSensorBase):
         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
@@ -608,6 +638,8 @@ class VirtualSensors():
                 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] == '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]}")