baseimage-tutorial-framework/lib/tfw/components/source_code_event_handler.py

128 lines
4.5 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

from os.path import isfile, join, relpath, exists, isdir
from glob import glob
from tfw.event_handler_base import TriggerlessEventHandler
from tfw.components.directory_monitor import DirectoryMonitor
from tfw.config.logs import logging
log = logging.getLogger(__name__)
class FileManager:
def __init__(self, working_directory, selected_file=None, exclude=None):
self.exclude = exclude if exclude is not None else []
self._workdir = None
self.workdir = working_directory
self.filename = selected_file or self.files[0]
@property
def workdir(self):
return self._workdir
@workdir.setter
def workdir(self, directory):
if not exists(directory) or not isdir(directory):
raise EnvironmentError('"{}" is not a directory!'.format(directory))
self._workdir = directory
def select_file(self, filename):
if not filename in self.files:
raise EnvironmentError('No such file in workdir!')
self.filename = filename
@property
def files(self):
return [self._relpath(file) for file in glob(join(self._workdir, '**/*'), recursive=True)
if isfile(file) and
not any(word in file for word in self.exclude)]
@property
def file_contents(self):
with open(self._filepath(self.filename), 'r', errors='surrogateescape') as ifile:
return ifile.read()
@file_contents.setter
def file_contents(self, value):
with open(self._filepath(self.filename), 'w', errors='surrogateescape') as ofile:
ofile.write(value)
def _filepath(self, filename):
return join(self._workdir, filename)
def _relpath(self, filename):
return relpath(self._filepath(filename), start=self._workdir)
class SourceCodeEventHandler(TriggerlessEventHandler):
def __init__(self, key, directory, selected_file=None, exclude=None):
super().__init__(key)
self.filemanager = FileManager(directory, selected_file=selected_file, exclude=exclude)
self.commands = {'read': self.read,
'write': self.write,
'select': self.select,
'selectdir': self.select_dir}
self._monitor = None
self.reload_monitor()
@property
def monitor(self):
return self._monitor
def reload_monitor(self):
if self._monitor:
try: self._monitor.stop()
except KeyError: logging.debug('Working directory was removed ignoring...')
self._monitor = DirectoryMonitor(self.filemanager.workdir)
self._monitor.watch() # This runs on a separate thread
def read(self, data):
try: data['content'] = self.filemanager.file_contents
except PermissionError: data['content'] = 'You have no permission to open that file :('
except FileNotFoundError: data['content'] = 'This file was removed :('
except Exception: data['content'] = 'Failed to read file :('
return data
def write(self, data):
self.monitor.eventhandler.ignore = 1
try: self.filemanager.file_contents = data['content']
except Exception: log.exception('Error writing file!')
del data['content']
return data
def select(self, data):
try: self.filemanager.select_file(data['filename'])
except EnvironmentError: log.exception('Failed to select file "{}"'.format(data['filename']))
return data
def select_dir(self, data):
try:
self.filemanager.workdir = data['directory']
self.reload_monitor()
try:
self.filemanager.select_file(self.filemanager.files[0])
self.read(data)
except IndexError:
data['content'] = 'No files in this directory :('
except EnvironmentError:
log.exception('Failed to select directory "{}"'.format(data['directory']))
return data
def attach_fileinfo(self, data):
data['filename'] = self.filemanager.filename
data['files'] = self.filemanager.files
data['directory'] = self.filemanager.workdir
def handle_event(self, key, message):
try:
data = message['data']
message['data'] = self.commands[data['command']](data)
self.attach_fileinfo(data)
return message
except KeyError:
log.error('IGNORING MESSAGE: Invalid message received: {}'.format(message))
def cleanup(self):
self.monitor.stop()