# Copyright (c) DataLab Platform Developers, BSD 3-Clause license, see LICENSE file.""".. Signal processor object (see parent package :mod:`cdl.core.gui.processor`)"""# pylint: disable=invalid-name # Allows short reference names like x, y, ...from__future__importannotationsimportrefromcollections.abcimportCallableimportguidata.datasetasgdsimportnumpyasnpfromguidata.qthelpersimportexec_dialogimportcdl.computation.baseascpbimportcdl.computation.signalascpsimportcdl.paramfromcdl.configimportConf,_fromcdl.core.gui.processor.baseimportBaseProcessorfromcdl.core.model.baseimportResultProperties,ResultShapefromcdl.core.model.signalimportROI1DParam,SignalObj,SignalROI,create_signalfromcdl.utils.qthelpersimportqt_try_exceptfromcdl.widgetsimportfitdialog,signalbaseline,signalpeak
[docs]classSignalProcessor(BaseProcessor[SignalROI]):"""Object handling signal processing: operations, processing, analysis"""# pylint: disable=duplicate-code
[docs]@qt_try_except()defcompute_sum(self)->None:"""Compute sum with :py:func:`cdl.computation.signal.compute_addition`"""self.compute_n1("Σ",cps.compute_addition,title=_("Sum"))
[docs]@qt_try_except()defcompute_addition_constant(self,param:cpb.ConstantParam|None=None)->None:"""Compute sum with a constant with :py:func:`cdl.computation.signal.compute_addition_constant`"""self.compute_11(cps.compute_addition_constant,param,paramclass=cpb.ConstantParam,title=_("Sum with constant"),)
[docs]@qt_try_except()defcompute_average(self)->None:"""Compute average with :py:func:`cdl.computation.signal.compute_addition` and divide by the number of signals"""deffunc_objs(new_obj:SignalObj,old_objs:list[SignalObj])->None:"""Finalize average computation"""new_obj.data=new_obj.data/float(len(old_objs))ifnew_obj.dyisnotNone:new_obj.dy=new_obj.dy/float(len(old_objs))self.compute_n1("μ",cps.compute_addition,func_objs=func_objs,title=_("Average"))
[docs]@qt_try_except()defcompute_product(self)->None:"""Compute product with :py:func:`cdl.computation.signal.compute_product`"""self.compute_n1("Π",cps.compute_product,title=_("Product"))
[docs]@qt_try_except()defcompute_product_constant(self,param:cpb.ConstantParam|None=None)->None:"""Compute product with a constant with :py:func:`cdl.computation.signal.compute_product_constant`"""self.compute_11(cps.compute_product_constant,param,paramclass=cpb.ConstantParam,title=_("Product with constant"),)
[docs]@qt_try_except()defcompute_swap_axes(self)->None:"""Swap data axes with :py:func:`cdl.computation.signal.compute_swap_axes`"""self.compute_11(cps.compute_swap_axes,title=_("Swap axes"))
[docs]@qt_try_except()defcompute_abs(self)->None:"""Compute absolute value with :py:func:`cdl.computation.signal.compute_abs`"""self.compute_11(cps.compute_abs,title=_("Absolute value"))
[docs]@qt_try_except()defcompute_re(self)->None:"""Compute real part with :py:func:`cdl.computation.signal.compute_re`"""self.compute_11(cps.compute_re,title=_("Real part"))
[docs]@qt_try_except()defcompute_im(self)->None:"""Compute imaginary part with :py:func:`cdl.computation.signal.compute_im`"""self.compute_11(cps.compute_im,title=_("Imaginary part"))
[docs]@qt_try_except()defcompute_astype(self,param:cdl.param.DataTypeSParam|None=None)->None:"""Convert data type with :py:func:`cdl.computation.signal.compute_astype`"""self.compute_11(cps.compute_astype,param,cps.DataTypeSParam,title=_("Convert data type"))
[docs]@qt_try_except()defcompute_log10(self)->None:"""Compute Log10 with :py:func:`cdl.computation.signal.compute_log10`"""self.compute_11(cps.compute_log10,title="Log10")
[docs]@qt_try_except()defcompute_exp(self)->None:"""Compute Log10 with :py:func:`cdl.computation.signal.compute_exp`"""self.compute_11(cps.compute_exp,title=_("Exponential"))
[docs]@qt_try_except()defcompute_sqrt(self)->None:"""Compute square root with :py:func:`cdl.computation.signal.compute_sqrt`"""self.compute_11(cps.compute_sqrt,title=_("Square root"))
[docs]@qt_try_except()defcompute_power(self,param:cps.PowerParam|None=None)->None:"""Compute power with :py:func:`cdl.computation.signal.compute_power`"""self.compute_11(cps.compute_power,param,cps.PowerParam,title="Power")
[docs]@qt_try_except()defcompute_arithmetic(self,obj2:SignalObj|None=None,param:cpb.ArithmeticParam|None=None)->None:"""Compute arithmetic operation between two signals with :py:func:`cdl.computation.signal.compute_arithmetic`"""self.compute_n1n(obj2,_("signal to operate with"),cps.compute_arithmetic,param=param,paramclass=cpb.ArithmeticParam,title=_("Arithmetic"),)
[docs]@qt_try_except()defcompute_difference(self,obj2:SignalObj|list[SignalObj]|None=None)->None:"""Compute difference between two signals with :py:func:`cdl.computation.signal.compute_difference`"""self.compute_n1n(obj2,_("signal to subtract"),cps.compute_difference,title=_("Difference"),)
[docs]@qt_try_except()defcompute_difference_constant(self,param:cpb.ConstantParam|None=None)->None:"""Compute difference with a constant with :py:func:`cdl.computation.signal.compute_difference_constant`"""self.compute_11(cps.compute_difference_constant,param,paramclass=cpb.ConstantParam,title=_("Difference with constant"),edit=True,)
[docs]@qt_try_except()defcompute_quadratic_difference(self,obj2:SignalObj|list[SignalObj]|None=None)->None:"""Compute quadratic difference between two signals with :py:func:`cdl.computation.signal.compute_quadratic_difference`"""self.compute_n1n(obj2,_("signal to subtract"),cps.compute_quadratic_difference,title=_("Quadratic difference"),)
[docs]@qt_try_except()defcompute_division(self,obj2:SignalObj|list[SignalObj]|None=None)->None:"""Compute division between two signals with :py:func:`cdl.computation.signal.compute_division`"""self.compute_n1n(obj2,_("divider"),cps.compute_division,title=_("Division"),)
[docs]@qt_try_except()defcompute_division_constant(self,param:cpb.ConstantParam|None=None)->None:"""Compute division by a constant with :py:func:`cdl.computation.signal.compute_division_constant`"""self.compute_11(cps.compute_division_constant,param,paramclass=cpb.ConstantParam,title=_("Division by constant"),)
[docs]@qt_try_except()defcompute_peak_detection(self,param:cdl.param.PeakDetectionParam|None=None)->None:"""Detect peaks from data with :py:func:`cdl.computation.signal.compute_peak_detection`"""obj=self.panel.objview.get_sel_objects(include_groups=True)[0]edit,param=self.init_param(param,cps.PeakDetectionParam,_("Peak detection"))ifedit:dlg=signalpeak.SignalPeakDetectionDialog(obj,parent=self.panel.parent())ifexec_dialog(dlg):param.threshold=int(dlg.get_threshold()*100)param.min_dist=dlg.get_min_dist()else:returnself.compute_11(cps.compute_peak_detection,param)
[docs]@qt_try_except()defcompute_reverse_x(self)->None:"""Reverse X axis with :py:func:`cdl.computation.signal.compute_reverse_x`"""self.compute_11(cps.compute_reverse_x,title=_("Reverse X axis"))
# ------Signal Processing
[docs]@qt_try_except()defcompute_normalize(self,param:cdl.param.NormalizeParam|None=None)->None:"""Normalize data with :py:func:`cdl.computation.signal.compute_normalize`"""self.compute_11(cps.compute_normalize,param,cps.NormalizeParam,title=_("Normalize"))
[docs]@qt_try_except()defcompute_derivative(self)->None:"""Compute derivative with :py:func:`cdl.computation.signal.compute_derivative`"""self.compute_11(cps.compute_derivative,title=_("Derivative"))
[docs]@qt_try_except()defcompute_integral(self)->None:"""Compute integral with :py:func:`cdl.computation.signal.compute_integral`"""self.compute_11(cps.compute_integral,title=_("Integral"))
[docs]@qt_try_except()defcompute_calibration(self,param:cdl.param.XYCalibrateParam|None=None)->None:"""Compute data linear calibration with :py:func:`cdl.computation.signal.compute_calibration`"""self.compute_11(cps.compute_calibration,param,cps.XYCalibrateParam,title=_("Linear calibration"),comment="y = a.x + b",)
[docs]@qt_try_except()defcompute_clip(self,param:cpb.ClipParam|None=None)->None:"""Compute maximum data clipping with :py:func:`cdl.computation.signal.compute_clip`"""self.compute_11(cps.compute_clip,param,cpb.ClipParam,title=_("Clipping"))
[docs]@qt_try_except()defcompute_offset_correction(self,param:ROI1DParam|None=None)->None:"""Compute offset correction with :py:func:`cdl.computation.signal.compute_offset_correction`"""obj=self.panel.objview.get_sel_objects(include_groups=True)[0]ifparamisNone:dlg=signalbaseline.SignalBaselineDialog(obj,parent=self.panel.parent())ifexec_dialog(dlg):param=ROI1DParam()param.xmin,param.xmax=dlg.get_x_range()else:returnself.compute_11(cps.compute_offset_correction,param)
[docs]@qt_try_except()defcompute_gaussian_filter(self,param:cpb.GaussianParam|None=None)->None:"""Compute gaussian filter with :py:func:`cdl.computation.signal.compute_gaussian_filter`"""self.compute_11(cps.compute_gaussian_filter,param,cpb.GaussianParam,title=_("Gaussian filter"),)
[docs]@qt_try_except()defcompute_moving_average(self,param:cpb.MovingAverageParam|None=None)->None:"""Compute moving average with :py:func:`cdl.computation.signal.compute_moving_average`"""self.compute_11(cps.compute_moving_average,param,cpb.MovingAverageParam,title=_("Moving average"),)
[docs]@qt_try_except()defcompute_moving_median(self,param:cpb.MovingMedianParam|None=None)->None:"""Compute moving median with :py:func:`cdl.computation.signal.compute_moving_median`"""self.compute_11(cps.compute_moving_median,param,cpb.MovingMedianParam,title=_("Moving median"),)
[docs]@qt_try_except()defcompute_wiener(self)->None:"""Compute Wiener filter with :py:func:`cdl.computation.signal.compute_wiener`"""self.compute_11(cps.compute_wiener,title=_("Wiener filter"))
def__freq_filter(self,param:cdl.param.LowPassFilterParam|cdl.param.HighPassFilterParam|cdl.param.BandPassFilterParam|cdl.param.BandStopFilterParam,paramclass:type[cdl.param.LowPassFilterParam|cdl.param.HighPassFilterParam|cdl.param.BandPassFilterParam|cdl.param.BandStopFilterParam],title:str,)->None:"""Compute frequency filter"""edit,param=self.init_param(param,paramclass,title)ifedit:obj=self.panel.objview.get_sel_objects(include_groups=True)[0]param.update_from_signal(obj)self.compute_11(cps.compute_filter,param,title=title,edit=edit)
[docs]@qt_try_except()defcompute_lowpass(self,param:cdl.param.LowPassFilterParam|None=None)->None:"""Compute high-pass filter with :py:func:`cdl.computation.signal.compute_filter`"""self.__freq_filter(param,cdl.param.LowPassFilterParam,_("Low-pass filter"))
[docs]@qt_try_except()defcompute_highpass(self,param:cdl.param.HighPassFilterParam|None=None)->None:"""Compute high-pass filter with :py:func:`cdl.computation.signal.compute_filter`"""self.__freq_filter(param,cdl.param.HighPassFilterParam,_("High-pass filter"))
[docs]@qt_try_except()defcompute_bandpass(self,param:cdl.param.BandPassFilterParam|None=None)->None:"""Compute band-pass filter with :py:func:`cdl.computation.signal.compute_filter`"""self.__freq_filter(param,cdl.param.BandPassFilterParam,_("Band-pass filter"))
[docs]@qt_try_except()defcompute_bandstop(self,param:cdl.param.BandStopFilterParam|None=None)->None:"""Compute band-stop filter with :py:func:`cdl.computation.signal.compute_filter`"""self.__freq_filter(param,cdl.param.BandStopFilterParam,_("Band-stop filter"))
[docs]@qt_try_except()defcompute_fft(self,param:cdl.param.FFTParam|None=None)->None:"""Compute FFT with :py:func:`cdl.computation.signal.compute_fft`"""ifparamisNone:param=cpb.FFTParam.create(shift=Conf.proc.fft_shift_enabled.get())self.compute_11(cps.compute_fft,param,title=_("FFT"),edit=False)
[docs]@qt_try_except()defcompute_ifft(self,param:cdl.param.FFTParam|None=None)->None:"""Compute iFFT with :py:func:`cdl.computation.signal.compute_ifft`"""ifparamisNone:param=cpb.FFTParam.create(shift=Conf.proc.fft_shift_enabled.get())self.compute_11(cps.compute_ifft,param,title=_("iFFT"),edit=False)
[docs]@qt_try_except()defcompute_magnitude_spectrum(self,param:cdl.param.SpectrumParam|None=None)->None:"""Compute magnitude spectrum with :py:func:`cdl.computation.signal.compute_magnitude_spectrum`"""self.compute_11(cps.compute_magnitude_spectrum,param,cdl.param.SpectrumParam,title=_("Magnitude spectrum"),)
[docs]@qt_try_except()defcompute_phase_spectrum(self)->None:"""Compute phase spectrum with :py:func:`cdl.computation.signal.compute_phase_spectrum`"""self.compute_11(cps.compute_phase_spectrum,title=_("Phase spectrum"))
[docs]@qt_try_except()defcompute_psd(self,param:cdl.param.SpectrumParam|None=None)->None:"""Compute power spectral density with :py:func:`cdl.computation.signal.compute_psd`"""self.compute_11(cps.compute_psd,param,cdl.param.SpectrumParam,title=_("PSD"))
[docs]@qt_try_except()defcompute_interpolation(self,obj2:SignalObj|None=None,param:cdl.param.InterpolationParam|None=None,):"""Compute interpolation with :py:func:`cdl.computation.signal.compute_interpolation`"""self.compute_n1n(obj2,_("signal for X values"),cps.compute_interpolation,param,cps.InterpolationParam,title=_("Interpolation"),)
[docs]@qt_try_except()defcompute_resampling(self,param:cdl.param.ResamplingParam|None=None):"""Compute resampling with :py:func:`cdl.computation.signal.compute_resampling`"""edit,param=self.init_param(param,cps.ResamplingParam,_("Resampling"))ifedit:obj=self.panel.objview.get_sel_objects(include_groups=True)[0]ifparam.xminisNone:param.xmin=obj.x[0]ifparam.xmaxisNone:param.xmax=obj.x[-1]ifparam.dxisNone:param.dx=obj.x[1]-obj.x[0]ifparam.nbptsisNone:param.nbpts=len(obj.x)self.compute_11(cps.compute_resampling,param,cps.ResamplingParam,title=_("Resampling"),edit=edit,)
[docs]@qt_try_except()defcompute_detrending(self,param:cdl.param.DetrendingParam|None=None):"""Compute detrending with :py:func:`cdl.computation.signal.compute_detrending`"""self.compute_11(cps.compute_detrending,param,cps.DetrendingParam,title=_("Detrending"),)
[docs]@qt_try_except()defcompute_convolution(self,obj2:SignalObj|None=None)->None:"""Compute convolution with :py:func:`cdl.computation.signal.compute_convolution`"""self.compute_n1n(obj2,_("signal to convolve with"),cps.compute_convolution,title=_("Convolution"),)
[docs]@qt_try_except()defcompute_windowing(self,param:cdl.param.WindowingParam|None=None)->None:"""Compute windowing with :py:func:`cdl.computation.signal.compute_windowing`"""self.compute_11(cps.compute_windowing,param,cdl.param.WindowingParam,title=_("Windowing"),)
[docs]@qt_try_except()defcompute_polyfit(self,param:cdl.param.PolynomialFitParam|None=None)->None:"""Compute polynomial fitting curve"""txt=_("Polynomial fit")edit,param=self.init_param(param,cps.PolynomialFitParam,txt)ifnoteditorparam.edit(self.panel.parent()):dlgfunc=fitdialog.polynomialfitdefpolynomialfit(x,y,parent=None):"""Polynomial fit dialog function"""returndlgfunc(x,y,param.degree,parent=parent)self.compute_fit(txt,polynomialfit)
def__row_compute_fit(self,obj:SignalObj,name:str,fitdlgfunc:Callable)->None:"""Curve fitting computing sub-method"""output=fitdlgfunc(obj.x,obj.y,parent=self.panel.parent())ifoutputisnotNone:y,params=outputparams:list[fitdialog.FitParam]pvalues={}forparaminparams:ifre.match(r"[\S\_]*\d{2}$",param.name):shname=param.name[:-2]value=pvalues.get(shname,np.array([]))pvalues[shname]=np.array(list(value)+[param.value])else:pvalues[param.name]=param.value# Creating new signalmetadata={fitdlgfunc.__name__:pvalues}signal=create_signal(f"{name}({obj.title})",obj.x,y,metadata=metadata)# Creating new plot itemself.panel.add_object(signal)
[docs]@qt_try_except()defcompute_fit(self,title:str,fitdlgfunc:Callable)->None:"""Compute fitting curve using an interactive dialog Args: title: Title of the dialog fitdlgfunc: Fitting dialog function """forobjinself.panel.objview.get_sel_objects():self.__row_compute_fit(obj,title,fitdlgfunc)
[docs]@qt_try_except()defcompute_multigaussianfit(self)->None:"""Compute multi-Gaussian fitting curve using an interactive dialog"""fitdlgfunc=fitdialog.multigaussianfitforobjinself.panel.objview.get_sel_objects():dlg=signalpeak.SignalPeakDetectionDialog(obj,parent=self.panel)ifexec_dialog(dlg):# Computing x, ypeaks=dlg.get_peak_indices()defmultigaussianfit(x,y,parent=None):"""Multi-Gaussian fit dialog function"""# pylint: disable=cell-var-from-loopreturnfitdlgfunc(x,y,peaks,parent=parent)self.__row_compute_fit(obj,_("Multi-Gaussian fit"),multigaussianfit)
@qt_try_except()def_extract_multiple_roi_in_single_object(self,group:gds.DataSetGroup)->None:"""Extract multiple Regions Of Interest (ROIs) from data in a single object"""self.compute_11(cps.extract_multiple_roi,group,title=_("Extract ROI"))@qt_try_except()def_extract_each_roi_in_separate_object(self,group:gds.DataSetGroup)->None:"""Extract each single Region Of Interest (ROI) from data in a separate object (keep the ROI in the case of a circular ROI, for example)"""self.compute_1n(cps.extract_single_roi,group.datasets,"ROI",edit=False)# ------Signal Analysis
[docs]@qt_try_except()defcompute_fwhm(self,param:cdl.param.FWHMParam|None=None)->dict[str,ResultShape]:"""Compute FWHM with :py:func:`cdl.computation.signal.compute_fwhm`"""returnself.compute_10(cps.compute_fwhm,param,cps.FWHMParam,title=_("FWHM"))
[docs]@qt_try_except()defcompute_fw1e2(self)->dict[str,ResultShape]:"""Compute FW at 1/e² with :py:func:`cdl.computation.signal.compute_fw1e2`"""returnself.compute_10(cps.compute_fw1e2,title=_("FW")+"1/e²")
[docs]@qt_try_except()defcompute_stats(self)->dict[str,ResultProperties]:"""Compute data statistics with :py:func:`cdl.computation.signal.compute_stats`"""returnself.compute_10(cps.compute_stats,title=_("Statistics"))
[docs]@qt_try_except()defcompute_histogram(self,param:cdl.param.HistogramParam|None=None)->dict[str,ResultShape]:"""Compute histogram with :py:func:`cdl.computation.signal.compute_histogram`"""returnself.compute_11(cps.compute_histogram,param,cps.HistogramParam,title=_("Histogram"))
[docs]@qt_try_except()defcompute_contrast(self)->dict[str,ResultProperties]:"""Compute contrast with :py:func:`cdl.computation.signal.compute_contrast`"""returnself.compute_10(cps.compute_contrast,title=_("Contrast"))
[docs]@qt_try_except()defcompute_x_at_minmax(self)->dict[str,ResultProperties]:"""Compute x at min/max with :py:func:`cdl.computation.signal.compute_x_at_minmax`"""returnself.compute_10(cps.compute_x_at_minmax,title="X @ min,max")
[docs]@qt_try_except()defcompute_sampling_rate_period(self)->dict[str,ResultProperties]:"""Compute sampling rate and period (mean and std) with :py:func:`cdl.computation.signal.compute_sampling_rate_period`"""returnself.compute_10(cps.compute_sampling_rate_period,title=_("Sampling rate and period"))
[docs]@qt_try_except()defcompute_bandwidth_3db(self)->None:"""Compute bandwidth at -3dB with :py:func:`cdl.computation.signal.compute_bandwidth_3db`"""self.compute_10(cps.compute_bandwidth_3db,title=_("Bandwidth"))