Source code for ccpn.core.ChemicalShift

"""
Module Documentation here
"""
#=========================================================================================
# Licence, Reference and Credits
#=========================================================================================
__copyright__ = "Copyright (C) CCPN project (https://www.ccpn.ac.uk) 2014 - 2022"
__credits__ = ("Ed Brooksbank, Joanna Fox, Victoria A Higman, Luca Mureddu, Eliza Płoskoń",
               "Timothy J Ragan, Brian O Smith, Gary S Thompson & Geerten W Vuister")
__licence__ = ("CCPN licence. See https://ccpn.ac.uk/software/licensing/")
__reference__ = ("Skinner, S.P., Fogh, R.H., Boucher, W., Ragan, T.J., Mureddu, L.G., & Vuister, G.W.",
                 "CcpNmr AnalysisAssign: a flexible platform for integrated NMR analysis",
                 "J.Biomol.Nmr (2016), 66, 111-124, http://doi.org/10.1007/s10858-016-0060-y")
#=========================================================================================
# Last code modification
#=========================================================================================
__modifiedBy__ = "$modifiedBy: Ed Brooksbank $"
__dateModified__ = "$dateModified: 2022-02-23 17:20:31 +0000 (Wed, February 23, 2022) $"
__version__ = "$Revision: 3.1.0 $"
#=========================================================================================
# Created
#=========================================================================================
__author__ = "$Author: Ed Brooksbank $"
__date__ = "$Date: 2021-08-02 10:55:54 +0100 (Mon, August 02, 2021) $"
#=========================================================================================
# Start of code
#=========================================================================================

from functools import partial
from typing import Optional, Union
from collections import namedtuple
import pandas as pd

from ccpn.core import _importOrder
from ccpn.core.Project import Project
from ccpn.core.NmrAtom import NmrAtom
from ccpn.core.ChemicalShiftList import ChemicalShiftList
from ccpn.core.lib.ContextManagers import ccpNmrV3CoreSetter, undoStackBlocking, deleteV3Object, ccpNmrV3CoreUndoBlock
from ccpn.core.lib.Pid import Pid, createId
from ccpn.core.ChemicalShiftList import CS_UNIQUEID, CS_ISDELETED, CS_STATIC, CS_VALUE, CS_VALUEERROR, \
    CS_FIGUREOFMERIT, CS_NMRATOM, CS_CHAINCODE, CS_SEQUENCECODE, CS_RESIDUETYPE, CS_ATOMNAME, \
    CS_COMMENT, CS_COLUMNS, ChemicalShiftState
from ccpn.core._implementation.V3CoreObjectABC import V3CoreObjectABC
from ccpn.util.Common import makeIterableList
from ccpn.util.decorators import logCommand


MINFOM = 0.0
MAXFOM = 1.0

ShiftParameters = namedtuple('ShiftParameters', f'{CS_UNIQUEID} {CS_ISDELETED} {CS_STATIC} '
                                                f'{CS_VALUE} {CS_VALUEERROR} {CS_FIGUREOFMERIT} '
                                                f'{CS_NMRATOM} {CS_CHAINCODE} {CS_SEQUENCECODE} {CS_RESIDUETYPE} {CS_ATOMNAME} '
                                                f'{CS_COMMENT} ')


[docs]class ChemicalShift(V3CoreObjectABC): """Chemical Shift, containing a ChemicalShift value for the NmrAtom they belong to. Chemical shift values are continuously averaged over peaks assigned to the NmrAtom, (unless this behaviour is turned off) ChemicalShift objects are sorted by uniqueId. """ #: Short class name, for PID. shortClassName = 'CS' # Attribute - necessary as subclasses must use superclass className className = 'ChemicalShift' _oldClassName = '_OldChemicalShift' _parentClass = ChemicalShiftList #: Name of plural link to instances of class _pluralLinkName = 'chemicalShifts' _childClasses = [] _isGuiClass = False # the attribute name used by current _currentAttributeName = 'chemicalShifts' def __init__(self, project, chemicalShiftList, _uniqueId): """Create a new instance of v3 Shift _unique Id links the shift to the dataFrame storage and MUST be specified before the shift can be used """ super().__init__(project, chemicalShiftList, _uniqueId) self._oldValue = self._oldValueError = None # All other properties are derived from the chemicalShiftList pandas dataframe #========================================================================================= # CCPN Properties #========================================================================================= @property def id(self) -> str: """Identifier for the object, used to generate the pid and longPid. Generated by combining the id of the containing object, i.e. the checmialShift instance, with the value of one or more key attributes that uniquely identify the object in context E.g. 'default.1' """ return self._deletedId if self._isDeleted else createId(self._wrapperList.name, str(self._uniqueId)) @property def chemicalShiftList(self): """ChemicalShiftList containing ChemicalShift. """ return self._wrapperList #========================================================================================= # Class Properties and methods #========================================================================================= @property def name(self) -> str: """Not allowed for ChemicalShift """ raise RuntimeError('ChemicalShift does not have attribute name') @name.setter @logCommand(get='self', isProperty=True) def name(self, value: str): """Not allowed for ChemicalShift """ raise RuntimeError('ChemicalShift name cannot be set') #~~~~~~~~~~~~~~~~ @property def _static(self): # Getter/setter for undo/redo return self._wrapperList._getAttribute(self._uniqueId, CS_STATIC, bool) @_static.setter @ccpNmrV3CoreSetter() def _static(self, value): self._wrapperList._setAttribute(self._uniqueId, CS_STATIC, value) @property def static(self) -> bool: """Static state of ChemicalShift. :return: True if chemicalShift or parent chemicalShiftTable is static """ return self._static or self.chemicalShiftList.static @property def dynamic(self) -> bool: """Dynamic state of ChemicalShift. :return: not chemicalShift.static """ return not self.static @property def orphan(self): """Orphan state of the chemicalShift :return: True if not static (i.e. dynamic), and has no associated peaks """ return not self.static and not self.assignedPeaks
[docs] def getStatic(self) -> bool: """Return the local static state of the chemicalShift """ return self._static
# @logCommand(get='self') # def setStatic(self, value: bool): # """Set the local static state for the chemicalShift. # """ # if not isinstance(value, bool): # raise ValueError(f'{self.className}.setStatic must be True/False') # # use setter above to handle undo/redo # self._static = value @property def state(self): """State of chemicalShift """ if self.static: return ChemicalShiftState.STATIC else: if self.orphan: return ChemicalShiftState.ORPHAN else: return ChemicalShiftState.DYNAMIC #~~~~~~~~~~~~~~~~ @property # NOTE:ED - added '@checkDeleted' but I don't think they are required # @checkDeleted() def value(self) -> Optional[float]: """shift value of ChemicalShift, in unit as defined in the ChemicalShiftList. """ return self._wrapperList._getAttribute(self._uniqueId, CS_VALUE, float) @value.setter @logCommand(get='self', isProperty=True) @ccpNmrV3CoreSetter() def value(self, val: Optional[float]): """Set the value for the chemicalShift. Integers will be cast as floats on the next get operation. """ if not isinstance(val, (float, int, type(None))): raise ValueError(f'{self.className}.value must be of type float, int or None') _nmrAtomPid = self._wrapperList._getAttribute(self._uniqueId, CS_NMRATOM, str) if _nmrAtomPid: raise ValueError(f'{self.className}.value cannot be changed with attached nmrAtom') self._wrapperList._setAttribute(self._uniqueId, CS_VALUE, val) #~~~~~~~~~~~~~~~~ @property def valueError(self) -> Optional[float]: """shift valueError of ChemicalShift, in unit as defined in the ChemicalShiftList. """ return self._wrapperList._getAttribute(self._uniqueId, CS_VALUEERROR, float) @valueError.setter @logCommand(get='self', isProperty=True) @ccpNmrV3CoreSetter() def valueError(self, value: Optional[float]): """Set the valueError for the chemicalShift. Integers will be cast as floats on the next get operation. """ if not isinstance(value, (float, int, type(None))): raise ValueError(f'{self.className}.valueError must be of type float, int or None') _nmrAtomPid = self._wrapperList._getAttribute(self._uniqueId, CS_NMRATOM, str) if _nmrAtomPid: raise ValueError(f'{self.className}.valueError cannot be changed with attached nmrAtom') self._wrapperList._setAttribute(self._uniqueId, CS_VALUEERROR, value) #~~~~~~~~~~~~~~~~ @property def figureOfMerit(self) -> Optional[float]: """Figure of Merit for ChemicalShift, between 0.0 and 1.0 inclusive. """ return self._wrapperList._getAttribute(self._uniqueId, CS_FIGUREOFMERIT, float) @figureOfMerit.setter @logCommand(get='self', isProperty=True) @ccpNmrV3CoreSetter() def figureOfMerit(self, value: Optional[float]): """Set the figureOfMerit for the chemicalShift. Integers will be cast as floats on the next get operation; only integers 0 and 1 are allowed. """ if not isinstance(value, (float, int, type(None))): raise ValueError(f'{self.className}.figureOfMerit must be of type float, int or None') if value is not None and not (MINFOM <= value <= MAXFOM): raise ValueError(f'{self.className}.figureOfMerit must be in range [{MINFOM} - {MAXFOM}]') self._wrapperList._setAttribute(self._uniqueId, CS_FIGUREOFMERIT, value) #~~~~~~~~~~~~~~~~ @property def nmrAtom(self) -> Optional[NmrAtom]: """Attached NmrAtom. """ _nmrAtomPid = self._wrapperList._getAttribute(self._uniqueId, CS_NMRATOM, str) return self.project.getByPid(_nmrAtomPid) def _nmrAtom(self, value: NmrAtom): """Set the nmrAtom inside core undoBlock """ nat = self.nmrAtom if value is None: if nat is None: return self._wrapperList._setAttribute(self._uniqueId, CS_NMRATOM, None) nat._chemicalShifts.remove(self) else: # nmrAtom and derived properties self._wrapperList._setAttribute(self._uniqueId, CS_NMRATOM, str(value.pid)) self._wrapperList._setAttributes(self._uniqueId, CS_CHAINCODE, CS_ATOMNAME, tuple(val or None for val in value.pid.fields)) if nat: nat._chemicalShifts.remove(self) value._chemicalShifts.append(self) self._recalculateShiftValue(value) @nmrAtom.setter @logCommand(get='self', isProperty=True) @ccpNmrV3CoreUndoBlock() def nmrAtom(self, value: Union[NmrAtom, str, Pid, None]): """Set the nmrAtom for the chemicalShift nmrAtom can be core object of type NmrAtom, or string pid or None :param value: new nmrAtom """ nat = self.nmrAtom _oldNmrAt = self.nmrAtom if value is None and _oldNmrAt is None: # trivial - ignore return if value is not None: # check the new value is valid - allow core object, str or Pid _nmrAtom = self._project.getByPid(value) if isinstance(value, str) else value if not _nmrAtom: raise ValueError(f'{self.className}.nmrAtom: {value} not found') if _nmrAtom == nat: # trivial - ignore return if not isinstance(_nmrAtom, NmrAtom): raise ValueError(f'{self.className}.nmrAtom: {value} must be of type NmrAtom') if self._wrapperList._searchChemicalShifts(nmrAtom=_nmrAtom): raise ValueError(f'{self.className}.nmrAtom: {_nmrAtom.pid} already exists') value = _nmrAtom with undoStackBlocking() as addUndoItem: if _oldNmrAt is None: # previously empty - remember the OLD settings - is only required by undo/redo to give previous state _oldSettings = self._wrapperList._getAttributes(self._uniqueId, CS_CHAINCODE, CS_ATOMNAME, (str, str, str, str)) addUndoItem(undo=partial(self._wrapperList._setAttributes, self._uniqueId, CS_CHAINCODE, CS_ATOMNAME, _oldSettings), ) # update the nmrAtom self._nmrAtom(value) addUndoItem(undo=partial(self._nmrAtom, _oldNmrAt), redo=partial(self._nmrAtom, value)) if value is None: # now empty - remember the NEW settings - is only required by undo/redo to give new state _newSettings = self._wrapperList._getAttributes(self._uniqueId, CS_CHAINCODE, CS_ATOMNAME, (str, str, str, str)) addUndoItem(redo=partial(self._wrapperList._setAttributes, self._uniqueId, CS_CHAINCODE, CS_ATOMNAME, _newSettings), ) #~~~~~~~~~~~~~~~~ @property def chainCode(self) -> Optional[str]: """chainCode for attached nmrAtom. Optional user value if nmrAtom is None """ return self._wrapperList._getAttribute(self._uniqueId, CS_CHAINCODE, str) @chainCode.setter @logCommand(get='self', isProperty=True) @ccpNmrV3CoreSetter() def chainCode(self, value: Optional[str]): """Set the chainCode for the chemicalShift Cannot be changed if there is an nmrAtom already attached Must be of type string or None :param value: new chainCode """ if not isinstance(value, (str, type(None))): raise ValueError(f'{self.className}.chainCode must be of type str or None') _nmrAtomPid = self._wrapperList._getAttribute(self._uniqueId, CS_NMRATOM, str) if _nmrAtomPid: raise RuntimeError(f'{self.className}.chainCode: derived value, cannot modify when nmrAtom is set') # only set if the nmrAtom has not been set self._wrapperList._setAttribute(self._uniqueId, CS_CHAINCODE, value) #~~~~~~~~~~~~~~~~ @property def sequenceCode(self) -> Optional[str]: """sequenceCode for attached nmrAtom. Optional user value if nmrAtom is None """ return self._wrapperList._getAttribute(self._uniqueId, CS_SEQUENCECODE, str) @sequenceCode.setter @logCommand(get='self', isProperty=True) @ccpNmrV3CoreSetter() def sequenceCode(self, value: Optional[str]): """Set the sequenceCode for the chemicalShift Cannot be changed if there is an nmrAtom already attached Must be of type string or None :param value: new sequenceCode """ if not isinstance(value, (str, type(None))): raise ValueError(f'{self.className}.sequenceCode must be of type str or None') _nmrAtomPid = self._wrapperList._getAttribute(self._uniqueId, CS_NMRATOM, str) if _nmrAtomPid: raise RuntimeError(f'{self.className}.sequenceCode: derived value, cannot modify when nmrAtom is set') # only set if the nmrAtom has not been set self._wrapperList._setAttribute(self._uniqueId, CS_SEQUENCECODE, value) #~~~~~~~~~~~~~~~~ @property def residueType(self) -> Optional[str]: """residueType for attached nmrAtom. Optional user value if nmrAtom is None """ return self._wrapperList._getAttribute(self._uniqueId, CS_RESIDUETYPE, str) @residueType.setter @logCommand(get='self', isProperty=True) @ccpNmrV3CoreSetter() def residueType(self, value: Optional[str]): """Set the residueType for the chemicalShift Cannot be changed if there is an nmrAtom already attached Must be of type string or None :param value: new residueType """ if not isinstance(value, (str, type(None))): raise ValueError(f'{self.className}.residueType must be of type str or None') _nmrAtomPid = self._wrapperList._getAttribute(self._uniqueId, CS_NMRATOM, str) if _nmrAtomPid: raise RuntimeError(f'{self.className}.residueType: derived value, cannot modify when nmrAtom is set') # only set if the nmrAtom has not been set self._wrapperList._setAttribute(self._uniqueId, CS_RESIDUETYPE, value) #~~~~~~~~~~~~~~~~ @property def atomName(self) -> Optional[str]: """atomName for attached nmrAtom Optional user value if nmrAtom is None """ return self._wrapperList._getAttribute(self._uniqueId, CS_ATOMNAME, str) @atomName.setter @logCommand(get='self', isProperty=True) @ccpNmrV3CoreSetter() def atomName(self, value: Optional[str]): """Set the atomName for the chemicalShift Cannot be changed if there is an nmrAtom already attached Must be of type string or None :param value: new atomName """ if not isinstance(value, (str, type(None))): raise ValueError(f'{self.className}.atomName must be of type str or None') _nmrAtomPid = self._wrapperList._getAttribute(self._uniqueId, CS_NMRATOM, str) if _nmrAtomPid: raise RuntimeError(f'{self.className}.atomName: derived value, cannot modify when nmrAtom is set') # only set if the nmrAtom has not been set self._wrapperList._setAttribute(self._uniqueId, CS_ATOMNAME, value) #~~~~~~~~~~~~~~~~ @property def assignedPeaks(self) -> tuple: """Assigned peaks for attached nmrAtom belonging to this chemicalShiftList. """ assigned = self.allAssignedPeaks if assigned is not None: return tuple(pp for pp in assigned if pp.chemicalShiftList == self.chemicalShiftList) return () @property def allAssignedPeaks(self) -> tuple: """All assigned peaks for attached nmrAtom. """ _nmrAtomPid = self._wrapperList._getAttribute(self._uniqueId, CS_NMRATOM, str) _nmrAtom = self.project.getByPid(_nmrAtomPid) if _nmrAtom: return tuple(set(makeIterableList(_nmrAtom.assignedPeaks))) return () @property def peakPpmPositions(self) -> tuple: """Return a tuple of the assigned peak positions (in ppm) """ return tuple(pos for pk in self.assignedPeaks for (pos, nmrAtom) in zip(pk.ppmPositions, pk.dimensionNmrAtoms)) #========================================================================================= # Implementation functions - necessary as there is no abstractWrapper object #=========================================================================================
[docs] def rename(self, value: str): """Not allowed for ChemicalShift """ raise RuntimeError('ChemicalShift cannot be renamed')
def _resetUniqueId(self, value): """Reset the uniqueId CCPN Internal - although not sure whether actually required here """ # if self._wrapperList._searchChemicalShifts(uniqueId=value): # raise ValueError(f'{self.className}._resetUniqueId: uniqueId {value} already exists') self._uniqueId = int(value) self._ccpnSortKey = (id(self.project), _importOrder.index(self._oldClassName), self._uniqueId)
[docs] def delete(self): """Delete the shift """ self._wrapperList.deleteChemicalShift(uniqueId=self._uniqueId)
# raise RuntimeError(f'{self.className}.delete: Please use ChemicalShiftList.deleteChemicalShift()') # optional error-trap def _updateNmrAtomShifts(self): """Restore the links to the nmrAtoms CCPN Internal - called from first creation from _restoreObject """ _nmrAtom = self.nmrAtom # must assume that the shift value is correct at this point if _nmrAtom: _nmrAtom._chemicalShifts.append(self) #========================================================================================= # CCPN functions #========================================================================================= def _getAsTuple(self): """Return the contents of the shift as a tuple. """ if self._isDeleted: raise RuntimeError(f'{self.className}._getAsTuple: shift is deleted') newRow = (self._uniqueId, self._isDeleted, self._static, # the local static self.value, self.valueError, self.figureOfMerit, ) + \ (str(self.nmrAtom.pid) if self.nmrAtom else None,) + \ (tuple(val or None for val in self.nmrAtom.pid.fields) if self.nmrAtom else (self.chainCode or None, self.sequenceCode or None, self.residueType or None, self.atomName or None) ) + \ (self.comment,) newRow = ShiftParameters(*newRow) return newRow def _getAsPandasSeries(self): """Return the contents of the shift as a pandas series """ _row = self._getAsTuple() return pd.DataFrame((_row,), columns=CS_COLUMNS) def _recalculateShiftValue(self, nmrAtom=None): """Calculate the shift value """ nmrAtom = nmrAtom or self.nmrAtom if not self.static: if nmrAtom: if nmrAtom.assignedPeaks: # remember the oldest setting self._oldValue, self._oldValueError = self.value, self.valueError else: self._oldValue = self._oldValueError = None value, valueError = nmrAtom._recalculateShiftValue(self._wrapperList.spectra) else: value, valueError = None, None valueError = valueError if value is not None else None if not (value is None and valueError is None): # update the dataframe self._wrapperList._setAttribute(self._uniqueId, CS_VALUE, value) self._wrapperList._setAttribute(self._uniqueId, CS_VALUEERROR, valueError) elif not (self._oldValue is None and self._oldValueError is None): # make sure that the empty value does not change self._wrapperList._setAttribute(self._uniqueId, CS_VALUE, self._oldValue) self._wrapperList._setAttribute(self._uniqueId, CS_VALUEERROR, self._oldValueError) def _renameNmrAtom(self, nmrAtom): """Update the values in the table for the renamed nmrAtom """ if nmrAtom and self._wrapperList._getAttribute(self._uniqueId, CS_NMRATOM, str) != nmrAtom.pid: # nmrAtom and derived properties self._wrapperList._setAttributes(self._uniqueId, CS_NMRATOM, CS_ATOMNAME, (str(nmrAtom.pid),) + tuple(val or None for val in nmrAtom.pid.fields)) #=========================================================================================== # new<Object> and other methods # Call appropriate routines in their respective locations #=========================================================================================== @deleteV3Object() def _deleteWrapper(self, chemicalShiftList, _newDeletedShifts, _newShifts, _oldDeletedShifts, _oldShifts): """Delete a pure V3 ChemicalShift object Method is wrapped with create/delete notifier CCPN Internal - Not standalone - requires functionality from ChemicalShiftList """ # add an undo/redo item to recover shifts with undoStackBlocking() as addUndoItem: # replace the contents of the internal list with the original/recovered items addUndoItem(undo=partial(chemicalShiftList._undoRedoShifts, _oldShifts), redo=partial(chemicalShiftList._undoRedoShifts, _newShifts)) self.nmrAtom = None with undoStackBlocking() as addUndoItem: addUndoItem(undo=partial(chemicalShiftList._undoRedoDeletedShifts, _oldDeletedShifts), redo=partial(chemicalShiftList._undoRedoDeletedShifts, _newDeletedShifts))
def _newChemicalShift(project: Project, chemicalShiftList, _uniqueId: Optional[int] = None ): """Create a new chemicalShift attached to the chemicalShiftList. :param chemicalShiftList: parent chemicalShiftList :param _uniqueId: _unique int identifier :return: a new ChemicalShift instance. """ result = ChemicalShift(project, chemicalShiftList, _uniqueId=_uniqueId) if result is None: raise RuntimeError('ChemicalShiftList._newChemicalShift: unable to generate new ChemicalShift item') return result def _getByTuple(chemicalShiftList, uniqueId: int = None, isDeleted: bool = False, static: bool = False, value: float = None, valueError: float = None, figureOfMerit: float = 1.0, nmrAtom: Union[NmrAtom, str, None] = None, chainCode: str = None, sequenceCode: str = None, residueType: str = None, atomName: str = None, comment: str = None): """Create a new tuple object from the supplied parameters Check whether a valid tuple can be created, otherwise raise the appropriate errors CCPN Internal """ # check whether the parameters are valid if nmrAtom and any((chainCode, sequenceCode, residueType, atomName)): # compare with parameter not the found nmrAtom raise ValueError('Cannot set nmrAtom and derived Properties at the same time') # now check with the 'found' nmrAtom nmrAtom = chemicalShiftList.project.getByPid(nmrAtom) if isinstance(nmrAtom, str) else nmrAtom if not isinstance(nmrAtom, (NmrAtom, type(None))): raise ValueError('nmrAtom must be of type nmrAtom or None') if not isinstance(static, bool): raise ValueError('static must be True/False') if not all(isinstance(val, (str, type(None))) for val in (chainCode, sequenceCode, residueType, atomName)): raise ValueError('chainCode, sequenceCode, residueType, atomName must be of type str or None') if not all(isinstance(val, (float, int, type(None))) for val in (value, valueError, figureOfMerit)): raise ValueError('value, valueError, figureOfMerit must be of type float, int or None') if figureOfMerit is not None and not (MINFOM <= figureOfMerit <= MAXFOM): raise ValueError(f'figureOfMerit must be in range [{MINFOM} - {MAXFOM}]') # create the row as defined in the pandas dataFrame newRow = (uniqueId, isDeleted, static, value, valueError, figureOfMerit, ) + \ (((str(nmrAtom.pid),) + tuple(val or None for val in nmrAtom.pid.fields)) if nmrAtom else (None, chainCode or None, sequenceCode or None, residueType or None, atomName or None)) + \ (comment,) newRow = ShiftParameters(*newRow) return newRow