From 191e69eb1d82a1336645fb86b79cacd981466ab3 Mon Sep 17 00:00:00 2001 From: Pierre Gironde Date: Sat, 26 Sep 2026 19:21:19 +0200 Subject: [PATCH] new event forwarder --- eventsforwarder.py | 68 ++++++++++++++++ unit_tests/test_events_forwarder.py | 116 ++++++++++++++++++++++++++++ 2 files changed, 184 insertions(+) create mode 100644 eventsforwarder.py create mode 100644 unit_tests/test_events_forwarder.py diff --git a/eventsforwarder.py b/eventsforwarder.py new file mode 100644 index 0000000..8d7c757 --- /dev/null +++ b/eventsforwarder.py @@ -0,0 +1,68 @@ +"""Forward HA events to conditional events and service calls. + +Each forwarders entry contains events_to_listen (EventHandler format), +optional events_to_send and services_to_call mappings. Both action types +accept event_condition, evaluated on every incoming event. Payloads are +static YAML data, as in the original forwarder. +""" +import appdaemon.plugins.hass.hassapi as hass +import ad_toolbox.smartcondition as SmartCondition +from ad_toolbox.eventhandler import EventHandler +from ad_toolbox.expressionparser import ParsingException +from ad_toolbox.logger_interface import LoggerInterface + + +class ConditionalAction: + def __init__(self, owner, config): + self.owner = owner + self.config = config + self.condition = None + if 'event_condition' in config: + self.condition = SmartCondition.Evaluator(owner, config['event_condition']) + + def is_allowed(self): + return (self.condition is None or + self.condition.evaluate(False) == SmartCondition.Result.Succeeded) + + +class EventSender(ConditionalAction): + def send_event(self, source_event, event_data): + if self.is_allowed(): + data = self.config.get('event_data', {}) + self.owner.log_info(f"{source_event} => {self.config['event_name']} data = {data}") + self.owner.fire_event(self.config['event_name'], **data) + + +class ServiceCaller(ConditionalAction): + def call_service(self, source_event, event_data): + if self.is_allowed(): + data = self.config.get('data', {}) + self.owner.log_info(f"{source_event} => Calling {self.config['service']} data = {data}") + self.owner.call_service(self.config['service'], **data) + + +class EventsForwarder(hass.Hass, LoggerInterface): + def initialize(self): + self.initialize_logger_interface(self.get_ad_api()) + self.mute_logger(self.args.get('mute', False)) + self.forwarders = {} + self.services_caller = {} + self.event_handlers = [] + for key, block in self.args.get('forwarders', {}).items(): + try: + senders = [EventSender(self, config) + for config in block.get('events_to_send', {}).values()] + callers = [ServiceCaller(self, config) + for config in block.get('services_to_call', {}).values()] + self.forwarders[key] = senders + self.services_caller[key] = callers + self.event_handlers.append(EventHandler( + self, block['events_to_listen'], self.on_event, key)) + except ParsingException as error: + self.log_error(f"Invalid forwarder {key}: {error}") + + def on_event(self, event_name, event_data, forwarder_key): + for sender in self.forwarders.get(forwarder_key, []): + sender.send_event(event_name, event_data) + for caller in self.services_caller.get(forwarder_key, []): + caller.call_service(event_name, event_data) diff --git a/unit_tests/test_events_forwarder.py b/unit_tests/test_events_forwarder.py new file mode 100644 index 0000000..1ff7f8e --- /dev/null +++ b/unit_tests/test_events_forwarder.py @@ -0,0 +1,116 @@ +"""Check volume routing with the real condition parser, without AppDaemon.""" +import ast +import sys +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock + +import yaml + +TOOLBOX = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(TOOLBOX)) +import smartcondition +from expressionparser import ParsingException + + +def load_classes(path, namespace): + nodes = [node for node in ast.parse(path.read_text()).body + if isinstance(node, ast.ClassDef)] + exec(compile(ast.Module(body=nodes, type_ignores=[]), str(path), 'exec'), namespace) + return namespace + + +handlers = load_classes(TOOLBOX / 'eventhandler.py', {}) +classes = load_classes(TOOLBOX / 'eventsforwarder.py', { + 'hass': SimpleNamespace(Hass=type('Hass', (), {})), 'LoggerInterface': type('LoggerInterface', (), {}), + 'SmartCondition': smartcondition, 'EventHandler': handlers['EventHandler'], + 'ParsingException': ParsingException, +}) + + +class ForwarderTests(unittest.TestCase): + def setUp(self): + self.owner = Mock() + self.states = {'media_player.bravia_theatre': 'off', + 'binary_sensor.is_tv_on': 'off'} + self.owner.entity_exists.side_effect = lambda entity: entity in self.states + self.owner.get_state.side_effect = lambda entity, **kwargs: self.states.get(entity) + config = yaml.safe_load((TOOLBOX.parents[0] / 'global/homecinema.yaml').read_text()) + self.blocks = config['homecinema_volume']['forwarders'] + + def test_priorities_and_repeated_commands(self): + for direction, block in self.blocks.items(): + senders = [classes['EventSender'](self.owner, value) + for value in block['events_to_send'].values()] + callers = [classes['ServiceCaller'](self.owner, value) + for value in block['services_to_call'].values()] + for bravia in ('on', 'playing', 'paused', 'idle', 'off', 'standby', 'unknown', 'unavailable'): + for tv in ('on', 'off'): + with self.subTest(direction=direction, bravia=bravia, tv=tv): + self.states.update({'media_player.bravia_theatre': bravia, + 'binary_sensor.is_tv_on': tv}) + self.owner.fire_event.reset_mock() + self.owner.call_service.reset_mock() + for _ in range(2): + for sender in senders: + sender.send_event('button', {}) + for caller in callers: + caller.call_service('button', {}) + if bravia in ('on', 'playing', 'paused', 'idle'): + target = 'media_player.bravia_theatre' + elif tv == 'on': + self.assertEqual(self.owner.fire_event.call_count, 2) + self.owner.fire_event.assert_called_with(f'the_frame_{direction}') + self.owner.call_service.assert_not_called() + continue + else: + target = 'media_player.living_room_sonos' + self.owner.fire_event.assert_not_called() + self.assertEqual(self.owner.call_service.call_count, 2) + self.owner.call_service.assert_called_with( + f'media_player/{direction}', entity_id=target) + + def test_unconditional_service_keeps_static_data(self): + caller = classes['ServiceCaller'](self.owner, { + 'service': 'media_player/volume_up', 'data': {'entity_id': 'media_player.test'}}) + caller.call_service('button', {'entity_id': 'ignored'}) + self.owner.call_service.assert_called_once_with( + 'media_player/volume_up', entity_id='media_player.test') + + +class SamsungVolumeTests(unittest.TestCase): + def setUp(self): + errors = SimpleNamespace(ConnectionFailure=ConnectionError, ResponseError=RuntimeError) + samsung = load_classes(TOOLBOX.parent / 'samsungtheframe.py', { + 'SmartObject': type('SmartObject', (), {}), 'tv_exceptions': errors, + 'websocket': SimpleNamespace(WebSocketException=ValueError), + }) + self.tv_app = samsung['SamsungTheFrame']() + self.tv_app.is_on_sensor = 'binary_sensor.is_tv_on' + self.tv_app.get_state = Mock(return_value='on') + self.connection = Mock() + self.tv_app.connect = Mock(return_value=self.connection) + self.tv_app.log = Mock() + + def test_volume_keys_and_connection_cleanup(self): + for direction, key in [('up', 'KEY_VOLUP'), ('down', 'KEY_VOLDOWN')]: + self.connection.reset_mock() + self.tv_app.on_volume_event('button', {}, direction) + self.connection.send_key.assert_called_once_with(key) + self.connection.close.assert_called_once_with() + + def test_art_mode_does_not_send_volume(self): + self.tv_app.get_state.return_value = 'off' + self.tv_app.on_volume_event('button', {}, 'up') + self.tv_app.connect.assert_not_called() + + def test_failure_is_logged_and_connection_closed(self): + self.connection.send_key.side_effect = ConnectionError('Disconnected') + self.tv_app.on_volume_event('button', {}, 'up') + self.tv_app.log.assert_called_once() + self.connection.close.assert_called_once_with() + + +if __name__ == '__main__': + unittest.main()