refactor: post-phase-3 consolidation (behaviour-preserving, witness-proven)

autoreduce/acquire/footprint.py

@@ -32,7 +32,9 @@ def covers_target(path: Path, ra: float, dec: float, margin_arcsec: float) -> bo
         for hdu in hdul:
             if hdu.name != "SCI" or hdu.data is None:
                 continue
-            wcs = WCS(hdu.header, naxis=2)
+            # fobj: HST headers carry lookup-table distortion (CPDIS/D2IM)
+            # that astropy can only resolve with the open HDUList in hand.
+            wcs = WCS(hdu.header, fobj=hdul, naxis=2)
             ny, nx = hdu.data.shape[-2], hdu.data.shape[-1]
             x, y = wcs.world_to_pixel_values(ra, dec)
             if not (np.isfinite(x) and np.isfinite(y)):

autoreduce/drizzle/_common.py

@@ -0,0 +1,61 @@
+"""
+Shared combine-backend plumbing: the scratch-directory discipline and the
+provenance fragment both backends emit identically.
+"""
+
+import os
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Dict, List
+
+import numpy as np
+
+from ..instruments import InstrumentAdapter
+from ..noise.rms import casertano_r
+from ..target import TargetSpec
+from .diagnostics import check_weight_uniformity
+
+
+@contextmanager
+def chdir_scratch(output_dir: Path):
+    """Resolve, create and chdir into a backend scratch dir; always restore."""
+    output_dir = Path(output_dir).resolve()
+    output_dir.mkdir(parents=True, exist_ok=True)
+    cwd = os.getcwd()
+    os.chdir(output_dir)
+    try:
+        yield output_dir
+    finally:
+        os.chdir(cwd)
+
+
+def combine_provenance(
+    spec: TargetSpec,
+    adapter: InstrumentAdapter,
+    exposures: List[Path],
+    wht: np.ndarray,
+    kwargs_key: str,
+    kwargs: Dict,
+    head: Dict = None,
+    tail: Dict = None,
+) -> Dict:
+    """
+    The provenance dict every combine backend records, assembled in the
+    canonical key order (kept stable — reduction.json is byte-compared by the
+    refactor witnesses). `head`/`tail` carry backend-specific extras.
+    """
+    out = dict(head or {})
+    out.update(
+        {
+            "n_exposures": len(exposures),
+            "exposures": [Path(p).name for p in exposures],
+            "single_exposure_branch": len(exposures) == 1,
+            kwargs_key: kwargs,
+            "correlated_noise_factor": casertano_r(
+                spec.final_pixfrac, adapter.scale_ratio(spec.final_scale)
+            ),
+            "weight_uniformity": check_weight_uniformity(wht),
+        }
+    )
+    out.update(tail or {})
+    return out

autoreduce/drizzle/combine.py

@@ -15,8 +15,6 @@
 
 from ..instruments import InstrumentAdapter
 from ..target import TargetSpec
-from .diagnostics import check_weight_uniformity
-from ..noise.rms import casertano_r
 
 
 def drizzle_kwargs_for(spec: TargetSpec, adapter: InstrumentAdapter, n_exposures: int) -> Dict:
@@ -66,28 +64,21 @@ def combine(
     from astropy.io import fits
     from drizzlepac import astrodrizzle
 
-    output_dir = Path(output_dir)
-    output_dir.mkdir(parents=True, exist_ok=True)
+    from ._common import chdir_scratch, combine_provenance
+
     # Drizzlepac lowercases output filenames internally, which breaks absolute
-    # paths containing capitals on case-sensitive filesystems — so chdir into
-    # the work dir and pass a relative, already-lowercase output root. This
+    # paths containing capitals on case-sensitive filesystems — so run inside
+    # the scratch dir with a relative, already-lowercase output root. This
     # also keeps AstroDrizzle's cwd scratch files contained.
     output_name = f"{spec.name}_{spec.filter_name}".lower()
-    output_root = str(output_dir / output_name)
-
     kwargs = drizzle_kwargs_for(spec, adapter, len(exposures))
-    import os
-
-    cwd = os.getcwd()
-    os.chdir(output_dir)
-    try:
+    with chdir_scratch(output_dir) as output_dir:
         astrodrizzle.AstroDrizzle(
             input=[str(p) for p in exposures],
             output=output_name,
             **kwargs,
         )
-    finally:
-        os.chdir(cwd)
+    output_root = str(output_dir / output_name)
 
     def _one(suffix: str) -> Path:
         hits = sorted(glob.glob(f"{output_root}*{suffix}"))
@@ -100,15 +91,12 @@ def _one(suffix: str) -> Path:
     sci = _one("_sci.fits")
     wht = _one("_wht.fits")
 
-    wht_data = fits.getdata(wht)
-    provenance = {
-        "n_exposures": len(exposures),
-        "exposures": [Path(p).name for p in exposures],
-        "single_exposure_branch": len(exposures) == 1,
-        "drizzle_kwargs": {k: kwargs[k] for k in sorted(kwargs)},
-        "correlated_noise_factor": casertano_r(
-            spec.final_pixfrac, adapter.scale_ratio(spec.final_scale)
-        ),
-        "weight_uniformity": check_weight_uniformity(wht_data),
-    }
+    provenance = combine_provenance(
+        spec,
+        adapter,
+        exposures,
+        fits.getdata(wht),
+        kwargs_key="drizzle_kwargs",
+        kwargs={k: kwargs[k] for k in sorted(kwargs)},
+    )
     return sci, wht, provenance

autoreduce/drizzle/jwst_combine.py

@@ -10,16 +10,13 @@
 stage (noise, psf, package) is backend-agnostic.
 """
 
-import json
 from pathlib import Path
 from typing import Dict, List, Tuple
 
 import numpy as np
 
 from ..instruments import InstrumentAdapter
 from ..target import TargetSpec
-from .diagnostics import check_weight_uniformity
-from ..noise.rms import casertano_r
 
 
 def combine(
@@ -34,23 +31,18 @@ def combine(
     from jwst.associations.lib.rules_level3_base import DMS_Level3_Base
     from jwst.pipeline import Image3Pipeline
 
-    # Resolve before the chdir below: relative paths would dangle afterwards.
-    output_dir = Path(output_dir).resolve()
-    output_dir.mkdir(parents=True, exist_ok=True)
-    product_name = f"{spec.name}_{spec.filter_name}".lower()
-
-    asn = asn_from_list(
-        [str(p) for p in exposures], rule=DMS_Level3_Base, product_name=product_name
-    )
-    asn_path = output_dir / f"{product_name}_asn.json"
-    _, serialized = asn.dump(format="json")
-    asn_path.write_text(serialized)
+    from ._common import chdir_scratch, combine_provenance
 
-    import os
+    product_name = f"{spec.name}_{spec.filter_name}".lower()
+    with chdir_scratch(output_dir) as output_dir:
+        asn = asn_from_list(
+            [str(p) for p in exposures], rule=DMS_Level3_Base,
+            product_name=product_name,
+        )
+        asn_path = output_dir / f"{product_name}_asn.json"
+        _, serialized = asn.dump(format="json")
+        asn_path.write_text(serialized)
 
-    cwd = os.getcwd()
-    os.chdir(output_dir)
-    try:
         Image3Pipeline.call(
             str(asn_path),
             output_dir=str(output_dir),
@@ -68,8 +60,6 @@ def combine(
                 },
             },
         )
-    finally:
-        os.chdir(cwd)
 
     i2d = output_dir / f"{product_name}_i2d.fits"
     if not i2d.exists():
@@ -92,22 +82,19 @@ def combine(
         fits.PrimaryHDU(data, header=header).writeto(path, overwrite=True)
         paths[name] = path
 
-    provenance = {
-        "backend": "jwst_image3",
-        "n_exposures": len(exposures),
-        "exposures": [Path(p).name for p in exposures],
-        "single_exposure_branch": len(exposures) == 1,
-        "resample_kwargs": {
+    provenance = combine_provenance(
+        spec,
+        adapter,
+        exposures,
+        wht,
+        kwargs_key="resample_kwargs",
+        kwargs={
             "pixel_scale": spec.final_scale,
             "pixfrac": spec.final_pixfrac,
             "kernel": spec.final_kernel,
             "rotation": 0.0,
         },
-        "correlated_noise_factor": casertano_r(
-            spec.final_pixfrac, adapter.scale_ratio(spec.final_scale)
-        ),
-        "weight_uniformity": check_weight_uniformity(wht),
-        "err_path": str(paths["err"]),
-        "i2d_path": str(i2d),
-    }
+        head={"backend": "jwst_image3"},
+        tail={"err_path": str(paths["err"]), "i2d_path": str(i2d)},
+    )
     return paths["sci"], paths["wht"], provenance