Newer
Older
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
import argparse
import os
import yaml
from detection import Detection
def parse_args():
'''
Parse command line arguments. Many of the following
can be set in the .yaml config file as well.
'''
parser = argparse.ArgumentParser()
parser.add_argument('config_file', help='.yaml config file for injection settings.')
parser.add_argument('-c', '--config_dir', help='Directory that houses the .yaml config file.')
# parser.add_argument('-d', '--data_dir', help='Directory that houses the input data.')
# parser.add_argument('-w', '--work_dir', help='The path for output.')
return parser.parse_args()
class MeasurementPipeline(object):
def __init__(self):
# Load configuration
args = parse_args()
if args.config_dir is None:
args.config_dir = ''
args.config_dir = os.path.abspath(args.config_dir)
args.config_file = os.path.join(args.config_dir, args.config_file)
with open(args.config_file, "r") as stream:
try:
self.config = yaml.safe_load(stream)
for key, value in self.config.items():
print (key + " : " + str(value))
except yaml.YAMLError as exc:
print(exc)
with open(self.config["measurement_setting"]["input_img_list"]) as input_list:
self.img_list = [line.rstrip() for line in input_list]
with open(self.config["measurement_setting"]["input_wht_list"]) as input_list:
self.wht_list = [line.rstrip() for line in input_list]
with open(self.config["measurement_setting"]["input_flg_list"]) as input_list:
self.flg_list = [line.rstrip() for line in input_list]
if self.config["measurement_setting"]["input_psf_list"]:
with open(self.config["measurement_setting"]["input_psf_list"]) as input_list:
self.psf_list = [line.rstrip() for line in input_list]
else:
self.psf_list = None
def run_pipeline(self):
# Detection
detection = Detection(
n_jobs=self.config["measurement_setting"]["n_jobs"],
output_dir=os.path.join(
self.config["measurement_setting"]["output_dir"],
self.config["run_name"]
)
)
detection.run_detection(
image_list=self.img_list,
weight_list=self.wht_list,
flag_list=self.flg_list,
psf_list=self.psf_list,
sex_config=self.config["measurement_setting"]["sex_config"],
sex_param=self.config["measurement_setting"]["sex_param"],
)
if __name__ == "__main__":
pipeline = MeasurementPipeline()
pipeline.run_pipeline()