Pierre/ad_toolbox
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add EventSensor class for temporary binary sensor activation

Browse Source
This commit is contained in:
Pierre Gironde
2026-07-09 23:07:30 +02:00
parent c71094641b
commit b997836401
2 changed files with 46 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
eventhandler.py
View File

@@ -60,8 +60,10 @@ import json
# =============================================================================
class EventDispatcher:
def __init__(self,ad_api,event_name,callback,event_data,reset_data,event_context,debounce_seconds=None):
self.ad_api = ad_api
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
@@ -126,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):
@@ -139,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):
@@ -183,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, 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):
@@ -198,6 +204,7 @@ class EventHandler:
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, debounce_seconds = 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,debounce_seconds)
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)

31
virtualsensors.py
View File

@@ -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):
@@ -525,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
@@ -609,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]}")