122 lines
3.9 KiB
Python
122 lines
3.9 KiB
Python
import ch.psi.pshell.epics.Positioner as Positioner
|
|
import ch.psi.pshell.epics.ChannelDouble as ChannelDouble
|
|
import_py("CPython/hfitoff", "hfitoff")
|
|
import_py("CPython/extremum", "extremum")
|
|
|
|
|
|
dry_run = False
|
|
do_elog = False
|
|
|
|
if get_exec_pars().args: # args is set by callin process (Qt panel)
|
|
und = args[0]
|
|
scan = args[1]
|
|
|
|
if dry_run:
|
|
und = "STEST01"
|
|
scan = "AR_PHI"
|
|
|
|
if scan == "AR_K" or scan == "AT_K":
|
|
x_channel = und + "-UIND030:K_SET"
|
|
message_channel = und + ":K-SCAN-MESSAGE"
|
|
x0 = caget(x_channel)
|
|
start = x0 - 0.005
|
|
end = x0 + 0.005
|
|
step = 10
|
|
lat = 0.1
|
|
nb = 90
|
|
|
|
if scan == "AR_PHI" or scan == "AT_PHI":
|
|
message_channel = und + ":PHI-SCAN-MESSAGE"
|
|
start = -180
|
|
end = 180
|
|
step = 12
|
|
lat = 0.1
|
|
nb = 90
|
|
|
|
if scan == "AR_K":
|
|
x_ok_channel = und + "-UPHS060:GAP-POS-OK" # UPHS is slower
|
|
y_channel = "SARFE10-PBIG050-EVR0:CALCI"
|
|
elif scan == "AT_K":
|
|
x_ok_channel = und + "-UIND030:RADIAL-POS-OK"
|
|
y_channel = "SATFE10-PEPG046-EVR0:CALCI"
|
|
elif scan == "AR_PHI":
|
|
x_channel = und + "-UPHS060:PHI_SET"
|
|
x_ok_channel = und + "-UPHS060:GAP-POS-OK"
|
|
y_channel = "SARFE10-PBIG050-EVR0:CALCI"
|
|
elif scan == "AT_PHI":
|
|
x_channel = und + "-UDLY060:PH-SHIFT-OP"
|
|
x_ok_channel = und + "-UDLY060:GAP-COMP-OP"
|
|
y_channel = "SATFE10-PEPG046-EVR0:CALCI"
|
|
|
|
if und == "STEST01" and scan == "AR_K":
|
|
y_channel = "STEST01:K-CALCI"
|
|
if und == "STEST01" and scan == "AR_PHI":
|
|
y_channel = "STEST01:PHI-CALCI"
|
|
|
|
x_val = ChannelDouble(x_channel, x_channel)
|
|
x_val.initialize()
|
|
y_val = ChannelDouble(y_channel, y_channel)
|
|
y_val.initialize()
|
|
x_val0 = x_val.read()
|
|
y_averager = create_averager(y_val, nb, lat)
|
|
|
|
#check mover position
|
|
def before():
|
|
time.sleep(1.0)
|
|
timeout = 10.0
|
|
start_time = time.time()
|
|
while (True):
|
|
mover_status = caget(x_ok_channel)
|
|
print(mover_status)
|
|
if mover_status=="OK" or mover_status=="True":
|
|
break
|
|
if time.time() - start_time > timeout:
|
|
raise Exception("mover timeout")
|
|
|
|
#update the plot dynamically
|
|
arr_x, arr_y = [],[]
|
|
def after(rec):
|
|
arr_x.append(rec.positions[0])
|
|
arr_y.append(rec.readables[0].mean)
|
|
caput("SF-SCAN-GLOBAL:X-ARRAY", to_array(arr_x, 'd'))
|
|
caput("SF-SCAN-GLOBAL:Y-ARRAY", to_array(arr_y, 'd'))
|
|
|
|
#scan and plot
|
|
try:
|
|
caput("SF-SCAN-GLOBAL:FIT-X-ARRAY", to_array([0.0], 'd'))
|
|
caput("SF-SCAN-GLOBAL:FIT-Y-ARRAY", to_array([0.0], 'd'))
|
|
x_val.write(start)
|
|
time.sleep(5.0)
|
|
caput(message_channel, "scanning ")
|
|
r = lscan(x_val, y_averager, start, end, step, latency=lat, before_read = before, after_read = after)
|
|
x_array = r.getPositions(0)
|
|
y_array = energy = [val.mean for val in r.getReadable(0)]
|
|
caput("SF-SCAN-GLOBAL:X-ARRAY", to_array(x_array, 'd'))
|
|
caput("SF-SCAN-GLOBAL:Y-ARRAY", to_array(y_array, 'd'))
|
|
try:
|
|
if scan == "AR_K" or scan == "AT_K":
|
|
x_max, y_max, fit_x, fit_y, R2 = extremum(x_array, y_array)
|
|
if scan == "AR_PHI" or scan == "AT_PHI":
|
|
fit_amplitude, fit_x_deg, fit_offset, x_max, fit_x, fit_y = hfitoff(y_array, x_array)
|
|
except:
|
|
raise Exception("Fit failure")
|
|
plt = plot(None, name="param scan")[0]
|
|
plt.getSeries(0).setData(to_array(x_array,'d'), to_array(y_array,'d'))
|
|
plt.getSeries(0).setPointSize(6)
|
|
plt.getSeries(0).setLinesVisible(False)
|
|
plt.addSeries(LinePlotSeries("fit"))
|
|
plt.getSeries(1).setData(fit_x, fit_y)
|
|
plt.getSeries(1).setPointsVisible(False)
|
|
plt.setLegendVisible(True)
|
|
caput(und + ":SCAN-CREST", x_max)
|
|
caput("SF-SCAN-GLOBAL:FIT-X-ARRAY", fit_x)
|
|
caput("SF-SCAN-GLOBAL:FIT-Y-ARRAY", fit_y)
|
|
if scan == "AR_K" or scan == "AT_K":
|
|
caput(message_channel, "on-crest: " + ("%.4f" % x_max))
|
|
if scan == "AR_PHI" or scan == "AT_PHI":
|
|
caput(message_channel, "on-crest: " + ("%.1f" % x_max))
|
|
finally:
|
|
x_val.write(x_val0)
|
|
x_val.close()
|
|
y_val.close()
|