"""
This module contains machinery for synchronizing models with various objects.
Various terms used in this module are defined here.
param: a data element of the type schrodinger.models.parameters.Param. Params
can be ints, bools, strings, etc., or more complex compound params that are
themselves composed of multiple params. There are two types of param references:
Abstract params and value params.
Abstract param: a param reference where the top level object is a class. For
example, MyModelClass.atom.coord would be an abstract param reference. As the
name suggests, the abstract param has no specific value, but is just a reference
to the kind of parameter.
Value param: a param reference where the top level object is an instance. For
example, my_model_object.atom.coord would a value param. The value param has a
distinct value.
model: an object with one or more params, each representing some data elements
of the model. The model can by synchronized to a target object via a mapper.
target: a target is any object that we want to keep in sync with a model param.
Targets are generally GUI widgets like spinboxes or line edits, but can be a
variety of other things, such as a specific signal we want a model param to
listen to, or a pair of setter/getter functions to sync to a model param's
value. A target could also be something like a command line argument, such that
each command line argument corresponds to a different param in a model.
access: a particular way of interacting with a target. A target can have one or
more accesses - a setter, a getter, or a signal.
default access: certain target types will have default accesses defined in this
module. The default accesses for QLineEdit, for example, are: QLineEdit.text as
the getter, QLineEdit.setText as the setter, and QLineEdit.textChanged as the
signal.
mapper: a manager object that is responsible for model/target synchronization.
mapping: a defined association between a target object and a model param. Note
that the mapping is always between a specific target instance (for example a
checkbox instance), and a model *class* param (ex. MyModel.myboolparam, where
MyModel is the class). By making the association with the model's class rather
than a model instance, the mapper is able to switch between different instances
of the same model. Consider, for example::
A model class Person, with params name and age
A GUI panel with panel.name_le and panel.age_sb
Mappings:
panel.name_le -> Person.name
panel.age_sb -> Person.age
Model instances amy, bob, and charlie
We can now user mapper.setModel to switch between model instances, and the GUI
state will change accordingly.
"""
import collections
import enum
import weakref
from schrodinger.infra import util
from schrodinger.models import parameters
from schrodinger.Qt import QtCore
from schrodinger.utils.scollections import IdDict
from schrodinger.utils.scollections import IdSet
DEFAULT = object()
NO_GETTER = object()
[docs]class WrappedObjDeletedError(RuntimeError):
"""
Custom exception for reporting "wrapped C/C++ object has been deleted"
"""
class _Connection:
def __init__(self, signal, slot):
self._signal = signal
self._slot = slot
self._connected = False
@property
def signal(self):
return self._signal
@signal.setter
def signal(self, signal):
connected = self._connected
if connected:
self.disconnect()
self._signal = signal
if connected:
self.connect()
@property
def slot(self):
return self._slot
@slot.setter
def slot(self, slot):
connected = self._connected
if connected:
self.disconnect()
self._slot = slot
if connected:
self.connect()
def connect(self):
if self._connected:
raise RuntimeError(f'{self._signal} is already connected to '
f'{self._slot}.')
if self.signal is None:
return
self._signal.connect(self._slot)
self._connected = True
def disconnect(self):
if not self._connected:
raise RuntimeError(f'{self._signal} is not connected to '
f'{self._slot}.')
if self.signal is None:
return
self._signal.disconnect(self._slot)
self._connected = False
[docs]class TargetSpec(QtCore.QObject):
"""
Describes a target that maps to a model param.
:ivar targetChanged: signal that gets emitted when a change in the target's
value is detected.
:vartype targetChanged: QtCore.pyqtSignal
"""
targetChanged = QtCore.pyqtSignal(object)
[docs] def __init__(self,
obj=None,
getter=DEFAULT,
setter=DEFAULT,
signal=DEFAULT,
datatype=DEFAULT,
slot=None,
auto_update_target=DEFAULT,
auto_update_model=DEFAULT):
"""
:param obj: this object, if specified, will be used to determine default
access for this target.
For example, passing in a QCheckBox, my_chk, will make the default
getter my_chk.isChecked, the default setter my_chk.setChecked, and the
default signal my_chk.stateChanged.
Passing in None will disable default access. In this case only
explicitly specified getters, setters, signals, and slots will be used.
:param getter: a function to get a value from the target. Overrides the
default getter in obj, if specified. Passing in None will result in the
target value always returning None.
:type getter: callable
:param setter: a function that sets the value on the target. Overrides
the default setter in obj, if specified. Passing in None will result in
the target value never being changed.
:type setter: callable
:param signal: the signal that indicates a change in target value. This
will override the default signal in obj, if specified. The target
signal is forwarded to targetChanged, to provide a common interface.
Pass in None to disable monitoring of target changes.
:type signal: QtCore.pyqtSignal
:param datatype: the type of data expected by the target object. If
set, this will be used to cast values being passed to the target
object via the setter. Ex. setting the datatype to str for a
QLineEdit allows the line edit to display IntParam data. Note: this
does not work the other way - mapping a QLineEdit to an IntParam
will cause the param to take on a string value.
:type datatype: type
:param slot: a function that will get called whenever the corresponding
model param is changed. Will get called regardless of whether a setter
or obj is specified. By default there is no slot set.
:type slot: callable
:param auto_update_target: whether the target should be automatically
(and immediately) updated when the param is changed. Default
behavior: if the obj has an attribute named auto_update_target, use
that, otherwise True.
:type auto_update_target: bool
:param auto_update_model: whether the model should be automatically
(and immediately) updated when the target is changed. Default
behavior: if the obj has an attribute named auto_update_model, use
that, otherwise True.
:type auto_update_model: bool
"""
super().__init__()
self._obj = obj
self._getter = self._getAccess(getter, _AccessType.getter)
self._setter = self._getAccess(setter, _AccessType.setter)
self._signal = self._getAccess(signal, _AccessType.signal)
self._datatype = self._getAccess(datatype, _AccessType.datatype)
self._slot = self._getAccess(slot, _AccessType.slot)
if self._signal is not None:
self._signal.connect(self.onTargetSignal)
self.auto_update_target = auto_update_target
self.auto_update_model = auto_update_model
@property
def auto_update_target(self):
"""
This property controls live updating of the target in response to model
value changes. This may be modified at any time. Set it to DEFAULT to
revert back to the original behavior.
"""
if self._auto_update_target is DEFAULT:
try:
return self._obj.auto_update_target
except RuntimeError as exc:
if "wrapped C/C++ object" in str(exc):
raise WrappedObjDeletedError
raise
except AttributeError:
return True
return self._auto_update_target
@auto_update_target.setter
def auto_update_target(self, value):
self._auto_update_target = value
@property
def auto_update_model(self):
"""
This property controls live updating of the model in response to target
value changes. This may be modified at any time. Set it to DEFAULT to
revert back to the original behavior.
"""
if self._auto_update_model is DEFAULT:
try:
return self._obj.auto_update_model
except AttributeError:
return True
return self._auto_update_model
@auto_update_model.setter
def auto_update_model(self, value):
self._auto_update_model = value
def _getAccess(self, access, access_type):
"""
Returns the way to access the target - either get its value, set its
value, or the signal to listen to that signifies a value change. This
takes in either a custom setter/getter/signal, DEFAULT, or None.
:param access_type: which type of access to return
:type access_type: _AccessType
"""
if access is None:
return
if access is DEFAULT:
return self._getDefaultAccess(self._obj, access_type)
return access
@staticmethod
def _getDefaultAccess(obj, access_type):
"""
Given an object, return the default setter/getter/signal for the object.
For example,
self._getDefaultAccess(my_line_edit, _AccessType.setter)
will return the function my_line_edit.setText, since that is the default
setter for a QLineEdit. Information about default access for each object
type is found in the DEFAULT_ACCESS_NAMES dictionary. Raises a
ValueError if the default access cannot be found.
:param obj: the target object
:param access_type: which sort of access we are looking for
:type access_type: _AccessType
"""
if obj is None:
return
if not isinstance(obj, TargetMixin):
# We only use the default access factory if the object doesn't
# subclass TargetMixin
DEFAULT_ACCESS_FACTORIES = _get_default_access_factories()
for obj_type, access_factory in DEFAULT_ACCESS_FACTORIES.items():
if isinstance(obj, obj_type):
access = access_factory(obj, access_type)
if access is DEFAULT:
break
else:
return access
DEFAULT_ACCESS_NAMES = _get_default_access_names()
for obj_type, info in DEFAULT_ACCESS_NAMES.items():
if isinstance(obj, obj_type):
break
else:
if access_type in (_AccessType.datatype, _AccessType.slot):
return None
err_msg = (
'A default getter and setter have not been defined for '
f'{type(obj).__name__}. Use TargetSpec to define a getter, '
'setter, and signal.')
raise ValueError(err_msg)
if access_type == _AccessType.datatype:
return info[access_type]
try:
name = info[access_type]
except IndexError:
name = None
if not name:
if access_type in (_AccessType.signal, _AccessType.datatype,
_AccessType.slot):
return
raise ValueError('No default %s for %s of type %s' %
(access_type.name, obj, obj_type))
return getattr(obj, name)
[docs] @QtCore.pyqtSlot()
def onTargetSignal(self):
"""
We connect this slot to the target's specific signal and emit the
generic targetChanged signal with the new value. This provides a uniform
interface for the mapper to connect to.
"""
value = self.getValue()
self.targetChanged.emit(value)
[docs] def onModelParamChanged(self, value):
if value == self.getValue():
return
self.setValue(value)
[docs] def slot(self):
if self._slot is not None:
self._slot()
[docs] def getValue(self):
"""
The standard method for getting a target's value, regardless of whether
this is using a default getter or a custom one.
"""
if self._getter is None:
return NO_GETTER
return self._getter()
[docs] def setValue(self, value):
"""
The standard method for setting a target's value, regardless of whether
this is using a default setter or a custom one.
"""
if self._setter is None:
return
if self._datatype is not None:
try:
value = self._datatype(value)
except TypeError:
err_msg = (f"Can't cast type {type(value).__name__} into "
f"{self._datatype.__name__}. Check that param "
"is compatible with target.")
raise TypeError(err_msg)
return self._setter(value)
def __repr__(self):
details = []
if self._obj is not None:
details.append('obj=' + repr(self._obj))
if self._getter is not None:
getter_name = getattr(self._getter, '__name__', str(self._getter))
details.append('getter=' + getter_name)
if self._setter is not None:
setter_name = getattr(self._setter, '__name__', str(self._setter))
details.append('setter=' + setter_name)
return '<TargetSpec: %s>' % ', '.join(details)
[docs]class ParamTargetSpec(TargetSpec):
"""
Class to allow a param to be synchronized to another param. Example:
target = ParamTargetSpec(target_model, MyModelClass.param)
This creates a target for synchronizing target_model.param, where
target_model is an instance of MyModelClass.
"""
[docs] def __init__(self, model, param):
"""
:param model: the model which contains the param to be mapped
:type model: parameters.CompoundParam
:param param: the abstract param to be mapped (i.e. MyModelClass.param)
:type param: parameters.Param
"""
super().__init__(obj=model,
getter=None,
setter=None,
datatype=None,
signal=None,
slot=None)
self.param = param
self.model = model
self._param_signals = []
self._connectParamSignals()
@QtCore.pyqtSlot()
def _connectParamSignals(self):
for signal in self._param_signals:
signal.disconnect(self.onTargetSignal)
try:
signal.disconnect(self._connectParamSignals)
except TypeError:
# valueChanged signal isn't connected to this slot.
pass
self._param_signals.clear()
model = self.model
# Get all the replaced signals from the owners and connect them
# to `onTargetSignal` to update the model and to `_connectParamSignals`
# to refresh the replaced signals.
for abs_param in self.param.ownerChain()[1:]:
if abs_param is not self.param:
replace_signal = abs_param.getParamSignal(
model, parameters.SignalType.Replaced)
replace_signal.connect(self.onTargetSignal)
replace_signal.connect(self._connectParamSignals)
valueChanged_signal = self.param.getParamSignal(model)
valueChanged_signal.connect(self.onTargetSignal)
self._param_signals.append(valueChanged_signal)
[docs] def getValue(self):
return self.param.getParamValue(self._obj)
[docs] def setValue(self, value):
self.param.setParamValue(self._obj, value)
[docs]class AttrTargetSpec(TargetSpec):
"""
Allows an attribute on any object to be synchronized to a param. Example:
target = AttrTargetSpec(my_obj, 'x_data')
This creates a target for synchronizing my_obj.x_data.
Note that attributes by default don't have a signal, so auto-updating of
the model param won't work unless the optional signal argument is supplied.
"""
[docs] def __init__(self, obj, name, signal=None):
"""
:param obj: the object that has the attribute to be mapped
:param name: the name of the target attribute on the object
:type name: str
:param signal: a Qt signal that indicates a change in the attribute's
value.
:type signal: QtCore.pyqtSignal
"""
self.name = name
super().__init__(obj=obj,
getter=None,
setter=None,
datatype=None,
signal=signal)
[docs] def getValue(self):
return getattr(self._obj, self.name)
[docs] def setValue(self, value):
setattr(self._obj, self.name, value)
[docs]class TargetMixin(object):
"""
Use this mixin to enable get default Target behavior from a custom object
the way it works for standard widgets like QCheckBox and QLineEdit. It is
up to the subclass to implement targetGetValue and targetSetValue as well as
to emit the targetValueChanged signal with the new value at the appropriate
time.
After subclassing, the new custom object can be passed in as the obj
argument to the Target constructor.
Using this mixin requires that the class also inherits from QObject.
The variables auto_update_target and auto_update_model can be set on the
instance at any time to turn on or off live-updating of the target/model
values.
"""
targetValueChanged = QtCore.pyqtSignal()
auto_update_target = True
auto_update_model = True
[docs] def targetGetValue(self):
pass
[docs] def targetSetValue(self, value):
pass
class _AccessType(enum.IntEnum):
"""
The different types of target access.
"""
getter = 0
setter = 1
signal = 2
datatype = 3
slot = 4
def _qbuttongroup_access_factory(obj, access_type):
if access_type == _AccessType.setter:
def setter(button_id):
obj.button(button_id).setChecked(True)
return setter
return DEFAULT
[docs]class TargetParamMapper(QtCore.QObject):
"""
A param mapper manages synchronization between target objects that represent
various params and a model object that contains those params.
:ivar setting_model: Context manager to set a flag indicating that the
model is being set. Intended for use by MapperMixin.
"""
TARGET_CLASS = TargetSpec
updating_values = util.flag_context_manager('_currently_updating', True)
setting_model = util.flag_context_manager('_setting_model', True)
_owner = util.WeakRefAttribute()
[docs] def __init__(self,
parent=None,
auto_update_target=True,
auto_update_model=True,
*,
_display_ok=True):
"""
:param auto_update_target: whether to update the target immediately when
the model is changed
:type auto_update_target: bool
:param auto_update_model: whether to update the model immediately when
the target is changed
:type auto_update_model: bool
:param _display_ok: whether it is safe to import modules that require
display (e.g. QtWidgets).
:type _display_ok: bool
"""
super().__init__(parent)
self._display_ok = _display_ok
self.model = None
self._currently_updating = False
self._target_slot_dict = IdDict()
self._param_slot_dict = IdDict()
self._saved_connections = []
self._signals_and_slots_callbacks = []
self.addGetSignalsAndSlotsCallback(self._getMapperSignalsAndSlots)
self.param_map = IdDict()
self.auto_update_model = auto_update_model
self.auto_update_target = auto_update_target
self._raw_targets = IdDict()
self._owner = None
self._setting_model = False
@property
def auto_update_target(self):
"""
This property controls live updating of the target in response to model
value changes. This may be modified at any time.
"""
return self._auto_update_target
@auto_update_target.setter
def auto_update_target(self, value):
self._auto_update_target = value
if value:
self.updateTarget()
@property
def auto_update_model(self):
"""
This property controls live updating of the model in response to target
value changes. This may be modified at any time.
"""
return self._auto_update_model
@auto_update_model.setter
def auto_update_model(self, value):
self._auto_update_model = value
if value:
self.updateModel()
def _addSingleMapping(self, target, param):
"""
Adds a mapping between a single target and param. Also makes sure the
mapping hasn't already been made. If so, it raises a ValueError.
:param target: the target object
:type target: self.TARGET_CLASS
:param param: an abstract param
:type param: parameters.Param
"""
mapped_params = self.param_map.setdefault(target, IdSet())
if param in mapped_params:
raise ValueError('%s is already mapped to %s' % (target, param))
else:
mapped_params.add(param)
def _getMappingList(self):
mapping_list = []
for target, params in self.param_map.items():
for param in params:
mapping_list.append((target, param))
return mapping_list
[docs] def mappedParams(self):
"""
Return a list of the abstract params that are mapped to.
"""
mapped_params = []
for params in self.param_map.values():
mapped_params.extend(params)
return mapped_params
[docs] def addMapping(self, target, param):
"""
Maps a target (or collection of targets) to an abstract param (or
collection of abstract params). An abstract param is a param that is
owned at the top level by the model's class rather than an instance of
the model. This allows the same mapping to be used on multiple model
instances.
The details of the target object are left to derived mapper classes.
Notes:
A target may be mapped to multiple params, and multiple targets
may be mapped to the same param. This is useful when the same
param is associated with multiple targets (e.g. multiple views on a
single data model) or vice versa (e.g. a single LineEdit sets the
value of multiple fields in the model).
If the target is not an instance of self.TARGET_CLASS already, it
will be automatically wrapped (i.e. self.TARGET_CLASS(target)). This
allows common targets such as Qt widgets to be passed in directly.
:param param: an abstract param (ex. Atom.coord.x) or collection of
abstract params
:type param: `parameters.Param` or tuple
:param target: the target or collection of targets mapped to a
parameter.
:type target: self.TARGET_CLASS or object that can be wrapped via
self.TARGET_CLASS(target) or tuple
"""
if isinstance(param, tuple):
params = param
else:
params = (param,)
if isinstance(target, tuple):
targets = target
else:
targets = (target,)
for target in targets:
if callable(target):
target = self._raw_targets.setdefault(
target, self.TARGET_CLASS(slot=target))
elif not isinstance(target, self.TARGET_CLASS):
target = self._raw_targets.setdefault(target,
self.TARGET_CLASS(target))
for param in params:
self._addSingleMapping(target, param)
[docs] def getSignalsAndSlots(self, model):
"""
Given a model object, return all signals and slots that need to be
connected to support auto updating. Override this method in subclasses.
Note that the returned slots will only be called when the specified
signal is emitted, and not when the model is changed using `setModel()`.
:return: a list of 2-tuples where each tuple is a signal, slot pair
"""
return []
def _connectModel(self, model):
"""
Make any signal connections between model and target needed for auto
updating.
"""
if model is None:
return
self._saved_connections = []
for get_signals_and_slots in self._signals_and_slots_callbacks:
for signal, slot in get_signals_and_slots(model):
self.connectSignalAndSlot(signal, slot)
# Connecting the replace slots last to avoid disconnecting replace
# slots connected in `getSignalsAndSlots`
for signal, slot in self._getReplaceSignalsAndSlots(model):
self.connectSignalAndSlot(signal, slot)
for signal, slot in self._getQObjectChangedSignalsAndSlots(model):
self.connectSignalAndSlot(signal, slot)
def _disconnectModel(self, model):
"""
Disconnect any signal connections between model and target made in
self._connectModel. This is needed when switching models via setModel so
that the mapper will not continue syncing the old model object.
"""
if model is None:
return
for connection in self._saved_connections:
connection.disconnect()
self._saved_connections = []
[docs] def setModel(self, model):
"""
Sets the model instance to map. This should be an instance of the model
class that is being used in addMapping().
:param model: the model instance
:type model: object
"""
prev_model = self.model
if prev_model is not None:
self._disconnectModel(prev_model)
self.model = model
if model is None:
for target in self.param_map:
if isinstance(target._obj, MapperMixin):
target._obj.setModel(None)
self._connectModel(model)
if self.auto_update_target:
self.updateTarget()
def _getTargetValue(self, target_obj):
return target_obj.getValue()
def _setTargetValue(self, target_obj, value):
target_obj.setValue(value)
def _callTargetSlot(self, target_obj):
# Don't call slot while MapperMixin is setting the model
if not self._setting_model:
target_obj.slot()
def _getModelValue(self, param):
return param.getParamValue(self.model)
def _setModelValue(self, param, value):
if value is NO_GETTER:
return
param.setParamValue(self.model, value)
def _updateModelParam(self, target_obj):
"""
Updates the param value on the model object from the target object. If
the new value is the same as the current value, the param will not be
set again.
:param param: the abstract param that defines the mapping
:type param: parameters.Param
"""
params = self.param_map[target_obj]
for param in params:
value = self._getTargetValue(target_obj)
old_value = self._getModelValue(param)
if value != old_value:
self._setModelValue(param, value)
def _updateTargetValue(self, target_obj):
"""
Updates the mapped object from the param value on the model object. If
the new value is the same as the target's current value, the value will
not be set again on the target object.
:param param: the abstract param that defines the mapping
:type param: parameters.Param
"""
params = self.param_map[target_obj]
for param in params:
value = self._getModelValue(param)
old_value = self._getTargetValue(target_obj)
if isinstance(value, parameters.CompoundParam):
# Use a stricter comparison for CompoundParams since a
# target CompoundParam should be identical to the CompoundParam
# it's mapped to.
is_same_value = value is old_value
else:
is_same_value = value == old_value
if not is_same_value:
self._setTargetValue(target_obj, value)
self._callTargetSlot(target_obj)
[docs] def resetMappedParams(self):
self.model.reset(*self.mappedParams())
[docs] def updateModel(self):
"""
Updates all mapped parameters on the model object from the target
objects. Any target values that are unchanged will be skipped.
"""
if self.model is None:
return
with self.updating_values():
for target_obj in self.param_map:
self._updateModelParam(target_obj)
[docs] def updateTarget(self):
"""
Updates all target objects from the mapped parameters on the model
object. Any param values that are unchanged will be skipped.
"""
if self.model is None:
return
with self.updating_values():
for target_obj in self.param_map:
self._updateTargetValue(target_obj)
[docs] def addGetSignalsAndSlotsCallback(self, callback):
"""
Adds a "getSignalsAndSlots" function that will be called whenever a
new model is set. See MapperMixin.getSignalsAndSlots for
information on parameters and return value for the callback.
"""
self._signals_and_slots_callbacks.append(callback)
def _formatMapperInfoMsg(self):
msg = f'\n\tMapper: {self}'
if self._owner is not None:
msg += f'\n\tMapper owner: {self._owner}'
return msg
[docs] def connectSignalAndSlot(self, signal, slot):
"""
Connects a signal/slot pair which will automatically be disconnected
when the model is changed. The connection is discarded once disconnected
and will not be reconnected when a new model is set.
"""
if not isinstance(signal, QtCore.pyqtBoundSignal):
msg = (f'{repr(signal)} was specified as a signal in '
'getSignalsAndSlots, but it is not a signal.')
msg += self._formatMapperInfoMsg()
raise TypeError(msg)
if not callable(slot):
msg = (f'{repr(slot)} was specified as a slot in '
'getSignalsAndSlots, but it is not callable.')
msg += self._formatMapperInfoMsg()
raise TypeError(msg)
connection = _Connection(signal, slot)
connection.connect()
self._saved_connections.append(connection)
[docs] def getTargetSlot(self, target):
"""
Gets the target-specific slot function for responding to param change.
If no slot exists for this target, a new one is created.
"""
# Closure slots with references to self or target cause problems with
# garbage collection. To avoid this, we replace them with weakrefs.
self = weakref.proxy(self)
target_proxy = weakref.proxy(target)
@QtCore.pyqtSlot(object)
def new_slot(value):
# do not use target inside the slot. Use target_proxy instead
try:
self.auto_update_target
target_proxy.auto_update_target
except (ReferenceError, WrappedObjDeletedError):
return
if (self.auto_update_target and target_proxy.auto_update_target and
not self._currently_updating):
try:
target_proxy.onModelParamChanged(value)
except Exception as e:
msg = self._formatTargetSlotErrorMsg(target_proxy, value, e)
raise RuntimeError(msg)
target_proxy.slot()
slot = self._target_slot_dict.get(target)
if slot is None:
self._target_slot_dict[target] = new_slot
slot = new_slot
return slot
def _formatTargetSlotErrorMsg(self, target_proxy, value, error):
mapping_list = self._getMappingList()
for mapped_target, mapped_param in mapping_list:
# must use equality check for weakref.proxy
if mapped_target == target_proxy:
param = mapped_param
break
else:
# Should never get here
msg = 'Target %s not found in mappings: %s' % (repr(target_proxy),
mapping_list)
msg += self._formatMapperInfoMsg()
msg += f'\nOriginal error:\n{type(error)}{error}\n'
return msg
msg = f'Error setting target {repr(target_proxy)}'
msg += self._formatMapperInfoMsg()
dtype = 'Not specified' if target_proxy._datatype is None else str(
target_proxy._datatype)
msg += (f'\n\tNew value: {value}\n'
f'\tParam: {repr(param)}\n'
f'\tTarget DataType: {dtype}\n'
f'\tError: {str(error)}')
return msg
[docs] def getParamSlot(self, param):
"""
Gets the param-specific slot function for responding to target change.
If no slot exists for this param, a new one is created.
"""
# Closure slots with references to self or target cause problems with
# garbage collection. To avoid this, we replace them with weakrefs.
self = weakref.proxy(self)
param = weakref.proxy(param)
@QtCore.pyqtSlot()
def new_slot(value):
target = self.sender()
if (self.auto_update_model and target.auto_update_model and
not self._currently_updating):
old_value = self._getModelValue(param)
if old_value != value:
try:
self._setModelValue(param, value)
except Exception as e:
msg = self._formatParamSlotErrorMsg(
param, value, target, e)
raise RuntimeError(msg)
slot = self._param_slot_dict.get(param)
if slot is None:
self._param_slot_dict[param] = new_slot
slot = new_slot
return slot
def _formatParamSlotErrorMsg(self, param, value, target, error):
msg = f'Error setting mapped param {repr(param)} on {repr(self.model)}.'
msg += self._formatMapperInfoMsg()
msg += (f'\n\tNew value: {value}\n\tTarget: {repr(target)}\n\tError: '
f'{str(error)}')
return msg
def _getMapperSignalsAndSlots(self, model):
# see parent class for documentation
ss = [] # list of signals and slots to connect/disconnect
if self._display_ok:
from schrodinger.ui.qt.mapperwidgets import plptable
else:
plptable = None
for target, params in self.param_map.items():
if plptable is not None and isinstance(target,
plptable.PLPTableWidget):
# PLPTableWidgets have different signals to handle updating
# FIXME: Can we fix this with auto_update_target?
continue
for param in params:
signal = param.getParamSignal(model)
slot = self.getTargetSlot(target)
ss.append((signal, slot))
signal = target.targetChanged
slot = self.getParamSlot(param)
ss.append((signal, slot))
return ss
def _getQObjectChangedSignalsAndSlots(self, model):
# Currently we just re-set the model whenever any concrete atomic param
# with signals is changed. May need to revisit this in the future if it
# causes perforamance problems.
ss = []
abstract_model = type(model)
for abs_param in parameters.get_all_atomic_subparams(abstract_model):
dc = abs_param.DataClass
if issubclass(dc, QtCore.QObject):
signal = abs_param.getParamSignal(model)
ss.append((signal, self._reSetModel))
return ss
def _getReplaceSignalsAndSlots(self, model):
# Currently we just re-set the model whenever any one param is replaced.
# May need to revisit this in the future if it causes perforamance
# problems. PANEL-12690
ss = []
replacement_signals = parameters.get_all_replaced_signals(model)
for signal in replacement_signals:
ss.append((signal, self._reSetModel))
return ss
def _reSetModel(self):
"""
Re-set the model, i.e. call `self.setModel(self.model)`. This is useful
for making sure the correct signals and slots are hooked up.
"""
self.setModel(self.model)
[docs]class MapperMixin:
"""
Mixin that can facilitate the use of parameters and mappers for storing the
state of its subclasses.
Works out of the box for widgetmixins.InitMixin or af2.baseapp.BasePanel
(which covers af2.App and af2.JobApp). To use with other base classes, call
`_setupMapperMixin()` during initialization.
By default, the mixin will attempt to create an empty model instance at
construction and set it as the model. If the model class' constructor
requires arguments, the model will be set to None instead. In this case a
model instance must be constructed and explicitly set using setModel before
the MapperMixin can be used.
:ivar mapper: an `AbstractParamMapper` instance that can be used to
keep track of data members of this mixin's subclasses.
:cvar model_class: to be defined in subclasses. The model class that stores
information about the subclass of this mixin (which can be though of as
a "view").
"""
model_class = None
_af2_setDefaults_called = False
_setupMapperMixin_called = False
setting_model = util.flag_context_manager('_setting_model', True)
[docs] def __init__(self, *args, **kwargs):
self._setting_model = False
self._model = None
super().__init__(*args, **kwargs)
@property
def model(self):
if self._setting_model:
msg = f"Cannot access {self}.model during setModel"
raise CantAccessModelError(msg)
return self._model
@model.setter
def model(self, value):
if self._setting_model:
msg = f"Cannot access {self}.model during setModel"
raise CantAccessModelError(msg)
self._model = value
[docs] def initLayOut(self):
"""
@overrides: widgetmixins.InitMixin
"""
super().initLayOut()
self._setupMapperMixin()
[docs] def initSetDefaults(self):
"""
@overrides: widgetmixins.InitMixin
"""
super().initSetDefaults()
self.__resetModel()
[docs] def setDefaults(self):
"""
@overrides: af2.App
"""
if not self._af2_setDefaults_called:
self._setupMapperMixin()
self._af2_setDefaults_called = True
super().setDefaults()
self.__resetModel()
[docs] def makeInitialModel(self):
return self.model_class()
def _setupMapperMixin(self):
"""
Performs inital setup of the object's state. This will be automatically
called at the right time if the mixin is used with an af2.App or
basewidgets.BaseWidget. Otherwise, it can be manually called during
initialization of the object.
"""
if self._setupMapperMixin_called:
raise RuntimeError('_setupMapperMixin() was called twice. Check '
'inheritance structure.')
self._setupMapperMixin_called = True
self.mapper = None
self._buildMapper()
self._clean_state = None
if self.model_class is None:
return
try:
model = self.makeInitialModel()
except Exception:
print("Error while constructing initial model. If the model's"
'constructor has required arguments, override '
'makeInitialModel to specify them.')
raise
self.setModel(model)
def __resetModel(self):
if self.model is not None:
self.model.reset()
[docs] def getSignalsAndSlots(self, model):
"""
Override this method to specify signal and slot pairs that need to be
connected/disconnected whenever the model instance is switched using
setModel. The model instance is provided as an argument so that
instance-specific signals can be used, but any pairs of signals and
slots may be returned from this method.
:return: a list of 2-tuples where each tuple is a signal, slot pair
"""
return []
def _getSignalsAndSlots(self, model):
"""
This private method is used to get signals/slots needed for
TargetMixin to function. This could have been done in the public
method setSignalsAndSlots, but that would've required every user of the
mixin to remember to call super().getSignalsAndSlots and append their
own signals/slots to the return value.
"""
ss = self.getSignalsAndSlots(model)
self._validateSignalsAndSlots(ss)
return ss
def _validateSignalsAndSlots(self, ss):
"""
Validates that the values used in getSignalsAndSlots are formatted
correctly. Checks that list contains 2-tuples
"""
# Check for list of 2-tuples
for idx, ss_tuple in enumerate(ss, 1):
if not (isinstance(ss_tuple, tuple) and len(ss_tuple) == 2):
raise SignalsAndSlotsException(
f"getSignalsAndSlots must return a list of (signal, slot) 2-tuples. Item {idx} is a {ss_tuple}."
)
[docs] def getModel(self):
return self.model
def _buildMapper(self):
mappings = self.defineMappings()
self._validateMappings(mappings)
if isinstance(self, QtCore.QObject):
parent = self
else:
parent = None
self.mapper = make_mapper(mappings, parent=parent)
self.mapper._owner = self
self.mapper.addGetSignalsAndSlotsCallback(self._getSignalsAndSlots)
[docs] @QtCore.pyqtSlot()
def resetMappedParams(self):
self.mapper.resetMappedParams()
[docs] def mappedParams(self):
"""
Return a list of the abstract params that are mapped to.
"""
return self.mapper.mappedParams()
[docs] def defineMappings(self):
"""
Override this in the subclass to define mappings. Should return a list
of tuples [(<target>, <param>)]. Targets can be:
1. a basic widget, like `QLineEdit` or `QComboBox`
2. a custom object that inherits `MapperMixin` or `TargetMixin`
3. a `TargetSpec` instance
4. a slot
For common widgets, standard signals and getter/setter methods will be
used, as defined in `mappers._get_default_access_names()`.
For more fine-grained custom control, instantiate a `TargetSpec` object,
which allows custom setters, getters, and signals to be specified.
Supplying a slot as the first element of the tuple is equivalent to
providing `TargetSpec(slot=my_slot)`.
Note that all target slots are triggered on `setModel()` as well as in
response to the specified signal.
The param is an abstract param reference, e.g. MyModel.my_param.
Example::
def defineMappings(self):
combo = self.style_combo
return [(self.name_le, MyModel.name),
(TargetSpec(combo,
getter=combo.currentText,
setter=combo.setCurrentText), MyModel.style),
(self.coord_widget, MyModel.coord),
(self._onASLTextChanged, MyModel.asl_text)]
"""
return []
def _validateMappings(self, mappings):
"""
Validates that the mappings of targets to their params are constructed correctly.
This should be a list of 2-tuples in the form [(<target>, <param>), ...]
Optionally, this can also be a dict of {target: param}
:param mappings: mapping of targets to params
:type: list | dict
"""
if isinstance(mappings, dict):
return
for idx, mappings_tuple in enumerate(mappings, 1):
self._validateMapping(idx, mappings_tuple)
def _validateMapping(self, idx, mapping):
if not isinstance(mapping, tuple):
raise DefineMappingsException(
f'defineMappings must return a list of tuples. Mapping {idx} '
f'is {mapping}.')
if not len(mapping) == 2:
raise DefineMappingsException(
f'defineMappings must return a list of (<target>, '
f'<abstract param>) 2-tuples. Mapping {idx}, {mapping}, is a '
f'{len(mapping)}-tuple.')
target, param = mapping
if (isinstance(target, tuple) or isinstance(param, tuple)):
if isinstance(target, tuple):
for t in target:
self._validateMapping(idx, (t, param))
if isinstance(param, tuple):
for p in param:
self._validateMapping(idx, (target, p))
return
if not (isinstance(param, parameters.Param) or
(isinstance(param, type) and
issubclass(param, parameters.Param))):
raise DefineMappingsException(
f'Second element in a defineMappings tuple must be an abstract '
f'Param. In mapping {idx}, {target} is mapped to a '
f'{type(param)}.')
if not param.isAbstract():
raise DefineMappingsException(
f'Second element in a defineMappings tuple must be an abstract '
f'Param. In item {idx}, {target} is mapped to a concrete '
f'param, {param}.')
[docs] @QtCore.pyqtSlot(object)
def setModel(self, model):
"""
Sets the model object for the mapper. Disconnects the old model, if one
is set, and connects the new model. Pass in None to have no model set.
:param model: the model instance or None
"""
with self.setting_model(), self.mapper.setting_model():
self.mapper.setModel(None)
if model is None:
self._model = None
return
self._setupModelClass(model)
# _setupModelClass may create a new mapper, so start a new context
with self.setting_model(), self.mapper.setting_model():
if model is not None and not isinstance(model, self.model_class):
raise TypeError(f'Model must be of type {self.model_class}. '
f'Got {model}.')
self.mapper.setModel(model)
self._model = model
if self.mapper.auto_update_target:
# TargetParamMapper won't call slots while MapperMixin is setting
# model, so call them explicitly
self.runAllSlots()
[docs] @QtCore.pyqtSlot(object)
def setModelWithoutSlots(self, model):
"""
This is called when this MapperMixin is a sub-widget of a parent
MapperMixin. Since the slots will all be called at the end of the parent
setModel, they shouldn't be called during the sub-widget's setModel.
"""
orig_auto_update = self.mapper.auto_update_target
try:
self.mapper.auto_update_target = False
self.setModel(model)
finally:
self.mapper.auto_update_target = orig_auto_update
[docs] @QtCore.pyqtSlot()
def runAllSlots(self):
for target in self.mapper.param_map:
target.slot()
def _setupModelClass(self, model):
pass
# Map an object's type to a 4-tuple of attribute names for the default getter,
# setter, signal, and datatype (indexes correspond to `_AccessType`)
def _get_default_access_names():
from schrodinger.Qt import QtWidgets
from schrodinger.Qt import QtGui
QW = QtWidgets
return collections.OrderedDict([
(TargetMixin, ('targetGetValue', 'targetSetValue', 'targetValueChanged', None)),
(MapperMixin, ('getModel', 'setModelWithoutSlots', None, None, 'runAllSlots')),
(QW.QLineEdit, ('text', 'setText', 'textChanged', str)),
(QW.QTextEdit, ('toPlainText', 'setText', 'textChanged', str)),
(QW.QCheckBox, ('isChecked', 'setChecked', 'stateChanged', bool)),
(QW.QAbstractButton, ('isChecked', 'setChecked', 'toggled', bool)),
(QtGui.QAction, ('isChecked', 'setChecked', 'toggled', bool)),
(QW.QGroupBox, ('isChecked', 'setChecked', 'toggled', int)),
(QW.QSpinBox, ('value', 'setValue', 'valueChanged', int)),
(QW.QDoubleSpinBox, ('value', 'setValue', 'valueChanged', float)),
(QW.QComboBox, ('currentIndex', 'setCurrentIndex', 'currentIndexChanged', int)),
(QW.QSlider, ('value', 'setValue', 'valueChanged', int)),
(QW.QLabel, ('text', 'setText', None, str)),
# See _qbuttongroup_access_factory() for setter
(QW.QButtonGroup, ('checkedId', None, 'buttonToggled', int))
]) # yapf: disable
# For QObjects that do not have acceptable pre-defined getter or setter methods,
# map them to a factory function that will define and return either a custom
# getter/setter method, or DEFAULT which instructs the mapper to fall back on
# the attributes defined in DEFAULT_ACCESS_NAMES.
# Each default access factory must accept as its arguments 1. a QObject instance
# and 2. the desired _AccessType enum, then return the appropriate function, or
# else DEFAULT.
def _get_default_access_factories():
from schrodinger.Qt import QtWidgets
QW = QtWidgets
return {
QW.QButtonGroup: _qbuttongroup_access_factory,
} # yapf: disable
[docs]class DefineMappingsException(ValueError):
"""
Exception to raise for improperly formatted DefineMappings in mappers
"""
pass
[docs]class SignalsAndSlotsException(ValueError):
"""
Exception to raise for improperly formatted SignalsAndSlots
"""
pass
[docs]class CantAccessModelError(RuntimeError):
"""
Exception to raise when accessing self.model during MapperMixin.setModel
"""
[docs]def make_mapper(mappings, model=None, mapper_class=None, parent=None):
"""
Convenience function for adding many mappings at once via a dictionary.
:param mappings: a list of (target, abstract param) tuples. The target may
be an actual Target object or an object that can be wrapped by Target.
:type mappings: list
:param model: an optional parameter for setting a specific model object to
this mapper. Doing so will also set this mapper as the model's primary
mapper, if possible
:type model: object
:param mapper_class: an optional parameter to use if the mapper is not meant
to be a SettingsParamMapper.
:type mapper_class: type
"""
if mapper_class is None:
mapper_class = TargetParamMapper
mapper = mapper_class(parent=parent)
if isinstance(mappings, dict):
mappings = [(target, param) for target, param in mappings.items()]
for target_obj, param in mappings:
mapper.addMapping(target_obj, param)
mapper.setModel(model)
return mapper