__all__ = [ 'PipelineImage', 'Pipeline' ]
# Imports STANDARD
import sys
import argparse
from functools import partial
import logging
from multiprocessing import Pool
import numpy as np
import nvtx
import pathlib
import re
import shutil
import tracemalloc
import uuid
# Imports ASTRO
from astropy.coordinates import SkyCoord
from astropy.table import Table
import astropy.units as u
from astropy.wcs.utils import skycoord_to_pixel
import fitsio
# Imports INTERNAL
import phrosty
from phrosty.imagesubtraction import sky_subtract, stampmaker
from sfft.SpaceSFFTFlow import SpaceSFFT_Flow
from snappl.dbclient import SNPITDBClient
from snappl.diaobject import DiaObject
from snappl.imagecollection import ImageCollection
from snappl.image import CompressedFITSImage
from snappl.lightcurve import Lightcurve
from snappl.provenance import Provenance
from snappl.psf import PSF
from snappl.config import Config
from snappl.logger import SNLogger
[docs]
class PipelineImage:
"""Holds a snappl.image.Image, with some other stuff the pipeline needs."""
def __init__( self, image, pipeline ):
"""Create a PipelineImage
Parameters:
-----------
image : snappl.image.Image
The image we're encapsulating. Pass either this or imagepath.
pipeline : phrosty.pipeline.Pipeline
The pipeline that owns this image.
"""
self.config = Config.get()
self.temp_dir = pipeline.temp_dir
self.keep_intermediate = self.config.value( 'photometry.phrosty.keep_intermediate' )
if self.keep_intermediate:
self.save_dir = pathlib.Path( self.config.value( 'system.paths.scratch_dir' ) )
elif not self.keep_intermediate:
self.save_dir = self.temp_dir
self.image = image
# if self.image.band != pipeline.band:
# raise ValueError( f"Image {self.image.path.name} has a band {self.image.band}, "
# f"which is different from the pipeline band {pipeline.band}" )
# Intermediate files
if self.keep_intermediate:
# Set to None. The path gets defined later on.
# They have to be defined here in __init__ so that they exist
# and are accessible in later functions.
self.skysub_img = {}
self.detmask_img = {}
self.input_sci_psf_path = {}
self.input_templ_psf_path = {}
self.aligned_templ_var_path = {}
self.aligned_templ_psf_path = {}
self.crossconv_sci_path = {}
self.crossconv_templ_path = {}
self.diff_path = {}
self.decorr_kernel_path = {}
# Always save and output these
self.decorr_psf_path = {}
self.decorr_zptimg_path = {}
self.decorr_diff_path = {}
self.zpt_stamp_path = {}
self.diff_var_path = {}
self.diff_var_stamp_path = {}
self.diff_stamp_path = {}
# Lauren added these for debugging...
self.aligned_templ_img_path = {}
self.aligned_templ_stamp_path = {}
self.diff_undecorr_img_path = {}
self.diff_undecorr_stamp_path = {}
# Held in memory
self.skyrms = None
self.psfobj = None
self.psf_data = None
# In case we fail...
self.fail_info = f'{image.band} {image.observation_id} {image.sca}'
self.failure_location = None
self.failure_pair = None
[docs]
def run_sky_subtract( self, mp=True ):
"""Run sky subtraction using Source Extractor.
Parameters
----------
mp : bool, optional
Toggle multiprocessing, by default True
Returns
-------
tuple
Tuple containing sky subtracted image, detection
mask array, and sky RMS value.
Output of phrosty.imagesubtraction.sky_subtract().
"""
try:
return sky_subtract( self.image, temp_dir=self.temp_dir )
except Exception as ex:
self.failure_location = 'skysub'
SNLogger.exception( ex )
raise
[docs]
def save_sky_subtract_info( self, info ):
"""Saves the sky-subtracted image, detection mask array,
and sky RMS values to attributes.
Parameters
----------
info : tuple
Output of self.run_sky_subtract(). See documentation
for phrosty.imagesubtraction.sky_subtract().
"""
SNLogger.debug( f"Saving sky_subtract info for path {info[0].path}" )
self.skysub_img = info[0]
self.detmask_img = info[1]
self.skyrms = info[2]
[docs]
def get_psf( self, ra, dec ):
"""Get the at the right spot on the image.
Parameters
----------
ra, dec : float
The coordinates in decimal degrees where we want the PSF.
Returns
-------
np.array
PSF stamp. If this function fails, None is returned.
"""
# TODO: right now snappl.psf.PSF.get_psf_object just
# passes the keyword arguments on to whatever makes
# the psf... and it's different for each type of
# PSF. We need to fix that... somehow....
try:
wcs = self.image.get_wcs()
x, y = wcs.world_to_pixel( ra, dec )
if self.psfobj is None:
psftype = self.config.value( 'photometry.phrosty.psf.type' )
psfparams = self.config.value( 'photometry.phrosty.psf.params' )
self.psfobj = PSF.get_psf_object( psftype, x=x, y=y,
band=self.image.band,
observation_id=self.image.observation_id,
sca=self.image.sca,
**psfparams )
stamp = self.psfobj.get_stamp( x, y )
if self.keep_intermediate:
outfile = self.save_dir / f"psf_{self.image.name}.fits"
fitsio.write( outfile, stamp, clobber=True )
return stamp
except Exception as ex:
self.failure_location = 'get_psf'
SNLogger.exception( ex )
raise
[docs]
def keep_psf_data( self, psf_data ):
"""Save PSF data to attribute.
Parameters
----------
psf_data : np.array
PSF stamp.
"""
self.psf_data = psf_data/np.sum(psf_data)
[docs]
def free( self, free_psf_data=False ):
"""Try to free memory. More might be done here."""
self.image.free()
self.skysub_img.free()
self.detmask_img.free()
if free_psf_data:
self.psf_data = None
[docs]
class Pipeline:
"""Phrosty's top-level pipeline"""
def __init__( self, diaobj, imgcol, band,
science_images=None,
template_images=None,
science_csv=None,
template_csv=None,
oid=None,
ltcv_prov_tag=None,
dbsave=False,
dbclient=None,
nprocs=1,
nwrite=5,
verbose=False,
memtrace=False,
catchfailures=False ):
"""Create the a pipeline object.
Parameters
----------
diaobj : DiaObject
The object we're building a lightcurve for
imgcol : ImageCollection
snappl.imagecollection.ImageCollection
band: str
One of R062, Z087, Y106, J129, H158, F184, K213
science_images: list of snappl.image.Image
The science images.
template_images: list of snappl.image.Image
The template images.
science_csv: Path or str
CSV file with the science images. The first line must be::
path observation_id sca mjd band
subsequent lines must have that information for all the
science images. path must be relative to ou24.images in
config. Pipeline will extract the images from this file
whose band matches the band of the pipeline (and ignore
the rest)
template_csv: Path or str
CSV file with template images. Same format as science_csv.
oid: str
Object ID. This is probably a temporary argument. It is only used
if diaobj is None, and is really only used to build a filepath
without "None" in it when saving the light curve parquet file.
ltcv_prov_tag: str
Provenance tag for light curve. Required to use SN PIT database.
dbsave: bool
Are we saving to the database?
Default False.
dbclient: snappl.dbclient.SNPITDBClient
nprocs: int, default 1
Number of cpus for the CPU multiprocessing segments of the pipeline.
(GPU segments will run a single process.)
nwrite: int, default 5
Number of asynchronous FITS writer processes.
verbose: bool, default True
Toggle verbose output.
memtrace: bool, default False
Toggle memory tracing.
catchfailures: bool, default False
Toggle collection of information for images that fail. If true, pipeline
will not crash if one image fails. If false, pipeline crashes if one image
in a given set fails (useful for debugging and tests).
"""
SNLogger.setLevel( logging.DEBUG if verbose else logging.INFO )
self.config = Config.get()
self.imgcol = imgcol
self.diaobj = diaobj
self.band = band
self.oid = oid
self.dia_out_dir = pathlib.Path( self.config.value( 'system.paths.dia_out_dir' ) )
self.scratch_dir = pathlib.Path( self.config.value( 'system.paths.scratch_dir' ) )
self.temp_dir_parent = pathlib.Path( self.config.value( 'system.paths.temp_dir' ) )
self.temp_dir = self.temp_dir_parent / str(uuid.uuid1())
self.temp_dir.mkdir()
self.ltcv_dir = pathlib.Path( self.config.value( 'system.paths.ltcv_dir' ) )
if ( science_images is None) == ( science_csv is None ):
raise ValueError( "Pass exactly one of science_images or science_csv" )
if science_csv is not None:
science_images = self._read_csv( science_csv )
if ( template_images is None ) == ( template_csv is None ):
raise ValueError( "Pass exactly one of template_images or template_csv" )
if template_csv is not None:
template_images = self._read_csv( template_csv )
if isinstance(science_images, list) or isinstance(science_images, tuple):
self.science_images = [ PipelineImage( i, self ) for i in science_images ]
else:
self.science_images = [ PipelineImage(science_images, self) ]
if isinstance(template_images, list) or isinstance(template_images, tuple):
self.template_images = [ PipelineImage( i, self ) for i in template_images ]
else:
self.template_images = [ PipelineImage(template_images, self) ]
# All of our failures.
self.catchfailures = catchfailures
self.failures = {'skysub': [],
'get_psf': [],
'align_and_preconvolve': [],
'find_decorrelation': [],
'subtract': [],
'variance': [],
'apply_decorrelation': [],
'make_stamps': []}
self.ltcv_prov_tag = ltcv_prov_tag
self.dbsave = dbsave
self.dbclient = dbclient
self.nprocs = nprocs
self.nwrite = nwrite
self.keep_intermediate = self.config.value( 'photometry.phrosty.keep_intermediate' )
self.remove_temp_dir = self.config.value( 'photometry.phrosty.remove_temp_dir' )
self.mem_trace = self.config.value( 'photometry.phrosty.mem_trace' )
# Debug LNA 20251202
# self.resid_img = None
# Debug LNA20260225
# self.aperture = None
def _read_csv( self, csvfile ):
"""Reads input csv files with columns:
'path observation_id sca mjd band'.
Parameters
----------
csvfile : str
Path to an input csv file.
Returns
-------
list of snappl.image.Image
Raises
------
ValueError
If the first line of the csv file doesn't match
'path observation_id sca mjd band', a ValueError is raised.
"""
imlist = []
with open( csvfile ) as ifp:
hdrline = ifp.readline()
if not re.search( r"^\s*path\s+observation_id\s+sca\s+mjd\s+band\s*$", hdrline ):
raise ValueError( f"First line of list file {csvfile} didn't match what was expected." )
for line in ifp:
path, observation_id, sca, _mjd, band = line.split()
if band == self.band:
# This should yell at us if the observation_id
# or sca doesn't match what is read from the path
imlist.append( self.imgcol.get_image( path=path,
observation_id=observation_id,
sca=sca,
band=band ) )
return imlist
[docs]
def sky_sub_all_images( self ):
"""Sky subtracts all snappl.image.Image objects in
self.science_images and self.template_images using
Source Extractor.
Contains its own error logging function, log_error().
"""
# Currently, this writes out a bunch of FITS files. Further refactoring needed
# to support more general image types.
all_imgs = self.science_images.copy() # shallow copy
all_imgs.extend( self.template_images )
def log_error( img, x ):
SNLogger.error( f"Sky subtraction failure on {img.image.path}: {x}" )
if self.catchfailures:
self.failures['skysub'].append(f'{self.image.band} \
{self.image.observation_id} \
{self.image.sca}')
if self.nprocs > 1:
with Pool( self.nprocs ) as pool:
for img in all_imgs:
pool.apply_async( img.run_sky_subtract, (), {},
callback=img.save_sky_subtract_info,
error_callback=partial(log_error, img) )
pool.close()
pool.join()
else:
for img in all_imgs:
img.save_sky_subtract_info( img.run_sky_subtract( mp=False ) )
[docs]
def get_psfs( self ):
"""Retrieve PSFs for all snappl.image.Image objects in
self.science_images and self.template_images.
Contains its own error logging function, log_error().
"""
all_imgs = self.science_images.copy() # shallow copy
all_imgs.extend( self.template_images )
def log_error( img, x ):
SNLogger.error( f"get_psf failure on {img.image.path}: {x}" )
img.failure_location = 'get_psf'
if self.nprocs > 1:
with Pool( self.nprocs ) as pool:
for img in all_imgs:
# callback_partial = partial( img.save_psf_path, all_imgs )
pool.apply_async( img.get_psf, (self.diaobj.ra, self.diaobj.dec), {},
callback=img.keep_psf_data,
error_callback=partial(log_error, img) )
pool.close()
pool.join()
for img in all_imgs:
if img.failure_location is not None:
self.failures[img.failure_location].append(img.fail_info)
else:
for img in all_imgs:
try:
img.keep_psf_data( img.get_psf(self.diaobj.ra, self.diaobj.dec) )
except OSError:
# OSError because if get_psf can't open a file, it does this
# (for example, if the file does not exist)
if self.catchfailures:
self.failures['get_psf'].append(img.fail_info)
[docs]
def align_and_pre_convolve(self, templ_image, sci_image ):
"""Align and pre convolve a single template/science pair.
Parameters
----------
sci_image: phrosty.PipelineImage
The science (new) image.
templ_image: phrosty.PipelineImage
The template (ref) image that will be subtracted from sci_image.
Returns
-------
sfftifier: SpaceSFFT_Flow
Use this object for futher SFFT work. Be sure to
dereference it to free the prodigious amount of memory it
allcoates.
"""
# SFFT needs FITS headers with a WCS and with NAXIS[12]
hdr_sci = sci_image.image.get_wcs().get_astropy_wcs().to_header( relax=True )
hdr_sci.insert( 0, ('NAXIS', 2) )
hdr_sci.insert( 'NAXIS', ('NAXIS1', sci_image.image.data.shape[1] ), after=True )
hdr_sci.insert( 'NAXIS1', ('NAXIS2', sci_image.image.data.shape[0] ), after=True )
data_sci = sci_image.skysub_img.data
noise_sci = sci_image.image.noise
var_sci = noise_sci ** 2
hdr_templ = templ_image.image.get_wcs().get_astropy_wcs().to_header( relax=True )
hdr_templ.insert( 0, ('NAXIS', 2) )
hdr_templ.insert( 'NAXIS', ('NAXIS1', templ_image.image.data.shape[1] ), after=True )
hdr_templ.insert( 'NAXIS1', ('NAXIS2', templ_image.image.data.shape[0] ), after=True )
data_templ = templ_image.skysub_img.data
noise_templ = templ_image.image.noise
var_templ = noise_templ ** 2
sci_psf = sci_image.psf_data
templ_psf = templ_image.psf_data
sci_detmask = sci_image.detmask_img.data
templ_detmask = templ_image.detmask_img.data
sfftifier = SpaceSFFT_Flow(
hdr_target=hdr_sci,
hdr_object=hdr_templ,
target_skyrms=sci_image.skyrms,
object_skyrms=templ_image.skyrms,
PixA_target=data_sci,
PixA_object=data_templ,
PixA_targetVar=var_sci,
PixA_objectVar=var_templ,
PixA_target_DMASK=sci_detmask,
PixA_object_DMASK=templ_detmask,
PSF_target=sci_psf,
PSF_object=templ_psf,
KerPolyOrder=Config.get().value('photometry.phrosty.kerpolyorder')
)
sfftifier.resample_image_mask_psf()
sfftifier.cross_convolve()
return sfftifier
[docs]
def phot_at_coords( self, img, psf, pxcoords=(50, 50), ap_r=3 ):
"""Do photometry at forced set of pixel coordinates.
Parameters
----------
img: snappl.image.Image
The image on which to do the photometry
psf: snappl.psf.PSF
The PSF.
pxcoords: tuple of (int, int)
The position on the image to do the photometry
ap_r: float
Radius of aperture.
Returns
-------
results: dict
Keys and values are:
* 'aperture_sum': flux in aperture of radius ap_r
* 'flux_fit': flux from PSF photometry
* 'flux_fit_err': uncertainty on flux_fit
* 'mag_fit': instrumental magnitude (i.e. no zeropoint) from flux_fit
* 'mag_fit_err': uncertainty on mag_fit
All values are floats.
"""
forcecoords = Table([[float(pxcoords[0])], [float(pxcoords[1])]], names=["x", "y"])
init = img.ap_phot( forcecoords, ap_r=ap_r )
init.rename_column( 'aperture_sum', 'flux_init' )
init.rename_column( 'xcenter', 'xcentroid' )
init.rename_column( 'ycenter', 'ycentroid' )
final = img.psf_phot( init_params=init,
psf=psf,
forced_phot=True
# return_resid_image=True
)
# Debug LNA 20250225
# self.aperture = img.apertures
# Debug LNA 20251202
# self.resid_img = resid_img
flux = final['flux_fit'][0]
flux_err = final['flux_err'][0]
mag = -2.5 * np.log10(final["flux_fit"][0])
mag_err = (2.5 / np.log(10)) * np.abs(final["flux_err"][0] / final["flux_fit"][0])
results_dict = {
'flux': flux,
'flux_err': flux_err,
'aperture_sum': init['flux_init'][0], # Has to be renamed and named back because photutils.
'mag': mag,
'mag_err': mag_err
}
return results_dict
[docs]
def make_phot_info_dict( self, sci_image, templ_image, ap_r=3 ):
""""
Do photometry on a difference image generated from sci_image
and templ_image. Collect the output in a dictionary.
Parmaeters
----------
sci_image: PipelineImage
science image wrapper
temp_image: PipelineImage
template image wrapper
ap_r: float, default 4
Radius of aperture to use in aperture photometry.
Returns
-------
results_dict: dict
Dictionary with keys sci_name, templ_name, success, ra,
dec, mjd, band, observation_id, sca, template_observation_id,
template_sca, zpt, aperture_sum, flux_fit, flux_fit_err,
mag_fit, and mag_fit_err.
"""
# Do photometry on stamp because it will read faster.
# (We hope. But CFS latency will kill you at 1 byte.)
SNLogger.debug( "...make_phot_info_dict reading stamp and psf" )
# Required results keys
req_results_keys = ['mjd', 'flux', 'flux_err', 'zpt', 'NEA', 'sky_rms',
'observation_id', 'sca', 'pix_x', 'pix_y']
phrosty_results_keys = ['science_name', 'template_name',
'science_id', 'template_id',
'template_observation_id', 'template_sca',
'aperture_sum', 'mag', 'mag_err', 'success']
results_keys = req_results_keys + phrosty_results_keys
results_dict = {key: np.nan for key in results_keys}
# Required results keys
results_dict['observation_id'] = str(sci_image.image.observation_id)
results_dict['sca'] = sci_image.image.sca
# phrosty results keys
results_dict['science_name'] = sci_image.image.name
results_dict['template_name'] = templ_image.image.name
results_dict['science_id'] = str(sci_image.image.id)
results_dict['template_id'] = str(templ_image.image.id)
results_dict['template_observation_id'] = str(templ_image.image.observation_id)
results_dict['template_sca'] = templ_image.image.sca
results_dict['success'] = False
try:
results_dict['mjd'] = sci_image.image.mjd
# Additional phrosty keys
diff_img = CompressedFITSImage( filepath=sci_image.diff_stamp_path[ templ_image.image.name ],
noisepath=sci_image.diff_var_stamp_path[ templ_image.image.name ]
)
psf_img = CompressedFITSImage( filepath=sci_image.decorr_psf_path[ templ_image.image.name ] )
# Make sure the files are there. Has side effect of loading the header and data.
diff_img.get_data( which='data', cache=True )
diff_img.get_data( which='noise', cache=True )
# The thing written to disk was actually variance, so fix that
diff_img.noise = np.sqrt( diff_img.noise )
psf_img.get_data( which='data', cache=True )
SNLogger.debug( "...make_phot_info_dict getting psf" )
coord = SkyCoord(ra=self.diaobj.ra * u.deg, dec=self.diaobj.dec * u.deg)
pxcoords = skycoord_to_pixel( coord, diff_img.get_wcs().get_astropy_wcs() )
psf = PSF.get_psf_object( 'OversampledImagePSF',
x=diff_img.data.shape[1]/2., y=diff_img.data.shape[1]/2.,
oversample_factor=1.,
data=psf_img.data )
SNLogger.debug( "...make_phot_info_dict doing photometry" )
results_dict.update( self.phot_at_coords( diff_img, psf, pxcoords=pxcoords, ap_r=ap_r) )
# Add additional info to the results dictionary so it can be merged into a nice file later.
SNLogger.debug( "...make_phot_info_dict getting zeropoint" )
results_dict['zpt'] = sci_image.image.zeropoint
results_dict['success'] = True
except Exception as e:
# results_dict['ap_zpt'] = np.nan
SNLogger.debug( f"...make_phot_info_dict failed for \
{sci_image.image.name} - {templ_image.image.name}. Reason: {e}" )
finally:
# Basically, make_lightcurve will never "fail". Instead, you will get a row of NaN
# with results_dict['success'] = False if something weird happened, here.
SNLogger.debug( "...make_phot_info_dict done." )
return results_dict
[docs]
def add_to_results_dict( self, one_pair ):
"""Record results from self.make_phot_info_dict() to the
aggregate dictionary for the entire light curve.
Parameters
----------
one_pair : dict
Dictionary output from self.make_phot_info_dict().
"""
for key, arr in self.results_dict.items():
arr.append( one_pair[ key ] )
if not one_pair['success']:
SNLogger.debug( "Failure in make_lightcurve!" )
SNLogger.debug( "Done adding to results dict" )
if self.mem_trace:
SNLogger.info( f"After adding to results dict for \
{one_pair['observation_id']} {one_pair['sca']}, memory usage = \
{tracemalloc.get_traced_memory()[1]/(1024**2):.2f} MB" )
[docs]
def save_stamp_paths( self, sci_image, templ_image, paths ):
"""Helper function for recording the stamp paths returned in
self.do_stamps.
Parameters
----------
sci_image : snappl.image.Image
Science image with supernova.
templ_image : snappl.image.Image
Template image without supernova.
paths : list of pathlib.Path
Output from self.do_stamps().
"""
try:
sci_image.zpt_stamp_path[ templ_image.image.name ] = paths[0]
sci_image.diff_stamp_path[ templ_image.image.name ] = paths[1]
sci_image.diff_var_stamp_path[ templ_image.image.name ] = paths[2]
# Lauren added these for debugging...
sci_image.aligned_templ_stamp_path[ templ_image.image.name ] = paths[3]
sci_image.diff_undecorr_stamp_path[ templ_image.image.name ] = paths[4]
except Exception as ex:
SNLogger.exception( f"Exception in self.save_stamp_paths: {ex}" )
if sci_image.failure_location is None:
sci_image.failure_location = 'make_stamps'
sci_image.failure_pair = templ_image.image.name
[docs]
def do_stamps( self, sci_image, templ_image ):
"""Make stamps from the zero point image, decorrelated
difference image, and variance image centered at the
location of the supernova.
Parameters
----------
sci_image : snappl.image.Image
Science image with supernova.
templ_image : snappl.image.Image
Template image without supernova.
Returns
-------
list of pathlib.Path
Paths to the stamps corresponding to the zero point image,
decorrelated difference image, and variance image centered
at the location of the supernova.
"""
try:
zptim = CompressedFITSImage( filepath=sci_image.decorr_zptimg_path[ templ_image.image.name ] )
zpt_stampname = stampmaker(
ra=self.diaobj.ra,
dec=self.diaobj.dec,
shape=np.array([100, 100]),
img=zptim,
savedir=self.dia_out_dir,
savename=f"stamp_{zptim.path.name}"
)
zpt_path = pathlib.Path( zpt_stampname )
zptim.free()
diffim = CompressedFITSImage( filepath=sci_image.decorr_diff_path[ templ_image.image.name ] )
diff_stampname = stampmaker(
ra=self.diaobj.ra,
dec=self.diaobj.dec,
shape=np.array([100, 100]),
img=diffim,
savedir=self.dia_out_dir,
savename=f"stamp_{diffim.path.name}"
)
diff_path = pathlib.Path( diff_stampname )
diffim.free()
diffvarim = CompressedFITSImage( filepath=sci_image.diff_var_path[ templ_image.image.name ] )
diffvar_stampname = stampmaker(
ra=self.diaobj.ra,
dec=self.diaobj.dec,
shape=np.array([100, 100]),
img=diffvarim,
savedir=self.dia_out_dir,
savename=f"stamp_{diffvarim.path.name}"
)
diffvar_path = pathlib.Path( diffvar_stampname )
diffvarim.free()
# Lauren added these for debugging...
alignedtemplim = CompressedFITSImage( filepath=sci_image.aligned_templ_img_path[ templ_image.image.name ] )
alignedtempl_stampname = stampmaker(
ra=self.diaobj.ra,
dec=self.diaobj.dec,
shape=np.array([100, 100]),
img=alignedtemplim,
savedir=self.dia_out_dir,
savename=f"stamp_{alignedtemplim.path.name}"
)
alignedtempl_path = pathlib.Path( alignedtempl_stampname )
alignedtemplim.free()
regdiffim = CompressedFITSImage( sci_image.diff_undecorr_img_path [ templ_image.image.name ] )
regdiffim_stampname = stampmaker(
ra=self.diaobj.ra,
dec=self.diaobj.dec,
shape=np.array([100, 100]),
img=regdiffim,
savedir=self.dia_out_dir,
savename=f"stamp_{regdiffim.path.name}"
)
regdiff_path = pathlib.Path( regdiffim_stampname )
regdiffim.free()
SNLogger.info(f"Decorrelated diff stamp path: {diff_path}")
SNLogger.info(f"Zpt image stamp path: {zpt_path}")
SNLogger.info(f"Decorrelated diff variance stamp path: {diffvar_path}")
SNLogger.info(f"Aligned template stamp path: {alignedtempl_path}")
SNLogger.info(f"Undecorrelated diff stamp path: {regdiff_path}")
return zpt_path, diff_path, diffvar_path, alignedtempl_path, regdiff_path
except Exception as e:
SNLogger.error( f"do_stamps failure for {sci_image.image.observation_id} \
{sci_image.image.sca} - \
{templ_image.image.observation_id} \
{templ_image.image.sca}" )
raise Exception(e)
[docs]
def make_lightcurve( self ):
"""Collect all results from photometry in one dictionary.
Write the output to a csv as a table.
Contains its own error logging function, log_error().
Returns
-------
pathlib.Path
Path to output csv file that contains a light curve.
"""
SNLogger.info( "Making lightcurve." )
self.metadata = {
'provenance_id': None,
'diaobject_id': self.diaobj.id,
'diaobject_position_id': None,
'iau_name': self.diaobj.iauname,
'band': self.band,
'ra': self.diaobj.ra,
'dec': self.diaobj.dec,
'ra_err': None,
'dec_err': None,
'ra_dec_covar': None,
f'local_surface_brightness_{self.band}': -999. # phrosty does not output this value!
# add it later!
}
self.results_dict = {
# Required keys in specified order
'mjd': [], # Days, float
'flux': [], # DN/s, float
'flux_err': [], # DN/s, float
'zpt': [], # AB mag of object is m = -2.5 * log(flux) + zpt, float
'NEA': [], # px^2, float
'sky_rms': [], # DN/s, float
'observation_id': [], # string, temporary name
'sca': [], # int
'pix_x': [], # x-position of SN on detector w/ 0-offset, float
'pix_y': [], # y-position of SN on detector w/ 0-offset, float
# Additional phrosty keys
'science_name': [],
'template_name': [],
'science_id': [],
'template_id': [],
'template_observation_id': [],
'template_sca': [],
'aperture_sum': [],
'mag': [],
'mag_err': [],
'success': []
}
def log_error( sci_image, templ_image, x ):
SNLogger.error( f"make_phot_info_dict failure for {sci_image.image.observation_id} \
{sci_image.image.sca} - \
{templ_image.image.observation_id} \
{templ_image.image.sca}: {x}" )
if self.nprocs > 1:
with Pool( self.nprocs ) as pool:
for sci_image in self.science_images:
for templ_image in self.template_images:
logerr_partial = partial(log_error, sci_image, templ_image)
pool.apply_async( self.make_phot_info_dict, (sci_image, templ_image), {},
self.add_to_results_dict,
error_callback=logerr_partial )
SNLogger.debug( f"pool.apply async done for \
{sci_image.image.observation_id} \
{sci_image.image.sca}" )
pool.close()
pool.join()
SNLogger.debug('Make phot info dict pool closed and joined.')
else:
for i, sci_image in enumerate( self.science_images ):
if sci_image.failure_location is None:
SNLogger.debug( f"Doing science image {i} of {len(self.science_images)}" )
for templ_image in self.template_images:
self.add_to_results_dict( self.make_phot_info_dict( sci_image, templ_image ) )
if self.dbsave:
SNLogger.debug('About to get image provenance.')
imgprov = Provenance.get_by_id( self.science_images[0].image.provenance_id, dbclient=self.dbclient )
SNLogger.debug('About to get object provenance.')
objprov = Provenance.get_by_id( self.diaobj.provenance_id, dbclient=self.dbclient )
phrosty_version = phrosty.__version__
major = int(phrosty_version.split('.')[0])
minor = int(phrosty_version.split('.')[1])
SNLogger.debug('About to make LC provenance.')
ltcvprov = Provenance( process='phrosty',
major=major,
minor=minor,
params=Config.get(),
keepkeys=[ 'photometry.phrosty' ],
omitkeys=None,
upstreams=[imgprov, objprov],
)
self.metadata['provenance_id'] = ltcvprov.id
SNLogger.debug('About to make LC object.')
lc_obj = Lightcurve(data=self.results_dict, meta=self.metadata)
lc_obj.diaobj = self.diaobj
lc_obj.provenance_object = ltcvprov
else:
lc_obj = Lightcurve(data=self.results_dict, meta=self.metadata)
if self.dbsave:
SNLogger.debug( "Saving results to database..." )
ltcvprov.save_to_db( tag=self.ltcv_prov_tag )
lc_obj.write()
lc_obj.save_to_db( dbclient=self.dbclient )
else:
SNLogger.debug( "Saving results using paths..." )
if self.diaobj.id is not None:
save_basename = str(self.diaobj.id)
else:
save_basename = str(self.oid)
filepath = pathlib.Path(f'data/{self.oid}/{save_basename}_{self.metadata["band"]}.pq')
results_savepath = f'{self.ltcv_dir}/{filepath}'
lc_obj.write( base_dir=self.ltcv_dir, filepath=filepath, overwrite=True)
SNLogger.info(f'Results saved to {results_savepath}.')
return results_savepath
[docs]
def write_fits_file( self, data, header, savepath ):
"""Helper function for writing fits files.
Parameters
----------
data : np.array
Image array.
header : _type_
FITS header.
savepath : str
Savepath for FITS file.
"""
# try:
if header is not None:
hdr_dict = dict(header.items())
else:
hdr_dict = None
fitsio.write( savepath, data, header=hdr_dict, clobber=True )
# except Exception as e:
# SNLogger.exception( f"Exception writing FITS image {savepath}: {e}" )
# raise
[docs]
def clear_contents( self, directory ):
"""Delete contents of a directory. Used to clear temporary
files.
Parameters
----------
directory : pathlib.Path
Path to directory to empty.
"""
for f in directory.iterdir():
try:
if f.is_dir():
shutil.rmtree( f )
else:
f.unlink()
except Exception as e:
print( f'Oops! Deleting {f} from {directory} did not work.\nReason: {e}' )
[docs]
def __call__( self, through_step=None ):
"""Run the pipeline.
Parameters
----------
through_step: str, default None
Which step to run thorough? Runs them all if not given.
Steps in order are:
* sky_subtract
* get_psfs
* align_and_preconvolve
* subtract
* find_decorrelation
* apply_decorrelation
* make_stamps
* make_lightcurve
Returns
-------
ltcvpath : pathlib.Path or None
The path to the output lightcurve file if make_lightcurve
was run, otherwise None.
"""
if self.mem_trace:
tracemalloc.start()
tracemalloc.reset_peak()
if through_step is None:
through_step = 'make_lightcurve'
steps = [ 'sky_subtract', 'get_psfs', 'align_and_preconvolve', 'subtract', 'find_decorrelation',
'apply_decorrelation', 'make_stamps', 'make_lightcurve' ]
stepdex = steps.index( through_step )
if stepdex < 0:
raise ValueError( f"Unknown step {through_step}" )
steps = steps[:stepdex+1]
if 'sky_subtract' in steps:
# After this step is done, all images (both science and template)
# will have the following fields set:
# .skysub_img : sub-subtracted image
# .detmask_img : deteciton mask image
# .skyrms : float, median of sky image calculated by SExtractor
SNLogger.info( "Running sky subtraction" )
with nvtx.annotate( "skysub", color=0xff8888 ):
self.sky_sub_all_images()
if self.mem_trace:
SNLogger.info( f"After sky_subtract, memory usage = {tracemalloc.get_traced_memory()[1]/(1024**2):.2f} MB" )
if 'get_psfs' in steps:
# After this step, all images (both science and template) will
# have the .psf_data field set with the image-resolution PSF stamp data
# (from a PSF.get_stamp() call.)
SNLogger.info( "Getting PSFs" )
with nvtx.annotate( "getpsfs", color=0xff8888 ):
self.get_psfs()
if self.mem_trace:
SNLogger.info( f"After get_psfs, memory usage = {tracemalloc.get_traced_memory()[1]/(1024**2):.2f} MB" )
# Create a process pool to write fits files
with Pool( self.nwrite ) as fits_writer_pool:
def log_fits_write_error( savepath, x ):
SNLogger.error( f"Exception writing FITS file {savepath}: {x}" )
# raise?
# Do the hardcore processing
for templ_image in self.template_images:
for sci_image in self.science_images:
SNLogger.info( f"Processing {sci_image.image.name} minus {templ_image.image.name}" )
sfftifier = None
i_failed_gpu = False # We haven't failed yet.
fail_info = {
'science': sci_image.fail_info,
'template': templ_image.fail_info
}
continuation_conditions = [
sci_image.fail_info in self.failures['skysub'],
sci_image.fail_info in self.failures['get_psf'],
templ_image.fail_info in self.failures['skysub'],
templ_image.fail_info in self.failures['get_psf']
]
if any(continuation_conditions):
i_failed_gpu = True
if 'align_and_preconvolve' in steps and not i_failed_gpu:
# After this step, sfftifier will be a SpaceSFFT_Flow object.
SNLogger.info( "...align_and_preconvolve" )
with nvtx.annotate( "align_and_pre_convolve", color=0x8888ff ):
try:
sfftifier = self.align_and_pre_convolve( templ_image=templ_image,
sci_image=sci_image
)
mess = f"{sci_image.image.name}-{templ_image.image.name}"
aligned_templ_path = self.dia_out_dir / f"aligned_{mess}"
sci_image.aligned_templ_img_path[ templ_image.image.name ] = aligned_templ_path
self.write_fits_file(sfftifier.op.asnumpy(sfftifier.PixA_resamp_object),
sfftifier.hdr_target, aligned_templ_path)
except Exception as e:
i_failed_gpu = True
SNLogger.debug(f'Failure in align_and_preconvolve! Reason: {e}')
if self.catchfailures:
self.failures['align_and_preconvolve'].append(fail_info)
if 'subtract' in steps and not i_failed_gpu:
# After this step is done, two more fields in sfftifier are set:
# Solution : Matching kernel parameterization (coefficients)
# PixA_DIFF : difference image
SNLogger.info( "...subtract" )
with nvtx.annotate( "subtraction", color=0x44ccff ):
try:
sfftifier.sfft_subtract()
mess = f"{sci_image.image.name}-{templ_image.image.name}"
undecorr_diff_path = self.dia_out_dir / f"diff_{mess}"
sci_image.diff_undecorr_img_path[ templ_image.image.name ] = undecorr_diff_path
self.write_fits_file(sfftifier.op.asnumpy(sfftifier.PixA_DIFF),
sfftifier.hdr_target, undecorr_diff_path)
except Exception as e:
i_failed_gpu = True
SNLogger.debug(f'Failure in subtraction! Failure is: {e}')
if self.catchfailures:
self.failures['subtract'].append(fail_info)
if 'find_decorrelation' in steps and not i_failed_gpu:
# This step does ...
# After it's done, the following fields of sfftifier are set:
# Solution : CPU copy of Solution
# FKDECO : result of PureCupy_Decorrelation_Calculator.PCDC
# In addition the two local varaibles diff_var and diff_var_path are set.
# diff_var : variance in difference image, on GPU
# diff_var_path : where we want to write diff_var in self.dia_out_dir
SNLogger.info( "...find_decorrelation" )
with nvtx.annotate( "find_decor", color=0xcc44ff ):
try:
sfftifier.find_decorrelation()
except Exception as e:
i_failed_gpu = True
SNLogger.debug(f'Failure in find_decorrelation! Reason: {e}')
if self.catchfailures:
self.failures['find_decorrelation'].append(fail_info)
SNLogger.info( "...generate variance image" )
with nvtx.annotate( "variance", color=0x44ccff ):
try:
diff_var = sfftifier.create_variance_image()
mess = f"{sci_image.image.name}-{templ_image.image.name}"
diff_var_path = self.dia_out_dir / f"diff_var_{mess}"
sci_image.diff_var_path[ templ_image.image.name ] = diff_var_path
self.write_fits_file(sfftifier.op.asnumpy(diff_var),
sfftifier.hdr_target, diff_var_path)
except Exception as e:
i_failed_gpu = True
SNLogger.debug(f'Failure in generate variance! Reason: {e}')
if self.catchfailures:
self.failures['variance'].append(fail_info)
if 'apply_decorrelation' in steps and not i_failed_gpu:
try:
mess = f"{sci_image.image.name}-{templ_image.image.name}"
decorr_psf_path = self.dia_out_dir / f"decorr_psf_{mess}"
decorr_zptimg_path = self.dia_out_dir / f"decorr_zptimg_{mess}"
decorr_diff_path = self.dia_out_dir / f"decorr_diff_{mess}"
images = [ sfftifier.PixA_DIFF,
sfftifier.PixA_Ctarget, sfftifier.PSF_Ctarget ]
savepaths = [ decorr_diff_path,
decorr_zptimg_path, decorr_psf_path ]
headers = [ sfftifier.hdr_target,
sfftifier.hdr_target, None ]
for i, (img, savepath, hdr) in enumerate(zip( images, savepaths, headers )):
with nvtx.annotate( "apply_decor", color=0xccccff ):
SNLogger.info( f"...apply_decor to {savepath}" )
decorimg = sfftifier.apply_decorrelation( img )
with nvtx.annotate( "submit writefits", color=0xff8888 ):
SNLogger.info( f"...writefits {savepath}" )
fits_writer_pool.apply_async( self.write_fits_file,
( sfftifier.op.asnumpy( decorimg ), hdr, savepath ), {},
error_callback=partial(log_fits_write_error,
savepath) )
sci_image.decorr_psf_path[ templ_image.image.name ] = decorr_psf_path
sci_image.decorr_zptimg_path[ templ_image.image.name ] = decorr_zptimg_path
sci_image.decorr_diff_path[ templ_image.image.name ] = decorr_diff_path
except Exception as e:
i_failed_gpu = True
SNLogger.debug( f"Failure in apply_decorrelation! Reason: {e}" )
if self.catchfailures:
self.failures['apply_decorrelation'].append(fail_info)
if self.keep_intermediate and not i_failed_gpu:
# Each key is the file prefix addition.
# Each list has [descriptive filetype, image file name, data, header].
# TODO: Include multiprocessing.
# In the future, we may want to write these things right after they happen
# instead of saving it all for the end of the SFFT stuff.
write_filepaths = {'aligned': [['img',
f'{templ_image.image.name}_-_{sci_image.image.name}',
sfftifier.op.asnumpy(sfftifier.PixA_resamp_object),
sfftifier.hdr_target],
['var',
f'{templ_image.image.name}_-_{sci_image.image.name}',
sfftifier.op.asnumpy(sfftifier.PixA_resamp_objectVar),
sfftifier.hdr_target],
['psf',
f'{templ_image.image.name}_-_{sci_image.image.name}',
sfftifier.op.asnumpy(sfftifier.PSF_target),
sfftifier.hdr_target],
['detmask',
f'{sci_image.image.name}_-_{templ_image.image.name}',
sfftifier.op.asnumpy(sfftifier.PixA_resamp_object_DMASK),
sfftifier.hdr_target]
],
'convolved': [['img',
f'{sci_image.image.name}_-_{templ_image.image.name}.fits',
sfftifier.op.asnumpy(sfftifier.PixA_Ctarget),
sfftifier.hdr_target],
['img',
f'{templ_image.image.name}_-_{sci_image.image.name}.fits',
sfftifier.op.asnumpy(sfftifier.PixA_Cresamp_object),
sfftifier.hdr_target]
],
'diff': [['img',
f'{sci_image.image.name}_-_{templ_image.image.name}.fits',
sfftifier.op.asnumpy(sfftifier.PixA_DIFF),
sfftifier.hdr_target]
],
'match_kernel': [['img',
f'{sci_image.image.name}_-_{templ_image.image.name}.fits',
sfftifier.op.asnumpy(sfftifier.MATCH_KERNEL),
sfftifier.hdr_target]
],
# LNA 20260131: There are issues with saving complex arrays using fitsio.
# But the imaginary part of the decorrelation kernel is real.
# Don't worry about it.
'decorr': [['kernel_real',
f'{sci_image.image.name}_-_{templ_image.image.name}.fits',
sfftifier.op.asnumpy(sfftifier.FKDECO).real,
sfftifier.hdr_target],
['kernel_imag',
f'{sci_image.image.name}_-_{templ_image.image.name}.fits',
sfftifier.op.asnumpy(sfftifier.FKDECO).imag,
sfftifier.hdr_target],
]
}
# Write the intermediate files
for key in write_filepaths.keys():
for (imgtype, name, data, header) in write_filepaths[key]:
savepath = self.scratch_dir / f'{key}_{imgtype}_{name}'
self.write_fits_file( data, header, savepath=savepath )
SNLogger.info( f"DONE processing {sci_image.image.name} minus {templ_image.image.name}" )
if self.mem_trace:
SNLogger.info( f"After preprocessing, subtracting, and postprocessing \
a science image, memory usage = \
{tracemalloc.get_traced_memory()[1]/(1024**2):.2f} MB" )
sci_image.free()
SNLogger.info( f"DONE with all science images for template {templ_image.image.name}" )
templ_image.free()
SNLogger.info( "Waiting for FITS writer processes to finish" )
with nvtx.annotate( "fits_write_wait", color=0xff8888 ):
fits_writer_pool.close()
fits_writer_pool.join()
SNLogger.info( "...FITS writer processes done." )
if 'make_stamps' in steps:
SNLogger.info( "Starting to make stamps..." )
with nvtx.annotate( "make stamps", color=0xff8888 ):
def log_stamp_err( sci_image, templ_image, x ):
SNLogger.error( f"do_stamps failure for {sci_image.image.observation_id} \
{sci_image.image.sca} - \
{templ_image.image.observation_id} \
{templ_image.image.sca}: {x} " )
partialstamp = partial(stampmaker, self.diaobj.ra, self.diaobj.dec, np.array([100, 100]))
# original sci path, savedir, savename
sci_orig_stamp_args = ( (si.image, self.dia_out_dir, f'stamp_{str(si.image.name)}', 'data')
for si in self.science_images)
# template path, savedir, savename
templstamp_args = ( (ti.image, self.dia_out_dir, f'stamp_{str(ti.image.name)}', 'data')
for ti in self.template_images )
if self.nwrite > 1:
# Save stamps for original template images.
with Pool( self.nwrite ) as templ_stamp_pool:
templ_stamp_pool.starmap_async( partialstamp, templstamp_args,
error_callback=log_stamp_err )
templ_stamp_pool.close()
templ_stamp_pool.join()
# Save stamps for original science images.
with Pool( self.nwrite ) as sci_orig_stamp_pool:
sci_orig_stamp_pool.starmap_async( partialstamp, sci_orig_stamp_args,
error_callback=log_stamp_err )
sci_orig_stamp_pool.close()
sci_orig_stamp_pool.join()
# Save stamps for decorrelated difference image, zero point image (decorrelated sky-subtracted
# science image), and variance image for decorrelated difference image
with Pool( self.nwrite ) as sci_stamp_pool:
for sci_image in self.science_images:
for templ_image in self.template_images:
pair = (sci_image, templ_image)
stamperr_partial = partial(log_stamp_err, sci_image, templ_image)
sci_stamp_pool.apply_async( self.do_stamps, pair, {},
callback = partial(self.save_stamp_paths,
sci_image, templ_image),
error_callback=stamperr_partial )
sci_stamp_pool.close()
sci_stamp_pool.join()
else:
# Make stamp of just the template image
for tsargs in templstamp_args:
try:
partialstamp(*tsargs)
except Exception:
self.failures['make_stamps'].append(tsargs[0].fail_info)
# Make stamp of just the science image
for sosargs in sci_orig_stamp_args:
try:
partialstamp(*sosargs)
except Exception:
self.failures['make_stamps'].append(sosargs[0].fail_info)
for sci_image in self.science_images:
for templ_image in self.template_images:
if templ_image.image.name in sci_image.decorr_diff_path.keys():
try:
stamp_paths = self.do_stamps( sci_image, templ_image)
self.save_stamp_paths( sci_image, templ_image, stamp_paths )
except Exception:
if self.catchfailures:
self.failures['make_stamps'].append({
'science': sci_image.fail_info,
'template': templ_image.fail_info
})
SNLogger.info('...finished making stamps.')
if self.mem_trace:
SNLogger.info( f"After make_stamps, memory usage = {tracemalloc.get_traced_memory()[1]/(1024**2):.2f} MB" )
lightcurve_path = None
if 'make_lightcurve' in steps:
with nvtx.annotate( "make_lightcurve", color=0xff8888 ):
lightcurve_path = self.make_lightcurve()
# Debugging stuff below for checking phot location...
# import matplotlib.pyplot as plt
# from astropy.visualization import ZScaleInterval
# check_image = self.science_images[0]
# check_templ = self.template_images[0]
# check_diff_path = check_image.diff_stamp_path[ check_templ.image.name ]
# check_diff = fitsio.FITS(check_diff_path)
# check_image_data = check_diff[0].read()
# fig, ax = plt.subplots(nrows=1, ncols=4, figsize=(15, 5))
# norm = ZScaleInterval().get_limits(check_image_data)
# ax[0].imshow(check_image_data, origin='lower', vmin=norm[0], vmax=norm[1])
# ax[1].imshow(check_image_data - self.resid_img, origin='lower', vmin=norm[0], vmax=norm[1])
# ax[2].imshow(check_image.psf_data, origin='lower')
# im = ax[3].imshow(self.resid_img, origin='lower', vmin=norm[0], vmax=norm[1])
# ax[0].set_title('Decorr diff')
# ax[1].set_title('Decorr PSF')
# ax[2].set_title('Original PSF')
# ax[3].set_title('Decorr diff - Decorr PSF')
# # ax[2].set_xlim(-25,75)
# # ax[2].set_ylim(-25,75)
# plt.tight_layout()
# plt.savefig('/home/psf_resids.pdf', format='pdf')
# fig, ax = plt.subplots(1,1)
# ax.imshow(check_image_data, origin='lower', vmin=norm[0], vmax=norm[1])
# self.aperture.plot(color='red', lw='2', ax=ax)
# plt.savefig('/home/aperture_stamp.pdf')
if self.mem_trace:
SNLogger.info( f"After make_lightcurve, memory usage = \
{tracemalloc.get_traced_memory()[1]/(1024**2):.2f} MB" )
if self.remove_temp_dir:
self.clear_contents( self.temp_dir )
if self.catchfailures:
if np.sum( [len(x) for x in self.failures.values()] ) > 0:
SNLogger.info( f"There were some failures here! They were:\n{self.failures}" )
else:
SNLogger.info( f"No failures here! Fail list:\n{self.failures}" )
if lightcurve_path is None:
SNLogger.info( "Light curves saved to database!" )
return lightcurve_path
# ======================================================================
def main():
# Run one arg pass just to get the config file, so we can augment
# the full arg parser later with config options
configparser = argparse.ArgumentParser( add_help=False )
configparser.add_argument( '-c', '--config-file', default=None,
help=( "Location of the .yaml config file; defaults to the value of the "
"SNPIT_CONFIG environment varaible." ) )
args, leftovers = configparser.parse_known_args()
try:
cfg = Config.get( args.config_file, setdefault=True )
except RuntimeError as e:
if str(e) == 'No default config defined yet; run Config.init(configfile)':
sys.stderr.write( "Error, no configuration file defined.\n"
"Either run phrosty with -c <configfile>\n"
"or set the SNPIT_CONFIG environment variable.\n" )
sys.exit(1)
else:
raise
parser = argparse.ArgumentParser()
# Put in the config_file argument, even though it will never be found, so it shows up in help
parser.add_argument( '-c', '--config-file', help="Location of the .yaml config file" )
# Running options
parser.add_argument( '-p', '--nprocs', type=int, default=1,
help="Number of process for multiprocessing steps (e.g. skysub)" )
parser.add_argument( '-w', '--nwrite', type=int, default=5,
help="Number of parallel FITS writing processes" )
parser.add_argument( '-v', '--verbose', action='store_true', default=False,
help="Show debug log info" )
parser.add_argument( '--through-step', default='make_lightcurve',
help="Stop after this step; one of (see above)" )
parser.add_argument( '--dbsave', action='store_true',
help="Toggle saving to the database." )
parser.add_argument( '--memtrace', action='store_true',
help="Toggle memory tracing with tracemalloc.")
parser.add_argument( '--catchfailures', action='store_true',
help="Toggle failure collection. If true, pipeline does not \
cancel if one image fails. If false, pipeline crashes if \
one image fails (useful for debugging)." )
# Object collections
parser.add_argument( '-oc', '--object-collection', default='snpitdb',
help='Collection of the object. Currently, "snpitdb", "ou2024", and "manual" supported.' )
parser.add_argument( '-os', '--object-subset', default=None,
help="Collection subset. Not used by all collections." )
# SN and observation information
parser.add_argument( '--oid', type=int, required=True,
help="Object ID. Meaning is collection-dependent." )
parser.add_argument( '-r', '--ra', type=float, default=None,
help="Object RA. By default, uses the one found for the object." )
parser.add_argument( '-d', '--dec', type=float, default=None,
help="Object Dec. By default, uses the one found for the object." )
parser.add_argument( '-b', '--band', type=str, required=True,
help="Band: R062, Z087, Y106, J129, H158, F184, or K213" )
# Required args for using SN PIT database
# DiaObj
parser.add_argument( '-did', '--diaobject-id', type=str, default=None,
help="ID for DiaObject. Required to use SN PIT database. \
Invalid if --image-collection is not snpitdb." )
parser.add_argument( '-dpt', '--diaobject-provenance-tag', type=str, default=None,
help="Provenance tag for DiaObject. Required to use SN PIT database. \
Invalid if --image-collection is not snpitdb." )
parser.add_argument( '-dp', '--diaobject-process', type=str, required=False, default=None,
help="Process for DiaObject. Required to use SN PIT database. \
Invalid if --image-collection is not snpitdb.")
parser.add_argument( '-dppt', '--diaobject-position-provenance-tag', type=str, default=None,
help="Provenance tag for the position of the DiaObject." )
parser.add_argument( '-dpp', '--diaobject-position-process', type=str, default=None,
help="The process where the DiaObject position originated." )
# Lightcurve
parser.add_argument( '-lpi', '--ltcv-provenance-id', type=str, default=None,
help="Provenance ID for lightcurve. Required to use SN PIT database. \
Invalid if --image-collection is not snpitdb." )
parser.add_argument( '-lpt', '--ltcv-provenance-tag', type=str, default=None,
help="Provenance tag for lightcurve. Required to use SN PIT database. \
Invalid if --image-collection is not snpitdb." )
parser.add_argument( '-lp', '--ltcv-process', type=str, default='phrosty',
help="Process for light curve. Required to use SN PIT database. \
Invalid if --image-collection is not snpitdb." )
# Image collection
parser.add_argument( '-ic', '--image-collection', default='snpitdb',
help="Collection of the images we're using. For SN PIT database, use snpitdb. \
Currently supported: ou2024, manual_fits, snpitdb (default)." )
parser.add_argument( '-is', '--image-subset', default=None,
help="Image collection subset. To use SN PIT database, must be None." )
# Image
parser.add_argument( '-ipt', '--image-provenance-tag', default=None,
help='Provenance tag for images. Required to use SN PIT database. \
Invalid if --image-collection is not snpitdb.' )
parser.add_argument( '-ip', '--image-process', default=None,
help='Image process. Required to use SN PIT database. \
Invalid if --image-collection is not snpitdb.' )
# Path-based options
parser.add_argument( '--base-path', type=str, default=None,
help='Base path for reading images. Required for "manual_fits" image collection.' )
parser.add_argument( '-t', '--template-images', type=str, default=None,
help="Path to file with, per line, ( path_to_image, observation_id, sca, mjd, band )" )
parser.add_argument( '-s', '--science-images', type=str, default=None,
help="Path to file with, per line, ( path_to_image, observation_id, sca, mjd, band )" )
cfg.augment_argparse( parser )
args = parser.parse_args( leftovers )
cfg.parse_args( args )
if args.base_path is None and args.image_collection == 'manual_fits':
SNLogger.error( 'Must provide --base-path if --image-collection is manual_fits.' )
raise ValueError( f'args.base_path is {args.base_path}.' )
if args.image_collection == 'snpitdb' and args.image_provenance_tag is None:
SNLogger.error( 'Must provide --image-provenance-tag if --image-collection is snpitdb.' )
raise ValueError( f'args.image_provenance_tag is {args.image_provenance_tag}.' )
dbclient = SNPITDBClient()
# Get the DiaObject, update the RA and Dec
if args.diaobject_id is None:
diaobjs = DiaObject.find_objects( collection=args.object_collection,
# subset=args.object_subset,
provenance_tag=args.diaobject_provenance_tag,
process=args.diaobject_process,
name=args.oid, ra=args.ra, dec=args.dec )
if len( diaobjs ) == 0:
raise ValueError( f"Could not find DiaObject with id={args.id}, ra={args.ra}, dec={args.dec}." )
if len( diaobjs ) > 1:
raise ValueError( f"Found multiple DiaObject with id={args.id}, ra={args.ra}, dec={args.dec}." )
diaobj = diaobjs[0]
if args.ra is not None:
if np.fabs( args.ra - diaobj.ra ) > 1. / 3600. / np.cos( diaobj.dec * np.pi / 180. ):
SNLogger.warning( f"Given RA {args.ra} is far from DiaObject nominal RA {diaobj.ra}" )
diaobj.ra = args.ra
if args.dec is not None:
if np.fabs( args.dec - diaobj.dec ) > 1. / 3600.:
SNLogger.warning( f"Given Dec {args.dec} is far from DiaObject nominal Dec {diaobj.dec}" )
diaobj.dec = args.dec
# Get the image collection
imgcol = ImageCollection.get_collection( collection=args.image_collection,
subset=args.image_subset,
provenance_tag=args.image_provenance_tag,
process=args.image_process,
base_path=args.base_path,
dbclient=dbclient
)
# if args.image_collection == 'snpitdb':
# _found_images = imgcol.find_images( ra=diaobj.ra,
# dec=diaobj.dec,
# band=args.band,
# dbclient=dbclient
# )
# else:
# _found_images = imgcol.find_images( ra=diaobj.ra,
# dec=diaobj.dec,
# band=args.band
# )
# fetched_prov = Provenance.get_provs_for_tag( tag=args.diaobject_provenance_tag,
# process=args.diaobject_process
# )
# Create and launch the pipeline
pipeline = Pipeline( diaobj, imgcol, args.band,
science_csv=args.science_images,
template_csv=args.template_images,
oid=args.oid,
ltcv_prov_tag=args.ltcv_provenance_tag,
dbsave=args.dbsave,
dbclient=dbclient,
nprocs=args.nprocs,
nwrite=args.nwrite,
verbose=args.verbose,
memtrace=args.memtrace,
catchfailures=args.catchfailures )
pipeline( args.through_step )
# ======================================================================
if __name__ == "__main__":
main()
