Reorganise swarp

mirar/pipelines/wirc/blocks.py

@@ -26,11 +26,11 @@
     wirc_mask_path,
 )
 from mirar.processors.alerts import AvroPacketMaker, SendToFritz
-from mirar.processors.astromatic import Scamp, Sextractor, Swarp
+from mirar.processors.astromatic import Scamp, Sextractor
 from mirar.processors.astromatic.psfex import PSFex
 from mirar.processors.astromatic.scamp.scamp import scamp_header_key
 from mirar.processors.astromatic.sextractor.sextractor import sextractor_checkimg_map
-from mirar.processors.astromatic.swarp.swarp import GetSwarpComponentImages
+from mirar.processors.astromatic.swarp import ReloadSwarpComponentImages, Swarp
 from mirar.processors.astrometry.autoastrometry import AutoAstrometry
 from mirar.processors.astrometry.utils import AstrometryFromFile
 from mirar.processors.candidates.candidate_detector import DetectCandidates
@@ -132,7 +132,7 @@
     ),
     ImageSaver(output_dir_name="mask2", write_mask=True),
     WriteMaskedCoordsToFile(output_dir="mask_stack"),
-    GetSwarpComponentImages(
+    ReloadSwarpComponentImages(
         load_image=load_raw_wirc_image,
         copy_header_keys=FITS_MASK_KEY,
     ),

mirar/processors/astromatic/swarp/__init__.py

@@ -1 +1,7 @@
-from mirar.processors.astromatic.swarp.swarp import run_swarp
+"""
+Processors for interfacing with the Swarp software.
+"""
+from mirar.processors.astromatic.swarp.component_images import (
+    ReloadSwarpComponentImages,
+)
+from mirar.processors.astromatic.swarp.swarp import Swarp

mirar/processors/astromatic/swarp/component_images.py

@@ -0,0 +1,96 @@
+"""
+Processor to get the component images used to make a swarp stack
+"""
+import logging
+from collections.abc import Callable
+from pathlib import Path
+
+import numpy as np
+from astropy.io.fits import Header
+
+from mirar.data import Image, ImageBatch
+from mirar.io import open_fits
+from mirar.paths import STACKED_COMPONENT_IMAGES_KEY
+from mirar.processors.astromatic.swarp.swarp import Swarp
+from mirar.processors.base_processor import BaseImageProcessor
+from mirar.processors.utils.image_saver import ImageSaver
+
+logger = logging.getLogger(__name__)
+
+
+class ReloadSwarpComponentImages(BaseImageProcessor):
+    """
+    Get the component images used to make a swarp stack
+    """
+
+    base_key = "swarp_component_images"
+
+    def __init__(
+        self,
+        load_image: Callable[[str], [np.ndarray, Header]] = open_fits,
+        header_key=STACKED_COMPONENT_IMAGES_KEY,
+        copy_header_keys: str | list[str] = None,
+    ):
+        super().__init__()
+        self.load_image = load_image
+        self.header_key = header_key
+
+        if isinstance(copy_header_keys, str):
+            copy_header_keys = [copy_header_keys]
+        self.copy_header_keys = copy_header_keys
+
+    def _apply_to_images(
+        self,
+        batch: ImageBatch,
+    ) -> ImageBatch:
+        if len(batch) > 1:
+            raise NotImplementedError(
+                "GetSwarpComponentImages only works on a batch containing a "
+                "single images. Consider adding an ImageDebatcher before "
+                "this processor."
+            )
+        component_batch = ImageBatch()
+        image = batch[0]
+        component_images_list = image[self.header_key].split(",")
+
+        for component_image_path in component_images_list:
+            if not Path(component_image_path).exists():
+                raise FileNotFoundError(
+                    f"Component image {component_image_path} not found. "
+                    f"Are you sure it was saved using ImageSaver to this path just "
+                    f"before the Swarp processor that stacked it?"
+                )
+            component_data, component_header = self.load_image(component_image_path)
+            component_image = Image(component_data, component_header)
+            if self.copy_header_keys is not None:
+                for key in self.copy_header_keys:
+                    if key in image.keys():
+                        component_image[key] = image[key]
+            component_batch.append(component_image)
+        logger.info(f"Loaded {len(component_batch)} component images")
+        return component_batch
+
+    def check_prerequisites(
+        self,
+    ):
+        mask = np.array([isinstance(x, Swarp) for x in self.preceding_steps])
+        if np.sum(mask) == 0:
+            err = (
+                f"{self.__module__} requires {Swarp} as a prerequisite. "
+                f"However, the following steps were found: {self.preceding_steps}."
+            )
+            logger.error(err)
+            raise ValueError(err)
+
+        index = np.argmax(mask)
+
+        preceding_step = self.preceding_steps[index - 1]
+
+        if not isinstance(preceding_step, ImageSaver):
+            err = (
+                f"{self.__module__} requires an {ImageSaver} to be used to save the "
+                f"component images immediately before {Swarp} is run. "
+                f"However, the following steps were found: {self.preceding_steps}."
+            )
+            logger.error(err)
+            raise ValueError(err)

mirar/processors/astromatic/swarp/swarp.py

@@ -2,18 +2,14 @@
 Module relating to `swarp <https://www.astromatic.net/software/swarp`_
 """
 import logging
-import os
-from collections.abc import Callable
 from pathlib import Path
 from typing import Optional
 
 import numpy as np
-from astropy.io.fits import Header
 from astropy.wcs import WCS
 
-from mirar.data import Image, ImageBatch
+from mirar.data import ImageBatch
 from mirar.errors import ProcessorError
-from mirar.io import open_fits
 from mirar.paths import (
     BASE_NAME_KEY,
     LATEST_SAVE_KEY,
@@ -27,9 +23,8 @@
     get_temp_path,
 )
 from mirar.processors.astromatic.scamp.scamp import scamp_header_key
+from mirar.processors.astromatic.swarp.swarp_wrapper import run_swarp
 from mirar.processors.base_processor import BaseImageProcessor
-from mirar.processors.utils.image_saver import ImageSaver
-from mirar.utils import execute
 
 logger = logging.getLogger(__name__)
 
@@ -38,125 +33,6 @@ class SwarpError(ProcessorError):
     """Error relating to swarp"""
 
 
-def run_swarp(
-    stack_list_path: str | Path,
-    swarp_config_path: str | Path,
-    out_path: str | Path,
-    weight_list_path: Optional[str | Path] = None,
-    weight_out_path: Optional[str | Path] = None,
-    pixscale: Optional[float] = None,
-    x_imgpixsize: Optional[float] = None,
-    y_imgpixsize: Optional[float] = None,
-    propogate_headerlist: Optional[list] = None,
-    center_ra: Optional[float] = None,
-    center_dec: Optional[float] = None,
-    combine: bool = True,
-    gain: Optional[float] = None,
-    subtract_bkg: bool = False,
-    flux_scaling_keyword: str = None,
-    cache: bool = False,
-    center_type: str = None,
-):
-    """
-    Wrapper to resample and stack images with swarp
-
-    Parameters
-    ----------
-    stack_list_path : string
-        Name of file containing the names of files to be stacked
-        One file name per line
-    weight_list_path : string
-        Name of file containing the names of weight files to be stacked
-        One file name per line
-    swarp_config_path: str
-        Path of Swarp config file
-    out_path : string
-        Path of stacked output file
-    weight_out_path: str
-        Path of output weight image
-    pixscale: float
-        Pixelscale in degrees
-    x_imgpixsize: float
-        X-dimension in pixels
-    y_imgpixsize: float
-        Y-dimension in pixels
-    propogate_headerlist: list
-        Headerlist to propagate from header
-    center_ra: float
-        Central RA
-    center_dec: float
-        Central Dec
-    combine: bool
-        Combine and coadd all images? For reasons internal to Swarp, it is strongly
-        advised to always set this to True (even if you are running Swarp on only
-        one image).
-    gain: float
-        Gain
-    subtract_bkg: bool
-        Background subtraction
-    flux_scaling_keyword: str
-        What flux scaling keyword do you want to use? If None, the default value in
-        the config will be used
-    """
-
-    swarp_command = (
-        f"swarp -c {swarp_config_path} "
-        f"@{stack_list_path} "
-        f"-IMAGEOUT_NAME {out_path} "
-        f"-RESAMPLE Y -RESAMPLE_DIR {os.path.dirname(out_path)} "
-    )
-
-    if subtract_bkg:
-        swarp_command += "-SUBTRACT_BACK Y "
-    else:
-        swarp_command += "-SUBTRACT_BACK N "
-    if combine:
-        swarp_command += "-COMBINE Y -COMBINE_TYPE MEDIAN "
-    else:
-        swarp_command += "-COMBINE N "
-
-    if weight_list_path is not None:
-        swarp_command += f" -WEIGHT_TYPE MAP_WEIGHT -WEIGHT_IMAGE @{weight_list_path} "
-
-    if weight_out_path is not None:
-        swarp_command += f" -WEIGHTOUT_NAME {weight_out_path}"
-
-    if pixscale is not None:
-        swarp_command += f" -PIXELSCALE_TYPE MANUAL -PIXEL_SCALE {pixscale}"
-
-    if propogate_headerlist is not None:
-        swarp_command += " -COPY_KEYWORDS "
-        for keyword in propogate_headerlist:
-            swarp_command += f"{keyword},"
-
-        # remove final comma
-        swarp_command = swarp_command[:-1]
-
-    if np.logical_and(center_ra is not None, center_dec is not None):
-        swarp_command += f" -CENTER_TYPE MANUAL -CENTER {center_ra},{center_dec}"
-
-    else:
-        swarp_command += f" -CENTER_TYPE {center_type}"
-
-    if x_imgpixsize is not None:
-        swarp_command += f" -IMAGE_SIZE {x_imgpixsize}"
-        if y_imgpixsize is not None:
-            swarp_command += f",{y_imgpixsize}"
-
-    if gain is not None:
-        swarp_command += f" -GAIN {gain}"
-
-    if flux_scaling_keyword is not None:
-        swarp_command += f" -FSCALE_KEYWORD {flux_scaling_keyword}"
-
-    if not cache:
-        swarp_command += " -DELETE_TMPFILES Y"
-    else:
-        swarp_command += " -DELETE_TMPFILES N"
-
-    execute(swarp_command)
-
-
 class Swarp(BaseImageProcessor):
     """
     Processor to apply Swarp
@@ -510,81 +386,3 @@ def _apply_to_images(
                 logger.debug(f"Deleted temporary file {temp_file}")
 
         return ImageBatch([new_image])
-
-
-class GetSwarpComponentImages(BaseImageProcessor):
-    """
-    Get the component images used to make a swarp stack
-    """
-
-    base_key = "swarp_component_images"
-
-    def __init__(
-        self,
-        load_image: Callable[[str], [np.ndarray, Header]] = open_fits,
-        header_key=STACKED_COMPONENT_IMAGES_KEY,
-        copy_header_keys: str | list[str] = None,
-    ):
-        super().__init__()
-        self.load_image = load_image
-        self.header_key = header_key
-
-        if isinstance(copy_header_keys, str):
-            copy_header_keys = [copy_header_keys]
-        self.copy_header_keys = copy_header_keys
-
-    def _apply_to_images(
-        self,
-        batch: ImageBatch,
-    ) -> ImageBatch:
-        if len(batch) > 1:
-            raise NotImplementedError(
-                "GetSwarpComponentImages only works on a batch containing a "
-                "single images. Consider adding an ImageDebatcher before "
-                "this processor."
-            )
-        component_batch = ImageBatch()
-        image = batch[0]
-        component_images_list = image[self.header_key].split(",")
-
-        for component_image_path in component_images_list:
-            if not Path(component_image_path).exists():
-                raise FileNotFoundError(
-                    f"Component image {component_image_path} not found. "
-                    f"Are you sure it was saved using ImageSaver to this path just "
-                    f"before the Swarp processor that stacked it?"
-                )
-            component_data, component_header = self.load_image(component_image_path)
-            component_image = Image(component_data, component_header)
-            if self.copy_header_keys is not None:
-                for key in self.copy_header_keys:
-                    if key in image.keys():
-                        component_image[key] = image[key]
-            component_batch.append(component_image)
-        logger.info(f"Loaded {len(component_batch)} component images")
-        return component_batch
-
-    def check_prerequisites(
-        self,
-    ):
-        mask = np.array([isinstance(x, Swarp) for x in self.preceding_steps])
-        if np.sum(mask) == 0:
-            err = (
-                f"{self.__module__} requires {Swarp} as a prerequisite. "
-                f"However, the following steps were found: {self.preceding_steps}."
-            )
-            logger.error(err)
-            raise ValueError(err)
-
-        index = np.argmax(mask)
-
-        preceding_step = self.preceding_steps[index - 1]
-
-        if not isinstance(preceding_step, ImageSaver):
-            err = (
-                f"{self.__module__} requires an {ImageSaver} to be used to save the "
-                f"component images immediately before {Swarp} is run. "
-                f"However, the following steps were found: {self.preceding_steps}."
-            )
-            logger.error(err)
-            raise ValueError(err)