# Copyright (c) DataLab Platform Developers, BSD 3-Clause license, see LICENSE file.""".. Common computation objects (see parent package :mod:`cdl.computation`)"""# pylint: disable=invalid-name # Allows short reference names like x, y, ...# Note:# ----# All dataset classes must also be imported in the cdl.computation.param module.from__future__importannotationsfromtypingimportTYPE_CHECKINGimportguidata.datasetasgdsimportnumpyasnpfromcdl.configimport_fromcdl.objimportResultProperties,create_signalifTYPE_CHECKING:fromtypingimportCallablefromcdl.objimportImageObj,SignalObj
[docs]defupdate_operation(self,_item,_value):# pylint: disable=unused-argument"""Update the operation item"""self.operation=self.get_operation()
operators=("+","-","×","/")operator=gds.ChoiceItem(_("Operator"),list(zip(operators,operators))).set_prop("display",callback=update_operation)factor=(gds.FloatItem(_("Factor"),default=1.0).set_pos(col=1).set_prop("display",callback=update_operation))constant=(gds.FloatItem(_("Constant"),default=0.0).set_pos(col=1).set_prop("display",callback=update_operation))operation=gds.StringItem(_("Operation"),default="").set_prop("display",active=False)restore_dtype=gds.BoolItem(_("Convert to `obj1` data type"),label=_("Result"),default=True)
HELP_MODE=_("""Mode of the filter:- 'reflect': Reflect the data at the boundary- 'constant': Pad with a constant value- 'nearest': Pad with the nearest value- 'mirror': Reflect the data at the boundary with the data itself- 'wrap': Circular boundary""")
[docs]classMovingAverageParam(gds.DataSet):"""Moving average parameters"""n=gds.IntItem(_("Size of the moving window"),default=3,min=1)modes=("reflect","constant","nearest","mirror","wrap")mode=gds.ChoiceItem(_("Mode"),list(zip(modes,modes)),default="reflect",help=HELP_MODE)
[docs]classMovingMedianParam(gds.DataSet):"""Moving median parameters"""n=gds.IntItem(_("Size of the moving window"),default=3,min=1,even=False)modes=("reflect","constant","nearest","mirror","wrap")mode=gds.ChoiceItem(_("Mode"),list(zip(modes,modes)),default="nearest",help=HELP_MODE)
[docs]classNormalizeParam(gds.DataSet):"""Normalize parameters"""methods=(("maximum",_("Maximum")),("amplitude",_("Amplitude")),("area",_("Area")),("energy",_("Energy")),("rms",_("RMS")),)method=gds.ChoiceItem(_("Normalize with respect to"),methods)
[docs]defget_suffix(self,data:np.ndarray)->str:"""Return suffix for the histogram computation Args: data: data array """suffix=f"bins={self.bins:d}"ifself.lowerisnotNone:suffix+=f", ymin={self.lower:.3f}"else:self.lower=np.min(data)ifself.upperisnotNone:suffix+=f", ymax={self.upper:.3f}"else:self.upper=np.max(data)
bins=gds.IntItem(_("Number of bins"),default=256,min=1)lower=gds.FloatItem(_("Lower limit"),default=None,check=False)upper=gds.FloatItem(_("Upper limit"),default=None,check=False)
[docs]classFFTParam(gds.DataSet):"""FFT parameters"""shift=gds.BoolItem(_("Shift"),help=_("Shift zero frequency to center"))
[docs]classConstantParam(gds.DataSet):"""Parameter used to set a constant value to used in operations"""value=gds.FloatItem(_("Constant value"))
# MARK: Helper functions for creating result objects -----------------------------------
[docs]defdst_11(src:SignalObj|ImageObj,name:str,suffix:str|None=None)->SignalObj|ImageObj:"""Create a result object, as returned by the callback function of the :func:`cdl.core.gui.processor.base.BaseProcessor.compute_11` method Args: src: source signal or image object name: name of the function. If provided, the title of the result object will be `{name}({src.short_id})|{suffix})`, unless the name is a single character, in which case the title will be `{src.short_id}{name}{suffix}` where `name` is an operator and `suffix` is the other term of the operation. suffix: suffix to add to the title. Optional. Returns: Result signal or image object """iflen(name)==1:# This is an operatortitle=f"{src.short_id}{name}"else:title=f"{name}({src.short_id})"ifsuffix:# suffix may be None or an empty stringtitle+="|"ifsuffix:# suffix may be None or an empty stringtitle+=suffixreturnsrc.copy(title=title)
[docs]defdst_n1n(src1:SignalObj|ImageObj,src2:SignalObj|ImageObj,name:str,suffix:str|None=None,)->SignalObj|ImageObj:"""Create a result object, as returned by the callback function of the :func:`cdl.core.gui.processor.base.BaseProcessor.compute_n1n` method Args: src1: input signal or image object src2: input signal or image object name: name of the processing function Returns: Output signal or image object """iflen(name)==1:# This is an operatortitle=f"{src1.short_id}{name}{src2.short_id}"else:title=f"{name}({src1.short_id}, {src2.short_id})"ifsuffixisnotNone:title+="|"+suffixreturnsrc1.copy(title=title)
[docs]defnew_signal_result(src:SignalObj|ImageObj,name:str,suffix:str|None=None,units:tuple[str,str]|None=None,labels:tuple[str,str]|None=None,)->SignalObj:"""Create new signal object as a result of a compute_11 function As opposed to the `dst_11` functions, this function creates a new signal object without copying the original object metadata, except for the "source" entry. Args: src: input signal or image object name: name of the processing function suffix: suffix to add to the title units: units of the output signal labels: labels of the output signal Returns: Output signal object """title=f"{name}({src.short_id})"dst=create_signal(title=title,units=units,labels=labels)ifsuffixisnotNone:dst.title+="|"+suffixif"source"insrc.metadata:dst.metadata["source"]=src.metadata["source"]# Keep track of the sourcereturndst
[docs]defcalc_resultproperties(title:str,obj:SignalObj|ImageObj,labeledfuncs:dict[str,Callable])->ResultProperties:"""Calculate result properties by executing a computation function on a signal/image object. Args: title: title of the result properties obj: signal or image object labeledfuncs: dictionary of labeled computation functions. The keys are the labels of the computation functions and the values are the functions themselves (each function must take a single argument - which is the data of the ROI or the whole signal/image - and return a float) Returns: Result properties object """ifnotall(isinstance(k,str)forkinlabeledfuncs.keys()):raiseValueError("Keys of labeledfuncs must be strings")ifnotall(callable(v)forvinlabeledfuncs.values()):raiseValueError("Values of labeledfuncs must be functions")res=[]fori_roiin[None]+list(obj.iterate_roi_indices()):data_roi=obj.get_data(i_roi)val_roi=-1ifi_roiisNoneelsei_roires.append([val_roi]+[fn(data_roi)forfninlabeledfuncs.values()])returnResultProperties(title,np.array(res),list(labeledfuncs.keys()))