import inspect
from qtpy.QtCore import QObject, Signal
from qtpyvcp.utilities.logger import getLogger, logLevelFromName
LOG = getLogger(__name__)
[docs]class Plugin(QObject):
"""QtPyVCP Plugin base class."""
def __init__(self):
super(Plugin, self).__init__()
self._log = None
self._initialized = False
self._postGuiInitialized = False
@property
def log(self):
if self._log is None:
self._log = getLogger(self.__module__)
return self._log
return self._log
[docs] def initialise(self):
"""Initialize the plugin.
This method is called after the main event loop has started. Any timers
or threads used by the plugin should be started here.
This method should set ``self._initialized`` to true if successfully.
"""
self._initialized = True
[docs] def postGuiInitialise(self, main_window):
"""Initialize the plugin after mainwindow is shown.
This method is called after the main window is shown.
Args:
main_window (VCPMainWindow) : the VCPMainWindow object.
This method should set ``self._postGuiInitialized`` to true if successfully.
"""
self._postGuiInitialized = True
[docs] def terminate(self):
"""Terminate the plugin.
This is called right before the main event loop exits. Any cleanup
of the plugin should be done here, such as saving persistent data.
"""
pass
def isDataChan(obj):
return isinstance(obj, DataChannel)
[docs]class DataPlugin(Plugin):
"""DataPlugin."""
def __init__(self):
super(DataPlugin, self).__init__()
self.channels = {name: obj for name, obj in
inspect.getmembers(self, isDataChan)}
[docs] def getChannel(self, url):
"""Get data channel from URL.
Args:
url (str) : The URL of the channel to get.
Returns:
tuple : (chan_obj, chan_exp)
"""
chan, sep, query = url.partition('?')
raw_args = query.split('&')
# print(url, chan, raw_args)
args = []
kwargs = {}
for arg in [a for a in raw_args if a != '']:
if '=' in arg:
key, val = arg.split('=')
kwargs[key] = val
else:
args.append(arg)
# print(chan, args, kwargs)
try:
chan_obj = self.channels[chan]
if len(args) > 0 and args[0] in ('string', 'text', 'str'):
chan_exp = lambda: chan_obj.getString(*args[1:], **kwargs)
else:
chan_exp = lambda: chan_obj.getValue(*args, **kwargs)
except (KeyError, SyntaxError):
return None, None
return chan_obj, chan_exp
[docs] def setLogLevel(self, level):
"""Set plugin log level.
Args:
level (str, int) : Log level (DEBUG | INFO | ERROR | CRITICAL)
"""
if level:
if isinstance(level, str):
level = logLevelFromName(level)
self.log.setLevel(level)
[docs]class DataChannel(QObject):
signal = Signal(object)
def __init__(self, fget=None, fset=None, fstr=None, data=None, settable=False,
doc = None):
super(DataChannel, self).__init__()
self.fget = fget
self.fset = fset
self.fstr = fstr
self.value = data
self.settable = settable
self.instance = None
if doc is None and fget is not None:
doc = fget.__doc__
self.__doc__ = doc
[docs] def getValue(self, *args, **kwargs):
"""Channel data getter method."""
if self.fget is None:
return self.value
return self.fget(self.instance, self, *args, **kwargs)
[docs] def getString(self, *args, **kwargs):
"""Channel data getter method."""
if self.fstr is None:
return str(self.value)
return self.fstr(self.instance, self, *args, **kwargs)
[docs] def setValue(self, value):
"""Channel data setter method."""
if self.fset is None:
self.value = value
self.signal.emit(value)
else:
self.fset(self.instance, self, value)
def getter(self, fget):
def inner(*args, **kwargs):
fget(*args, **kwargs)
self.fget = inner
return self
def setter(self, fset):
def inner(*args, **kwargs):
fset(*args, **kwargs)
self.fset = inner
return self
def tostring(self, fstr):
def inner(*args, **kwargs):
return fstr(*args, **kwargs)
self.fstr = inner
return self
def notify(self, slot, *args, **kwargs):
# print('Connecting %s to slot %s' % (self._signal, slot))
if len(args) == 0 and len(kwargs) == 0:
self.signal.connect(slot)
else:
if args[0] in ['string', 'str']:
self.signal.connect(lambda: slot(self.getString(*args[1:], **kwargs)))
else:
self.signal.connect(lambda: slot(self.getValue(*args, **kwargs)))
# fixme
onValueChanged = notify
def __get__(self, instance, owner):
self.instance = instance
return self
def __call__(self, *args, **kwargs):
return self.getValue(*args, **kwargs)
def __set__(self, instance, value):
return self.setValue(value)
def __getitem__(self, item):
return self.value[item]
def __str__(self):
return self.getString()