Source code for pipeline.infrastructure.new_pipeline_task.verify_new_pipeline_task

"""
Verify a new CASA pipeline task.

The purpose of this check a newly created pipeline task.
"""
import os
import sys
import argparse

import pipeline
pipeline.initcli()

[docs]class NewTask(): def __init__(self): self.repository_path = os.environ['SCIPIPE_HEURISTICS']
[docs] def parse_command_line(self, argv): script_args = '' for idx, arg in enumerate(argv): if 'verify_new_pipeline_task.py' in arg: script_args = argv[idx + 1:] parser = argparse.ArgumentParser() parser.add_argument('--package', help="Pipeline package. One of 'h', 'hif', 'hifa', 'hifas', 'hifv', or 'hsd'.", type=str, choices=['h', 'hif', 'hifa', 'hifas', 'hifv', 'hsd'], required=True) parser.add_argument('--task', help='New task name', type=str, required=True) parser.add_argument('--test-ms', dest='vis', help='Test Measurement Set', type=str, required=True) try: args = parser.parse_args(script_args) except: print('Problem with new_pipeline_task arguments.') return '', '', '' area = args.package task_name = args.task vis = args.vis return area, task_name, vis
[docs] def verify(self, area, task_name, vis): # --------------------------- # prepare to check the task # --------------------------- h_init() h_save() hifv_importdata(vis=[vis], session=['session_1'], overwrite=False) h_save() context = pipeline.Pipeline(context='last').context # --------------------------- # execute the task # --------------------------- inputs = eval('pipeline.{area}.tasks.{task}.{captask}.Inputs(context)'.format( area=area, task=task_name, captask=task_name.capitalize())) task = eval('pipeline.{area}.tasks.{task}.{captask}(inputs)'.format( area=area, task=task_name, captask=task_name.capitalize())) result = task.execute(dry_run=False) result.accept(context) context.save() # -------------------------------- # run as a registered CASA task # -------------------------------- try: eval('{area}_{task}()'.format(area=area, task=task_name)) h_save() except NameError as ee: print('ERROR: {msg}'.format(msg=str(ee))) print('\tTry using runsetup to register the new task with CASA first.')
if __name__ == '__main__': new_task = NewTask() area, task_name, vis = new_task.parse_command_line(sys.argv) if not area or not task_name or not vis: print('area, task_name and test-ms all need to be defined') else: new_task.verify(area, task_name, vis)