mirror of
https://github.com/avatao-content/baseimage-tutorial-framework
synced 2024-11-15 03:37:17 +00:00
102 lines
2.9 KiB
Python
102 lines
2.9 KiB
Python
import logging
|
|
|
|
from tfw.internals.networking import Scope
|
|
from tfw.internals.inotify import InotifyObserver
|
|
|
|
from .file_manager import FileManager
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
BUILD_ARTIFACTS = (
|
|
"*.a",
|
|
"*.class",
|
|
"*.dll",
|
|
"*.dylib",
|
|
"*.elf",
|
|
"*.exe",
|
|
"*.jar",
|
|
"*.ko",
|
|
"*.la",
|
|
"*.lib",
|
|
"*.lo",
|
|
"*.o",
|
|
"*.obj",
|
|
"*.out",
|
|
"*.py[cod]",
|
|
"*.so",
|
|
"*.so.*",
|
|
"*.tar.gz",
|
|
"*.zip",
|
|
"*__pycache__*"
|
|
)
|
|
|
|
|
|
class IdeHandler:
|
|
keys = ['ide']
|
|
type_id = 'ControlEventHandler'
|
|
|
|
def __init__(self, *, patterns, initial_file=None):
|
|
self.connector = None
|
|
self.filemanager = FileManager(patterns)
|
|
self._initial_file = initial_file or ''
|
|
|
|
self.monitor = InotifyObserver(
|
|
path=self.filemanager.parents,
|
|
exclude=BUILD_ARTIFACTS
|
|
)
|
|
self.monitor.on_modified = self._reload_frontend
|
|
self.monitor.start()
|
|
|
|
self.commands = {
|
|
'ide.read' : self.read,
|
|
'ide.write' : self.write
|
|
}
|
|
|
|
def _reload_frontend(self, event): # pylint: disable=unused-argument
|
|
self.send_message({'key': 'ide.reload'})
|
|
|
|
@property
|
|
def initial_file(self):
|
|
if not self.filemanager.is_allowed(self._initial_file):
|
|
self._initial_file = self.filemanager.files[0]
|
|
return self._initial_file
|
|
|
|
def send_message(self, message):
|
|
self.connector.send_message(message, scope=Scope.WEBSOCKET)
|
|
|
|
def handle_event(self, message, _):
|
|
try:
|
|
self.commands[message['key']](message)
|
|
message['files'] = self.filemanager.files
|
|
self.send_message(message)
|
|
except KeyError:
|
|
LOG.error('IGNORING MESSAGE: Invalid message received: %s', message)
|
|
|
|
def read(self, message):
|
|
if 'patterns' in message:
|
|
self.filemanager.patterns = message['patterns']
|
|
try:
|
|
message['filename'] = self.filemanager.find_file(
|
|
message.get('filename') or self.initial_file
|
|
)
|
|
message['content'] = self.filemanager.read_file(message['filename'])
|
|
except (PermissionError, ValueError):
|
|
message['content'] = 'You have no permission to open that file :('
|
|
except FileNotFoundError:
|
|
message['content'] = 'This file does not exist :('
|
|
except Exception: # pylint: disable=broad-except
|
|
message['content'] = 'Failed to read file :('
|
|
LOG.exception('Error reading file!')
|
|
|
|
def write(self, message):
|
|
try:
|
|
self.filemanager.write_file(message['filename'], message['content'])
|
|
except KeyError:
|
|
LOG.error('You must provide a filename to write!')
|
|
except Exception: # pylint: disable=broad-except
|
|
LOG.exception('Error writing file!')
|
|
del message['content']
|
|
|
|
def cleanup(self):
|
|
self.monitor.stop()
|