###################################################################################################
# OPERATION FUNCTIONS
###################################################################################################
#
# Contains functions to operate the FOFB application.
#
###################################################################################################
import tango
import logging
import numpy as np
import time
import FofbTool.Utils
# Get the module logger
logger = logging.getLogger("FofbTool")
###################################################################################################
# OPERATIONS ON CCN
###################################################################################################
def align_ccn(nodename):
"""
Align FA sequence number on a cellnode
PARAMETERS
----------
nodename: str
The target fofbnode, ie 'cellnode-c09' or 'centralnode'
"""
prx=FofbTool.Utils.get_prx_from_nodename(nodename)
if prx is None:
logger.error("Failed to align CCN on {}".format(nodename))
return
logger.info("Trying to detect and align FA sequences")
logger.debug("Reset offsets on comlbp")
set_comlbp_seqoffset(nodename, 0)
# Loop until two consecutives identical measures
N=[7,3]
while N[-2] != N[-1]:
logger.debug("Latch two sequence numbers without offset")
prx.ccnpack0_control=1 # let it roll a bit to collect data
time.sleep(2)
prx.ccnpack0_control=2 # latch it
time.sleep(2)
N.append(prx.ccnpack0_latchedseq1-prx.ccnpack0_latchedseq2)
logger.debug("seq ({}, {}, {})".format(prx.ccnpack0_latchedseq1, prx.ccnpack0_latchedseq2, N[-1]))
if len(N) > 5:
logger.error("Could not measure sequence offset.")
return False
N=N[-1]
if N in (-1, 0, 1):
logger.warning("Sequence offset measured = {}, something might be wrong".format(N))
# handle diff going from 17 bits to 16bits offset
if N > 0x7FFF:
N=N-0x10000
if -N > 0x8000:
N=N+0x10000
set_comlbp_seqoffset(nodename, N)
seqoffset=N
logger.debug("Perform autocheck")
logger.debug("Latch two sequence numbers without offset")
prx.ccnpack0_control=1
time.sleep(2)
prx.ccnpack0_control=2
time.sleep(2)
N=prx.ccnpack0_latchedseq2-prx.ccnpack0_latchedseq1
if not N in (-1, 1):
logger.warning("Corrected sequence offset measured = {}, something might be wrong. Run it again".format(N))
return False
return seqoffset
def set_comlbp_seqoffset(nodename, N):
""" Set sequence offset for all comlbp interfaces """
prx=FofbTool.Utils.get_prx_from_nodename(nodename)
if prx is None:
logger.error("Failed to align CCN on {}".format(nodename))
return
logger.info("Program sequence offset {} in {}".format(N, nodename))
for n in range(4):
prx["comlbp{}_seqoffset".format(n)]=N
def stop_ccn(nodename):
"""
Stop the communication with cellnode on the specified fofbnode.
PARAMETERS
----------
nodename: str
The target fofbnode, ie 'cellnode-c09' or 'centralnode'
"""
prx=FofbTool.Utils.get_prx_from_nodename(nodename)
if prx is None:
logger.error("Failed to stop CCN on {}".format(nodename))
return
logger.info("Stopping CCN on {}".format(nodename))
nint=1
if 'central' in nodename:
nint = 4
for n in range(nint):
prx["ccnpack{}_control".format(n)] = False
prx["ccnunpack{}_control".format(n)] = False
def reset_ccn(nodename):
"""
Reset the communication with cellnode on the specified fofbnode.
PARAMETERS
----------
nodename: str
The target fofbnode, ie 'cellnode-c09' or 'centralnode'
"""
prx=FofbTool.Utils.get_prx_from_nodename(nodename)
if prx is None:
logger.error("Failed to reset CCN on {}".format(nodename))
return
logger.info("Reset CCN on {}".format(nodename))
nint=1
if 'central' in nodename:
nint = 4
for n in range(nint):
prx["ccneth{}_reset".format(n)] = 0x60000001 # impossible to write 0xE0000001
prx["ccneth{}_gt_reset".format(n)] = 1
time.sleep(2)
for n in range(nint):
prx["ccneth{}_gt_reset".format(n)] = 0
prx["ccneth{}_reset".format(n)] = 0
ack_ccn(nodename)
def start_ccn(nodename):
"""
Start the communication with cellnode on the specified fofbnode.
PARAMETERS
----------
nodename: str
The target fofbnode, ie 'cellnode-c09' or 'centralnode'
"""
prx=FofbTool.Utils.get_prx_from_nodename(nodename)
if prx is None:
logger.error("Failed to start CCN on {}".format(nodename))
return
logger.info("Starting CCN on {}".format(nodename))
nint=1
if 'central' in nodename:
nint = 4
for n in range(nint):
prx["ccnpack{}_control".format(n)] = True
prx["ccnunpack{}_control".format(n)] = True
def ack_ccn(nodename):
"""
Start the communication with cellnode on the specified fofbnode.
PARAMETERS
----------
nodename: str
The target fofbnode, ie 'cellnode-c09' or 'centralnode'
"""
prx=FofbTool.Utils.get_prx_from_nodename(nodename)
if prx is None:
logger.error("Failed to ack CCN on {}".format(nodename))
return
logger.info("Ack CCN error on {}".format(nodename))
nint=1
if 'central' in nodename:
nint = 4
for n in range(nint):
prx["ccnpack{}_reset_error".format(n)] = True
prx["ccnunpack{}_reset_error".format(n)] = True
time.sleep(1)
for n in range(nint):
prx["ccnpack{}_reset_error".format(n)] = False
prx["ccnunpack{}_reset_error".format(n)] = False
###################################################################################################
# OPERATIONS ON COMBPM
###################################################################################################
def stop_combpm(cellnodename):
"""
Stop the communication with bpm on the specified cellnode.
PARAMETERS
----------
cellnodename: str
The target fofbnode, ie 'cellnode-c09'
"""
prx=FofbTool.Utils.get_prx_from_nodename(cellnodename)
if prx is None:
logger.error("Failed to stop ComBPM on {}".format(p))
return
logger.info("Stopping ComBPM on {}".format(cellnodename))
prx["combpm_reset"] = 1
prx["combpm_gt_control"] = 0x5
def start_combpm(cellnodename):
"""
Start the communication with bpm on the specified cellnode.
PARAMETERS
----------
cellnodename: str
The target fofbnode, ie 'cellnode-c09'
"""
prx=FofbTool.Utils.get_prx_from_nodename(cellnodename)
if prx is None:
logger.error("Failed to start ComBPM on {}".format(cellnodename))
return
logger.info("Starting ComBpm on {}".format(cellnodename))
prx["combpm_reset"] = 0
prx["combpm_gt_control"] = 0x1
time.sleep(1)
ack_combpm(cellnodename)
def ack_combpm(cellnodename):
"""
Ack errors on the communication with bpm on the specified cellnode.
PARAMETERS
----------
cellnodename: str
The target fofbnode, ie 'cellnode-c09'
"""
prx=FofbTool.Utils.get_prx_from_nodename(cellnodename)
if prx is None:
logger.error("Failed to start ComBPM on {}".format(cellnodename))
return
logger.info("Ack ComBpm on {}".format(cellnodename))
prx["combpm_reset_error"] = True
time.sleep(1)
prx["combpm_reset_error"] = False
###################################################################################################
# OPERATIONS ON COMLBP
###################################################################################################
def stop_comlbp(cellnodename):
"""
Stop the communication with LBP on the specified cellnode.
PARAMETERS
----------
cellnodename: str
The target fofbnode, ie 'cellnode-c09'
"""
prx=FofbTool.Utils.get_prx_from_nodename(cellnodename)
if prx is None:
logger.error("Failed to stop ComLBP on {}".format(p))
return
logger.info("Stopping ComLBP on {}".format(cellnodename))
for n in range(4):
prx["comlbp{}_control".format(n)] = 0
def start_comlbp(cellnodename):
"""
Start the communication with LBP on the specified cellnode.
PARAMETERS
----------
cellnodename: str
The target fofbnode, ie 'cellnode-c09'
"""
prx=FofbTool.Utils.get_prx_from_nodename(cellnodename)
if prx is None:
logger.error("Failed to start ComLBP on {}".format(p))
return
logger.info("Starting ComLBP on {}".format(cellnodename))
for n in range(4):
prx["comlbp{}_control".format(n)] = 0x10
def reset_comlbp(cellnodename):
"""
Reset the communication with LBP on the specified cellnode.
PARAMETERS
----------
cellnodename: str
The target fofbnode, ie 'cellnode-c09'
"""
prx=FofbTool.Utils.get_prx_from_nodename(cellnodename)
if prx is None:
logger.error("Failed to reset ComLBP on {}".format(p))
return
logger.info("Reset ComLBP on {}".format(cellnodename))
for n in range(4):
prx["comlbp{}_control".format(n)] = 0x3
time.sleep(1)
for n in range(4):
prx["comlbp{}_control".format(n)] = 0x0
###################################################################################################
# OPERATIONS ON BPM ELECTRONICS
###################################################################################################
# Some local constants
ADDR_CCCFG=0 # Register address of Libera Electron
ADDR_FAICFG=0x2000 # Register address of Libera Electron
def electron_group_writefadata(group, listargs):
args = tango.DeviceData()
args.insert(tango.DevVarLongArray, listargs)
r=group.command_inout("WriteFAData", args)
return r
def electron_init_fa(bpmlist):
"""
Configure FA register of Libera Electron
PARAMETERS
----------
bpmlist: list of (bpmid, bpmpath)
ID to apply and path of BPM tango device
RETURNS
-------
success: boolean
True on success
"""
# Init BPMs, put for each the ID number
# Cannot do that with group
logger.info("Initialize BPMs")
for bpmid, bpmpath in bpmlist:
logger.debug("Initialize BPM {}".format(bpmpath))
pbpm = tango.DeviceProxy(bpmpath)
try:
pbpm.ping()
except tango.ConnectionFailed:
logger.error("Failed to connect to {}".format(bpmpath))
return False
# Configure : bpmid, tframe, mgtpower, mgtlb, bfrclr
electron_group_writefadata(pbpm, [ADDR_CCCFG, 4, 5, bpmid, 6000, 0, 0, 6000])
# Ack rise
electron_group_writefadata(pbpm, [ADDR_FAICFG, 4, 1, 9])
# Ack fall
electron_group_writefadata(pbpm, [ADDR_FAICFG, 4, 1, 8])
return True
def electron_sync_next_trigger(bpmlist):
"""
Prepare Libera Electron to sync on next trigger
PARAMETERS
----------
bpmlist: list of str
List of Tango paths to Libera devices
RETURNS
-------
success: boolean
True on success
"""
logger.info("Prepare Libera Electron to sync at next trigger")
bpms = tango.Group('bpms')
bpms.add(bpmlist)
r=bpms.command_inout("settimeonnexttrigger")
for _r in r:
if _r.has_failed():
logger.error("Failed to apply command SetTimeOnNextTrigger on bpm {}".format(_r.dev_name()))
return False
return True
def electron_start_next_trigger(bpmlist):
"""
Prepare Libera Electron to start on next trigger
PARAMETERS
----------
bpmlist: list of str
List of Tango paths to Libera devices
RETURNS
-------
success: boolean
True on success
"""
logger.info("Prepare Libera Electron to start at next trigger")
bpms = tango.Group('bpms')
bpms.add(bpmlist)
# Write start, tframelim=zero, no debug data
r=electron_group_writefadata(bpms, [ADDR_FAICFG, 4, 1, 8])
for _r in r:
if _r.has_failed():
logger.error("Failed to apply command WriteFAData on bpm {}".format(_r.dev_name()))
return False
# Write Enable ITech FAI
r=electron_group_writefadata(bpms, [ADDR_FAICFG+4, 4, 1,1])
for _r in r:
if _r.has_failed():
logger.error("Failed to apply command WriteFAData on bpm {}".format(_r.dev_name()))
return False
return True
def electron_stop_com(bpmlist):
"""
Stop Libera Electron FA communication
PARAMETERS
----------
bpmlist: list of str
List of Tango paths to Libera devices
RETURNS
-------
success: boolean
True on success
"""
logger.info("Stop Libera Electron FA communication")
bpms = tango.Group('bpms')
bpms.add(bpmlist)
# Write Disable ITech FAI
r=electron_group_writefadata(bpms, [ADDR_FAICFG+4, 4, 1, 0])
for _r in r:
if _r.has_failed():
logger.error("Failed to apply command WriteFAData on bpm {}".format(_r.dev_name()))
return False
# Soft stop
r=electron_group_writefadata(bpms, [ADDR_FAICFG+8, 4, 1, 0])
for _r in r:
if _r.has_failed():
logger.error("Failed to apply command WriteFAData on bpm {}".format(_r.dev_name()))
return False
# Reset User FAI
r=electron_group_writefadata(bpms, [ADDR_FAICFG, 4, 1, 0])
for _r in r:
if _r.has_failed():
logger.error("Failed to apply command WriteFAData on bpm {}".format(_r.dev_name()))
return False
return True
###################################################################################################
# OPERATIONS FOR LBP and Electron SYNCHRONIZATION
###################################################################################################
def sync_all_bpm():
"""
Synchronize all BPM electronics, Electron and Brillance Plus.
This will use the FofbManager, but add sync on LiberaBrillancePlus before and sequence alignement after.
"""
# Get FofbManager proxy
FofbManager = tango.DeviceProxy(FofbTool.Configuration.config["tangopath"]["fofb-manager"])
# Set a group of Libera Brillance Plus EVRX board, from FofbTool configuration
lbpevrx = tango.Group('lbpevrx')
lbpevrx.add(FofbTool.Configuration.config["tangopath"]["lbpevrx"].split())
# ---------------------------------------------------------------------------------------------------------------
# Step 1 : Init BPMs
logger.info("Configure Libera Electron with FofbManager")
lastlog = FofbManager.logs[-1]
FofbManager.StartStep01ConfigLiberaCom() # Non blocking call, shame
# Wait for the logs to contain the magic words...
while True:
# Read the logs, crop to get only from the last log read
readlogs = FofbManager.logs
readlogs = readlogs[readlogs.index(lastlog):]
lastlog = readlogs[-1]
if '[INFO] FOFB com. successfully configured on Libera bpms' in "".join(readlogs):
break
time.sleep(1)
# ---------------------------------------------------------------------------------------------------------------
# Step 2 : Synchronize bpm
logger.info("Prepare Libera Brillance Plus to start on next trigger")
r=lbpevrx.write_attribute("synchronize", 0)
for _r in r:
if _r.has_failed():
logger.error("Failed to write synchronize on LBP EVRX {}".format(_r.dev_name()))
return
logger.info("Synchronize Libera Electron with FofbManager")
FofbManager.StartStep02SynchronizeLibera() # Non blocking call, shame
# Wait for the logs to contain the magic words...
while True:
# Read the logs, crop to get only from the last log read
readlogs = FofbManager.logs
readlogs = readlogs[readlogs.index(lastlog):]
lastlog = readlogs[-1]
if '[INFO] Libera bpms successfully triggered' in "".join(readlogs):
break
time.sleep(1)
# ---------------------------------------------------------------------------------------------------------------
# Step 3 : Start communication
logger.info("Start Libera Electron with FofbManager")
FofbManager.StartStep03StartLiberaCom() # Non blocking call, shame
# Wait for the logs to contain the magic words...
while True:
# Read the logs, crop to get only from the last log read
readlogs = FofbManager.logs
readlogs = readlogs[readlogs.index(lastlog):]
lastlog = readlogs[-1]
if '[INFO] FOFB com. successfully started on Libera bpms' in "".join(readlogs):
break
time.sleep(1)