75 lines
1.5 KiB
Python
75 lines
1.5 KiB
Python
from pathlib import Path
|
|
|
|
from importing import load_module
|
|
from task import Task
|
|
from utils import printable_exception
|
|
|
|
|
|
class Module:
|
|
|
|
def __init__(self, fname, entry):
|
|
fname = Path(fname)
|
|
self.fname = fname
|
|
self.entry = entry
|
|
self.name = fname.stem.replace("-", "_")
|
|
self.mtime = self.get_mtime()
|
|
self.mod = self.func = self.task = None
|
|
|
|
|
|
def start(self):
|
|
if not self.func:
|
|
raise MissingEntryFunctionError
|
|
|
|
self.task = task = Task(self.func, self.name)
|
|
task.start()
|
|
return task
|
|
|
|
|
|
def kill(self):
|
|
if self.task:
|
|
self.task.kill()
|
|
|
|
@property
|
|
def is_running(self):
|
|
if self.task:
|
|
return self.task.is_running
|
|
|
|
|
|
def load(self):
|
|
try:
|
|
self.mod = mod = load_module(self.name, self.fname)
|
|
except Exception as e:
|
|
raise ModuleLoadError(printable_exception(e)) from e
|
|
|
|
try:
|
|
self.func = getattr(mod, self.entry)
|
|
except AttributeError as e:
|
|
raise MissingEntryFunctionError from e
|
|
|
|
|
|
def file_exists(self):
|
|
return self.fname.is_file()
|
|
|
|
def has_changed(self):
|
|
return self.mtime != self.get_mtime()
|
|
|
|
def get_mtime(self):
|
|
try:
|
|
return self.fname.stat().st_mtime
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
def __str__(self):
|
|
return f"{self.fname} -> {self.name} ({self.mtime})"
|
|
|
|
|
|
|
|
class ModuleLoadError(Exception):
|
|
pass
|
|
|
|
class MissingEntryFunctionError(Exception):
|
|
pass
|
|
|
|
|
|
|