baseimage-tutorial-framework/src/event_handlers/source_code_event_handler.py

99 lines
3.3 KiB
Python
Raw Normal View History

from shutil import copy, rmtree, copytree
from os.path import splitext, isfile, join, basename
from glob import glob
2018-01-31 14:50:52 +00:00
from tfw.util import SupervisorMixin
from tfw.config import LOGIN_APP_DIR
from tfw.event_handler_base import EventHandlerBase
from tfw.config.logs import logging
log = logging.getLogger(__name__)
class FileManager:
def __init__(self, source_directory, working_directory, selected_file=None):
self._sourcedir = source_directory
self._workdir = working_directory
self._reload_files()
self.filename = selected_file or basename(self.files[0])
self.language = map_file_extension_to_language(self.filename)
def select_file(self, filename):
if not filename in self.files:
raise EnvironmentError('No such file in workdir!')
self.filename = filename
self.language = map_file_extension_to_language(self.filename)
@property
def files(self):
return [basename(file) for file in glob(join(self._workdir, '**/*'), recursive=True) if isfile(file)]
@property
def file_contents(self):
with open(self._filepath(self.filename), 'r') as ifile:
return ifile.read()
@file_contents.setter
def file_contents(self, value):
with open(self._filepath(self.filename), 'w') as ofile:
ofile.write(value)
def _filepath(self, filename):
return join(self._workdir, filename)
def _reload_files(self):
rmtree(self._workdir, ignore_errors=True)
copytree(self._sourcedir, self._workdir)
class SourceCodeEventHandler(EventHandlerBase, SupervisorMixin):
def __init__(self, anchor, directory, process_name=None, selected_file=None):
super().__init__(anchor)
self.filemanager = FileManager(directory, LOGIN_APP_DIR, selected_file=selected_file)
self.process_name = process_name
2018-01-18 13:35:53 +00:00
self.commands = {
'read': self.read,
'write': self.write,
'select': self.select
2018-01-18 13:35:53 +00:00
}
# Supervisor needs these to run the login program
copy('source_code_server/server.py', LOGIN_APP_DIR)
copy('source_code_server/users.db', LOGIN_APP_DIR)
def read(self, data_json):
data_json['data'] = {
'filename': self.filemanager.filename,
'content': self.filemanager.file_contents,
'language': self.filemanager.language,
'files': self.filemanager.files
}
def write(self, data_json):
self.filemanager.file_contents = data_json['data']['content']
self.restart_process()
def select(self, data_json):
try:
self.filemanager.select_file(data_json['data']['filename'])
except EnvironmentError:
log.critical('Failed to select file "{}"'.format(data_json['data']['filename']))
def handle_event(self, anchor, data_json):
data = data_json['data']
self.commands[data['command']](data_json)
return data_json
def map_file_extension_to_language(filename):
language_map = {
# TODO: extend, maybe auto-generate???
'.py': 'python',
'.js': 'javascript'
}
_, extension = splitext(filename)
language = 'javascript'
try: language = language_map[extension]
except KeyError: log.debug('No such extension in list, falling back to default: "{}".'.format(language))
return language