208 lines
7.2 KiB
Python
208 lines
7.2 KiB
Python
"""
|
|
PIX_PATHA_CHN = "X04SA-ES3-CCD:PATHa"
|
|
PIX_PATHB_CHN = "X04SA-ES3-CCD:PATHb"
|
|
PIX_FNAM_FMT_CHN = "X04SA-ES3-CCD:FNAM_FMT"
|
|
PIX_FNUM_CHN = "X04SA-ES3-CCD:FNUM"
|
|
PIX_FNAM_CHN = "X04SA-ES3-CCD:FNAM"
|
|
PIX_EXPOSE_CHN = "X04SA-ES3-CCD:EXPOSE"
|
|
PIX_STATUS_CHN = "X04SA-ES3-CCD:STATUS"
|
|
PIX_TRIG_CHN = "X04SA-ES3-CCD:TRIG"
|
|
"""
|
|
|
|
class ImageFilename(DeviceBase, Readable, Readable.StringType):
|
|
def __init__(self, pixel):
|
|
DeviceBase.__init__(self, "image filename")
|
|
self.setParent(pixel)
|
|
|
|
def read(self):
|
|
return self.getParent().get_image_filename()
|
|
|
|
class Pixel(DeviceBase, Readable, Readable.IntegerType):
|
|
def __init__(self, name, prefix):
|
|
DeviceBase.__init__(self, name)
|
|
self.prefix = prefix
|
|
self.image_root_folder = "/sls/X04SA/data/x04sa/ES3/pixel/images/"
|
|
self.image_filename = ImageFilename(self)
|
|
|
|
def doInitialize(self):
|
|
self.PIX_TYPE = "PilatusII"
|
|
self.PIX_FILE_HEADER_LENGTH = 0
|
|
self.PIX_COLOR_DEPTH = 32
|
|
self.PIX_XDIM = 487
|
|
self.PIX_YDIM = 195
|
|
self.PIX_THRESH1 = 40
|
|
self.PIX_THRESH2 = 500
|
|
self.PIX_THRESH3 = 8000
|
|
self.PIX_THRESH4 = 100000
|
|
self.PIX_EXP_START_TIME = 0
|
|
self._update_header_length()
|
|
self.show()
|
|
|
|
|
|
def _update_header_length(self):
|
|
format = self.get_image_format()
|
|
ext = format.split('.')[-1]
|
|
if ext == "tif":
|
|
self.IMAGE_HEADER_LENGTH = 4096
|
|
elif ext == "edf":
|
|
self.IMAGE_HEADER_LENGTH = 1024
|
|
else:
|
|
self.IMAGE_HEADER_LENGTH = 0
|
|
|
|
def get_image_filename(self):
|
|
return self.image_root_folder + self.get_path() + self.get_file_short_name()
|
|
|
|
def get_next_image_filename(self):
|
|
return self.image_root_folder + self.get_path() + ( self.get_image_format() % (self.get_count_id()+1))
|
|
|
|
def get_image_format(self):
|
|
return caget(self.prefix + ":FNAM_FMT")
|
|
|
|
def set_image_format(self, fmt):
|
|
caput(self.prefix + ":FNAM_FMT", fmt)
|
|
self._update_header_length();
|
|
|
|
def get_count_id(self):
|
|
ret = caget (self.prefix + ":FNUM", 'i')
|
|
self.setCache(ret, None)
|
|
return ret
|
|
|
|
def set_count_id(self, val):
|
|
caput (self.prefix + ":FNUM", int(val))
|
|
self.setCache(int(val), None)
|
|
|
|
def get_expose(self):
|
|
return caget (self.prefix + ":EXPOSE", 'd') / 1000.0
|
|
|
|
def set_expose(self, val):
|
|
val = max(float(val), 0.000005)
|
|
caput (self.prefix + ":EXPOSE", val * 1000)
|
|
|
|
def get_status(self):
|
|
ret = caget (self.prefix + ":STATUS", 's')
|
|
if ret in ["Ready0", "Ready1"]:
|
|
self.setState(State.Ready)
|
|
elif ret in ["Snapping", "Writing", "Updating"]:
|
|
self.setState(State.Busy)
|
|
elif ret in ["Error"]:
|
|
self.setState(State.Error)
|
|
elif ret in ["Time-out", "Disconnected"]:
|
|
self.setState(State.Offline)
|
|
elif ret in ["Undefined","Connected",]:
|
|
self.setState(State.Invalid)
|
|
else:
|
|
self.setState(State.Invalid)
|
|
return ret
|
|
|
|
def is_status_ready(self, status=None):
|
|
if status is None:
|
|
status = self.get_status()
|
|
return status in ["Ready0", "Ready1"]
|
|
|
|
def set_path(self, patha, pathb=""):
|
|
if not patha.endswith("/"):
|
|
patha=patha + "/"
|
|
if pathb!="":
|
|
if not pathb.endswith("/"):
|
|
pathb=pathb + "/"
|
|
|
|
caput(self.prefix + ":PATHa", patha)
|
|
caput(self.prefix + ":PATHb", pathb)
|
|
|
|
def get_path(self):
|
|
return caget(self.prefix + ":PATHa") + caget(self.prefix + ":PATHb")
|
|
|
|
|
|
def get_file_short_name(self):
|
|
return caget(self.prefix + ":FNAM")
|
|
|
|
#Trigger Snap Increment Write Inc+Wr Inc+Sn+Wr Connect Disconnect Update
|
|
def set_trigger(self, val):
|
|
caput (self.prefix + ":TRIG", val)
|
|
|
|
def get_trigger(self):
|
|
return caget (self.prefix + ":TRIG")
|
|
|
|
|
|
def get_trigger_id(self):
|
|
return caget (self.prefix + ":TRIG", 'i')
|
|
|
|
def start(self):
|
|
pixwait() # Wait until pixel detector is ready
|
|
PIX_EXPOSE = epics_get(PIX_EXPOSE_CHN)
|
|
slef.set_trigger ("Inc+Sn+Wr")
|
|
self.PIX_EXP_START_TIME = time.time()
|
|
|
|
|
|
def wait_finished(self):
|
|
cnt = 0
|
|
expose = self.get_expose()
|
|
#
|
|
# First wait for "TRIG" to go back to idle. This should
|
|
# ensure that execution of the command has started.
|
|
#
|
|
trig = epics_get (PIX_TRIG_CHN, "long")
|
|
while (get_trigger_id() != 0):
|
|
cnt = cnt + 1
|
|
if (cnt > 1000):
|
|
raise Exception("Trigger signal still busy, pixwait timed-out!\n")
|
|
time.sleep (0.05)
|
|
|
|
#
|
|
# Then wait for exposure time to elapse.
|
|
# (PIX_EXPOSE is in milliseconds!)
|
|
#
|
|
while time.time() <= self.PIX_EXP_START_TIME + (expose * 1000) :
|
|
time.sleep(0.05)
|
|
|
|
#
|
|
# Then wait for the pixel detector to become ready.
|
|
# Wait time is 1000 x 0.05 sec = 50 sec.
|
|
#
|
|
sms_flag = 0
|
|
while (True):
|
|
stat = self.get_status()
|
|
if is_status_ready(stat):
|
|
break
|
|
cnt = cnt + 1
|
|
if cnt > 1000 :
|
|
if (smsFlag == 0):
|
|
print("pixwait: Pixel status did not return to 'Ready' within 50 seconds.")
|
|
notify ("Pixel detector problem in pixwait.")
|
|
sms_flag = 1
|
|
|
|
if stat in ("Undefined", "Time-out", "Error"):
|
|
raise Exception ("Invalid PIX status:", stat)
|
|
time.sleep (0.05)
|
|
|
|
|
|
if sms_flag == 1:
|
|
notify ("Pixel detector seems to have recovered from error.")
|
|
|
|
|
|
def logwrite(self):
|
|
pass
|
|
|
|
def show(self):
|
|
print ("\nThe current pixel detector settings are:")
|
|
print " The frame number: ", self.get_count_id()
|
|
print " The file path is:", self.get_path()
|
|
print " The file template is: ", self.get_image_format()
|
|
print " The last file name was: ", self.get_image_filename()
|
|
print " The next file name will be: ", self.get_next_image_filename()
|
|
print " Exposure time = %5d sec" % (self.get_expose(), )
|
|
print " Image dimensions: %d x %d pixels (total = %d)" % (self.PIX_XDIM, self.PIX_YDIM, self.PIX_XDIM*self.PIX_YDIM)
|
|
print " Threshold values: Thresh1 = %d, Thresh2 = %d, Thresh3 = %d, Thresh4 = %d" % (self.PIX_THRESH1,self.PIX_THRESH2,self.PIX_THRESH3,self.PIX_THRESH4)
|
|
|
|
def doUpdate(self):
|
|
self.get_status()
|
|
self.get_count_id()
|
|
|
|
def read(self):
|
|
#Readable interface: current
|
|
return self.get_count_id()
|
|
|
|
add_device( Pixel("pixel", "X04SA-ES3-CCD"), True)
|
|
pixel.polling = 1000
|
|
pixel.update()
|