280 lines
9.3 KiB
Python
280 lines
9.3 KiB
Python
|
|
POL_IDS={ "Circ_Plus": 1, "Circ_Minus": 2, "Lin_Hor":3, "Lin_Ver": 4, "Lin": 5}
|
|
|
|
channel_id1_status=Channel("X11MA-ID1-USER:STATUS", 'i', monitored=True)
|
|
channel_id2_status=Channel("X11MA-ID2-USER:STATUS", 'i', monitored=True)
|
|
channel_id1_error=Channel("X11MA-ID1-USER:ERROR-SOURCE", 'i', monitored=True)
|
|
channel_id2_error=Channel("X11MA-ID2-USER:ERROR-SOURCE", 'i', monitored=True)
|
|
channel_id1_control=Channel("X11MA-ID1-GAP:SCTRL", 'i', monitored=True)
|
|
channel_id2_control=Channel("X11MA-ID2-GAP:SCTRL", 'i', monitored=True)
|
|
channel_ring_status=Channel("ACOAU-ACCU:OP-MODE", 'i', monitored=True)
|
|
|
|
def get_id_status(id):
|
|
# 0 = ok
|
|
# 1 = Warning
|
|
# 2 = Error
|
|
# 3 = Not available
|
|
if id<1 or id>2:
|
|
raise Exception("Invalid id")
|
|
return channel_id1_status.get() if (id==1) else channel_id2_status.get()
|
|
|
|
def get_id_error(id):
|
|
# B0 (1) = PLC (Taper, Motor controller, Limitswitch....)
|
|
# B1 (2) = Encoders (cabel, not referenced)
|
|
# B2 (4) = FeedForward (PS off, PS defect, FF ausgeschaltet)
|
|
# B3 (8) = Operator COntrol
|
|
# B4 (16)= Moving Timeout (moving longer then 120 sec)
|
|
# B5 (32)= Interlock (Orbit, Temperature...)\
|
|
if id<1 or id>2:
|
|
raise Exception("Invalid id")
|
|
return channel_id1_error.get() if (id==1)else channel_id2_error.get()
|
|
|
|
|
|
def get_id_control(id):
|
|
#0 = SLS control, 1 = experiment
|
|
if id<1 or id>2: raise Exception("Invalid id")
|
|
return channel_id1_control.get() if (id==1) else channel_id2_control.get()
|
|
|
|
def get_ring_status():
|
|
#ACOAU-ACCU:OP-MODE kann Werte zwischen 0 und 5 annehmen:
|
|
# rule from 24.8.04
|
|
#If (PCT beam current <0.09) -> State 0: "Machine Down"
|
|
#Else If (GUN Trigger OFF) -> State 1: "Inj. Stopped"
|
|
#Else If (INJ_MODE == Normal) -> State 2: "Accumulating."
|
|
#Else If !(reached Top-Up Current) -> State 3: "Accumulating"
|
|
#Else If (OP-READY == Wait) -> State 4: "Top-up ready"
|
|
#Else If (OFB off) -> State 5: "Light-Available" (former State 4)
|
|
#Else -> State 6: "Light Available" (former State 5)
|
|
return channel_ring_status.get()
|
|
|
|
|
|
|
|
def get_pol_as_string(pol):
|
|
for (k,v) in POL_IDS.items():
|
|
if v==pol:
|
|
return k
|
|
return "Unknown"
|
|
|
|
|
|
def get_id_pol(id, as_string=False):
|
|
# 1: Circ+
|
|
# 2: Circ-
|
|
# 3: Lin H
|
|
# 4: Lin V
|
|
# 5: LIn
|
|
if id<1 or id>2: raise Exception("Invalid id")
|
|
#ret = caget ("X11MA-ID1:MODE" if id==1 else "X11MA-ID2:MODE", 's' if as_string else 'i')
|
|
mode = id1_mode.read() if (id==1) else id2_mode.read()
|
|
if mode=="LINEAR":
|
|
#linear mode
|
|
alpha = get_alpha_id(id)
|
|
if alpha==0 :
|
|
ret = 3
|
|
elif alpha==90:
|
|
ret = 4
|
|
else:
|
|
ret = 5
|
|
elif mode == "CIRC +":
|
|
ret = 1
|
|
elif mode == "CIRC -":
|
|
ret = 2
|
|
else:
|
|
ret = -1
|
|
if as_string:
|
|
return get_pol_as_string(ret)
|
|
return ret
|
|
|
|
def get_alpha_id(id):
|
|
if id<1 or id>2: raise Exception("Invalid id")
|
|
return id1_alpha.read() if (id==1) else id2_alpha.read()
|
|
|
|
|
|
def put_id_offset(id, v):
|
|
if id == 1:
|
|
id1_offset.write(v)
|
|
elif id == 2:
|
|
id2_offset.write(v)
|
|
else:
|
|
raise Exception("Invalid id")
|
|
|
|
def get_id_offset(id):
|
|
if id<1 or id>2:
|
|
raise Exception("Invalid id")
|
|
return id1_offset.read() if (id==1) else id2_offset.read()
|
|
|
|
|
|
def put_id_pol(id, pol, alpha=None):
|
|
#pol = 1 : Mode =1, alpha = 0 circ +
|
|
#pol = 2 : Mode =2, alpha = 0 circ -
|
|
#pol = 3 : Mode =0, alpha = 0 lin hor
|
|
#pol = 4 : Mode =0, alpha = 90 lin vert
|
|
#pol = 5 : Mode =0, alpha = ? lin rot
|
|
print "Set id" + str(id) + " Pol=" + str(pol) + " Alpha=" + str(alpha)
|
|
|
|
if id<1 or id>2:
|
|
raise Exception("Invalid id")
|
|
|
|
if get_dry_run():
|
|
return
|
|
|
|
mode_dev = id1_mode if (id==1) else id2_mode
|
|
alpha_dev = id1_alpha if (id==1) else id2_alpha
|
|
|
|
if is_string(pol):
|
|
pol=POL_IDS[pol]
|
|
|
|
if pol == 1:
|
|
mode_dev.write("CIRC +")
|
|
elif pol == 2:
|
|
mode_dev.write("CIRC -")
|
|
elif pol == 3:
|
|
if mode_dev.read() != "LINEAR":
|
|
mode_dev.write("LINEAR")
|
|
time.sleep(0.5)
|
|
alpha_dev.write(0)
|
|
elif pol == 4:
|
|
if mode_dev.read() != "LINEAR":
|
|
mode_dev.write("LINEAR")
|
|
time.sleep(0.5)
|
|
alpha_dev.write(90)
|
|
elif pol == 5:
|
|
if mode_dev.read() != "LINEAR":
|
|
mode_dev.write("LINEAR")
|
|
time.sleep(0.5)
|
|
if alpha is not None:
|
|
alpha_dev.write(alpha)
|
|
|
|
|
|
def tune_detune(active_id):
|
|
if get_dry_run():
|
|
return
|
|
|
|
offset_1 = float(get_setting("OFFSET_ID_1"))
|
|
offset_2 = float(get_setting("OFFSET_ID_2"))
|
|
|
|
if active_id ==1:
|
|
print "Tune ID1"
|
|
put_id_offset(2, offset_2-40) #detuneID2
|
|
put_id_offset(1, offset_1) #tuneID1
|
|
else:
|
|
print "Tune ID2"
|
|
put_id_offset(1, offset_1-40) #detuneID1
|
|
put_id_offset(2, offset_2) #tuneID2
|
|
|
|
class IdPol(ReadonlyRegisterBase):
|
|
def __init__(self, id):
|
|
self.id = id
|
|
def doRead(self):
|
|
return get_id_pol(self.id, True)
|
|
def getName(self):
|
|
return "id" + str(self.id) + "_pol"
|
|
|
|
add_device(IdPol(1), True)
|
|
add_device(IdPol(2), True)
|
|
id1_pol.update()
|
|
id2_pol.update()
|
|
id1_pol.setPolling(2000)
|
|
id2_pol.setPolling(2000)
|
|
|
|
|
|
class EnOptDesc(ReadonlyRegisterBase):
|
|
def doRead(self):
|
|
opt = energy_opt.read()
|
|
if opt == "PGM": return "PGM"
|
|
if opt == "PGM+ID1": return "ID1"
|
|
if opt == "PGM+ID2": return "ID2"
|
|
if opt == "PGM+ID1+ID2": return "ID1_ID2"
|
|
return ""
|
|
|
|
def getName(self):
|
|
return "energy_opt_desc"
|
|
|
|
add_device(EnOptDesc(), True)
|
|
energy_opt_desc.setPolling(2000)
|
|
|
|
|
|
#Polarizarion Switch Tools
|
|
|
|
_switching_type = None
|
|
_switching_active_id = None
|
|
_switching_current_pol = None
|
|
_switching_pol_id1 = None
|
|
_switching_pol_id2 = None
|
|
_switching_id = None
|
|
|
|
|
|
def init_pol_switch(switching_type, sid = None, pol_id1=None, pol_id2=None):
|
|
global _switching_type, _switching_active_id, _switching_current_pol, _switching_pol_id1, _switching_pol_id2, _switching_id
|
|
if sid is None:
|
|
sid = get_setting("ID")
|
|
if pol_id1 is None:
|
|
#pol_id1 = POL_IDS[get_setting("POL_ID_1")] #From config
|
|
pol_id1=get_id_pol(1) #Current
|
|
if pol_id2 is None:
|
|
#pol_id2 =POL_IDS[get_setting("POL_ID_2")] #From config
|
|
pol_id2 =get_id_pol(2) #Current
|
|
_switching_type = switching_type
|
|
_switching_id =sid
|
|
_switching_pol_id1=pol_id1
|
|
_switching_pol_id2=pol_id2
|
|
print "Init pol switch: %s id:%s pol_id1:%s pol_id2:%s" %(_switching_type, _switching_id, _switching_pol_id1, _switching_pol_id2)
|
|
_switching_active_id = 1
|
|
if _switching_id == "ID1":
|
|
_switching_current_pol = _switching_pol_id1
|
|
elif _switching_id == "ID2":
|
|
_switching_current_pol = _switching_pol_id2
|
|
elif _switching_id == "ID1_ID2":
|
|
_switching_current_pol=_switching_pol_id1
|
|
|
|
if switching_type == "Tune_Detune":
|
|
tune_detune(1) #Tune ID1, Detune ID2
|
|
else:
|
|
if _switching_id == "ID1_ID2":
|
|
put_id_pol(2, _switching_current_pol) #Force both IDs to same polarization
|
|
if get_dry_run():
|
|
return
|
|
wait_channel("X11PHS:alldone", 1)
|
|
|
|
|
|
def nextpol():
|
|
global _switching_type, _switching_active_id, _switching_current_pol, _switching_pol_id1, _switching_pol_id2
|
|
if _switching_type == "Normal":
|
|
if _switching_current_pol==1: return 2 #circ+ -> circ-
|
|
elif _switching_current_pol==2: return 1 #circ+ -> circ-
|
|
elif _switching_current_pol==3: return 4 #lin hor -> lin vert
|
|
elif _switching_current_pol==4: return 3 #lin vert -> lin hor
|
|
elif _switching_current_pol==5: return5 #lin rot -> lin rot
|
|
else: raise "Invalid pol: " + str(_switching_current_pol)
|
|
elif _switching_type == "Tune_Detune":
|
|
if _switching_active_id==1: return _switching_pol_id2
|
|
elif _switching_active_id==2: return _switching_pol_id1
|
|
else: raise "Invalid ID: " + str(_switching_active_id)
|
|
else: raise "Invalid switching: " + str(_switching_type)
|
|
|
|
def switch_pol():
|
|
global _switching_type, _switching_active_id, _switching_current_pol, _switching_pol_id1, _switching_pol_id2, _switching_id
|
|
newpol=nextpol()
|
|
print "Switch pol: ", get_pol_as_string(newpol)
|
|
if _switching_id == "ID1":
|
|
put_id_pol(1,newpol)
|
|
elif _switching_id == "ID2":
|
|
put_id_pol(2,newpol)
|
|
elif ID == "ID1_ID2":
|
|
if _switching_type == "Normal":
|
|
put_id_pol(1, newpol)
|
|
put_id_pol(2, newpol)
|
|
elif _switching_type == "Tune_Detune":
|
|
if _switching_active_id==1:
|
|
_switching_active_id=2
|
|
else:
|
|
_switching_active_id=1
|
|
tune_detune(_switching_active_id)
|
|
|
|
if _switching_type == "Normal":
|
|
_switching_current_pol = newpol
|
|
|
|
time.sleep(1.0)
|
|
wait_channel("X11PHS:alldone", 1)
|
|
|
|
def get_cur_pol():
|
|
return _switching_current_pol |