test-tutorial-framework/solvable/src/event_handler_main.py

from re import match

from tornado.ioloop import IOLoop

from tfw.components import WebideEventHandler, TerminadoEventHandler, ProcessManagingEventHandler, BashMonitor
from tfw.networking import MessageSender, TFWServerConnector
from tfw.config import TFWENV
from tfw.config.logs import logging

LOG = logging.getLogger(__name__)


def cenator(history):
    LOG.debug('User executed command: "%s"', history[-1])
    MessageSender().send('JOHN CENA', 'You\'ve executed "{}"'.format(history[-1]))


class Commands:
    def __init__(self):
        self.command_impls = {self._parse_command_name(fun): getattr(self, fun) for fun in dir(self)
                              if callable(getattr(self, fun))
                              and self._is_command_implementation(fun)}

    @staticmethod
    def _is_command_implementation(method_name):
        return bool(Commands._match_command_regex(method_name))

    @staticmethod
    def _parse_command_name(method_name):
        try:
            return Commands._match_command_regex(method_name).groups()[0]
        except AttributeError:
            return ''

    @staticmethod
    def _match_command_regex(string):
        command_impl_regex = r'^_command_impl_(.+)$'
        return match(command_impl_regex, string)

    def callback(self, history):
        parts = history[-1].split()
        command = parts[0]
        if command in self.command_impls.keys():
            try:
                self.command_impls[command](*parts[1:])
            except IndexError:
                LOG.debug('Command "%s" failed!', command)

    def _command_impl_selectdir(self, *args):
        TFWServerConnector().send_to_eventhandler('webide',
                                                  {'data': {'command': 'selectdir',
                                                            'directory': args[0]}})

    def _command_impl_trigger(self, *args):
        TFWServerConnector().send('selectdir_needs_no_key',
                                  {'trigger': args[0]})

    def _command_impl_togglenext(self, *args):
        if not hasattr(self, 'togglenext_visible'):
            self.togglenext_visible = True
        TFWServerConnector().send('messagecontrol',
                                  {'data': {'command': 'showbutton',
                                            'next_visibility': self.togglenext_visible}})
        self.togglenext_visible = not self.togglenext_visible


if __name__ == '__main__':
    # pylint: disable=invalid-name
    ide = WebideEventHandler(key='webide', allowed_directories=[TFWENV.WEBIDE_WD],
                             directory=TFWENV.WEBIDE_WD, exclude=['*.pyc'])
    terminado = TerminadoEventHandler(key='shell', monitor=BashMonitor(TFWENV.HISTFILE))
    terminado.historymonitor.subscribe_callback(cenator)
    commands = Commands()
    terminado.historymonitor.subscribe_callback(commands.callback)
    processmanager = ProcessManagingEventHandler(key='processmanager', dirmonitor=ide.monitor)

    eventhandlers = {ide, terminado, processmanager}
    try:
        IOLoop.instance().start()
    finally:
        for eh in eventhandlers:
            eh.cleanup()