#=========================================================================================
# 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: Luca Mureddu $"
__dateModified__ = "$dateModified: 2022-04-04 15:19:14 +0100 (Mon, April 04, 2022) $"
__version__ = "$Revision: 3.1.0 $"
#=========================================================================================
# Created
#=========================================================================================
__author__ = "$Author: Luca Mureddu $"
__date__ = "$Date: 2017-05-28 10:28:42 +0000 (Sun, May 28, 2017) $"
#=========================================================================================
# Start of code
#=========================================================================================
#### GUI IMPORTS
from ccpn.AnalysisScreen.gui.widgets import HitFinderWidgets as hw
from ccpn.ui.gui.widgets.PipelineWidgets import GuiPipe
from ccpn.ui.gui.widgets.Label import Label
from ccpn.ui.gui.widgets.CheckBox import CheckBox
from ccpn.ui.gui.widgets.RadioButtons import RadioButtons
from ccpn.ui.gui.widgets.DoubleSpinbox import ScientificDoubleSpinBox
#### NON GUI IMPORTS
from ccpn.framework.lib.pipeline.PipeBase import SpectraPipe, PIPE_SCREEN
from ccpn.core.lib.peakUtils import snap1DPeaksToExtrema, snap1DPeaksAndRereferenceSpectrum, recalculatePeaksHeightAtPosition
from ccpn.util.Logging import getLogger
########################################################################################################################
### Attributes:
### Used in setting the dictionary keys on _kwargs either in GuiPipe and Pipe
########################################################################################################################
PipeName = 'Copy PeakLists'
ReferenceSpectrumGroup = 'Origin_SpectrumGroup'
TargetSpectrumGroup = 'Destination_SpectrumGroup'
SnapNewPeak = 'Snap_Peaks'
SnapNegPeak = 'Snap_To_Neg_Peaks'
RefitPeak = 'Refit_Peaks'
Tolerance = 'Max_snap_distance_(ppm)'
ToleranceFoM = 'Skip_Below_FoM'
OptimiseReferencing = 'Optimise_ReferencingPoints'
UsePeakList = 'Destination_PeakList'
PeakHeight = 'Peak_Position-Height'
Original, AtPosition, Snapped = 'All_Original', 'Recalculate_Height_At_Original_Position', 'Snap_To_Best_Fit'
HeightOptions = [Original, AtPosition, Snapped]
Last = 'Last'
New = 'New'
PlistOptions = [Last, New]
SGVarNames = [ReferenceSpectrumGroup, TargetSpectrumGroup]
defaultPeakListIndice = -1
DefaultSnap = True
DefaultRefit = True
DefaultTolerance = 0.1 #ppm
DefaultToleranceFoM = 0.5 #ppm
DefaultOptimiseReferencing = True
DefaultSnapToNegative = False
########################################################################################################################
########################################## ALGORITHM ########################################################
########################################################################################################################
########################################################################################################################
########################################## GUI PIPE #############################################################
########################################################################################################################
[docs]class CopyPeakListsBetweenSGroupsGuiPipe(GuiPipe):
pipeName = PipeName
def __init__(self, name=pipeName, parent=None, project=None, **kw):
super(CopyPeakListsBetweenSGroupsGuiPipe, self)
GuiPipe.__init__(self, parent=parent, name=name, project=project, **kw)
self.parent = parent
row = 0
hw._addSGpulldowns(self, row, SGVarNames)
row += len(SGVarNames)
self.copyToPeaksLabel = Label(self.pipeFrame, UsePeakList, grid=(row, 0))
setattr(self, UsePeakList, RadioButtons(self.pipeFrame,
texts=PlistOptions, selectedInd=0,
direction='v',
callback=None, grid=(row, 1)))
row += 1
self.peakHeightLabel = Label(self.pipeFrame, PeakHeight, grid=(row, 0))
setattr(self, PeakHeight, RadioButtons(self.pipeFrame,
texts=HeightOptions, selectedInd=2,
direction='v',
callback=None, grid=(row, 1)))
row += 1
self.tolleranceLabel = Label(self.pipeFrame, text=Tolerance, grid=(row, 0))
setattr(self, Tolerance, ScientificDoubleSpinBox(self.pipeFrame, value=DefaultTolerance, min=0,
step=0.01, decimals=3, grid=(row, 1)))
row += 1
self.optimiseReferencingLabel = Label(self.pipeFrame, OptimiseReferencing, grid=(row, 0))
setattr(self, OptimiseReferencing, CheckBox(self.pipeFrame, checked=DefaultOptimiseReferencing,
callback=None, grid=(row, 1)))
row += 1
self.tolleranceFoMLabel = Label(self.pipeFrame, text=ToleranceFoM, grid=(row, 0))
setattr(self, ToleranceFoM, ScientificDoubleSpinBox(self.pipeFrame, value=DefaultToleranceFoM, min=0,
step=0.1, decimals=3, grid=(row, 1)))
row += 1
self.snapToNeg = Label(self.pipeFrame, SnapNegPeak, grid=(row, 0))
setattr(self, SnapNegPeak, CheckBox(self.pipeFrame, checked=DefaultSnapToNegative,
callback=None, grid=(row, 1)))
self._updateWidgets()
def _updateWidgets(self):
self._setSpectrumGroupPullDowns(SGVarNames)
########################################################################################################################
########################################## PIPE #############################################################
########################################################################################################################
[docs]class CopyPeakListsBetweenSGroupsPipe(SpectraPipe):
"""
Apply phasing to all the spectra in the pipeline
"""
guiPipe = CopyPeakListsBetweenSGroupsGuiPipe
pipeName = PipeName
pipeCategory = PIPE_SCREEN
_kwargs = {
ReferenceSpectrumGroup: 'ReferenceSpectrumGroup.pid', #this will be replaced by the SG pid in the gui
TargetSpectrumGroup: 'TargetSpectrumGroup.pid',
UsePeakList: Last,
Tolerance : DefaultTolerance,
ToleranceFoM : DefaultToleranceFoM,
OptimiseReferencing : DefaultOptimiseReferencing,
PeakHeight : Snapped,
SnapNegPeak: DefaultSnapToNegative
}
[docs] def runPipe(self, spectra):
'''
:param spectra: inputData
:return: spectra
'''
controlSpectrumGroup = self._getSpectrumGroup(self._kwargs[ReferenceSpectrumGroup])
targetSpectrumGroup = self._getSpectrumGroup(self._kwargs[TargetSpectrumGroup])
createNewPL = True if self._kwargs[UsePeakList] == New else False
snap = self._kwargs[PeakHeight] == Snapped
original = self._kwargs[PeakHeight] == Original
keepPos = self._kwargs[PeakHeight] == AtPosition
snapToNeg = self._kwargs[SnapNegPeak]
if self.project is not None:
if None not in [controlSpectrumGroup, targetSpectrumGroup]:
if not len(controlSpectrumGroup.spectra) == len(targetSpectrumGroup.spectra):
getLogger().warning('%s: Spectra count missmatched between SpectrumGroups. Copy aborted ' %PipeName)
return spectra
shifts = []
for controlSpectrum, targetSpectrum in zip(controlSpectrumGroup.spectra, targetSpectrumGroup.spectra):
controlPeakList = controlSpectrum.peakLists[defaultPeakListIndice]
if createNewPL:
targetPeakList=None # if None a new peakList will be created
else:
targetPeakList = targetSpectrum.peakLists[-1]
targetPeakList = controlPeakList.copyTo(targetSpectrum, targetPeakList=targetPeakList)
if targetSpectrum.noiseLevel is None:
targetSpectrum.noiseLevel = controlSpectrum.noiseLevel
if targetSpectrum.negativeNoiseLevel is None:
targetSpectrum.negativeNoiseLevel = controlSpectrum.negativeNoiseLevel
tolerance = self._kwargs[Tolerance]
toleranceFoM = self._kwargs[ToleranceFoM]
if targetSpectrum.dimensionCount == 1:
optimise = self._kwargs[OptimiseReferencing]
if snap:
if optimise:
shift = snap1DPeaksAndRereferenceSpectrum(targetPeakList.peaks, tolerance,
figOfMeritLimit=toleranceFoM, doNeg=snapToNeg)
shifts.append(shift)
recalculatePeaksHeightAtPosition(targetPeakList.peaks)
snap1DPeaksToExtrema(targetPeakList.peaks, tolerance, figOfMeritLimit=toleranceFoM,
doNeg=snapToNeg)
if original:
continue #do nothing
if keepPos:
recalculatePeaksHeightAtPosition(targetPeakList.peaks)
else:
peaks = targetPeakList.peaks, self.application
for peak in peaks:
if snap:
peak.snapToExtremum(halfBoxSearchWidth=4, halfBoxFitWidth=4,
minDropFactor=self.application.preferences.general.peakDropFactor,
searchBoxMode=self.application.preferences.general.searchBoxMode,
searchBoxDoFit=self.application.preferences.general.searchBoxDoFit,
fitMethod=self.application.preferences.general.peakFittingMethod)
if original:
continue #do nothing
if keepPos:
recalculatePeaksHeightAtPosition(targetPeakList.peaks)
return spectra
else:
getLogger().warning('Spectra not present. Add spectra first')
return spectra
CopyPeakListsBetweenSGroupsPipe.register() # Registers the pipe in the pipeline