Source code for ccpn.AnalysisScreen.pipes.CopyPeakListsBetweenSG

#=========================================================================================
# 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