forked from ajstewart/PiLL
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathLBA_pipeline.py
executable file
·173 lines (147 loc) · 6.5 KB
/
LBA_pipeline.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
#! /usr/bin/env python
# -*- coding: utf-8 -*-
"""
See http://www.astron.nl/citt/genericpipeline/ for further information on parsets.
"""
import os, sys
import logging
import resource
import optparse
_version = '1.0'
os.system('clear')
print '\033[30;1m################################################'
print '## LOFAR LBA calibration and imaging pipeline ##'
print '################################################\033[0m'
def add_coloring_to_emit_ansi(fn):
def new(*args):
levelno = args[0].levelno
if(levelno>=50):
color = '\x1b[31m' # red
pass
elif(levelno>=40):
color = '\x1b[31m' # red
pass
elif(levelno>=30):
color = '\x1b[33m' # yellow
pass
elif(levelno>=20):
color = '\x1b[32m' # green
pass
elif(levelno>=10):
color = '\x1b[35m' # pink
pass
else:
color = '\x1b[0m' # normal
pass
args[0].msg = color + args[0].msg + '\x1b[0m' # normal
return fn(*args)
pass
return new
pass
def create_clusterdesc(working_directory):
os.system('sed "s/\/any/localhost/g" $LOFARROOT/share/local.clusterdesc > ' + working_directory + '/pipeline.clusterdesc')
pass
def create_pipeline_config(working_directory):
try:
default_config = os.environ['LOFARROOT'] + '/share/pipeline/pipeline.cfg'
default_runtime = os.popen('grep runtime_directory ' + default_config + ' | cut -f2- -d"="').readlines()[0].rstrip('\n').replace(' ','')
default_working = os.popen('grep working_directory ' + default_config + ' | cut -f2- -d"="').readlines()[0].rstrip('\n').replace(' ','')
default_clusterdesc = os.popen('grep clusterdesc ' + default_config + ' | cut -f2- -d"="').readlines()[0].rstrip('\n').replace(' ','')
default_logfile = os.popen('grep log_file ' + default_config + ' | cut -f2- -d"="').readlines()[0].rstrip('\n').replace(' ','')
default_xml = os.popen('grep xml_stat_file ' + default_config + ' | cut -f2- -d"="').readlines()[0].rstrip('\n').replace(' ','')
pipeline_cfg = working_directory + '/pipeline.cfg'
with open(pipeline_cfg, 'w') as outfile:
with open(default_config, 'r') as infile:
for line in infile:
outfile.write(line.replace(default_runtime, working_directory)\
.replace(default_working, '%(runtime_directory)s')\
.replace(default_clusterdesc, '%(working_directory)s/pipeline.clusterdesc')\
.replace(default_logfile, '%(runtime_directory)s/%(job_name)s/logs/%(start_time)s/pipeline.log')\
.replace(default_xml, '%(runtime_directory)s/%(job_name)s/logs/%(start_time)s/statistics.xml'))
pass
try:
max_per_node = os.popen('nproc').readlines()[0].rstrip('\n')
os.system('echo >> ' + pipeline_cfg)
os.system('echo [remote] >> ' + pipeline_cfg)
os.system('echo method = local >> ' + pipeline_cfg)
os.system('echo max_per_node = ' + max_per_node + ' >> ' + pipeline_cfg)
pass
except IndexError:
logging.error('The number of available CPUs could not be determined. Please check your installation of nproc.')
sys.exit(1)
pass
except IOError or IndexError:
logging.error('LOFAR pipeline configuration not found. Please check your installation.')
sys.exit(1)
pass
infile.close()
outfile.close()
pass
if __name__=='__main__':
# Get command-line options.
opt = optparse.OptionParser(usage='%prog <pipeline.parset> <output_directory> ', version='%prog '+_version, description=__doc__)
opt.add_option('-c', '--clobber', help='clobber output directory', action='store_true', default=False)
(options, args) = opt.parse_args()
logging.root.setLevel(logging.INFO)
log = logging.StreamHandler()
format = logging.Formatter('\033[1m%(levelname)s\033[0m: %(message)s')
log.setFormatter(format)
log.emit = add_coloring_to_emit_ansi(log.emit)
logging.root.addHandler(log)
# Get inputs
if len(args) != 2:
logging.error('Wrong number of arguments.')
opt.print_help()
sys.exit(1)
pass
logging.info('Checking pipeline parset: \033[34m' + args[0])
if not os.path.isfile(args[0]):
logging.error('Pipeline parset does not exist.')
sys.exit(1)
pass
working_directory = args[1].rstrip('.').rstrip('/')
logging.info('Checking working directory: \033[34m' + working_directory)
if os.path.isdir(working_directory) and not options.clobber:
prompt = "\033[1;35mWARNING\033[0m: Output directory already exists. Press enter to clobber or 'q' to quit : "
answer = raw_input(prompt)
while answer != '':
if answer == 'q':
sys.exit(0)
pass
answer = raw_input(prompt)
pass
#logging.info('Cleaning working directory \033[5m...')
#os.system('rm -rfv ' + working_directory)
#pass
os.system('mkdir -pv ' + working_directory)
# checking number of files limit
try:
nof_files_limits = resource.getrlimit(resource.RLIMIT_NOFILE)
logging.info('Setting limit for number of open files to: {}.'.format(nof_files_limits[1]))
resource.setrlimit(resource.RLIMIT_NOFILE,(nof_files_limits[1],nof_files_limits[1]))
nof_files_limits = resource.getrlimit(resource.RLIMIT_NOFILE)
logging.info('Active limit for number of open files is {0}, maximum limit is {1}.'.format(nof_files_limits[0],nof_files_limits[1]))
if nof_files_limits[0] < 2048:
logging.error('The limit for number of open files is small, this could results in a "Too many open files" problem when running PiLL.')
logging.error('The active limit can be increased to the maximum for the user with: "ulimit -Sn <number>" (bash) or "limit descriptors 1024" (csh).')
pass
pass
except resource.error:
logging.error('Cannot check limits for number of open files.')
pass
# creating cluster description file
create_clusterdesc(working_directory)
logging.info('Created local cluster description file: \033[34m' + working_directory + '/pipeline.clusterdesc')
# creating pipeline configuration
create_pipeline_config(working_directory)
logging.info('Created pipeline configuration file: \033[34m' + working_directory + '/pipeline.cfg')
## creating pipeline parameter set
os.system('cp ' + args[0] + ' ' + working_directory + '/pipeline.parset')
logging.info('Created pipeline configuration file: \033[34m' + working_directory + '/pipeline.parset')
# starting of generic pipeline
logging.info('Calibration is starting \033[5m...')
os.system('genericpipeline.py ' + working_directory + '/pipeline.parset -v -c ' + working_directory + '/pipeline.cfg')
# calibration has been finished
logging.info('\033[30;4mCalibration has been finished.')
sys.exit(0)
pass