new event forwarder

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)

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()