From 3e92054efa40ed71c460c3c387d64f4ecab64681 Mon Sep 17 00:00:00 2001
From: Wilson Beebe <wbeebe@uw.edu>
Date: Tue, 15 Oct 2024 11:08:53 -0700
Subject: [PATCH] Add multi_night_workflow for ingesting multi-night
 ImageCollections

---
 src/kbmod_wf/multi_night_workflow.py          | 161 ++++++++++
 .../resource_configs/klone_configuration.py   |  14 +-
 ...roject_multi_chip_multi_night_from_uris.py | 289 ++++++++++++++++++
 .../reproject_multi_chip_multi_night_wu.py    | 170 +----------
 src/kbmod_wf/workflow_tasks/kbmod_search.py   |   2 +-
 src/kbmod_wf/workflow_tasks/reproject_wu.py   |   2 +-
 .../reproject_wu_multi_night.py               |  56 ++++
 7 files changed, 523 insertions(+), 171 deletions(-)
 create mode 100644 src/kbmod_wf/multi_night_workflow.py
 create mode 100644 src/kbmod_wf/task_impls/reproject_multi_chip_multi_night_from_uris.py
 create mode 100644 src/kbmod_wf/workflow_tasks/reproject_wu_multi_night.py

diff --git a/src/kbmod_wf/multi_night_workflow.py b/src/kbmod_wf/multi_night_workflow.py
new file mode 100644
index 0000000..4b176f6
--- /dev/null
+++ b/src/kbmod_wf/multi_night_workflow.py
@@ -0,0 +1,161 @@
+import argparse
+import os
+
+import toml
+import parsl
+from parsl import python_app, File
+import parsl.executors
+
+from kbmod_wf.utilities import (
+    apply_runtime_updates,
+    get_resource_config,
+    get_executors,
+    get_configured_logger,
+)
+
+from kbmod_wf.workflow_tasks import create_manifest, ic_to_wu, reproject_wu_multi_night, kbmod_search
+
+
+@python_app(
+    cache=True,
+    executors=get_executors(["local_dev_testing", "sharded_reproject"]),
+    ignore_for_cache=["logging_file"],
+)
+def reproject_wu(inputs=(), outputs=(), runtime_config={}, logging_file=None):
+    from kbmod_wf.utilities.logger_utilities import get_configured_logger, ErrorLogger
+
+    logger = get_configured_logger("task.reproject_wu", logging_file.filepath)
+
+    from kbmod_wf.task_impls.reproject_multi_chip_multi_night_wu import reproject_wu
+    guess_dist = inputs[1] # heliocentric guess distance in AU
+    logger.info(f"Starting reproject_ic for guess distance {guess_dist}")
+    with ErrorLogger(logger):
+        reproject_wu(
+            guess_dist,
+            original_wu_filepath=inputs[0].filepath,
+            reprojected_wu_filepath=outputs[0].filepath,
+            runtime_config=runtime_config,
+            logger=logger,
+        )
+    logger.info("Completed reproject_ic")
+    return outputs[0]
+
+
+def workflow_runner(env=None, runtime_config={}):
+    """This function will load and configure Parsl, and run the workflow.
+
+    Parameters
+    ----------
+    env : str, optional
+        Environment string used to define which resource configuration to use,
+        by default None
+    runtime_config : dict, optional
+        Dictionary of assorted runtime configuration parameters, by default {}
+    """
+    resource_config = get_resource_config(env=env)
+    resource_config = apply_runtime_updates(resource_config, runtime_config)
+
+    app_configs = runtime_config.get("apps", {})
+
+    dfk = parsl.load(resource_config)
+    if dfk:
+        logging_file = File(os.path.join(dfk.run_dir, "kbmod.log"))
+        logger = get_configured_logger("workflow.workflow_runner", logging_file.filepath)
+
+        if runtime_config is not None:
+            logger.info(f"Using runtime configuration definition:\n{toml.dumps(runtime_config)}")
+
+        logger.info("Starting workflow")
+
+        # gather all the *.collection files that are staged for processing
+        create_manifest_config = app_configs.get("create_manifest", {})
+        manifest_file = File(
+            os.path.join(create_manifest_config.get("output_directory", os.getcwd()), "manifest.txt")
+        )
+        create_manifest_future = create_manifest(
+            inputs=[],
+            outputs=[manifest_file],
+            runtime_config=app_configs.get("create_manifest", {}),
+            logging_file=logging_file,
+        )
+
+        wu_filenames =[]
+        with open(create_manifest_future.result(), "r") as f:
+            # process each .collection file in the manifest into a .wu file
+            original_work_unit_futures = []
+            collection_files = []
+            for line in f:
+                collection_file = File(line.strip())
+                collection_files.append(collection_file)
+                wu_filename = line + ".wu"
+                wu_filenames.append(wu_filename)
+                original_work_unit_futures.append(
+                    ic_to_wu(
+                        inputs=[collection_file],
+                        outputs=[File(wu_filename)],
+                        runtime_config=app_configs.get("ic_to_wu", {}),
+                        logging_file=logging_file,
+                    )
+                )
+
+        # reproject each WorkUnit for a range of distances
+        reproject_futures = []
+        repro_wu_filenames = []
+        for i in range(len(original_work_unit_futures)):
+            f = original_work_unit_futures[i]
+            for distance in [46.7, 30.6]: # The reprojected distance in AU
+                output_filename=wu_filenames[i]+ f".{distance}.repro"
+                repro_wu_filenames.append(output_filename)
+                reproject_futures.append(
+                    reproject_wu(
+                        inputs=[f, distance],
+                        outputs=[File(output_filename)],
+                        runtime_config=app_configs.get("reproject_wu", {}),
+                        logging_file=logging_file,
+                    )
+                )
+
+        # run kbmod search on each reprojected WorkUnit
+        search_futures = []
+        for i in range(len(reproject_futures)):
+            f = reproject_futures[i]
+            search_futures.append(
+                kbmod_search(
+                    inputs=[f],
+                    outputs=[File(repro_wu_filenames[i] + ".search.ecsv")],
+                    runtime_config=app_configs.get("kbmod_search", {}),
+                    logging_file=logging_file,
+                )
+            )
+
+        [f.result() for f in search_futures]
+
+        logger.info("Workflow complete")
+
+    parsl.clear()
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--env",
+        type=str,
+        choices=["dev", "klone"],
+        help="The environment to run the workflow in.",
+    )
+
+    parser.add_argument(
+        "--runtime-config",
+        type=str,
+        help="The complete runtime configuration filepath to use for the workflow.",
+    )
+
+    args = parser.parse_args()
+
+    # if a runtime_config file was provided and exists, load the toml as a dict.
+    runtime_config = {}
+    if args.runtime_config is not None and os.path.exists(args.runtime_config):
+        with open(args.runtime_config, "r") as toml_runtime_config:
+            runtime_config = toml.load(toml_runtime_config)
+
+    workflow_runner(env=args.env, runtime_config=runtime_config)
diff --git a/src/kbmod_wf/resource_configs/klone_configuration.py b/src/kbmod_wf/resource_configs/klone_configuration.py
index 953622b..da72c0d 100644
--- a/src/kbmod_wf/resource_configs/klone_configuration.py
+++ b/src/kbmod_wf/resource_configs/klone_configuration.py
@@ -18,9 +18,9 @@ def klone_resource_config():
         app_cache=True,
         checkpoint_mode="task_exit",
         checkpoint_files=get_all_checkpoints(
-            os.path.join("/gscratch/dirac/kbmod/workflow/run_logs", datetime.date.today().isoformat())
+            os.path.join("/gscratch/dirac/kbmod/workflow/run_logs/wbeebe", datetime.date.today().isoformat())
         ),
-        run_dir=os.path.join("/gscratch/dirac/kbmod/workflow/run_logs", datetime.date.today().isoformat()),
+        run_dir=os.path.join("/gscratch/dirac/kbmod/workflow/run_logs/wbeebe", datetime.date.today().isoformat()),
         retries=1,
         executors=[
             HighThroughputExecutor(
@@ -46,8 +46,8 @@ def klone_resource_config():
                 label="large_mem",
                 max_workers=1,
                 provider=SlurmProvider(
-                    partition="ckpt-g2",
-                    account="astro",
+                    partition="gpu-a40",
+                    account="escience",
                     min_blocks=0,
                     max_blocks=2,
                     init_blocks=0,
@@ -65,8 +65,8 @@ def klone_resource_config():
                 label="sharded_reproject",
                 max_workers=1,
                 provider=SlurmProvider(
-                    partition="ckpt-g2",
-                    account="astro",
+                    partition="gpu-a40",
+                    account="escience",
                     min_blocks=0,
                     max_blocks=2,
                     init_blocks=0,
@@ -84,7 +84,7 @@ def klone_resource_config():
                 label="gpu",
                 max_workers=1,
                 provider=SlurmProvider(
-                    partition="ckpt-g2",
+                    partition="gpu-a40",
                     account="escience",
                     min_blocks=0,
                     max_blocks=2,
diff --git a/src/kbmod_wf/task_impls/reproject_multi_chip_multi_night_from_uris.py b/src/kbmod_wf/task_impls/reproject_multi_chip_multi_night_from_uris.py
new file mode 100644
index 0000000..49fe440
--- /dev/null
+++ b/src/kbmod_wf/task_impls/reproject_multi_chip_multi_night_from_uris.py
@@ -0,0 +1,289 @@
+import kbmod
+from kbmod.work_unit import WorkUnit
+from kbmod.reprojection_utils import transform_wcses_to_ebd
+
+import kbmod.reprojection as reprojection
+from astropy.wcs import WCS
+from astropy.io import fits
+from astropy.coordinates import EarthLocation
+from astropy.time import Time
+import numpy as np
+import os
+import time
+from logging import Logger
+
+
+def reproject_wu(
+    original_wu_filepath: str = None,
+    uri_filepath: str = None,
+    reprojected_wu_filepath: str = None,
+    runtime_config: dict = {},
+    logger: Logger = None,
+):
+    """This task will perform reflex correction and reproject a WorkUnit to a common WCS.
+
+    Parameters
+    ----------
+    original_wu_filepath : str, optional
+        The fully resolved filepath to the input WorkUnit file, by default None
+    uri_filepath : str, optional
+        The fully resolved filepath to the original uri file. This is used
+        exclusively for the header contents, by default None
+    reprojected_wu_filepath : str, optional
+        The fully resolved filepath to the resulting WorkUnit file after reflex
+        and reprojection, by default None
+    runtime_config : dict, optional
+        Additional configuration parameters to be used at runtime, by default {}
+    logger : Logger, optional
+        Primary logger for the workflow, by default None
+
+    Returns
+    -------
+    str
+        The fully resolved filepath of the resulting WorkUnit file after reflex
+        and reprojection.
+    """
+    wu_reprojector = WUReprojector(
+        original_wu_filepath=original_wu_filepath,
+        uri_filepath=uri_filepath,
+        reprojected_wu_filepath=reprojected_wu_filepath,
+        runtime_config=runtime_config,
+        logger=logger,
+    )
+
+    return wu_reprojector.reproject_workunit()
+
+
+class WUReprojector:
+    def __init__(
+        self,
+        original_wu_filepath: str = None,
+        uri_filepath: str = None,
+        reprojected_wu_filepath: str = None,
+        runtime_config: dict = {},
+        logger: Logger = None,
+    ):
+        self.original_wu_filepath = original_wu_filepath
+        self.uri_filepath = uri_filepath
+        self.reprojected_wu_filepath = reprojected_wu_filepath
+        self.runtime_config = runtime_config
+        self.logger = logger
+        kbmod._logging.basicConfig(level=self.logger.level)
+
+        self.overwrite = self.runtime_config.get("overwrite", False)
+        self.search_config = self.runtime_config.get("search_config", None)
+
+        # Default to 8 workers if not in the config. Value must be 0<num workers<65.
+        self.n_workers = max(1, min(self.runtime_config.get("n_workers", 8), 64))
+
+        #! In the long run, we likely won't have the URI files to start from
+        #! So we'll need to rethink how we get these parameters.
+        self.uri_params = self._get_params_from_uri_file()
+        self.patch_size = self.uri_params["patch_size"]
+        self.pixel_scale = self.uri_params["pixel_scale"]
+        self.guess_dist = self.uri_params.get("dist_au", 40)
+        self.patch_corners = self.uri_params["patch_box"]
+
+        if "patch_size" not in self.uri_params:
+            raise KeyError(
+                f"Must supply image dimensions (image_width, image_height) or #patch_size= must be in a specified URI file."
+            )
+        if self.pixel_scale == None:
+            raise KeyError(
+                f"When patch pixel dimensions are not specifified, the user must supply a pixel scale via the command line or the uri file."
+            )
+
+        self.image_width, self.image_height = self._patch_arcmin_to_pixels()
+
+        self.point_on_earth = EarthLocation.of_site(self.runtime_config.get("observation_site", "ctio"))
+
+    def reproject_workunit(self):
+        # Create a WCS object for the patch. This will be our common reprojection WCS
+        self.logger.debug(f"Creating WCS from patch")
+        patch_wcs = self._create_wcs_from_corners(
+            self.patch_corners,
+            self.image_width,
+            self.image_height,
+            pixel_scale=self.pixel_scale,
+        )
+
+        last_time = time.time()
+        self.logger.info(f"Lazy reading existing WorkUnit from disk: {self.original_wu_filepath}")
+        directory_containing_shards, wu_filename = os.path.split(self.original_wu_filepath)
+        wu = WorkUnit.from_sharded_fits(wu_filename, directory_containing_shards, lazy=True)
+        elapsed = round(time.time() - last_time, 1)
+        self.logger.debug(
+            f"Required {elapsed}[s] to lazy read original WorkUnit {self.original_wu_filepath}."
+        )
+
+        #! This method to get image dimensions won't hold if the images are different sizes.
+        image_height, image_width = wu._per_image_wcs[0].array_shape
+
+        # Find the EBD (estimated barycentric distance) WCS for each image
+        last_time = time.time()
+        ebd_per_image_wcs, geocentric_dists = transform_wcses_to_ebd(
+            wu._per_image_wcs,
+            image_width,
+            image_height,
+            self.guess_dist,
+            Time(wu.get_all_obstimes(), format="mjd"),
+            self.point_on_earth,
+            npoints=10,
+            seed=None,
+        )
+        elapsed = round(time.time() - last_time, 1)
+        self.logger.debug(f"Required {elapsed}[s] to transform WCS objects to EBD..")
+
+        if len(wu._per_image_wcs) != len(ebd_per_image_wcs):
+            raise ValueError(
+                f"Number of barycentric WCS objects ({len(ebd_per_image_wcs)}) does not match the original number of images ({len(wu._per_image_wcs)})."
+            )
+
+        wu._per_image_ebd_wcs = ebd_per_image_wcs
+        wu.heliocentric_distance = self.guess_dist
+        wu.geocentric_distances = geocentric_dists
+
+        # Reproject to a common WCS using the WCS for our patch
+        self.logger.debug(f"Reprojecting WorkUnit with {self.n_workers} workers...")
+        last_time = time.time()
+
+        directory_containing_reprojected_shards, reprojected_wu_filename = os.path.split(
+            self.reprojected_wu_filepath
+        )
+        reprojection.reproject_lazy_work_unit(
+            wu,
+            patch_wcs,
+            directory_containing_reprojected_shards,
+            reprojected_wu_filename,
+            frame="ebd",
+            max_parallel_processes=self.n_workers,
+        )
+        elapsed = round(time.time() - last_time, 1)
+        self.logger.debug(f"Required {elapsed}[s] to create the sharded reprojected WorkUnit.")
+
+        return self.reprojected_wu_filepath
+
+    def _get_params_from_uri_file(self):
+        """
+        Get parameters we place into URI file as comments at the top.
+        Example start of URI file (6/6/2024 COC):
+        #desired_dates=['2019-04-02', '2019-05-07']
+        #dist_au=42.0
+        #patch_size=[20, 20]
+        #patch_id=5845
+        #patch_center_coords=(216.49999999999997, -13.500000000000005)
+        #patch_box=[[216.33333333333331, -13.666666666666671], [216.33333333333331, -13.333333333333337], [216.66666666666666, -13.333333333333337], [216.66666666666666, -13.666666666666671], [216.33333333333331, -13.666666666666671]]
+        /gscratch/dirac/DEEP/repo/DEEP/20190507/A0b/science#step6/20240425T145342Z/differenceExp/20190508/VR/VR_DECam_c0007_6300.0_2600.0/855719/differenceExp_DECam_VR_VR_DECam_c0007_6300_0_2600_0_855719_S12_DEEP_20190507_A0b_scienceHASHstep6_20240425T145342Z.fits
+        6/6/2024 COC
+        """
+        results = {}
+        with open(self.uri_filepath, "r") as f:
+            for line in f:
+                line = line.strip()
+                if line == "":
+                    continue  # deal with rogue blank lines or invisible double line endings
+                if not line.startswith("#"):
+                    break  # comments section is done
+                line = line.lstrip("#").split("=")
+                lhs = line[0].strip()
+                rhs = line[1].strip()
+                try:
+                    rhs = eval(rhs)
+                except ValueError:
+                    self.logger.debug(f"Unable to eval() {lhs} field with value {rhs}.")
+                    continue
+                results[lhs] = rhs
+        return results
+
+    def _patch_arcmin_to_pixels(self):
+        """Operate on the self.patch_size array (with size (2,1)) to convert to
+        pixels. Uses self.pixel_scale to do the conversion.
+
+        Returns
+        -------
+        nd.array
+            A 2d array with shape (2,1) containing the width and height of the
+            patch in pixels.
+        """
+        patch_pixels = np.ceil(np.array(self.patch_size) * 60 / self.pixel_scale).astype(int)
+
+        self.logger.debug(
+            f"Derived patch_pixels (w, h) = {patch_pixels} from patch_size={self.patch_size}[arcmin] and pixel_scale={self.pixel_scale}[arcsec/pixel]."
+        )
+
+        return patch_pixels
+
+    def _create_wcs_from_corners(
+        self,
+        corners=None,
+        image_width=None,
+        image_height=None,
+        pixel_scale=None,
+        verbose=True,
+    ):
+        """
+        Create a WCS object given the RA, Dec coordinates of the four corners of the image
+        and the dimensions of the image in pixels. Optionally save as a FITS file.
+
+        Parameters:
+        corners (list of lists): [[RA1, Dec1], [RA2, Dec2], [RA3, Dec3], [RA4, Dec4]]
+        image_width (int): Width of the image in pixels; if none, pixel_scale will be used to determine size.
+        image_height (int): Height of the image in pixels; if none, pixel_scale will be used to determine size.
+        filename (str): The name of the FITS file to save
+        pixel_scale (float): The pixel scale (in units of arcseconds per pixel); used if image_width or image_height is None.
+        verbose (bool): Print more messages.
+
+        Returns:
+        WCS: The World Coordinate System object for the image
+
+        5/6/2024 COC + ChatGPT4
+        """
+        # TODO switch to https://docs.astropy.org/en/stable/api/astropy.wcs.utils.fit_wcs_from_points.html
+        # Extract the corners
+        if verbose:
+            self.logger.debug(f"At the start, corners={corners}, type={type(corners)}")
+        if type(corners) == type((0, 1)) or len(corners) == 1:
+            corners = corners[0]
+            self.logger.debug(f"After un-tuple corners are {corners}.")
+        corners = np.unique(
+            corners, axis=0
+        )  # eliminate duplicate coords in case someone passes the repeat 1st corner used for plotting 6/5/2024 COC/DO
+        if len(corners) != 4:
+            raise ValueError(f"There should be four (4) corners. We saw: {corners}")
+        ra = [corner[0] for corner in corners]
+        dec = [corner[1] for corner in corners]
+        #
+        # Calculate the central position (average of the coordinates)
+        center_ra = np.mean(ra)
+        center_dec = np.mean(dec)
+        #
+        # Calculate the pixel scale in degrees per pixel
+        if pixel_scale is not None:
+            pixel_scale_ra = pixel_scale / 60 / 60
+            pixel_scale_dec = pixel_scale / 60 / 60
+        else:
+            ra_range = max(ra) - min(
+                ra
+            )  # * np.cos(np.radians(center_dec))  # Adjust RA difference for declination; do not use cos(), results in incorrect pixel scale 6/6/2024 COC
+            dec_range = max(dec) - min(dec)
+            pixel_scale_ra = ra_range / image_width
+            pixel_scale_dec = dec_range / image_height
+        if verbose:
+            self.logger.debug(
+                f'Saw (RA,Dec) pixel scales ({pixel_scale_ra*60*60},{pixel_scale_dec*60*60})"/pixel. User-supplied: {pixel_scale}"/pixel.'
+            )
+        # Initialize a WCS object with 2 axes (RA and Dec)
+        wcs = WCS(naxis=2)
+        wcs.wcs.crpix = [image_width / 2, image_height / 2]
+        wcs.wcs.crval = [center_ra, center_dec]
+        wcs.wcs.cdelt = [
+            -pixel_scale_ra,
+            pixel_scale_dec,
+        ]  # RA pixel scale might need to be negative (convention)
+        # Rotation matrix, assuming no rotation
+        wcs.wcs.pc = [[1, 0], [0, 1]]
+        # Define coordinate frame and projection
+        wcs.wcs.ctype = ["RA---TAN", "DEC--TAN"]
+        wcs.array_shape = (image_height, image_width)
+
+        return wcs
diff --git a/src/kbmod_wf/task_impls/reproject_multi_chip_multi_night_wu.py b/src/kbmod_wf/task_impls/reproject_multi_chip_multi_night_wu.py
index 49fe440..5981fa9 100644
--- a/src/kbmod_wf/task_impls/reproject_multi_chip_multi_night_wu.py
+++ b/src/kbmod_wf/task_impls/reproject_multi_chip_multi_night_wu.py
@@ -14,8 +14,8 @@
 
 
 def reproject_wu(
+    guess_dist: float,
     original_wu_filepath: str = None,
-    uri_filepath: str = None,
     reprojected_wu_filepath: str = None,
     runtime_config: dict = {},
     logger: Logger = None,
@@ -24,11 +24,10 @@ def reproject_wu(
 
     Parameters
     ----------
+    guess_dist: float
+        The heliocentric guess distance to reproject to in AU.
     original_wu_filepath : str, optional
         The fully resolved filepath to the input WorkUnit file, by default None
-    uri_filepath : str, optional
-        The fully resolved filepath to the original uri file. This is used
-        exclusively for the header contents, by default None
     reprojected_wu_filepath : str, optional
         The fully resolved filepath to the resulting WorkUnit file after reflex
         and reprojection, by default None
@@ -44,8 +43,8 @@ def reproject_wu(
         and reprojection.
     """
     wu_reprojector = WUReprojector(
+        guess_dist=guess_dist,
         original_wu_filepath=original_wu_filepath,
-        uri_filepath=uri_filepath,
         reprojected_wu_filepath=reprojected_wu_filepath,
         runtime_config=runtime_config,
         logger=logger,
@@ -57,14 +56,14 @@ def reproject_wu(
 class WUReprojector:
     def __init__(
         self,
+        guess_dist: float,
         original_wu_filepath: str = None,
-        uri_filepath: str = None,
         reprojected_wu_filepath: str = None,
         runtime_config: dict = {},
         logger: Logger = None,
     ):
+        self.guess_dist = guess_dist
         self.original_wu_filepath = original_wu_filepath
-        self.uri_filepath = uri_filepath
         self.reprojected_wu_filepath = reprojected_wu_filepath
         self.runtime_config = runtime_config
         self.logger = logger
@@ -76,37 +75,9 @@ def __init__(
         # Default to 8 workers if not in the config. Value must be 0<num workers<65.
         self.n_workers = max(1, min(self.runtime_config.get("n_workers", 8), 64))
 
-        #! In the long run, we likely won't have the URI files to start from
-        #! So we'll need to rethink how we get these parameters.
-        self.uri_params = self._get_params_from_uri_file()
-        self.patch_size = self.uri_params["patch_size"]
-        self.pixel_scale = self.uri_params["pixel_scale"]
-        self.guess_dist = self.uri_params.get("dist_au", 40)
-        self.patch_corners = self.uri_params["patch_box"]
-
-        if "patch_size" not in self.uri_params:
-            raise KeyError(
-                f"Must supply image dimensions (image_width, image_height) or #patch_size= must be in a specified URI file."
-            )
-        if self.pixel_scale == None:
-            raise KeyError(
-                f"When patch pixel dimensions are not specifified, the user must supply a pixel scale via the command line or the uri file."
-            )
-
-        self.image_width, self.image_height = self._patch_arcmin_to_pixels()
-
         self.point_on_earth = EarthLocation.of_site(self.runtime_config.get("observation_site", "ctio"))
 
     def reproject_workunit(self):
-        # Create a WCS object for the patch. This will be our common reprojection WCS
-        self.logger.debug(f"Creating WCS from patch")
-        patch_wcs = self._create_wcs_from_corners(
-            self.patch_corners,
-            self.image_width,
-            self.image_height,
-            pixel_scale=self.pixel_scale,
-        )
-
         last_time = time.time()
         self.logger.info(f"Lazy reading existing WorkUnit from disk: {self.original_wu_filepath}")
         directory_containing_shards, wu_filename = os.path.split(self.original_wu_filepath)
@@ -125,7 +96,7 @@ def reproject_workunit(self):
             wu._per_image_wcs,
             image_width,
             image_height,
-            self.guess_dist,
+            wu.heliocentric_dist,
             Time(wu.get_all_obstimes(), format="mjd"),
             self.point_on_earth,
             npoints=10,
@@ -152,7 +123,7 @@ def reproject_workunit(self):
         )
         reprojection.reproject_lazy_work_unit(
             wu,
-            patch_wcs,
+            wu.wcs, # Use the common WCS of the WorkUnit
             directory_containing_reprojected_shards,
             reprojected_wu_filename,
             frame="ebd",
@@ -162,128 +133,3 @@ def reproject_workunit(self):
         self.logger.debug(f"Required {elapsed}[s] to create the sharded reprojected WorkUnit.")
 
         return self.reprojected_wu_filepath
-
-    def _get_params_from_uri_file(self):
-        """
-        Get parameters we place into URI file as comments at the top.
-        Example start of URI file (6/6/2024 COC):
-        #desired_dates=['2019-04-02', '2019-05-07']
-        #dist_au=42.0
-        #patch_size=[20, 20]
-        #patch_id=5845
-        #patch_center_coords=(216.49999999999997, -13.500000000000005)
-        #patch_box=[[216.33333333333331, -13.666666666666671], [216.33333333333331, -13.333333333333337], [216.66666666666666, -13.333333333333337], [216.66666666666666, -13.666666666666671], [216.33333333333331, -13.666666666666671]]
-        /gscratch/dirac/DEEP/repo/DEEP/20190507/A0b/science#step6/20240425T145342Z/differenceExp/20190508/VR/VR_DECam_c0007_6300.0_2600.0/855719/differenceExp_DECam_VR_VR_DECam_c0007_6300_0_2600_0_855719_S12_DEEP_20190507_A0b_scienceHASHstep6_20240425T145342Z.fits
-        6/6/2024 COC
-        """
-        results = {}
-        with open(self.uri_filepath, "r") as f:
-            for line in f:
-                line = line.strip()
-                if line == "":
-                    continue  # deal with rogue blank lines or invisible double line endings
-                if not line.startswith("#"):
-                    break  # comments section is done
-                line = line.lstrip("#").split("=")
-                lhs = line[0].strip()
-                rhs = line[1].strip()
-                try:
-                    rhs = eval(rhs)
-                except ValueError:
-                    self.logger.debug(f"Unable to eval() {lhs} field with value {rhs}.")
-                    continue
-                results[lhs] = rhs
-        return results
-
-    def _patch_arcmin_to_pixels(self):
-        """Operate on the self.patch_size array (with size (2,1)) to convert to
-        pixels. Uses self.pixel_scale to do the conversion.
-
-        Returns
-        -------
-        nd.array
-            A 2d array with shape (2,1) containing the width and height of the
-            patch in pixels.
-        """
-        patch_pixels = np.ceil(np.array(self.patch_size) * 60 / self.pixel_scale).astype(int)
-
-        self.logger.debug(
-            f"Derived patch_pixels (w, h) = {patch_pixels} from patch_size={self.patch_size}[arcmin] and pixel_scale={self.pixel_scale}[arcsec/pixel]."
-        )
-
-        return patch_pixels
-
-    def _create_wcs_from_corners(
-        self,
-        corners=None,
-        image_width=None,
-        image_height=None,
-        pixel_scale=None,
-        verbose=True,
-    ):
-        """
-        Create a WCS object given the RA, Dec coordinates of the four corners of the image
-        and the dimensions of the image in pixels. Optionally save as a FITS file.
-
-        Parameters:
-        corners (list of lists): [[RA1, Dec1], [RA2, Dec2], [RA3, Dec3], [RA4, Dec4]]
-        image_width (int): Width of the image in pixels; if none, pixel_scale will be used to determine size.
-        image_height (int): Height of the image in pixels; if none, pixel_scale will be used to determine size.
-        filename (str): The name of the FITS file to save
-        pixel_scale (float): The pixel scale (in units of arcseconds per pixel); used if image_width or image_height is None.
-        verbose (bool): Print more messages.
-
-        Returns:
-        WCS: The World Coordinate System object for the image
-
-        5/6/2024 COC + ChatGPT4
-        """
-        # TODO switch to https://docs.astropy.org/en/stable/api/astropy.wcs.utils.fit_wcs_from_points.html
-        # Extract the corners
-        if verbose:
-            self.logger.debug(f"At the start, corners={corners}, type={type(corners)}")
-        if type(corners) == type((0, 1)) or len(corners) == 1:
-            corners = corners[0]
-            self.logger.debug(f"After un-tuple corners are {corners}.")
-        corners = np.unique(
-            corners, axis=0
-        )  # eliminate duplicate coords in case someone passes the repeat 1st corner used for plotting 6/5/2024 COC/DO
-        if len(corners) != 4:
-            raise ValueError(f"There should be four (4) corners. We saw: {corners}")
-        ra = [corner[0] for corner in corners]
-        dec = [corner[1] for corner in corners]
-        #
-        # Calculate the central position (average of the coordinates)
-        center_ra = np.mean(ra)
-        center_dec = np.mean(dec)
-        #
-        # Calculate the pixel scale in degrees per pixel
-        if pixel_scale is not None:
-            pixel_scale_ra = pixel_scale / 60 / 60
-            pixel_scale_dec = pixel_scale / 60 / 60
-        else:
-            ra_range = max(ra) - min(
-                ra
-            )  # * np.cos(np.radians(center_dec))  # Adjust RA difference for declination; do not use cos(), results in incorrect pixel scale 6/6/2024 COC
-            dec_range = max(dec) - min(dec)
-            pixel_scale_ra = ra_range / image_width
-            pixel_scale_dec = dec_range / image_height
-        if verbose:
-            self.logger.debug(
-                f'Saw (RA,Dec) pixel scales ({pixel_scale_ra*60*60},{pixel_scale_dec*60*60})"/pixel. User-supplied: {pixel_scale}"/pixel.'
-            )
-        # Initialize a WCS object with 2 axes (RA and Dec)
-        wcs = WCS(naxis=2)
-        wcs.wcs.crpix = [image_width / 2, image_height / 2]
-        wcs.wcs.crval = [center_ra, center_dec]
-        wcs.wcs.cdelt = [
-            -pixel_scale_ra,
-            pixel_scale_dec,
-        ]  # RA pixel scale might need to be negative (convention)
-        # Rotation matrix, assuming no rotation
-        wcs.wcs.pc = [[1, 0], [0, 1]]
-        # Define coordinate frame and projection
-        wcs.wcs.ctype = ["RA---TAN", "DEC--TAN"]
-        wcs.array_shape = (image_height, image_width)
-
-        return wcs
diff --git a/src/kbmod_wf/workflow_tasks/kbmod_search.py b/src/kbmod_wf/workflow_tasks/kbmod_search.py
index b572c31..0320834 100644
--- a/src/kbmod_wf/workflow_tasks/kbmod_search.py
+++ b/src/kbmod_wf/workflow_tasks/kbmod_search.py
@@ -36,7 +36,7 @@ def kbmod_search(inputs=(), outputs=(), runtime_config={}, logging_file=None):
     with ErrorLogger(logger):
         kbmod_search(
             wu_filepath=inputs[0].filepath,
-            result_filepath=outputs[0].filepath,
+            result_filepath=inputs[0].filepath + ".search.ecsv",
             runtime_config=runtime_config,
             logger=logger,
         )
diff --git a/src/kbmod_wf/workflow_tasks/reproject_wu.py b/src/kbmod_wf/workflow_tasks/reproject_wu.py
index 4d5911b..0476edf 100644
--- a/src/kbmod_wf/workflow_tasks/reproject_wu.py
+++ b/src/kbmod_wf/workflow_tasks/reproject_wu.py
@@ -39,7 +39,7 @@ def reproject_wu(inputs=(), outputs=(), runtime_config={}, logging_file=None):
 
     logger = get_configured_logger("task.ic_to_wu", logging_file)
 
-    from kbmod_wf.task_impls.reproject_multi_chip_multi_night_wu import reproject_wu
+    from kbmod_wf.task_impls.reproject_multi_chip_multi_night_wu_from_uris import reproject_wu
 
     logger.info("Starting reproject_ic")
     with ErrorLogger(logger):
diff --git a/src/kbmod_wf/workflow_tasks/reproject_wu_multi_night.py b/src/kbmod_wf/workflow_tasks/reproject_wu_multi_night.py
new file mode 100644
index 0000000..b2cd3b6
--- /dev/null
+++ b/src/kbmod_wf/workflow_tasks/reproject_wu_multi_night.py
@@ -0,0 +1,56 @@
+from parsl import python_app
+from kbmod_wf.utilities.executor_utilities import get_executors
+
+
+@python_app(
+    cache=True,
+    executors=get_executors(["local_dev_testing", "sharded_reproject"]),
+    ignore_for_cache=["logging_file"],
+)
+def reproject_wu(inputs=(), outputs=(), runtime_config={}, logging_file=None):
+    """This app will call the reproject_wu function to reproject and reflex correct
+    a given WorkUnit file.
+
+    Parameters
+    ----------
+    inputs : tuple, optional
+        A tuple with a single parsl.File object that references the original WorkUnit
+        file, by default ()
+    outputs : tuple, optional
+        A tuple with a single parsl.File object that references the reprojected
+        WorkUnit file, by default ()
+    runtime_config : dict, optional
+        A dictionary of configuration setting specific to this task, by default {}
+    logging_file : parsl.File, optional
+        The parsl.File object the defines where the logs are written, by default None
+
+    Returns
+    -------
+    parsl.File
+        The file object that points to the resulting WorkUnit file that was created.
+
+    Raises
+    ------
+    e
+        Reraises any exceptions that occur during the execution of the reproject_wu
+        function.
+    """
+    from kbmod_wf.utilities.logger_utilities import get_configured_logger, ErrorLogger
+
+    logger = get_configured_logger("task.ic_to_wu", logging_file)
+
+    from kbmod_wf.task_impls.reproject_multi_chip_multi_night_wu import reproject_wu
+
+
+    logger.info("Starting reproject_ic")
+    with ErrorLogger(logger):
+        reproject_wu(
+            original_wu_filepath=inputs[0].filepath,
+            guess_dist=inputs[1],
+            reprojected_wu_filepath=outputs[0].filepath,
+            runtime_config=runtime_config,
+            logger=logger,
+        )
+    logger.info("Completed reproject_ic")
+
+    return outputs[0]