# Copyright (c) DataLab Platform Developers, BSD 3-Clause license, see LICENSE file.""".. Base processor object (see parent package :mod:`cdl.core.gui.processor`)"""# pylint: disable=invalid-name # Allows short reference names like x, y, ...from__future__importannotationsimportabcimportmultiprocessingimporttimeimportwarningsfromcollections.abcimportCallablefrommultiprocessing.poolimportPoolfromtypingimportTYPE_CHECKING,Any,Generic,Unionimportguidata.datasetasgdsimportnumpyasnpfromguidata.datasetimportupdate_datasetfromguidata.qthelpersimportexec_dialogfromguidata.widgets.arrayeditorimportArrayEditorfromqtpyimportQtCoreasQCfromqtpyimportQtWidgetsasQWfromcdlimportenvfromcdl.algorithms.datatypesimportis_complex_dtypefromcdl.configimportConf,_fromcdl.core.gui.processor.catcherimportCompOut,wng_err_funcfromcdl.core.model.baseimportResultProperties,ResultShape,TypeROIfromcdl.utils.qthelpersimportcreate_progress_bar,qt_try_exceptfromcdl.widgets.warningerrorimportshow_warning_errorifTYPE_CHECKING:frommultiprocessing.poolimportAsyncResultfromplotpy.plotimportPlotWidgetfromcdl.computation.baseimport(ArithmeticParam,ClipParam,ConstantParam,GaussianParam,MovingAverageParam,MovingMedianParam,NormalizeParam,)fromcdl.core.gui.panel.imageimportImagePanelfromcdl.core.gui.panel.signalimportSignalPanelfromcdl.core.model.imageimportImageObjfromcdl.core.model.signalimportSignalObjObj=Union[SignalObj,ImageObj]# Enable multiprocessing support for Windows, with frozen executable (e.g. PyInstaller)multiprocessing.freeze_support()# Set start method to 'spawn' for Linux (default is 'fork' which is not safe here# because of the use of Qt and multithreading) - for other OS, the default is# 'spawn' anywaytry:multiprocessing.set_start_method("spawn")exceptRuntimeError:# This exception is raised if the method is already set (this may happen because# this module is imported more than once, e.g. when running tests)passCOMPUTATION_TIP=_("DataLab relies on various libraries to perform the computation. During the ""computation, errors may occur because of the data (e.g. division by zero, ""unexpected data type, etc.) or because of the libraries (e.g. memory error, ""etc.). If you encounter an error, before reporting it, please ensure that ""the computation is correct, by checking the data and the parameters.")POOL:Pool|None=None
[docs]classWorker:"""Multiprocessing worker, to run long-running tasks in a separate process"""def__init__(self)->None:self.asyncresult:AsyncResult=Noneself.result:Any=None
[docs]@staticmethoddefcreate_pool()->None:"""Create multiprocessing pool"""globalPOOL# pylint: disable=global-statement# Create a pool with one processPOOL=Pool(processes=1)# pylint: disable=not-callable,consider-using-with
[docs]@staticmethoddefterminate_pool(wait:bool=False)->None:"""Terminate multiprocessing pool. Args: wait: wait for all tasks to finish. Defaults to False. """globalPOOL# pylint: disable=global-statementifPOOLisnotNone:ifwait:# Close the pool properly (wait for all tasks to finish)POOL.close()else:# Terminate the pool and stop the timerPOOL.terminate()POOL.join()POOL=None
[docs]defrestart_pool(self)->None:"""Terminate and recreate the pool"""# Terminate the process and stop the timerself.terminate_pool(wait=False)# Recreate the pool for the next computationself.create_pool()
[docs]defrun(self,func:Callable,args:tuple[Any])->None:"""Run computation. Args: func: function to run args: arguments """globalPOOL# pylint: disable=global-statement,global-variable-not-assignedassertPOOLisnotNoneself.asyncresult=POOL.apply_async(wng_err_func,(func,args))
[docs]defclose(self)->None:"""Close worker: close pool properly and wait for all tasks to finish"""# Close multiprocessing Pool properly, but only if no computation is running,# to avoid blocking the GUI at exit (so, when wait=True, we wait for the# task to finish before closing the pool but there is actually no task running,# so the pool is closed immediately but *properly*)self.terminate_pool(wait=self.asyncresultisNone)
[docs]defis_computation_finished(self)->bool:"""Return True if computation is finished. Returns: bool: True if computation is finished """returnself.asyncresult.ready()
[docs]defget_result(self)->CompOut:"""Return computation result. Returns: CompOut: computation result """self.result=self.asyncresult.get()self.asyncresult=Nonereturnself.result
[docs]defis_pairwise_mode()->bool:"""Return True if operation mode is pairwise. Returns: bool: True if operation mode is pairwise """state=Conf.proc.operation_mode.get()=="pairwise"returnstate
[docs]defset_process_isolation_enabled(self,enabled:bool)->None:"""Set process isolation enabled. Args: enabled: enabled """ifenabled:ifself.workerisNone:self.worker=Worker()self.worker.create_pool()else:ifself.workerisnotNone:self.worker.terminate_pool()self.worker=None
[docs]defhas_param_defaults(self,paramclass:type[gds.DataSet])->bool:"""Return True if parameter defaults are available. Args: paramclass: parameter class Returns: bool: True if parameter defaults are available """returnparamclass.__name__inself.PARAM_DEFAULTS
[docs]definit_param(self,param:gds.DataSet,paramclass:type[gds.DataSet],title:str,comment:str|None=None,)->tuple[bool,gds.DataSet]:"""Initialize processing parameters. Args: param: parameter paramclass: parameter class title: title comment: comment Returns: Tuple (edit, param) where edit is True if parameters have been edited, False otherwise. """edit=paramisNoneifedit:param=paramclass(title,comment)self.update_param_defaults(param)returnedit,param
[docs]defcompute_11(self,func:Callable,param:gds.DataSet|None=None,paramclass:gds.DataSet|None=None,title:str|None=None,comment:str|None=None,edit:bool|None=None,)->None:"""Compute 11 function: 1 object in → 1 object out. Args: func: function param: parameter paramclass: parameter class title: title comment: comment edit: edit parameters """if(editisNoneorparamisNone)andparamclassisnotNone:edit,param=self.init_param(param,paramclass,title,comment)ifparamisnotNone:ifeditandnotparam.edit(parent=self.panel.parent()):returnself._compute_11_subroutine([func],[param],title)
[docs]defcompute_1n(self,funcs:list[Callable]|Callable,params:list|None=None,title:str|None=None,edit:bool|None=None,)->None:"""Compute 1n function: 1 object in → n objects out. Args: funcs: list of functions params: list of parameters title: title edit: edit parameters """ifparamsisNone:assertnotisinstance(funcs,Callable)params=[None]*len(funcs)else:group=gds.DataSetGroup(params,title=_("Parameters"))ifeditandnotgroup.edit(parent=self.panel.parent()):returnifisinstance(funcs,Callable):funcs=[funcs]*len(params)else:assertlen(funcs)==len(params)self._compute_11_subroutine(funcs,params,title)
[docs]defhandle_output(self,compout:CompOut,context:str,progress:QW.QProgressDialog)->SignalObj|ImageObj|ResultShape|ResultProperties|None:"""Handle computation output: if error, display error message, if warning, display warning message. Args: compout: computation output context: context (e.g. "Computing: Gaussian filter") progress: progress dialog Returns: Output object: a signal or image object, or a result shape object, or None if error """ifcompout.error_msgorcompout.warning_msg:mindur=progress.minimumDuration()progress.setMinimumDuration(1000000)ifcompout.error_msg:show_warning_error(self.panel,"error",context,compout.error_msg,COMPUTATION_TIP)ifcompout.warning_msg:show_warning_error(self.panel,"warning",context,compout.warning_msg)progress.setMinimumDuration(mindur)ifcompout.error_msg:returnNonereturncompout.result
def__exec_func(self,func:Callable,args:tuple,progress:QW.QProgressDialog,)->CompOut|None:"""Execute function, eventually in a separate process. Args: func: function to execute args: function arguments progress: progress dialog Returns: Computation output object or None if canceled """QW.QApplication.processEvents()ifnotprogress.wasCanceled():ifself.workerisNone:returnwng_err_func(func,args)self.worker.run(func,args)whilenotself.worker.is_computation_finished():QW.QApplication.processEvents()time.sleep(0.1)ifprogress.wasCanceled():self.worker.restart_pool()breakifself.worker.is_computation_finished():returnself.worker.get_result()returnNonedef_compute_11_subroutine(self,funcs:list[Callable],params:list,title:str)->None:"""Compute 11 subroutine: used by compute 11 and compute 1n methods. Args: funcs: list of functions to execute params: list of parameters title: title of progress bar """assertlen(funcs)==len(params)objs=self.panel.objview.get_sel_objects(include_groups=True)grps=self.panel.objview.get_sel_groups()new_gids={}withcreate_progress_bar(self.panel,title,max_=len(objs)*len(params))asprogress:fori_row,objinenumerate(objs):fori_param,(param,func)inenumerate(zip(params,funcs)):name=func.__name__.replace("compute_","")i_title=f"{title} ({i_row+1}/{len(objs)})"progress.setLabelText(i_title)pvalue=(i_row+1)*(i_param+1)pvalue=0ifpvalue==1elsepvalueprogress.setValue(pvalue)args=(obj,)ifparamisNoneelse(obj,param)result=self.__exec_func(func,args,progress)ifresultisNone:breaknew_obj=self.handle_output(result,_("Computing: %s")%i_title,progress)ifnew_objisNone:continue# Is new object a native object (i.e. a Signal object for a Signal# Panel, or an Image object for an Image Panel) ?# (example of non-native object use case: image profile extraction)is_new_obj_native=isinstance(new_obj,self.panel.PARAMCLASS)new_gid=Noneifgrpsandis_new_obj_native:# If groups are selected, then it means that there is no# individual object selected: we work on groups onlyold_gid=self.panel.objmodel.get_object_group_id(obj)new_gid=new_gids.get(old_gid)ifnew_gidisNone:# Create a new group for each selected groupold_g=self.panel.objmodel.get_group(old_gid)new_g=self.panel.add_group(f"{name}({old_g.short_id})")new_gids[old_gid]=new_gid=new_g.uuidifis_new_obj_native:self.panel.add_object(new_obj,group_id=new_gid)else:self.panel.mainwindow.add_object(new_obj)# Select newly created groups, if anyforgroup_idinnew_gids.values():self.panel.objview.set_current_item_id(group_id,extend=True)
[docs]defcompute_10(self,func:Callable,param:gds.DataSet|None=None,paramclass:gds.DataSet|None=None,title:str|None=None,comment:str|None=None,edit:bool|None=None,)->dict[str,ResultShape|ResultProperties]:"""Compute 10 function: 1 object in → 0 object out (the result of this method is stored in original object's metadata). Args: func: function to execute param: parameters. Defaults to None. paramclass: parameters class. Defaults to None. title: title of progress bar. Defaults to None. comment: comment. Defaults to None. edit: if True, edit parameters. Defaults to None. Returns: Dictionary of results (keys: object uuid, values: ResultShape or ResultProperties objects) """if(editisNoneorparamisNone)andparamclassisnotNone:edit,param=self.init_param(param,paramclass,title,comment)ifparamisnotNone:ifeditandnotparam.edit(parent=self.panel.parent()):returnNoneobjs=self.panel.objview.get_sel_objects(include_groups=True)current_obj=self.panel.objview.get_current_object()title=func.__name__.replace("compute_","")iftitleisNoneelsetitlewithcreate_progress_bar(self.panel,title,max_=len(objs))asprogress:results:dict[str,ResultShape|ResultProperties]={}xlabels=Noneylabels=[]foridx,objinenumerate(objs):pvalue=idx+1pvalue=0ifpvalue==1elsepvalueprogress.setValue(pvalue)args=(obj,)ifparamisNoneelse(obj,param)# Execute functioncompout=self.__exec_func(func,args,progress)ifcompoutisNone:breakresult=self.handle_output(compout,_("Computing: %s")%title,progress)ifresultisNone:continue# Add result shape to object's metadataresult.add_to(obj)ifparamisnotNone:obj.metadata[f"{result.title}Param"]=str(param)results[obj.uuid]=resultxlabels=result.headersifobjiscurrent_obj:self.panel.selection_changed(update_items=True)else:self.panel.SIG_REFRESH_PLOT.emit(obj.uuid,True)fori_row_resinrange(result.array.shape[0]):ylabel=f"{result.title}({obj.short_id})"i_roi=int(result.array[i_row_res,0])ifi_roi>=0:ylabel+=f"|ROI{i_roi}"ylabels.append(ylabel)ifresults:withwarnings.catch_warnings():warnings.simplefilter("ignore",RuntimeWarning)dlg=ArrayEditor(self.panel.parent())title=_("Results")res=np.vstack([result.shown_arrayforresultinresults.values()])dlg.setup_and_check(res,title,readonly=True,xlabels=xlabels,ylabels=ylabels)dlg.setObjectName(f"{objs[0].PREFIX}_results")dlg.resize(750,300)exec_dialog(dlg)returnresults
def__get_src_grps_gids_objs_nbobj_valid(self)->tuple[list,list,dict,int]:"""Get source groups, group ids, objects, and number of objects, for pairwise mode, and check if the number of objects is valid. Returns: Tuple (source groups, group ids, objects, number of objects, valid) """# In pairwise mode, we need to create a new object for each pair of objectsobjs=self.panel.objview.get_sel_objects(include_groups=True)objmodel=self.panel.objmodelsrc_grps=sorted({objmodel.get_group_from_object(obj)forobjinobjs},key=lambdax:x.number,)src_gids=[grp.uuidforgrpinsrc_grps]# [src_objs dictionary] keys: old group id, values: list of old objectssrc_objs:dict[str,list[Obj]]={}forsrc_gidinsrc_gids:src_objs[src_gid]=[objforobjinobjsifobjmodel.get_object_group_id(obj)==src_gid]nbobj=len(src_objs[src_gids[0]])valid=len(src_grps)>1ifnotvalid:# In pairwise mode, we need selected objects in at least two groups.ifenv.execenv.unattended:raiseValueError("Pairwise mode: objects must be selected in at least two groups")QW.QMessageBox.warning(self.panel.parent(),_("Warning"),_("In pairwise mode, you need to select objects ""in at least two groups."),)ifvalid:valid=all(len(src_objs[src_gid])==nbobjforsrc_gidinsrc_gids)ifnotvalid:ifenv.execenv.unattended:raiseValueError("Pairwise mode: invalid number of objects in each group")QW.QMessageBox.warning(self.panel.parent(),_("Warning"),_("In pairwise mode, you need to select ""the same number of objects in each group."),)returnsrc_grps,src_gids,src_objs,nbobj,valid
[docs]defcompute_n1(self,name:str,func:Callable,param:gds.DataSet|None=None,paramclass:gds.DataSet|None=None,title:str|None=None,comment:str|None=None,func_objs:Callable|None=None,edit:bool|None=None,)->None:"""Compute n1 function: N(>=2) objects in → 1 object out. Args: name: name of function func: function to execute param: parameters. Defaults to None. paramclass: parameters class. Defaults to None. title: title of progress bar. Defaults to None. comment: comment. Defaults to None. func_objs: function to execute on objects. Defaults to None. edit: if True, edit parameters. Defaults to None. """if(editisNoneorparamisNone)andparamclassisnotNone:edit,param=self.init_param(param,paramclass,title,comment)ifparamisnotNone:ifeditandnotparam.edit(parent=self.panel.parent()):returnobjs=self.panel.objview.get_sel_objects(include_groups=True)objmodel=self.panel.objmodelpairwise=is_pairwise_mode()ifpairwise:src_grps,src_gids,src_objs,_nbobj,valid=(self.__get_src_grps_gids_objs_nbobj_valid())ifnotvalid:returndst_gname=(f"{name}({','.join([grp.short_idforgrpinsrc_grps])})|pairwise")group_exclusive=len(self.panel.objview.get_sel_groups())!=0ifnotgroup_exclusive:# This is not a group exclusive selectiondst_gname+="[...]"dst_gid=self.panel.add_group(dst_gname).uuidn_pairs=len(src_objs[src_gids[0]])max_i_pair=min(n_pairs,max(len(src_objs[grp.uuid])forgrpinsrc_grps))withcreate_progress_bar(self.panel,title,max_=n_pairs)asprogress:fori_pair,src_obj1inenumerate(src_objs[src_gids[0]][:max_i_pair]):src_obj1:SignalObj|ImageObjprogress.setValue(i_pair+1)progress.setLabelText(title)src_dtype=src_obj1.data.dtypedst_dtype=complexifis_complex_dtype(src_dtype)elsefloatdst_obj=src_obj1.copy(dtype=dst_dtype)src_objs_pair=[src_obj1]forsrc_gidinsrc_gids[1:]:src_obj=src_objs[src_gid][i_pair]src_objs_pair.append(src_obj)ifparamisNone:args=(dst_obj,src_obj)else:args=(dst_obj,src_obj,param)result=self.__exec_func(func,args,progress)ifresultisNone:breakdst_obj=self.handle_output(result,_("Calculating: %s")%title,progress)ifdst_objisNone:breakdst_obj.update_resultshapes_from(src_obj)ifsrc_obj.roiisnotNone:ifdst_obj.roiisNone:dst_obj.roi=src_obj.roi.copy()else:dst_obj.roi.add_roi(src_obj.roi)iffunc_objsisnotNone:func_objs(dst_obj,src_objs_pair)short_ids=[obj.short_idforobjinsrc_objs_pair]dst_obj.title=f'{name}({", ".join(short_ids)})'self.panel.add_object(dst_obj,group_id=dst_gid)else:# In single operand mode, we create a single object for all selected objects# [new_objs dictionary] keys: old group id, values: new objectdst_objs:dict[str,Obj]={}# [src_dtypes dictionary] keys: old group id, values: old data typesrc_dtypes:dict[str,np.dtype]={}# [src_objs dictionary] keys: old group id, values: list of old objectssrc_objs:dict[str,list[Obj]]={}withcreate_progress_bar(self.panel,title,max_=len(objs))asprogress:forindex,src_objinenumerate(objs):progress.setValue(index+1)progress.setLabelText(title)src_gid=objmodel.get_object_group_id(src_obj)dst_obj=dst_objs.get(src_gid)ifdst_objisNone:src_dtypes[src_gid]=src_dtype=src_obj.data.dtypedst_dtype=complexifis_complex_dtype(src_dtype)elsefloatdst_objs[src_gid]=dst_obj=src_obj.copy(dtype=dst_dtype)dst_obj.roi=Nonesrc_objs[src_gid]=[src_obj]else:src_objs[src_gid].append(src_obj)ifparamisNone:args=(dst_obj,src_obj)else:args=(dst_obj,src_obj,param)result=self.__exec_func(func,args,progress)ifresultisNone:breakdst_obj=self.handle_output(result,_("Calculating: %s")%title,progress)ifdst_objisNone:breakdst_objs[src_gid]=dst_objdst_obj.update_resultshapes_from(src_obj)ifsrc_obj.roiisnotNone:ifdst_obj.roiisNone:dst_obj.roi=src_obj.roi.copy()else:dst_obj.roi.add_roi(src_obj.roi)grps=self.panel.objview.get_sel_groups()ifgrps:# (Group exclusive selection)# At least one group is selected: create a new groupdst_gname=f"{name}({','.join([grp.short_idforgrpingrps])})"dst_gid=self.panel.add_group(dst_gname).uuidelse:# (Object exclusive selection)# No group is selected: use each object's groupdst_gid=Noneforsrc_gid,dst_objindst_objs.items():iffunc_objsisnotNone:func_objs(dst_obj,src_objs[src_gid])short_ids=[obj.short_idforobjinsrc_objs[src_gid]]dst_obj.title=f'{name}({", ".join(short_ids)})'group_id=dst_gidifdst_gidisnotNoneelsesrc_gidself.panel.add_object(dst_obj,group_id=group_id)# Select newly created group, if anyifdst_gidisnotNone:self.panel.objview.set_current_item_id(dst_gid)
[docs]defcompute_n1n(self,obj2:Obj|list[Obj]|None,obj2_name:str,func:Callable,param:gds.DataSet|None=None,paramclass:gds.DataSet|None=None,title:str|None=None,comment:str|None=None,edit:bool|None=None,)->None:"""Compute n1n function: N(>=1) objects + 1 object in → N objects out. Examples: subtract, divide Args: obj2: second object (or list of objects in case of pairwise operation mode) obj2_name: name of second object func: function to execute param: parameters. Defaults to None. paramclass: parameters class. Defaults to None. title: title of progress bar. Defaults to None. comment: comment. Defaults to None. edit: if True, edit parameters. Defaults to None. """if(editisNoneorparamisNone)andparamclassisnotNone:edit,param=self.init_param(param,paramclass,title,comment)objs=self.panel.objview.get_sel_objects(include_groups=True)objmodel=self.panel.objmodelpairwise=is_pairwise_mode()ifobj2isNone:objs2=[]elifisinstance(obj2,list):objs2=obj2assertpairwiseelse:objs2=[obj2]dlg_title=_("Select %s")%obj2_nameifpairwise:group_exclusive=len(self.panel.objview.get_sel_groups())!=0src_grps,src_gids,src_objs,nbobj,valid=(self.__get_src_grps_gids_objs_nbobj_valid())ifnotvalid:returnifnotobjs2:objs2=self.panel.get_objects_with_dialog(dlg_title,_("<u>Note:</u> operation mode is <i>pairwise</i>: ""%s object(s) expected (i.e. as many as in the first group)")%nbobj,nbobj,)ifobjs2isNone:returnname=func.__name__.replace("compute_","")n_pairs=len(src_objs[src_gids[0]])max_i_pair=min(n_pairs,max(len(src_objs[grp.uuid])forgrpinsrc_grps))grp2_id=objmodel.get_object_group_id(objs2[0])grp2=objmodel.get_group(grp2_id)withcreate_progress_bar(self.panel,title,max_=len(src_gids))asprogress:fori_group,src_gidinenumerate(src_gids):progress.setValue(i_group+1)progress.setLabelText(title)ifgroup_exclusive:# This is a group exclusive selectionsrc_grp=objmodel.get_group(src_gid)grp_short_ids=[grp.short_idforgrpin(src_grp,grp2)]dst_gname=f"{name}({','.join(grp_short_ids)})|pairwise"else:dst_gname=f"{name}[...]"dst_gid=self.panel.add_group(dst_gname).uuidfori_pairinrange(max_i_pair):args=[src_objs[src_gid][i_pair],objs2[i_pair]]ifparamisnotNone:args.append(param)result=self.__exec_func(func,tuple(args),progress)ifresultisNone:breaknew_obj=self.handle_output(result,_("Calculating: %s")%title,progress)ifnew_objisNone:continueself.panel.add_object(new_obj,group_id=dst_gid)else:ifnotobjs2:objs2=self.panel.get_objects_with_dialog(dlg_title,_("<u>Note:</u> operation mode is <i>single operand</i>: ""1 object expected"),)ifobjs2isNone:returnobj2=objs2[0]withcreate_progress_bar(self.panel,title,max_=len(objs))asprogress:forindex,objinenumerate(objs):progress.setValue(index+1)progress.setLabelText(title)args=(obj,obj2)ifparamisNoneelse(obj,obj2,param)result=self.__exec_func(func,args,progress)ifresultisNone:breaknew_obj=self.handle_output(result,_("Calculating: %s")%title,progress)ifnew_objisNone:continuegroup_id=objmodel.get_object_group_id(obj)self.panel.add_object(new_obj,group_id=group_id)
[docs]@abc.abstractmethod@qt_try_except()defcompute_addition_constant(self,param:ConstantParam)->None:"""Compute sum with a constant"""
[docs]@abc.abstractmethod@qt_try_except()defcompute_difference_constant(self,param:ConstantParam)->None:"""Compute difference with a constant"""
[docs]@abc.abstractmethod@qt_try_except()defcompute_product_constant(self,param:ConstantParam)->None:"""Compute product with a constant"""
[docs]@abc.abstractmethod@qt_try_except()defcompute_division_constant(self,param:ConstantParam)->None:"""Compute division by a constant"""
[docs]@qt_try_except()defcompute_roi_extraction(self,roi:TypeROI|None=None)->None:"""Extract Region Of Interest (ROI) from data with: - :py:func:`cdl.computation.image.extract_single_roi` for single ROI - :py:func:`cdl.computation.image.extract_multiple_roi` for multiple ROIs"""# Expected behavior:# -----------------# * If `roi` is not None or not empty, skip the ROI dialog# * If first selected obj has a ROI, use this ROI as default but open# ROI Editor dialog anyway# * If multiple objs are selected, then apply the first obj ROI to allifroiisNoneorroi.is_empty():roi=self.edit_regions_of_interest(extract=True)ifroiisNoneorroi.is_empty():returnobj=self.panel.objview.get_sel_objects(include_groups=True)[0]group=roi.to_params(obj)ifroi.singleobjandlen(group.datasets)>1:# Extract multiple ROIs into a single object (remove all the ROIs),# if the "Extract all ROIs into a single image object"# option is checked and if there are more than one ROIself._extract_multiple_roi_in_single_object(group)else:# Extract each ROI into a separate object (keep the ROI in the case of# a circular ROI), if the "Extract all ROIs into a single image object"# option is not checked or if there is only one ROI (See Issue #31)self._extract_each_roi_in_separate_object(group)
@abc.abstractmethod@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"""@abc.abstractmethod@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)"""# ------Analysis-------------------------------------------------------------------
[docs]defedit_regions_of_interest(self,extract:bool=False,)->TypeROI|None:"""Define Region Of Interest (ROI). Args: extract: If True, ROI is extracted from data. Defaults to False. Returns: ROI object or None if ROI dialog has been canceled. """# Expected behavior:# -----------------# * If first selected obj has a ROI, use this ROI as default but open# ROI Editor dialog anyway# * If multiple objs are selected, then apply the first obj ROI to allresults=self.panel.get_roi_editor_output(extract=extract)ifresultsisNone:returnNoneedited_roi,modified=resultsobj=self.panel.objview.get_sel_objects(include_groups=True)[0]group=edited_roi.to_params(obj)if(env.execenv.unattended# Unattended mode (automated unit tests)oredited_roi.is_empty()# No ROI has been definedorgroup.edit(parent=self.panel.parent())# ROI dialog has been accepted):ifmodified:# If ROI has been modified, save ROI (even in "extract mode")ifedited_roi.is_empty():obj.roi=Noneelse:edited_roi=edited_roi.from_params(obj,group)obj.roi=edited_roiself.SIG_ADD_SHAPE.emit(obj.uuid)self.panel.selection_changed(update_items=True)returnedited_roi
[docs]defdelete_regions_of_interest(self)->None:"""Delete Regions Of Interest"""forobjinself.panel.objview.get_sel_objects():ifobj.roiisnotNone:obj.roi=Noneself.panel.selection_changed(update_items=True)
[docs]@abc.abstractmethod@qt_try_except()defcompute_stats(self)->dict[str,ResultShape]:"""Compute data statistics"""