"""
A module for decorating AF2 methods as Maestro callbacks. The decorators may be
used outside of Maestro, but they will have no effect.
"""
import collections
import functools
import inspect
import schrodinger
from schrodinger import project
from schrodinger.infra import util
from schrodinger.project import utils as project_utils
from schrodinger.Qt import QtCore
from schrodinger.Qt import QtGui
from schrodinger.Qt import QtWidgets
from schrodinger.Qt.QtCore import Qt
maestro = schrodinger.get_maestro()
class _WorkspaceEntriesObserver(QtCore.QObject):
"""
An instance of this class watches the workspace for changes and emits events
whenever the entries in the workspace change.
Only one instance of the class needs to exist, and it is instantiated below.
In the typical case (af2 panels), no reference to the instance is necessary.
Rather, the decorators, entries_included and entries_excluded, defined below
can be used to with any method on an af2 panel.
"""
entriesIncluded = QtCore.pyqtSignal(set)
entriesExcluded = QtCore.pyqtSignal(set)
def __init__(self):
"""
Initialize object and start listening for changes in the workspace
We do a check to see if maestro exists because a single instance of this
class is created below and we want to be able to import this outside of
maestro (e.g., in tests).
"""
QtCore.QObject.__init__(self)
self._entries = set()
if maestro:
maestro.workspace_changed_function_add(self._onWorkspaceChanged)
def _emitEntryChanges(self, current_entries):
"""
Checks a set of entry ids against the previous set and emits signals
for both new and departing entries
:param current_entries: A set of entry ids currently in the workspace
:type current_entries: set
"""
new_entries = current_entries - self._entries
if new_entries:
self.entriesIncluded.emit(new_entries)
departing_entries = self._entries - current_entries
if departing_entries:
self.entriesExcluded.emit(departing_entries)
self._entries = current_entries
def _onWorkspaceChanged(self, what_changed):
"""
Listens for changes in the Workspace and passes a list of currently
included entry ids to _emitEntryChanges
"""
if what_changed in (maestro.WORKSPACE_CHANGED_EVERYTHING,
maestro.WORKSPACE_CHANGED_APPEND):
try:
current_entries = maestro.get_included_entry_ids()
except project.ProjectException:
# This happens if the project has been closed
current_entries = set()
self._emitEntryChanges(current_entries)
def currentEntryIds(self):
"""
Return a set of entry ids for all entries currently in the workspace
"""
# Return a copy of the set. That way, if the calling scope modifies the
# return value, it won't affect our set.
return self._entries.copy()
workspace_entries_observer = _WorkspaceEntriesObserver()
ENTRIES_INCLUDED_CALLBACK = 'entries_included'
ENTRIES_EXCLUDED_CALLBACK = 'entries_excluded'
PROJECT_CHANGE_CALLBACK = 'project_change'
WORKSPACE_CHANGED_ACTIVE_PROJ_CALLBACK = 'workspace_changed_active_proj'
# CallbackInfo simply collects together a triplet of a function to add a
# callback, a function to remove a callback and a Bool indicating whether we
# should check that the callback is a properly registered maestro callback. We
# use CallbackInfo in check_callbacks
CallbackInfo = collections.namedtuple(
'CallbackInfo', ['add', 'remove', 'maestro_check_callback'])
if maestro:
from schrodinger.maestro.maestro import HOVER_CALLBACK
from schrodinger.maestro.maestro import PROJECT_CLOSE_CALLBACK
from schrodinger.maestro.maestro import PROJECT_UPDATE_CALLBACK
from schrodinger.maestro.maestro import WORKSPACE_CHANGED_CALLBACK
CALLBACK_FUNCTIONS = {
PROJECT_UPDATE_CALLBACK: CallbackInfo(
add=maestro.project_update_callback_add,
remove=maestro.project_update_callback_remove,
maestro_check_callback=True),
PROJECT_CLOSE_CALLBACK: CallbackInfo(
add=maestro.project_close_callback_add,
remove=maestro.project_close_callback_remove,
maestro_check_callback=True),
WORKSPACE_CHANGED_CALLBACK: CallbackInfo(
add=maestro.workspace_changed_function_add,
remove=maestro.workspace_changed_function_remove,
maestro_check_callback=True),
HOVER_CALLBACK: CallbackInfo(add=maestro.hover_callback_add,
remove=maestro.hover_callback_remove,
maestro_check_callback=True),
ENTRIES_INCLUDED_CALLBACK: CallbackInfo(
add=workspace_entries_observer.entriesIncluded.connect,
remove=workspace_entries_observer.entriesIncluded.disconnect,
maestro_check_callback=False),
ENTRIES_EXCLUDED_CALLBACK: CallbackInfo(
add=workspace_entries_observer.entriesExcluded.connect,
remove=workspace_entries_observer.entriesExcluded.disconnect,
maestro_check_callback=False)
}
else:
# Dummy values so we don't get NameErrors when decorating a method outside
# of Maestro. Note that only the tests rely on the specific values set
# here.
PROJECT_UPDATE_CALLBACK = 'project_update'
PROJECT_CLOSE_CALLBACK = 'project_close'
WORKSPACE_CHANGED_CALLBACK = 'workspace_changed'
HOVER_CALLBACK = 'hover'
CALLBACK_FUNCTIONS = {}
# a decorator for methods that should be skipped if we're inside an
# ignoreMaestroCallbacks() context
skip_if_ignoring_maestro_callbacks = util.skip_if("_ignoring_maestro_callbacks")
# The following decorators may be applied to methods in an AF2 class (or any
# other class that inherits from MaestroCallbackMixin). Note, however, that
# these decorator functions will be passed a function object, not a method
# object, hence the variable name "func". Because these decorators do not have
# access to the method object, the project_changed and
# workspace_changed_active_project wrappers are applied when the class is
# instantiated rather than when the decorator is run.
[docs]def project_changed(func):
"""
A decorator for methods that should be called when the project updates but
not when the project closes. Decorated methods that take one argument will
be called with the active project (`schrodinger.project.Project`).
Decorated methods may also take no arguments.
:note: Decorated methods will not be called when the panel is closed. If
a callback occurred while the panel was closed, then the decorated
method will be called when the panel is re-opened.
"""
func.maestro_callback = PROJECT_CHANGE_CALLBACK
func.maestro_callback_wrapper = (_project_changed_wrapper,
PROJECT_UPDATE_CALLBACK)
return skip_if_ignoring_maestro_callbacks(func)
[docs]def project_close(func):
"""
A decorator for methods that should be called immediately before the project
closes. Decorated methods will be called with no arguments.
:note: Decorated methods will not be called when the panel is closed. If
a callback occurred while the panel was closed, then the decorated method
will be called when the panel is re-opened.
"""
func.maestro_callback = PROJECT_CLOSE_CALLBACK
return skip_if_ignoring_maestro_callbacks(func)
[docs]def project_updated(func):
"""
A decorator for methods that should be called when the project updates,
regardless of whether the project was closed. Decorated methods will be
called with no arguments. Consider using `project_changed` and/or
`project_close` instead.
:note: Decorated methods will not be called when the panel is closed. If
a callback occurred while the panel was closed, then the decorated method
will be called when the panel is re-opened.
"""
func.maestro_callback = PROJECT_UPDATE_CALLBACK
return skip_if_ignoring_maestro_callbacks(func)
[docs]def workspace_changed(func):
"""
A decorator for methods that should be called when the workspace changes,
regardless of whether the workspace change was triggered by a project
closure. Decorated methods will be called with the what changed flag from
Maestro.
:note: Decorated methods will not be called when the panel is closed. If a
callback occurred while the panel was closed, then the decorated method
will be called when the panel is re-opened.
"""
func.maestro_callback = WORKSPACE_CHANGED_CALLBACK
return skip_if_ignoring_maestro_callbacks(func)
[docs]def workspace_changed_active_project(func):
"""
A decorator for methods that should be called when the workspace changes,
but not when the workspace change was triggered by a project closure.
Decorated methods that take one argument will be called with:
- The what changed flag from Maestro
Decorated methods that take two arguments will be called with:
- The what changed flag from Maestro
- The active project (`schrodinger.project.Project`)
:note: Decorated methods will not be called when the panel is closed. If a
callback occurred while the panel was closed, then the decorated method
will be called when the panel is re-opened.
"""
func.maestro_callback = WORKSPACE_CHANGED_ACTIVE_PROJ_CALLBACK
func.maestro_callback_wrapper = (_ws_no_close_wrapper,
WORKSPACE_CHANGED_CALLBACK)
return skip_if_ignoring_maestro_callbacks(func)
[docs]def entries_included(func):
"""
A decorator for methods that should be called when an entry enters the
Workspace
The decorated method is passed a set of newly included entries when the
Workspace changes
:note: Decorated methods will not be called when the panel is closed. If
a callback occurred while the panel was closed, then the decorated method
will be called when the panel is re-opened.
"""
func.maestro_callback = ENTRIES_INCLUDED_CALLBACK
return skip_if_ignoring_maestro_callbacks(func)
[docs]def entries_excluded(func):
"""
A decorator for methods that should be called when an entry exits the
Workspace
The decorated method is passed a set of newly excluded entries when the
Workspace changes
:note: Decorated methods will not be called when the panel is closed. If
a callback occurred while the panel was closed, then the decorated method
will be called when the panel is re-opened.
"""
func.maestro_callback = ENTRIES_EXCLUDED_CALLBACK
return skip_if_ignoring_maestro_callbacks(func)
[docs]class AbstractMaestroCallbackMixin(object):
"""
A mixin that allows the above decorators to be used for maestro callbacks.
Any callbacks that occur while the panel is closed will be monitored and the
appropriate callback methods will be called when the panel is re-opened.
This class may only be mixed in to `PyQt5.QtWidgets.QWidget`s. Note this
widget should not be used directly. Instead, use `MaestroCallbackMixin` or
`MaestroCallbackWidgetMixin`.
:cvar IGNORE_DELAYED_CALLBACKS: Whether or not delayed Maestro callbacks
(e.g. those triggered when a panel is closed) should be ignored by the
panel.
:vartype IGNORE_DELAYED_CALLBACKS: bool
"""
IGNORE_DELAYED_CALLBACKS = False
ignoreMaestroCallbacks = util.flag_context_manager(
"_ignoring_maestro_callbacks")
ignoreMaestroCallbacks.__doc__ = """
A context manager for temporarily disabling Maestro callbacks created
using the decorators above. (Note that callbacks that have been manually
added using maestro.*_callback_add() will not be disabled.)
Example::
def includeEntry(self, entry_id):
proj = maestro.project_table_get()
with self.ignoreMaestroCallbacks():
proj[entry_id].in_workspace = project.IN_WORKSPACE
@maestro_callback.project_changed
def onProjectChanged(self):
print "This method will not be called during includeEntry."
@maestro_callback.workspace_changed
def onWorkspaceChanged(self):
print "Neither will this one."
"""
[docs] def __init__(self, *args, **kwargs):
self._ignoring_maestro_callbacks = False
self._maestro_callbacks = {}
self._maestro_callbacks_wrapped = {}
self._callbacks_registered = False
self._callback_monitor = CallbackMonitor()
super(AbstractMaestroCallbackMixin, self).__init__(*args, **kwargs)
if maestro:
self.buildCallbackDicts()
def _showEvent(self, event):
"""
When the panel is shown, add callbacks for the decorated methods and
trigger any callbacks that occurred while the panel was closed
"""
# Ignore spontaneous events (i.e. un-minimizing the window)
if maestro and not event.spontaneous():
self._addCallbacks()
if not self.IGNORE_DELAYED_CALLBACKS:
calls = self._callback_monitor.stopMonitoring()
self._delayedCallbacks(calls)
def _closeEvent(self, event):
"""
When the panel is closed, remove callbacks for the decorated methods but
continue to monitor callbacks (so the appropriate methods can be called
when the panel is re-opened.
"""
if maestro:
self._removeCallbacks()
if (not self.IGNORE_DELAYED_CALLBACKS and
not (self.testAttribute(Qt.WA_DeleteOnClose) or
self.window().testAttribute(Qt.WA_DeleteOnClose))):
# Make sure we don't add callback functions if the panel object
# is about to be deleted
self._callback_monitor.startMonitoring()
def _addCallbacks(self):
"""
Add maestro callbacks for all decorated methods. The callbacks will not
be added if they are already present.
"""
wrapped = self._maestro_callbacks_wrapped
for callback_type, functions in wrapped.items():
callback_info = CALLBACK_FUNCTIONS[callback_type]
add_callback = callback_info.add
check_callback = callback_info.maestro_check_callback
for cur_func in functions:
if not check_callback:
add_callback(cur_func)
elif not maestro.is_function_registered(callback_type,
cur_func):
add_callback(cur_func)
self._callbacks_registered = True
def _removeCallbacks(self):
"""
Remove maestro callbacks for all decorated methods. If the callbacks
are not present, an exception will be raised.
"""
if not self._callbacks_registered:
# When parent panel is closed before child panel, multiple
# close events can be issued by Qt, causing multiple calls to
# _removeCallback() - honor first call only
return
wrapped = self._maestro_callbacks_wrapped
for callback_type, functions in wrapped.items():
remove_callback = CALLBACK_FUNCTIONS[callback_type].remove
for cur_func in functions:
remove_callback(cur_func)
self._callbacks_registered = False
[docs] def buildCallbackDicts(self):
"""
Create a dictionary of all methods that have a maestro_callback
decorator.
"""
unwrapped = self._maestro_callbacks
wrapped = self._maestro_callbacks_wrapped
for attr in dir(self):
func = getattr(self, attr)
if hasattr(func, "maestro_callback") and inspect.ismethod(func):
callback_type = func.maestro_callback
callback_list = unwrapped.setdefault(callback_type, [])
callback_list.append(func)
if hasattr(func, "maestro_callback_wrapper"):
wrapper, callback_type = func.maestro_callback_wrapper
func = functools.partial(wrapper, func)
callback_list = wrapped.setdefault(callback_type, [])
callback_list.append(func)
def _delayedCallbacks(self, calls):
"""
Call the appropriate methods for any callbacks that occurred while the
panel was closed
:param calls: A description of the callbacks that occurred
:type calls: `MonitoredCallbacks`
"""
unwrapped = self._maestro_callbacks
proj = project_utils.get_PT()
if proj is None:
# The project was closed
return
if calls.project_close:
for func in unwrapped.get(PROJECT_CLOSE_CALLBACK, []):
func()
if calls.project_changed:
for func in unwrapped.get(PROJECT_CHANGE_CALLBACK, []):
# proj argument is optional
if _num_args(func) == 1:
func()
else:
func(proj)
if calls.project_updated:
for func in unwrapped.get(PROJECT_UPDATE_CALLBACK, []):
func()
if calls.workspace_changed:
for what_changed in calls.workspace_changed:
for func in unwrapped.get(WORKSPACE_CHANGED_CALLBACK, []):
func(what_changed)
if calls.workspace_changed_active_proj:
for what_changed in calls.workspace_changed_active_proj:
for func in unwrapped.get(
WORKSPACE_CHANGED_ACTIVE_PROJ_CALLBACK, []):
# proj argument is optional
if _num_args(func) == 2:
func(what_changed)
else:
func(what_changed, proj)
if calls.removed_eids:
for func in unwrapped.get(ENTRIES_EXCLUDED_CALLBACK, []):
func(calls.removed_eids)
if calls.added_eids:
for func in unwrapped.get(ENTRIES_INCLUDED_CALLBACK, []):
func(calls.added_eids)
[docs]class MaestroCallbackMixin(AbstractMaestroCallbackMixin):
"""
A mixin that allows the maestro callback decorators to be used in a panel
class or in any `QtWidgets.QWidget` that will be its own window.
"""
[docs] def showEvent(self, event):
# See Qt documentation for method documentation.
self._showEvent(event)
super(MaestroCallbackMixin, self).showEvent(event)
[docs] def closeEvent(self, event):
# See Qt documentation for method documentation.
self._closeEvent(event)
super(MaestroCallbackMixin, self).closeEvent(event)
[docs]class MaestroCallbackModelMixin(MaestroCallbackWidgetMixin):
"""
A mixin that allows the maestro callback decorators to be used on a
`QtCore.QAbstractItemModel`. Any model that uses this mixin must have a
`QtWidgets.QWidget` parent.
"""
[docs] def window(self):
# QAbstractProxyModel.parent(index) shadows QObject.parent(), so we have
# to call QObject.parent() directly
parent = QtCore.QObject.parent(self)
return parent.window()
def _closeEvent(self, event):
"""
When the panel is closed, remove callbacks for the decorated methods but
continue to monitor callbacks (so the appropriate methods can be called
when the panel is re-opened.
This method overrides the parent class method since
`QtCore.QAbstractItemModel`s don't have a `testAttribute` method.
"""
if maestro:
self._removeCallbacks()
if not self.window().testAttribute(Qt.WA_DeleteOnClose):
# Make sure we don't add callback functions if the panel object
# is about to be deleted
self._callback_monitor.startMonitoring()
def _project_changed_wrapper(func):
"""
A wrapper for methods that should be called when the project updates but
not when the project closes.
:param func: The function to wrap and call
:type func: function
:return: The function return value
"""
proj = project_utils.get_PT()
if proj is None:
# The project was closed
return
if _num_args(func) == 1:
return func()
else:
return func(proj)
def _ws_no_close_wrapper(func, what_changed):
"""
A wrapper for methods that should be called when the workspace changes but
not when the project was closed.
:param func: The function to wrap and call
:type func: function
:param what_changed: The what changed flag from Maestro
:type what_changed: str
:return: The function return value
"""
try:
proj = maestro.project_table_get()
except project.ProjectException:
# The project was closed
pass
else:
if _num_args(func) == 2:
return func(what_changed)
else:
return func(what_changed, proj)
[docs]class MonitoredCallbacks(object):
"""
Data describing which callbacks have occurred since a panel was closed
"""
[docs] def __init__(self):
self.reset()
[docs] def reset(self):
self.project_changed = False
self.project_close = False
self.project_updated = False
self.workspace_changed = set()
self.workspace_changed_active_proj = set()
self.added_eids = set()
self.removed_eids = set()
[docs]class CallbackMonitor(object):
"""
Monitoring for Maestro callbacks that occur while a panel is closed
"""
[docs] def __init__(self):
self._calls = MonitoredCallbacks()
self._eids_start = None
self._currently_monitoring = False
self._monitor_funcs = {
PROJECT_UPDATE_CALLBACK: self._projectUpdated,
PROJECT_CLOSE_CALLBACK: self._projectClosed,
WORKSPACE_CHANGED_CALLBACK: self._workspaceChanged,
}
[docs] def startMonitoring(self):
"""
Start monitoring all callbacks
"""
if self._currently_monitoring:
# So that monitoring is not re-started on secondary close events.
# e.g. 2nd event can be triggered when parent window is closed if
# this panel is a dialog.
return
self._calls.reset()
self._eids_start = workspace_entries_observer.currentEntryIds()
self._addCallbacks()
self._currently_monitoring = True
def _addCallbacks(self):
for callback_type, func in self._monitor_funcs.items():
add = CALLBACK_FUNCTIONS[callback_type].add
add(func)
[docs] def stopMonitoring(self):
"""
Stop monitoring all callbacks
:return: The callbacks that have occurred since monitoring was started
:rtype: `MonitoredCallbacks`
"""
for callback_type, func in self._monitor_funcs.items():
self._removeCallback(callback_type, func)
self._trimWorkspaceChanged()
self._determineEids()
self._currently_monitoring = False
return self._calls
def _removeCallback(self, callback_type, func):
"""
Remove the specified callback monitoring
:param callback_type: The callback type to stop monitoring
:type callback_type: str
:param func: The callback to remove
:type func: function
"""
if maestro.is_function_registered(callback_type, func):
remove = CALLBACK_FUNCTIONS[callback_type].remove
remove(func)
def _removeCallbackDelayed(self, callback_type):
"""
Remove the specified callback monitoring after the callback finishes.
If we remove a callback from the callback itself, Maestro crashes, but
the delay avoids the crash.
:param callback_type: The callback type to stop monitoring
:type callback_type: str
"""
func = self._monitor_funcs[callback_type]
timer_func = lambda: self._removeCallback(callback_type, func)
QtCore.QTimer.singleShot(0, timer_func)
def _trimWorkspaceChanged(self):
"""
If the workspace changed callback was called with "everything", remove
all other calls
"""
everything = maestro.WORKSPACE_CHANGED_EVERYTHING
if everything in self._calls.workspace_changed:
self._calls.workspace_changed = {everything}
if everything in self._calls.workspace_changed_active_proj:
self._calls.workspace_changed_active_proj = {everything}
def _determineEids(self):
"""
Determine which entry IDs were added and removed since the panel was
closed
"""
if self._eids_start is not None:
eids_stop = workspace_entries_observer.currentEntryIds()
self._calls.added_eids = eids_stop - self._eids_start
self._calls.removed_eids = self._eids_start - eids_stop
def _isActiveProj(self):
"""
Determine whether a Maestro project is available. (I.e. are we in the
middle of a project close callback?)
:return: True if it's possible to get the Maestro project. False
otherwise.
:rtype: bool
"""
try:
maestro.project_table_get()
except project.ProjectException:
return False
else:
return True
def _workspaceChanged(self, what_changed):
"""
A workspace changed callback that records the workspace_changed and
workspace_changed_active_proj calls. Stops recording if "everything"
has changed, since `_trimWorkspaceChanged` will remove all other calls
if "everything" is present.
:param what_changed: A description of what has changed. Provided by
the Maestro callback.
:type what_changed: str
"""
self._calls.workspace_changed.add(what_changed)
if self._isActiveProj():
self._calls.workspace_changed_active_proj.add(what_changed)
if (maestro.WORKSPACE_CHANGED_EVERYTHING
in self._calls.workspace_changed_active_proj):
self._removeCallbackDelayed(WORKSPACE_CHANGED_CALLBACK)
def _projectUpdated(self):
"""
A project update callback that records whether it has been called and
whether there is an active Maestro project. Stops recording if there is
an active Maestro project.
"""
self._calls.project_updated = True
if self._isActiveProj():
self._calls.project_changed = True
self._removeCallbackDelayed(PROJECT_UPDATE_CALLBACK)
def _projectClosed(self):
"""
A project closed callback that records whether it has been called.
Immediately stops recording once called.
"""
self._calls.project_close = True
self._removeCallbackDelayed(PROJECT_CLOSE_CALLBACK)
def _num_args(func):
"""
Determine the number of positional arguments for the specified function. If
the function accepts args, then infinity will be returned.
:param func: The function to check
:type func: function
:return: The number of positional arguments
:rtype: int or float
"""
argspec = inspect.getfullargspec(func)
if argspec.varargs:
return float("inf")
else:
return len(argspec.args)
[docs]class InclusionStateMixin(object):
"""
A mixin for AF2 panels that emits a single signal when an entry is included
or excluded from the workspace.
Only one signal is emitted even if multiple inclusions/exclusions are
happening - such as when including a new entry excludes the current entry.
The signal is emitted only when the PT state is stable - ie. checking if
entries are included or not works fine. This is accomplished by emitting the
signal in a different thread.
The signal does not contain information about what entries changed state or
what entries are included.
The signal obeys the same rules as the normal callbacks do - it is not
emitted if the panel is closed, but is emitted upon panel show if a state
changed while the panel is hidden
Inclusion/Exclusion actions generated by the inclusionStateChanged *slot*
will be ignored and will not generate a new signal.
"""
inclusionStateChanged = QtCore.pyqtSignal()
[docs] def __init__(self, *args, **kwargs):
"""
Create an InclusionStateMixin instance
"""
# Used to ignore multiple callbacks generated by the same action, such
# as both an inclusion and an exclusion
self.ignore_inclusion_callbacks = False
super().__init__(*args, **kwargs)
@entries_excluded
def _newWSExclusion(self, *args):
""" Start a new inclusion state when entries are excluded """
# A separate function is needed for this because the entries_included
# and entries_excluded decorators don't both operate when decorating the
# same function
self._newInclusionState()
@entries_included
def _newInclusionState(self, *args):
""" Start a new inclusion state """
if self.ignore_inclusion_callbacks:
# Don't do anything if multiple callbacks are generated from the
# same event
return
# Ignore all other inclusion callbacks until this thread completes
self.ignore_inclusion_callbacks = True
# Emit the signal in a new thread so that it is emitted after all
# inclusion/exclusion activities are complete
short = 1
QtCore.QTimer.singleShot(short, self.inclusionStateChanged.emit)
# Reset the ignore state after the signal is emitted and any attached
# slot is processed
QtCore.QTimer.singleShot(short * 2, self._resetIgnoreInclusionCallbacks)
def _resetIgnoreInclusionCallbacks(self):
""" Reset the ignore state so callbacks are paid attention to again """
self.ignore_inclusion_callbacks = False