Merge pull request #54 from des-science/desdm-meds

ENH retry on race condition, make obj map for MEDS, and add DESMEDS

bin/run-eastlake-sim

@@ -65,8 +65,9 @@ def parse_args():
     parser.add_argument(
         '--resume', action='store_true', default=False,
         help=(
-            "If passed, implies --skip_completed_steps and "
-            "the restart file must be in the default location "
+            "If passed and --step_names is not given, "
+            "implies --skip_completed_steps and "
+            "The restart file must be in the default location "
             "(i.e., 'BASE_DIR/job_record.pkl')."
         ),
     )
@@ -115,7 +116,10 @@ def main():
                 )
             )
     if args.resume:
-        args.skip_completed_steps = True
+        if args.step_names is None:
+            args.skip_completed_steps = True
+        else:
+            args.skip_completed_steps = False
 
     # Setup the logger
     # If we're resuming a pipeline (ie. if args.job_record_file is not None),

eastlake/config/Y6A1_v1_meds-desdm-Y6A1v11.yaml

@@ -0,0 +1,81 @@
+# Originally https://github.com/beckermr/des-y6-analysis/blob/main/2022_09_20_color_dep_meds/Y6A1_test_piff.yaml
+# Color ranges (gi_color_range and iz_color_range have been updated)
+# MRB - changed to use astrorefine WCS for use in sims
+medsconf: "Y6A1_v1_meds-desdm-Y6A1v11"
+campaign: "Y6A2_COADD"
+piff_campaign: "Y6A2_PIFF_V3"
+
+joblib:
+  max_workers: 8
+  backend: loky
+
+psf:
+  se:
+    type: piff
+    stamp_size: 25
+    use_color: True
+    gi_color_range: [0.0, 3.5]
+    iz_color_range: [0.0, 0.65]
+
+  coadd:
+    type: psfex
+    use_color: False
+
+# extensions for single-epoch images
+# the ngwint files
+se_image_ext:  "sci"
+se_weight_ext: "wgt"
+se_bmask_ext:  "msk"
+
+se_astrom:
+  use_color: False
+
+# separate bkg files
+se_bkg_ext:    "sci"
+
+# separate seg files
+se_seg_ext:    "sci"
+
+# coadd images
+coadd_image_ext:  "sci"
+coadd_weight_ext: "wgt"
+coadd_bmask_ext:  "msk"
+
+# separate segmap file
+coadd_seg_ext:    "sci"
+
+# coadds are already background subtracted
+coadd_bkg_ext: "none"
+
+# DESDM prefers to have the compression keywords instead
+# of fpack command line arguments
+
+fpack_pars:
+  FZQVALUE: 4
+  FZTILE: "(10240,1)"
+  FZALGOR: "RICE_1"
+  FZQMETHD: "SUBTRACTIVE_DITHER_2"
+
+source_type: "finalcut"
+
+use_astro_refine: True
+coadd_astrom:
+  use_color: False
+
+# allowed_box_sizes: [32, 48, 64, 96, 128, 192, 256]
+allowed_box_sizes: [40, 56, 72, 104, 136, 200, 264]
+min_box_size: 40
+max_box_size: 264
+
+# special bmask flags that we zero out in the weight map. This should be
+# restricted to BADAMP (8) currently, which is designed to deal with the fact
+# that the good half of ccd 31 was used for coadds but not the bad half.  The
+# bad part will be included in the MEDS files and potentially misused.  Of
+# course users should check the bitmask but in the previous MEDS files ccd 31
+# was not included so codes may not be checking for BADAMP
+
+
+unusable_bmask: 8
+
+comment: |
+  Updated for piff color dep

eastlake/pipeline.py

@@ -26,6 +26,7 @@
     MetadetectRunner,
     StashPrep,
     BalrogRunner,
+    DESDMMEDSRunner,
 )
 from .utils import get_logger, safe_mkdir, pushd
 from .stash import Stash
@@ -49,6 +50,7 @@
     ('stash_prep', StashPrep),
     ('balrog', BalrogRunner),
     ('delete_sources', DeleteSources),
+    ('desdm_meds', DESDMMEDSRunner),
 ])
 
 STEP_IS_GALSIM = set(["galsim"])

eastlake/stash.py

@@ -414,39 +414,110 @@ def get_input_pizza_cutter_yaml(self, tilename, band):
 
     def _make_lists_psfmaps_symlinks(
         self, base_dir_or_imsim_data, tilename, band, pyml, skip_existing=False,
+    ):
+        # retry here for race conditions
+        import random
+        import time
+
+        ntry = 10
+        for i in range(ntry):
+            stime = 1.25**i * random.uniform(0.9, 1.0)
+            try:
+                self._make_lists_psfmaps_symlinks_impl(
+                    base_dir_or_imsim_data, tilename, band, pyml, skip_existing=skip_existing,
+                )
+            except Exception as e:
+                if i == ntry-1:
+                    raise e
+                else:
+                    time.sleep(stime)
+
+    def _make_lists_psfmaps_symlinks_impl(
+        self, base_dir_or_imsim_data, tilename, band, pyml, skip_existing=False,
     ):
         odir = os.path.join(base_dir_or_imsim_data, self["desrun"], tilename)
 
         ##############################################
-        # make bkg and nullwt flist files
-        os.makedirs(os.path.join(odir, "lists"), exist_ok=True)
-        bkg_file = os.path.join(
-            odir, "lists", f"{tilename}_{band}_bkg-flist-{self['desrun']}.dat",
+        # make lists and file conf stuff
+        meds_path = os.path.join(
+            odir,
+            "%s_%s_meds-%s.fits.fz" % (tilename, band, self["desrun"]),
         )
-        if not (skip_existing and os.path.exists(bkg_file)):
-            with open(bkg_file, "w") as fp_bkg:
-                for i in range(len(pyml["src_info"])):
-                    fp_bkg.write(pyml["src_info"][i]["bkg_path"] + "\n")
-
-        nw_file = os.path.join(
-            odir, "lists", f"{tilename}_{band}_nullwt-flist-{self['desrun']}.dat",
+        fileconf_data = {
+            "band": band,
+            "tilename": tilename,
+            "coadd_image_url": pyml["image_path"],
+            "coadd_cat_url": pyml["cat_path"],
+            "coadd_seg_url": pyml["seg_path"],
+            "coadd_psf_url": pyml["psf_path"],
+            "coadd_magzp": 30,
+            "coadd_object_map": self.get_filepaths(
+                "coadd_object_map", tilename, band=band, keyerror=False
+            ),
+            "meds_url": meds_path,
+            "coaddinfo": get_pizza_cutter_yaml_path(
+                self["base_dir"],
+                self["desrun"],
+                tilename,
+                band,
+            ),
+        }
+
+        # there are two versions of the lists files, one in a single list dir and one per band
+        for ldir in ["lists", f"lists-{band}"]:
+            for flist_name, key in [
+                ("bkg_flist", "bkg_path"),
+                ("piff_flist", "piff_path"),
+                ("psf_flist", "psf_path"),
+                ("seg_flist", "seg_path"),
+                ("nullwt_flist", "coadd_nwgint_path"),
+                ("nwgint_flist", "coadd_nwgint_path"),
+                ("finalcut_flist", "image_path"),
+            ]:
+                fname = os.path.join(
+                    odir,
+                    ldir,
+                    f"{tilename}_{band}_{flist_name.replace('_', '-')}-{self['desrun']}.dat",
+                )
+                if (
+                    (not (skip_existing and os.path.exists(fname)))
+                    and any(
+                        key in pyml["src_info"][i]
+                        for i in range(len(pyml["src_info"]))
+                    )
+                ):
+                    os.makedirs(os.path.dirname(fname), exist_ok=True)
+                    with open(fname, "w") as fp:
+                        if flist_name in ["finalcut_flist", "nullwt_flist", "nwgint_flist"]:
+                            for i in range(len(pyml["src_info"])):
+                                if ldir == "lists":
+                                    fp.write(
+                                            "%s %r\n" % (
+                                                pyml["src_info"][i][key],
+                                                pyml["src_info"][i]["magzp"],
+                                            )
+                                        )
+                                else:
+                                    fp.write(
+                                            "%s %s %r\n" % (
+                                                pyml["src_info"][i][key],
+                                                pyml["src_info"][i]["head_path"],
+                                                pyml["src_info"][i]["magzp"],
+                                            )
+                                        )
+                        else:
+                            for i in range(len(pyml["src_info"])):
+                                fp.write("%s\n" % (pyml["src_info"][i][key],))
+
+                if ldir != "lists":
+                    fileconf_data[flist_name] = fname
+
+        fname = os.path.join(
+            odir, f"lists-{band}", f"{tilename}_{band}_fileconf-{self['desrun']}.yaml",
         )
-        if (
-            (not (skip_existing and os.path.exists(nw_file)))
-            and any(
-                "coadd_nwgint_path" in pyml["src_info"][i]
-                for i in range(len(pyml["src_info"]))
-            )
-        ):
-            with open(nw_file, "w") as fp_nw:
-                for i in range(len(pyml["src_info"])):
-                    if "coadd_nwgint_path" in pyml["src_info"][i]:
-                        fp_nw.write(
-                            "%s %r\n" % (
-                                pyml["src_info"][i]["coadd_nwgint_path"],
-                                pyml["src_info"][i]["magzp"],
-                            )
-                        )
+        with open(fname, "w") as fp:
+            yaml.dump(fileconf_data, fp)
+        self.set_filepaths("desdm-fileconf", fname, tilename, band=band)
 
         # make psf map files
         pmap_file = os.path.join(odir, f"{tilename}_{band}_psfmap-{self['desrun']}.dat")

eastlake/steps/__init__.py

@@ -12,3 +12,4 @@
 from .stash_prep import StashPrep  # noqa
 from .balrog import BalrogRunner  # noqa
 from .delete_sources import DeleteSources  # noqa
+from .desdm_meds import DESDMMEDSRunner  # noqa

eastlake/steps/balrog.py

@@ -150,7 +150,7 @@ def execute(self, stash, new_params=None):
                     )
 
         # update the stash with PSF info for downstream w/ MEDS
-        stash["psf_config"] = {"type": "DES_PSFEx"}
+        stash["psf_config"] = {"type": "DES_Piff"}
         stash["draw_method"] = "no_pixel"
 
         return 0, stash