linse/frappy
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

safety

Browse Source
This commit is contained in:
last_d
2025-08-05 10:33:59 +02:00
parent 1777e4b7b1
commit b3d243d831
6 changed files with 94 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
frappy_psi/tnmr/OTFModule.py
View File

@@ -27,7 +27,7 @@ import time
import os
import traceback
class ProgrammedSequence(fc.Readable):
class ProgrammedSequence(fc.Drivable): # Drivable only for kill() funcitonality
"""An NMR device being driven by an instance of TNMR. Requires that an instance of TNMR is opened before creation.
Use
@@ -49,7 +49,7 @@ class ProgrammedSequence(fc.Readable):
pre_acquisition_time: float (usecs) which describes the length of time to wait after ringdown finishes (1u is okay)
post_acquisition_time: float (ms) which describes the length of time to wait after finishing acquisition
acq_phase_cycle: str, the phase cycle to run on acquisition (eg., '0 1 1 2', '0 1 2 3', '1 1 2 2 0 0 3 3 1 2 3 4', ...)
num_scans: int (ct), the number of 1D scans to take per sequence
num_acqs: int (ct), the number of 1D scans to take per sequence
obs_freq: float (MHz), the NMR frequency
Commands
@@ -72,7 +72,12 @@ class ProgrammedSequence(fc.Readable):
imags=fc.ArrayOf(fc.FloatRange(), maxlen=4096), # imag values
t =fc.ArrayOf(fc.FloatRange(), maxlen=4096)), # times (starting from zero)
default={ 'reals': [], 'imags': [], 't': [] })
status = fc.Parameter(datatype=frappy.datatypes.StatusType(fc.Readable, "DISABLED", 'PREPARED', 'BUSY'), default=('IDLE', 'ok - uncompiled'))
target = fc.Parameter('dummy', fc.StructOf(reals=fc.ArrayOf(fc.FloatRange(), maxlen=4096), # real values
imags=fc.ArrayOf(fc.FloatRange(), maxlen=4096), # imag values
t =fc.ArrayOf(fc.FloatRange(), maxlen=4096)), # times (starting from zero)
default={ 'reals': [], 'imags': [], 't': [] }, readonly=True, visibility='w--')
status = fc.Parameter(datatype=frappy.datatypes.StatusType(fc.Drivable, "DISABLED", 'PREPARED', 'BUSY'), default=('IDLE', 'ok - uncompiled'))
pollinterval = fc.Parameter(default=1)
# basic
@@ -81,10 +86,12 @@ class ProgrammedSequence(fc.Readable):
comments = fc.Parameter('comments', fc.StringType(), default='', readonly=False)
nucleus = fc.Parameter('nucleus', fc.StringType(), default='', readonly=False)
sequence_length = fc.Parameter('sequence_length', fc.IntRange(), default=0, readonly=True)
sequence_data = fc.Parameter('sequence_config', fc.ArrayOf(fc.StructOf(pulse_width=fc.FloatRange(unit='usecs'),
pulse_height=fc.FloatRange(unit='%'),
delay_time=fc.FloatRange(unit='usecs'),
phase_cycle=fc.StringType())), default=[], readonly=False)
phase_cycle=fc.StringType()), minlen=0), default=[{'pulse_width':0,'pulse_height':0,'delay_time':0,'phase_cycle':''}]*100, readonly=False)
num_acqs_actual = fc.Parameter('num_acqs', fc.IntRange(), readonly=True, default=0)
# final details
acquisition_time = fc.Parameter('acquisition_time', fc.FloatRange(unit='usecs'), readonly=False, group='sequence_editor', default=204.8) # this is a limit set by the dwell limit and number of acquisition points
@@ -92,11 +99,12 @@ class ProgrammedSequence(fc.Readable):
pre_acquisition_time = fc.Parameter('pre_acquisition_time', fc.FloatRange(unit='usecs'), readonly=False, group='sequence_editor', default=1)
post_acquisition_time = fc.Parameter('post_acquisition_time', fc.FloatRange(unit='msecs'), readonly=False, group='sequence_editor', default=500)
acq_phase_cycle = fc.Parameter('acq_phase_cycle', fc.StringType(), readonly=False, group='sequence_editor', default='')
num_scans = fc.Parameter('num_scans', fc.IntRange(), readonly=False, group='sequence_editor', default=16)
num_acqs = fc.Parameter('num_acqs', fc.IntRange(), readonly=False, group='sequence_editor', default=16)
obs_freq = fc.Parameter('obs_freq', fc.FloatRange(unit='MHz'), readonly=False, group='sequence_editor', default=213.16)
compiled_parameters = {} # so that we can store the values of parameters only when compiling, effectively giving us an instance of each parameter loaded into TNMR, as well as "targets" (those above)
inited = False
starting = False
approx_sequence_length = 0
### SETUP
@@ -148,6 +156,20 @@ class ProgrammedSequence(fc.Readable):
### READ/WRITE
def read_status(self):
if not(self.inited):
self.status = ('ERROR', 'TNMR disconnected!')
else:
if(self.starting):
self.status = ('BUSY', 'starting')
else:
if(self.tnmr().acquisition_running()):
self.status = ('BUSY', 'acquiring')
elif(self.status[1] == 'acquiring'):
# we've just finished acquiring, in frappy's perspective
self.status = ('PREPARED', 'compiled')
return self.status
def write_title(self, t):
self.title = t
self.status = ('IDLE', 'ok - uncompiled')
@@ -193,14 +215,14 @@ class ProgrammedSequence(fc.Readable):
self.status = ('IDLE', 'ok - uncompiled')
return self.read_acq_phase_cycle()
def read_num_scans(self):
return self.tnmr().get_nmrparameter('Scans 1D')
def read_num_acqs(self):
return int(self.tnmr().get_nmrparameter('Scans 1D'))
def write_num_scans(self, t):
def write_num_acqs(self, t):
if(self.status[0] != 'BUSY'):
self.tnmr().set_nmrparameter('Scans 1D', t)
self.status = ('IDLE', 'ok - uncompiled')
return self.read_num_scans()
return self.read_num_acqs()
def read_obs_freq(self):
return self.tnmr().get_nmrparameter('Observe Freq.')
@@ -210,6 +232,36 @@ class ProgrammedSequence(fc.Readable):
self.tnmr().set_nmrparameter('Observe Freq.', t)
self.status = ('IDLE', 'ok - uncompiled')
return self.read_obs_freq()
def write_sequence_data(self, t):
self.sequence_length = len(t)
seq = []
seq += t
print(seq)
seq += [{'pulse_width':0,'pulse_height':0,'delay_time':0,'phase_cycle':''}] * (100-self.sequence_length) # because nicos will only send the smallest size it has ever sent...
self.sequence_data = seq
return self.read_sequence_data()
def read_value(self):
newvals = {}
#try:
d = self.tnmr().get_data()
newvals['reals'] = d[0]
newvals['imags'] = d[1]
newvals['t'] = [ self.compiled_parameters['acquisition_time'] * i/len(d[0]) for i in range(0, len(d[0])) ]
#except:
# newvals['reals'] = []
# newvals['imags'] = []
# newvals['t'] = []
return newvals
def read_num_acqs_actual(self):
try:
n = self.tnmr().get_nmrparameter('Actual Scans 1D')
return int(n)
except:
return 0
### PRIVATE (Utility)
def __compile_sequence(self):
@@ -229,7 +281,8 @@ class ProgrammedSequence(fc.Readable):
seq = seq_gen.get_initial_block()
i = 0
self.approx_sequence_length = 0
for s in self.sequence_data:
for si in range(self.sequence_length):
s = self.sequence_data[si]
seq = seq_gen.combine_blocks(seq, seq_gen.get_single_pulse_block(f'pulse_{i}', str(s['pulse_width']) + 'u',
str(s['pulse_height']),
str(s['delay_time']) + 'u',
@@ -252,10 +305,9 @@ class ProgrammedSequence(fc.Readable):
filename = filepath + '/sequences/' + filename.replace('.','')
seq_gen.save_sequence(filename, seq)
seq_gen.save_sequence_cfg(filename, seq)
print(filename)
dashboard_params = { 'Observe Freq.': self.read_obs_freq(),
'Scans 1D': self.read_num_scans(),
'Scans 1D': self.read_num_acqs(),
}
self.compiled_parameters['ringdown_time'] = self.ringdown_time
@@ -263,7 +315,7 @@ class ProgrammedSequence(fc.Readable):
self.compiled_parameters['acquisition_time'] = self.acquisition_time
self.compiled_parameters['post_acquisition_time'] = self.post_acquisition_time
self.compiled_parameters['acq_phase_cycle'] = self.acq_phase_cycle
self.compiled_parameters['num_scans'] = self.read_num_scans()
self.compiled_parameters['num_acqs'] = self.read_num_acqs()
self.compiled_parameters['obs_freq'] = self.read_obs_freq()
self.compiled_parameters['title'] = self.read_title()
self.compiled_parameters['comments'] = self.read_comments()
@@ -290,15 +342,8 @@ class ProgrammedSequence(fc.Readable):
def __zero_go(self):
'''Tells TNMR to acquire data. Only call after __compile_sequence().'''
if(self.status[0] != 'BUSY'):
self.status = ('BUSY', 'acquiring')
self.tnmr().ZeroGo(lock=True, interval=0.5, check_time=max(self.approx_sequence_length*5, 5))
newvals = {}
newvals['reals'] = self.tnmr().get_data()[0]
newvals['imags'] = self.tnmr().get_data()[1]
newvals['t'] = [ self.compiled_parameters['acquisition_time'] * i/1024 for i in range(0, 1024) ]
self.value = newvals
self.status = ('PREPARED', 'compiled')
if(self.status[0] != 'BUSY' or self.starting):
self.tnmr().ZeroGo(lock=False, check_time=max(int(self.approx_sequence_length*1.5), 5))
def __compile_and_run(self, thread=True):
'''Compiles and runs the currently-loaded sequence
@@ -307,9 +352,11 @@ class ProgrammedSequence(fc.Readable):
----------
thread: bool, determines if this should open a child thread and detach the process
'''
self.starting = True
self.__compile_sequence()
time.sleep(1.0)
self.__zero_go()
self.starting = False