[docs]defacquire_dataset(self,data):""" Acquires the dataset. """# data.sequence = self.cur_expdata.attrs['time']=datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')returndata
[docs]deflaunch(self,sequence,savename:str):"""Launches the experiment and initialises autosaving. Parameters ---------- sequence : Sequence The sequence to be launched savename : str The savename for this measurement. A timestamp will be added to the value. """timestamp=datetime.datetime.now().strftime(r'%Y%m%d_%H%M_')self.savename=timestamp+savename+'.h5'pass
[docs]defterminate(self)->None:""" Terminates the experiment immediately. """pass
[docs]defterminate_at(self,criterion,test_interval=2,keep_running=True,verbosity=0,autosave=True):"""Terminates the experiment upon a specific condition being satisified. Parameters ---------- criterion : _type_ The criteria to be tested. test_interval : int, optional How often should the criteria be tested in minutes, by default 10. keep_running : bool, optional If True, an error will not be raised if the experiment finishes before the criteria is met, by default True. verbosity : int, optional The verbosity level, by default 0. autosave : bool, optional If True, the data will be autosaved, by default True. """test_interval_seconds=test_interval*60condition=Falselast_scan=0whilenotcondition:ifnotself.isrunning():ifkeep_running:self.terminate()returnNoneelse:msg="Experiments has finished before criteria met."raiseRuntimeError(msg)start_time=time.time()data=self.acquire_dataset()ifautosave:self.log.debug(f"Autosaving to {os.path.join(self.savefolder,self.savename)}")data.to_netcdf(os.path.join(self.savefolder,self.savename),engine='h5netcdf',invalid_netcdf=True)try:# nAvgs = data.num_scans.valuenAvgs=data.attrs['nAvgs']exceptAttributeErrororKeyError:self.log.warning("WARNING: Dataset missing number of averages(nAvgs)!")nAvgs=1finally:ifnAvgs<1:time.sleep(30)# Replace with single scan timecontinueelifnAvgs<=last_scan:time.sleep(30)continuelast_scan=nAvgsifverbosity>0:print("Testing")ifisinstance(criterion,list):conditions=[crit.test(data,verbosity)forcritincriterion]condition=any(conditions)else:condition=criterion.test(data,verbosity)ifnotcondition:end_time=time.time()if(end_time-start_time)<test_interval_seconds:ifverbosity>0:print("Sleeping")time.sleep(test_interval_seconds-(end_time-start_time))ifisinstance(criterion,list):fori,critinenumerate(criterion):ifconditions[i]:ifcallable(crit.end_signal):crit.end_signal()else:ifcallable(criterion.end_signal):criterion.end_signal()self.terminate()pass
[docs]classParameter:""" Represents a sequence or pulse parameter. """def__init__(self,name,value,unit="",description="",virtual=False,**kwargs)->None:"""A general parameter. Parameters ---------- name : str The parameter name value : float or int The parameter value, eithe initial or static unit : str, optional The unit of parameter, by default None. Leave as None if unitless. description : str, optional A brief description of the parameter, by default None axis : np.ndarray, optional The difference from the intial value for each position in a dynamic axis. Can be n-dimensional, by default None. ax_id : list, optional virtual: bool, optional A virtual paramter is only used to vary other parameters, it is not varied itself and will not be directly passed to a spectrometer. This parameter is **never** inherited. By default, False Attributes ---------- progressive : bool Is the parameter used in any progression or is it constant prog : dict A dict containing progressive programs for this parameter. This list has two elements. 1) The axis_id"s and 2) the "axis" of values. Parameter Arthimatic -------------------- Examples -------- Creating a static parameter ``` Par1 = Parameter( name="Par1", value=10, unit="us", description="The first parameter") ``` Creating a dynamic parameter ``` Par1 = Parameter( name="Par1", value=10, unit="us", description="The first parameter", axis=np.arange(0,10,1), axis_id=0) ``` Adding a parameter and a number: ``` Par1 = Parameter( name="Par1", value=10, unit="us", description="The first parameter") Par2 = Par1 + 2 """
if"link"inkwargs:ifnotisinstance(kwargs["link"],Parameter):raiseValueError("The linked parameter must be a Parmater object")self.uuid=kwargs["link"].uuidelse:self.uuid=uuid.uuid1()if"step"inkwargs:step=kwargs["step"]dim=kwargs["dim"]if"axis_id"inkwargs:axis_id=kwargs["axis_id"]else:axis_id=0if"start"inkwargs:start=kwargs["start"]else:start=0ifstep==0:axis=np.zeros(dim)else:axis=np.arange(start=start,stop=dim*step+start,step=step)self.add_axis(axis=axis,axis_id=axis_id)pass
[docs]defadd_axis(self,axis_id,axis):# if self.axis == []:# self.axis.append(np.array(axis))# self.ax_id.append(axis_id)self.axis.append({"axis":axis,"uuid":self.uuid})
[docs]def__eq__(self,__o:object)->bool:iftype(__o)isnotParameter:raiseValueError("Equivalence only works between Parameter classes")returnself.value==__o.value
[docs]def__add__(self,__o:object):iftype(__o)isParameter:ifself.unit!=__o.unit:raiseRuntimeError("Both parameters must have the same unit")new_value=self.value+__o.valuenew_name=f"{self.name} + {__o.name}"new_description=new_namenew_parameter=Parameter(name=new_name,value=new_value,unit=self.unit,description=new_description)ifnotself.is_static():ifnot__o.is_static():# Dynamic parmaters can only be summed and multiplied if the axis has the same uuid. I.e. they were linked when created or are deriratives of each other. new_ax_id=[]new_axis=[]# a_ax_ids:list = self.ax_ida_ax_ids:list=[self.axis[i]["uuid"]foriinrange(len(self.axis))]# b_ax_ids:list = __o.ax_idb_ax_ids:list=[__o.axis[i]["uuid"]foriinrange(len(__o.axis))]ab_ax_ids=list(set(a_ax_ids+b_ax_ids))foridinab_ax_ids:ifidnotinb_ax_ids:# I.e. only in Aa_index=a_ax_ids.index(id)new_axis.append(self.axis[a_index])new_ax_id.append(id)elifidnotina_ax_ids:# I.e. only in Bb_index=b_ax_ids.index(id)new_axis.append(__o.axis[b_index])new_ax_id.append(id)else:# in botha_index=a_ax_ids.index(id)b_index=b_ax_ids.index(id)b_ax_ids.remove(id)new_axis.append({"axis":self.axis[a_index]["axis"]+__o.axis[b_index]["axis"],"uuid":id})new_ax_id.append(id)else:new_axis=self.axisnew_ax_id=self.ax_idelse:ifnot__o.is_static():new_axis=__o.axisnew_ax_id=__o.ax_idelse:new_axis=[]new_ax_id=[]new_parameter.axis=new_axisnew_parameter.ax_id=new_ax_idreturnnew_parameterelifisinstance(__o,numbers.Number):new_value=self.value+__onew_name=f"{self.name} + {__o}"new_parameter=Parameter(name=new_name,value=new_value,unit=self.unit)ifnotself.is_static():new_axis=self.axisnew_ax_id=self.ax_idnew_parameter.axis=new_axisnew_parameter.ax_id=new_ax_idreturnnew_parameterelifisinstance(__o,np.ndarray):ifself.axis.shape!=__o.shape:raiseRuntimeError("Both parameters axis and the array must have the same shape")
[docs]def__sub__(self,__o:object):iftype(__o)isParameter:ifself.unit!=__o.unit:raiseRuntimeError("Both parameters must have the same unit")new_value=self.value-__o.valuenew_name=f"{self.name} - {__o.name}"new_parameter=Parameter(name=new_name,value=new_value,unit=self.unit)ifnotself.is_static():ifnot__o.is_static():# Dynamic parmaters can only be summed and multiplied if the axis has the same uuid. I.e. they were linked when created or are deriratives of each other. new_ax_id=[]new_axis=[]# a_ax_ids:list = self.ax_ida_ax_ids:list=[self.axis[i]["uuid"]foriinrange(len(self.axis))]# b_ax_ids:list = __o.ax_idb_ax_ids:list=[__o.axis[i]["uuid"]foriinrange(len(self.__o))]ab_ax_ids=list(set(a_ax_ids+b_ax_ids))foridinab_ax_ids:ifidnotinb_ax_ids:# I.e. only in Aa_index=a_ax_ids.index(id)new_axis.append({"axis":self.axis[a_index],"uuid":self.uuid})new_ax_id.append(id)elifidnotina_ax_ids:# I.e. only in Bb_index=b_ax_ids.index(id)new_axis.append({"axis":__o.axis[b_index],"uuid":__o.uuid})new_ax_id.append(id)else:# in botha_index=a_ax_ids.index(id)b_index=b_ax_ids.index(id)b_ax_ids.remove(id)new_axis.append({"axis":self.axis[a_index]-__o.axis[b_index],"uuid":id})new_ax_id.append(id)else:new_axis=self.axisnew_ax_id=self.ax_idelse:ifnot__o.is_static():new_axis=__o.axisnew_ax_id=__o.ax_idelse:new_axis=[]new_ax_id=[]new_parameter.axis=new_axisnew_parameter.ax_id=new_ax_idreturnnew_parameterelifisinstance(__o,numbers.Number):new_value=self.value-__onew_name=f"{self.name} - {__o}"new_parameter=Parameter(name=new_name,value=new_value,unit=self.unit)ifself.axisisnotNone:new_axis=self.axisnew_ax_id=self.ax_idnew_parameter.axis=new_axisnew_parameter.ax_id=new_ax_idreturnnew_parameterelifisinstance(__o,np.ndarray):ifself.axis.shape!=__o.shape:raiseRuntimeError("Both parameters axis and the array must have the same shape")
[docs]def__mul__(self,__o:object):iftype(__o)isParameter:ifself.unit!=__o.unit:raiseRuntimeError("Both parameters must have the same unit")# if not __o.is_static():# raise RuntimeError("Multiplictaion of two dynamic parameters is not supported")new_value=self.value*__o.valuenew_name=f"{self.name} * {__o.name}"new_parameter=Parameter(name=new_name,value=new_value,unit=self.unit)# if self.axis is not None:# new_axis = [np.array([item * __o.value for item in axis]) for axis in self.axis ]# new_ax_id = self.ax_id# new_parameter.axis = new_axis# new_parameter.ax_id = new_ax_id# return new_parameterifnotself.is_static():ifnot__o.is_static():# Dynamic parmaters can only be summed and multiplied if the axis has the same uuid. I.e. they were linked when created or are deriratives of each other. new_ax_id=[]new_axis=[]# a_ax_ids:list = self.ax_ida_ax_ids:list=[self.axis[i]["uuid"]foriinrange(len(self.axis))]# b_ax_ids:list = __o.ax_idb_ax_ids:list=[__o.axis[i]["uuid"]foriinrange(len(self.__o))]ab_ax_ids=list(set(a_ax_ids+b_ax_ids))foridinab_ax_ids:ifidnotinb_ax_ids:# I.e. only in Aa_index=a_ax_ids.index(id)new_axis.append({"axis":self.axis[a_index],"uuid":self.uuid})new_ax_id.append(id)elifidnotina_ax_ids:# I.e. only in Bb_index=b_ax_ids.index(id)new_axis.append({"axis":__o.axis[b_index],"uuid":__o.uuid})new_ax_id.append(id)else:# in botha_index=a_ax_ids.index(id)b_index=b_ax_ids.index(id)b_ax_ids.remove(id)new_axis.append({"axis":self.axis[a_index]*__o.axis[b_index],"uuid":id})new_ax_id.append(id)else:new_axis=self.axisnew_ax_id=self.ax_idelse:ifnot__o.is_static():new_axis=__o.axisnew_ax_id=__o.ax_idelse:new_axis=[]new_ax_id=[]new_parameter.axis=new_axisnew_parameter.ax_id=new_ax_idreturnnew_parameterelifisinstance(__o,numbers.Number):new_value=self.value*__onew_name=f"{self.name} + {__o}"new_parameter=Parameter(name=new_name,value=new_value,unit=self.unit)ifself.axisisnot[]:new_axis=copy.deepcopy(self.axis)fori,axisinenumerate(new_axis):new_axis[i]["axis"]=np.array(axis["axis"])*__o# new_axis = [np.array([item * __o for item in axis]) for axis in self.axis ]new_ax_id=self.ax_idnew_parameter.axis=new_axisnew_parameter.ax_id=new_ax_idreturnnew_parameterelifisinstance(__o,np.ndarray):ifself.axis.shape!=__o.shape:raiseRuntimeError("Both parameters axis and the array must have the same shape")
[docs]defsave(self,filename):"""Save the parameter to a JSON file. Parameters ---------- filename : str Path to the JSON file. Returns ------- None Raises ------ TypeError If the object cannot be serialized to JSON. Example ------- >>> obj = Parameter() >>> obj.save("my_parameter.json") """withopen(filename,"w")asf:f.write(self._to_json())
[docs]defload(cls,filename):"""Load a Parameter object from a JSON file. Parameters ---------- filename : str Path to the JSON file. Returns ------- obj : Parameter The Pulse loaded from the JSON file. Raises ------ FileNotFoundError If the file does not exist. Example ------- >>> obj = Parameter.load("my_parameter.json") """withopen(filename,"r")asf:file_buffer=f.read()returncls._from_json(file_buffer)