run_measurement.py

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()