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

from ast import literal_eval
from functools import partial
from signal import signal, SIGTERM, SIGINT

from tornado.ioloop import IOLoop

from tfw import YamlFSM, FSMAwareEventHandler, EventHandlerBase
from tfw.components import IdeEventHandler, TerminalEventHandler
from tfw.components import ProcessManagingEventHandler, BashMonitor
from tfw.components import TerminalCommands, LogMonitoringEventHandler
from tfw.components import FSMManagingEventHandler
from tfw.networking import MessageSender, TFWServerConnector
from tfw.config import TFWENV
from tfw.config.logs import logging
from tao.config import TAOENV

LOG = logging.getLogger(__name__)


def cenator(history):
    """
    Logs commands executed in terminal to messages.
    !! Please remove from production code. !!
    """
    LOG.debug('User executed command: "%s"', history[-1])
    MessageSender().send('JOHN CENA', f'You\'ve executed "{history[-1]}"')


class TestCommands(TerminalCommands):
    """
    Some example commands useful for debugging.
    !! Please remove from production code !! and inherit your own
    class from TerminalCommands if you need to define custom
    commands in your challenge.
    """
    # pylint: disable=unused-argument, attribute-defined-outside-init, no-self-use
    def command_sendmessage(self, *args):
        """
        Insert TFW message template as first argument if executed without args.

        Evaluate first argumen as a dict and send it to the frontend.
        This is useful for playing around with frontend APIs.
        """
        if not args:
            message_template = """'{"key": "", "data": {"command": ""}}'"""
            TFWServerConnector().send_to_eventhandler({
                'key': 'shell',
                'data': {
                    'command': 'write',
                    'value': f'sendmessage {message_template}'
                }
            })
        else:
            TFWServerConnector().send(literal_eval(args[0]))

    def command_seppuku_tfw(self, *args):
        """
        Restart tfw_server.py and event_handler_main.py.
        This can speed up development when combined with mounting
        volumes from host to container.
        """
        seppuku = (
            'nohup sh -c "supervisorctl restart tfwserver event_handler_main" &> /dev/null & '
            'clear && echo "Committed seppuku! :)" && sleep infinity'
        )
        uplink = TFWServerConnector()
        uplink.send_to_eventhandler({
            'key': 'shell',
            'data': {
                'command': 'write',
                'value': f'{seppuku}\n'
            }
        })
        uplink.send({
            'key': 'dashboard',
            'data': {
                'command': 'reloadFrontend'
            }
        })


class MessageFSMStepsEventHandler(FSMAwareEventHandler):
    """
    This example EventHandler is capable of detecting FSM state.
    !! Please remove from production code !!
    """
    def handle_event(self, message):
        pass

    def handle_fsm_step(self, from_state, to_state, trigger):
        """
        When the FSM steps this method is invoked.
        """
        MessageSender().send(
            'FSM info',
            f'FSM has stepped from state "{from_state}" '
            f'to state "{to_state}" in response to trigger "{trigger}"'
        )


if __name__ == '__main__':
    # TFW component EventHandlers (builtins, required for their respective functionalities)
    fsm = FSMManagingEventHandler( # TFW FSM
        key='fsm',
        fsm_type=partial(YamlFSM, 'test_fsm.yml')
    )
    ide = IdeEventHandler(  # Web IDE backend
        key='ide',
        allowed_directories=[TFWENV.IDE_WD, TFWENV.WEBSERVICE_DIR],
        directory=TFWENV.IDE_WD,
        exclude=['*.pyc']
    )
    terminal = TerminalEventHandler(  # Web shell backend
        key='shell',
        monitor=BashMonitor(TFWENV.HISTFILE)
    )
    processmanager = ProcessManagingEventHandler(  # Handles 'deploy' button clicks
        key='processmanager',
        dirmonitor=ide.monitor,
        log_tail=2000
    )
    logmonitor = LogMonitoringEventHandler(  # Sends live logs of webservice process to frontend
        key='logmonitor',
        process_name='webservice',
        log_tail=2000
    )

    # Your custom event handlers
    message_fsm_steps = MessageFSMStepsEventHandler(
        key='test'
    )

    # Terminal command handlers
    commands = TestCommands(bashrc=f'/home/{TAOENV.USER}/.bashrc')
    terminal.historymonitor.subscribe_callback(commands.callback)

    # Example terminal command callback
    terminal.historymonitor.subscribe_callback(cenator)

    event_handlers = EventHandlerBase.get_local_instances()
    def cleanup(sig, frame):
        for eh in event_handlers:
            eh.cleanup()
        exit(0)
    signal(SIGTERM, cleanup)
    signal(SIGINT, cleanup)

    IOLoop.instance().start()