72 lines
2.5 KiB
Python
72 lines
2.5 KiB
Python
import logging
|
|
from ast import literal_eval
|
|
|
|
from tfw.components import MessageSender
|
|
from tfw.builtins import TFWServerUplinkConnector
|
|
from tfw.builtins import EventHandler, FSMAwareEventHandler, TerminalCommandsEventHandler
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
|
|
class CenatorEventHandler(EventHandler):
|
|
"""
|
|
Logs commands executed in terminal to messages and invokes an
|
|
additional callback function to handle special commands.
|
|
!! Please remove from production code. !!
|
|
"""
|
|
def __init__(self, key):
|
|
super().__init__(key)
|
|
|
|
def handle_event(self, message):
|
|
command = message['value']
|
|
LOG.debug('User executed command: "%s"', command)
|
|
MessageSender(self.server_connector).send('JOHN CENA', f'You\'ve executed "{command}"')
|
|
|
|
|
|
class TestCommandsEventHandler(TerminalCommandsEventHandler):
|
|
"""
|
|
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": ""}}'"""
|
|
TFWServerUplinkConnector().send_message({
|
|
'key': 'shell',
|
|
'data': {
|
|
'command': 'write',
|
|
'value': f'sendmessage {message_template}'
|
|
}
|
|
})
|
|
else:
|
|
TFWServerUplinkConnector().send_message(literal_eval(args[0]))
|
|
|
|
|
|
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, **kwargs):
|
|
"""
|
|
When the FSM steps this method is invoked.
|
|
Receives a 'data' field from an fsm_update message as kwargs.
|
|
"""
|
|
MessageSender(self.server_connector).send(
|
|
'FSM info',
|
|
f'FSM has stepped from state "{kwargs["last_event"]["from_state"]}" '
|
|
f'to state "{kwargs["current_state"]}" in response to trigger "{kwargs["last_event"]["trigger"]}"'
|
|
)
|