diff --git a/romancal/regtest/regtestdata.py b/romancal/regtest/regtestdata.py index 0cdd8393e..b3464b6c1 100644 --- a/romancal/regtest/regtestdata.py +++ b/romancal/regtest/regtestdata.py @@ -25,6 +25,8 @@ from deepdiff.operator import BaseOperator from gwcs.wcstools import grid_from_bounding_box +from romancal.associations.load_asn import load_asn + # from romancal.lib.suffix import replace_suffix from romancal.stpipe import RomanStep @@ -225,51 +227,51 @@ def get_truth(self, path=None, docopy=None): return self.truth - # def get_asn(self, path=None, docopy=True, get_members=True): - # """Copy association and association members from Artifactory remote - # resource to the CWD/truth. - # - # Updates self.input and self.input_remote upon completion - # - # Parameters - # ---------- - # path: str - # The remote path - # - # docopy : bool - # Switch to control whether or not to copy a file - # into the test output directory when running the test. - # If you wish to open the file directly from remote - # location or just to set path to source, set this to `False`. - # Default: `True` - # - # get_members: bool - # If an association is the input, retrieve the members. - # Otherwise, do not. - # """ - # if path is None: - # path = self.input_remote - # else: - # self.input_remote = path - # if docopy is None: - # docopy = self.docopy - # - # # Get the association JSON file - # self.input = get_bigdata(self._inputs_root, self._env, path, - # docopy=docopy) - # with open(self.input) as fp: - # asn = load_asn(fp) - # self.asn = asn - # - # # Get each member in the association as well - # if get_members: - # for product in asn['products']: - # for member in product['members']: - # fullpath = os.path.join( - # os.path.dirname(self.input_remote), - # member['expname']) - # get_bigdata(self._inputs_root, self._env, fullpath, - # docopy=self.docopy) + def get_asn(self, path=None, docopy=True, get_members=True): + """Copy association and association members from Artifactory remote + resource to the CWD/truth. + + Updates self.input and self.input_remote upon completion + + Parameters + ---------- + path: str + The remote path + + docopy : bool + Switch to control whether or not to copy a file + into the test output directory when running the test. + If you wish to open the file directly from remote + location or just to set path to source, set this to `False`. + Default: `True` + + get_members: bool + If an association is the input, retrieve the members. + Otherwise, do not. + """ + if path is None: + path = self.input_remote + else: + self.input_remote = path + if docopy is None: + docopy = self.docopy + + # Get the association JSON file + self.input = get_bigdata(self._inputs_root, self._env, path, docopy=docopy) + with open(self.input) as fp: + asn = load_asn(fp) + self.asn = asn + + # Get each member in the association as well + if get_members: + for product in asn["products"]: + for member in product["members"]: + fullpath = os.path.join( + os.path.dirname(self.input_remote), member["expname"] + ) + get_bigdata( + self._inputs_root, self._env, fullpath, docopy=self.docopy + ) def to_asdf(self, path): tree = eval(str(self)) diff --git a/romancal/regtest/test_mos_pipeline.py b/romancal/regtest/test_mos_pipeline.py index 885dc90a0..460df26df 100644 --- a/romancal/regtest/test_mos_pipeline.py +++ b/romancal/regtest/test_mos_pipeline.py @@ -23,19 +23,7 @@ def passfail(bool_expr): @metrics_logger("DMS356", "DMS374") def test_level3_mos_pipeline(rtdata, ignore_asdf_paths): """Tests for level 3 processing requirements DMS356""" - - cal_files = [ - "WFI/image/r0000101001001001001_01101_0001_WFI01_cal.asdf", - "WFI/image/r0000101001001001001_01101_0002_WFI01_cal.asdf", - "WFI/image/r0000101001001001001_01101_0003_WFI01_cal.asdf", - ] - - for cal_file in cal_files: - rtdata.get_data(cal_file) - - input_asn = "L3_regtest_asn.json" - rtdata.get_data(f"WFI/image/{input_asn}") - rtdata.input = input_asn + rtdata.get_asn("WFI/image/L3_regtest_asn.json") # Test Pipeline output = "r0099101001001001001_F158_visit_i2d.asdf" @@ -135,19 +123,7 @@ def test_level3_mos_pipeline(rtdata, ignore_asdf_paths): @metrics_logger("DMS373") def test_hlp_mosaic_pipeline(rtdata, ignore_asdf_paths): """Tests for level 3 mosaic requirements DMS373""" - - cal_files = [ - "WFI/image/r0000101001001001001_01101_0001_WFI01_cal.asdf", - "WFI/image/r0000101001001001001_01101_0002_WFI01_cal.asdf", - "WFI/image/r0000101001001001001_01101_0003_WFI01_cal.asdf", - ] - - for cal_file in cal_files: - rtdata.get_data(cal_file) - - input_asn = "L3_mosaic_asn.json" - rtdata.get_data(f"WFI/image/{input_asn}") - rtdata.input = input_asn + rtdata.get_asn("WFI/image/L3_mosaic_asn.json") # Test Pipeline output = "r0099101001001001001_r274dp63x31y81_prompt_F158_i2d.asdf" diff --git a/romancal/regtest/test_resample.py b/romancal/regtest/test_resample.py index 5fdda5130..bcdc588a4 100644 --- a/romancal/regtest/test_resample.py +++ b/romancal/regtest/test_resample.py @@ -12,18 +12,11 @@ @metrics_logger("DMS342", "DMS343", "DMS344", "DMS345") @pytest.mark.bigdata def test_resample_single_file(rtdata, ignore_asdf_paths): - input_data = [ - "r0000101001001001001_01101_0001_WFI01_cal.asdf", - "r0000101001001001001_01101_0002_WFI01_cal.asdf", - ] output_data = "mosaic_resamplestep.asdf" - [rtdata.get_data(f"WFI/image/{data}") for data in input_data] - asnfn = "mosaic_asn.json" - rtdata.get_data(f"WFI/image/{asnfn}") + rtdata.get_asn("WFI/image/mosaic_asn.json") rtdata.get_truth(f"truth/WFI/image/{output_data}") - rtdata.input = asnfn rtdata.output = output_data # instantiate ResampleStep (for running and log access)