import os as bash
import subprocess
import copy
from pydiana.tools import pydianaIO
from time import sleep
from pydiana.shell.logging_theme import logging,coloredlogs
logging.config.dictConfig({'disable_existing_loggers': True,'version': 1})
logger = logging.getLogger(__name__)
coloredlogs.install(level='WARNING', logger=logger)
[docs]
class sequencecfg():
def __init__(self,cfg_file,other_flags:str=''):
self.name=''
self.variables=None
self.write_vars=None
self.call = None
self.cfg_file = self.findcfg(cfg_file)
if self.cfg_file is not None:
self.loadfile()
self.remove_newline()
self.remove_tabs()
self.remove_comments()
self.remove_spaces()
self.drop_empty()
if not self.sequence_name():
logger.warning("Sequence not found")
self.read_framework_vars()
self.read_vars()
self.compare_vars()
self.other_flags = other_flags
[docs]
def findcfg(self,cfg) -> str:
"""
This function searces the cfg in the DIANA_EXT_INSTALL/cfg and DIANA_INSTALL/cfg folders. If a path is provided (relative or absolute) then the search is not performed
Parameters
----------
cfg (str) : string with name of cfg file
Returns
-------
String with path to cfg. If not Found returns None.
"""
if cfg[-4:] !='.cfg':
logger.warning("Not a valid cfg")
return None
if "/" in cfg:
if not bash.path.exists(cfg):
logger.warning(f" '{cfg}' is not a valid path to a cfg file")
return None
return cfg
diana_install = subprocess.check_output('echo $DIANA_INSTALL',shell=True, universal_newlines=True)[:-1]
diana_ext_install = subprocess.check_output('echo $DIANA_EXT_INSTALL',shell=True, universal_newlines=True)[:-1]
found = False
trypath = diana_install+'/cfg/'+cfg
if bash.path.exists(trypath):
found=True
cfg = trypath
trypath = diana_ext_install+'/cfg/'+cfg
if bash.path.exists(trypath):
found=True
cfg = trypath
if not found:
logger.warning(f" '{cfg}' not found")
return None
return cfg
[docs]
def sequence_name(self) -> bool:
"""
Read sequence name from cfg
Returns
-------
Returns True if the name was found else returns False
"""
found =False
for line in self.content:
if line[0] =='sequence':
self.name=line[1]
found =True
break
return found
[docs]
def loadfile(self):
"""
Function for loading cfg file content
"""
with open(self.cfg_file,'r') as cfg:
self.content = cfg.readlines()
[docs]
def remove_spaces(self):
"""
This function removes spaces from the file content
"""
new_content=[]
for line in self.content:
split_line = line.split(' ')
new_content.append([w for w in split_line if w!=''])
self.content=new_content
[docs]
def remove_tabs(self):
"""
This function removes the new line at the end of each line from the file
"""
self.content = ["".join(line.split("\t")) for line in self.content]
[docs]
def remove_newline(self):
"""
This function removes the new line at the end of each line from the file
"""
self.content = [line[:-1] for line in self.content]
[docs]
def drop_empty(self):
"""
Remove empty lines
"""
self.content = [line for line in self.content if len(line)>0]
[docs]
def read_framework_vars(self):
"""
This function reads the cfg parameters
"""
start=0
stop=-1
for idx,line in enumerate(self.content):
if 'framework' in line:
start = idx+1
if 'endfw' in line:
stop = idx
self.framework ={}
for line in self.content[start:stop]:
line = ''.join([w for w in line if w!='cfgvar']).split('=')
if len(line)==1:
line.append(None)
if line[1]=='' or line[1]=="\"\"" or line[1]=="''":
line[1]=None
self.framework[line[0]]=line[1]
[docs]
def find_var_pos_in_line(self,line:str):
"""
Utility function for finding variable position in the line of the cfg.
"""
idxs_dollar = self.find_all_substr_occur(line,"${")
idxs_closed = [line[i:].find('}')+i+1 for i in idxs_dollar]
if len(idxs_dollar) != len(idxs_closed):
logger.warning("Found something wierd when reading variables")
return None,None
return idxs_dollar,idxs_closed
[docs]
def read_vars_in_line(self,line:str):
"""
Utility function for reading all variables in the line of the cfg.
"""
vars=[]
idxs_dollar,idxs_closed =self.find_var_pos_in_line(line)
if idxs_dollar is not None and idxs_closed is not None:
for start,stop in zip(idxs_dollar,idxs_closed):
vars.append(line[start:stop-1])
return vars
[docs]
def read_vars(self):
"""
This function reads all the variables in the cfg.
"""
self.variables=[]
for line in self.content:
line = ''.join(line)
vars = self.read_vars_in_line(line)
if vars is None:
break
[self.variables.append(vv) for vv in vars]
[docs]
def compare_vars(self):
"""
Not sure....
"""
for fwv in self.framework:
if fwv not in self.variables:
self.variables.append(fwv)
self.variables = [vv for vv in self.variables if vv not in ['DIANA_INSTALL','DIANA_EXT_INSTALL']]
varswithdef = {}
for vv in self.variables:
if vv in self.framework:
varswithdef[vv]=self.framework[vv]
else:
varswithdef[vv]=None
self.variables = varswithdef
[docs]
def complete_vars(self):
"""
Function for completing the variable contents.
"""
repeat=True
max_iterations=100
i=0
self.write_vars = copy.deepcopy(self.variables)
while(repeat):
for vv in self.write_vars:
if self.write_vars[vv] is not None:
self.write_vars[vv] = self.complete_line(self.write_vars[vv])
found_var=False
for vv in self.write_vars:
try:
if len(self.read_vars_in_line(self.write_vars[vv]))>0:
found_var=True
logger.info(self.write_vars[vv])
except:
logger.error(f"Could not write {vv} (value: {self.write_vars[vv]})")
return False
repeat=found_var
i+=1
if i >=max_iterations:
repeat=False
return False
return True
[docs]
def find_all_substr_occur(self,line:str,substr:str):
"""
Function for finding a substring in a string.
Parameters
----------
line (str) : string where to search for the substring
substr (str) : string to search in line
Returns
-------
List of indexes with the positions in which the substring was found
"""
idxs = []
idx =line.find(substr)
stop=False
if idx !=-1:
idxs.append(idx+len(substr))
else:
stop=True
while(not stop):
if idxs[-1]> len(line):
stop=True
continue
idx = line[idxs[-1]:].find(substr)
if(idx!=-1):
idxs.append(idx+idxs[-1]+len(substr))
else:
stop=True
return idxs
[docs]
def set_vars(self,params):
"""
Function for setting the cfg parameters
Parameters
----------
params (dict) : dictionary containing the name of the cfg variables as keys as their value as the content (should be strings).
Returns
-------
True if everything goes well else returns False
"""
for vv in params:
if vv in self.variables:
self.variables[vv] = params[vv]
else:
logger.error(f"Variable '{vv}' not in sequence")
return False
return True
[docs]
def set_var(self,name:str,value):
"""
Function for setting a single cfg variable.
Parameters
----------
name (str) : string with the name of the variable
value (undefined): value to which the variable must be set. In the function it is converted to string.
Returns
-------
True if everything goes well else returns False
"""
return self.set_vars({name:str(value)})
[docs]
def get_var(self,name:str):
"""
Function for getting the content of a cfg variable
Parameters
----------
name (str) : string with the name of the parameter to get.
Returns
-------
Value of the parameter or None if the parameter is not found
"""
if name in list(self.variables.keys()):
return self.variables[name]
else:
logger.error(f"Variables {name} not found")
return None
[docs]
def make_diana_call(self,optional_args:str=None):
"""
Function for making the diana bash call for this sequence.
Parameters
----------
optional_args (str) : string with the optional flags to add to the call, like '-O'
Returns
-------
True if everything is successfull otherwise False
"""
self.call = 'diana -C '+self.cfg_file+' '
if not self.complete_vars():
logger.error(f"Could not complete all variables")
return False
for vv in self.write_vars:
if self.write_vars[vv] is not None:
self.call+="-V "+vv+" "+self.write_vars[vv]+' '
else:
logger.error(f"Please set '{vv}'. Call making aborted.")
self.call=None
return False
self.call+= ' '+self.other_flags
if optional_args is not None:
self.call+=' '+optional_args
return True
[docs]
def complete_line(self,line:str):
"""
Function for completing the line in the cfg by inserting the value of the variable.
Parameters
----------
line (str) : line to complete
Returns
-------
The completed line or None if not successful.
"""
vars = self.read_vars_in_line(line)
for vv in vars:
if self.variables[vv] is None:
logger.error(f"Please specify '{vv}'.")
return None
compline=[]
idxs_dollar,idxs_closed =self.find_var_pos_in_line(line)
if idxs_dollar is not None and idxs_closed is not None:
if len(idxs_dollar)>0 and len(idxs_closed)>0:
compline.append(line[:idxs_dollar[0]-2])
last_part = line[idxs_closed[-1]:]
idxs_dollar = idxs_dollar[1:]
idxs_closed = idxs_closed[:-1]
for start,stop in zip(idxs_dollar,idxs_closed):
compline.append(line[stop:start-2])
compline.append(last_part)
cc = [compline[i]+self.variables[vars[i]] for i in range(len(vars))]
for i in range(len(compline)-len(cc)):
cc.append(compline[len(cc)+i])
cc = ''.join(cc)
else:
cc = line
return cc
[docs]
def get_calls(self):
"""
Get all the calls from the cfg.
Returns
-------
Returns list with all the generated bash calls.
"""
return [self.call]
[docs]
def run_cfg(self):
"""
Function for running the cfg bash call using subprocess.
Returns
-------
The result of the diana call
"""
if len(self.get_calls())<1:
self.make_diana_call()
calls=self.get_calls()
result=[]
for cc in calls:
child = subprocess.Popen(cc,shell=True)
result.append({"PID":child.pid,"Return":child.wait(),'Command':cc})
return result
[docs]
def get_module(self,name):
"""
Getting a specific module from the cfg.
TODO: probably needs perfecting.
Parameters
----------
name (str) : name of the module to return
Returns
-------
List containing the lines regarding the module
"""
start=0
stop=-1
for idx,line in enumerate(self.content):
if name in line:
start = idx
if 'endmod' in line:
stop = idx
self.framework ={}
return self.content[start:stop]
[docs]
def get_reader(self):
"""
Function for getting reader of the cfg.
Returns
-------
List containing the lines regarding the module
"""
return self.get_module('reader')
[docs]
def get_writer(self):
"""
Function for getting writer of the cfg.
Returns
-------
List containing the lines regarding the module
"""
return self.get_module('writer')
[docs]
def find_line_value(self,line,content):
"""
Find value after the '=' sign in a cfg line
Parameters
----------
line (str) : line where to search for the value
Returns
-------
String with the parameter value, the string is empty if the parameter is not found.
"""
for cc in content:
if line in ''.join(cc):
return self.complete_line(''.join(cc).split('=')[-1])
return ''
[docs]
def get_output_file(self):
"""
Get outputfile line from writer of cfg.
Returns
-------
string containing the output file value.
"""
tmp_writer = self.get_writer()
if self.find_line_value('AppendToInput',tmp_writer) != '':
return self.get_input_file()
tmp_reader = self.get_reader()
name=''
name += self.find_line_value('OutputFilePrefix',tmp_writer)
if name !='':
name+='_'
if self.get_core_filename() is not None:
name+=self.get_core_filename()
if self.find_line_value('MeasType',tmp_reader)!='':
name+='_'+self.find_line_value('MeasType',tmp_reader)
name+='.list'
return name
[docs]
def get_core_filename(self):
"""
Get run formatted for output file
Returns
-------
None if run is not set else returns the string of the run.
"""
if self.get_run() is None:
return None
return f"{self.get_run():06d}"
[docs]
def get_run(self):
"""
Getting run of the cfg
Returns
-------
run (int) : integer with the run number
"""
return self.runnumber
[docs]
def set_run(self,run:int=None):
"""
Setting run for the cfg
Parameters
----------
run (int) : integer with the run number
"""
self.runnumber=run
def __str__(self):
"""
Function for converting to string an overview of the cfg. Called when printing the module.
Returns
-------
String with an overview of the class
"""
os=''
if self.variables is not None:
vars = copy.deepcopy(self.variables)
for vv in vars:
if vars[vv] is None:
vars[vv] = 'None'
maxlen1 = max([len(vv) for vv in vars]) + 3
maxlen1 = max((maxlen1,len('Variables')+3))
maxlen2 = max([len(vars[vv]) for vv in vars]) + 3
maxlen2 = max((maxlen2,len('Values')+3))
first_line = "| Variables"+' '*(maxlen1-len('Variables'))+"| Values"+' '*(maxlen2-len('Values'))+'|'
interrupt ="-"*(len(first_line)-2)
lines = ["| "+vv+' '*(maxlen1-len(vv))+'| '+vars[vv]+' '*(maxlen2-len(vars[vv]))+'|' for vv in vars]
seqname = "Sequence "+self.name
first_len = (len(first_line)-len(seqname))//2
os +="\n"
os += "-"+interrupt+'-\n'
os += "|"+" "*(first_len-1) + seqname+" "*(len(first_line)-first_len-len(seqname)-1)+"|\n"
os += "|"+interrupt+'|\n'
os += first_line+'\n'
os += "|"+interrupt+'|\n'
os += '\n'.join(lines)+"\n"
os += "-"+interrupt+'-\n\n'
else:
os = "Sequence "+self.name +'\n'
if self.call is not None:
os += "Call:\t"+self.call+"\n"
return os
[docs]
class averagecfg(sequencecfg):
"""
Class for dealing with sequences that build average quantites. Made mainly for averagepulse and averagenoise.
"""
def __init__(self,cfg_file,other_flags:str='',all=False):
super().__init__(cfg_file,other_flags)
self.all = all
self.total_calls=None
self.use=None
[docs]
def find_all_files(self):
"""
Finds all the files in the .list file
Returns
-------
List with the files in the list.
"""
input_file = self.find_input_file()
ll = pydianaIO.find_all_files(input_file)
lf = [l[:-10]+'.list' for l in ll]
lf = list(set(lf))
lf.sort()
return lf
[docs]
def make_diana_call(self,optional_args:str=None,exclude:list=None):
"""
Function for making the diana bash call for this sequence.
Parameters
----------
optional_args (str) : string with the optional flags to add to the call, like '-O'
exclude (list) : List of strings with the files to exclude when making the calls for all the files in the .list
Returns
-------
True if everything is successfull otherwise False
"""
super().make_diana_call(optional_args)
if self.all:
self.total_calls=[]
call = self.call
orig_file = self.variables['FILENAME']
lf = self.find_all_files()
self.use = [True]*len(lf)
if exclude is not None:
excl_idx=[]
for ex in exclude:
if isinstance(ex,int):
excl_idx.append(ex)
elif isinstance(ex,str):
exi=None
if "/" in ex:
exi = [kk for kk in range(len(lf)) if ex == lf[kk]]
else:
exi = [kk for kk in range(len(lf)) if ex in lf[kk]]
if len(exi)==0:
exi=None
if exi is not None:
excl_idx+=exi
for ex in excl_idx:
self.use[ex]=False
for f in lf:
idx = f[::-1].find('/')
f = f[-idx:]
self.variables['FILENAME']=f
super().make_diana_call(optional_args)
self.total_calls.append(self.call)
self.variables['FILENAME']=orig_file
self.call=call
[docs]
def get_calls(self):
"""
Get all the calls from the cfg.
Returns
-------
Returns list with all the generated bash calls.
"""
cc = super().get_calls()
return cc + [tc for tc,valid in zip(self.total_calls,self.use) if valid]
def __str__(self):
"""
Function for converting to string an overview of the cfg. Called when printing the module.
Returns
-------
String with an overview of the class
"""
os=super().__str__()
if self.all and self.total_calls is not None:
os+='\nTotal Calls:\n'
for i,tc in enumerate(self.total_calls):
os+=f'{i})'
if not self.use[i]:
os+=' [EXCLUDED] '
else:
os+=' '
os+=tc+"\n\n"
return os