import logging from sys import stderr from functools import partial from tornado.ioloop import IOLoop from tfw.fsm import YamlFSM from tfw.builtins import IdeEventHandler, TerminalEventHandler, FrontendEventHandler from tfw.builtins import LogMonitoringEventHandler, ProcessManagingEventHandler from tfw.builtins import DirectorySnapshottingEventHandler, FSMManagingEventHandler from tfw.config import TFWENV from tfw.logging import Log, Logger, LogFormatter, VerboseLogFormatter from tao.config import TAOENV from custom_event_handlers import MessageFSMStepsEventHandler from custom_event_handlers import TerminalCallbackEventHandler, TestCommands from signal_handling import setup_signal_handlers LOG = logging.getLogger(__name__) def main(): # pylint: disable=unused-variable Logger([ Log(stderr, LogFormatter(20)), Log(TFWENV.LOGFILE, VerboseLogFormatter()) ]).start() # TFW component EventHandlers (builtins, required for their respective functionalities) fsm = FSMManagingEventHandler( # TFW FSM key='fsm', fsm_type=partial( YamlFSM, 'test_fsm.yml', {} # jinja2 variables, use empty dict to enable jinja2 parsing without any variables ) ) 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' ) commands = TerminalCallbackEventHandler( # Reacts to terminal commands 'history.bash', TestCommands(bashrc=f'/home/{TAOENV.USER}/.bashrc').callback ) processmanager = ProcessManagingEventHandler( # Handles 'deploy' button clicks key='processmanager', log_tail=2000 ) logmonitor = LogMonitoringEventHandler( # Sends live logs of webservice process to frontend key='logmonitor', process_name='webservice', log_tail=2000 ) snapshot = DirectorySnapshottingEventHandler( # Manages filesystem snapshots of directories key='snapshot', directories=[ TFWENV.IDE_WD, TFWENV.WEBSERVICE_DIR ] ) frontend = FrontendEventHandler() # Proxies frontend API calls to frontend # Your custom event handlers message_fsm_steps_eh = MessageFSMStepsEventHandler( key='test' ) setup_signal_handlers() IOLoop.instance().start() if __name__ == '__main__': main()