"""Association Definitions: DMS Level3 product associations."""
import logging
from jwst.associations.lib.dms_base import (
Constraint_TargetAcq,
Constraint_TSO,
nissoss_calibrated_filter,
nrccoron_valid_detector,
nrsfss_valid_detector,
nrsifu_valid_detector,
)
from jwst.associations.lib.process_list import ListCategory
from jwst.associations.lib.rules_level3_base import (
ASN_SCHEMA, # noqa: F401
AsnMixin_AuxData,
AsnMixin_Coronagraphy,
AsnMixin_Science,
AsnMixin_Spectrum,
Constraint,
Constraint_IFU,
Constraint_Image,
Constraint_MSA,
Constraint_Optical_Path,
Constraint_Target,
DMS_Level3_Base,
DMSAttrConstraint,
SimpleConstraint,
Utility, # noqa: F401
dms_product_name_coronimage,
dms_product_name_noopt,
dms_product_name_nrsfs_sources,
dms_product_name_sources,
dms_product_name_wfss,
format_product,
)
from jwst.associations.registry import RegistryMarker
__all__ = [
"Asn_Lv3ACQ_Reprocess",
"Asn_Lv3AMI",
"Asn_Lv3Image",
"Asn_Lv3ImageMosaic",
"Asn_Lv3ImageBackground",
"Asn_Lv3MIRCoron",
"Asn_Lv3MIRMRS",
"Asn_Lv3MIRMRSBackground",
"Asn_Lv3NRCCoron",
"Asn_Lv3NRCCoronImage",
"Asn_Lv3NRSFSS",
"Asn_Lv3NRSIFU",
"Asn_Lv3NRSIFUBackground",
"Asn_Lv3SlitlessSpectral",
"Asn_Lv3SpecAux",
"Asn_Lv3SpectralSource",
"Asn_Lv3SpectralTarget",
"Asn_Lv3TSO",
"Asn_Lv3WFSCMB",
"Asn_Lv3WFSSNIS",
"Asn_Lv3WFSSNRC",
]
# Configure logging
logger = logging.getLogger(__name__)
# --------------------------------
# Start of the User-level rules
# --------------------------------
[docs]
@RegistryMarker.rule
class Asn_Lv3ACQ_Reprocess(DMS_Level3_Base):
"""
Level 3 Gather Target Acquisitions.
Characteristics:
- Association type: Not applicable
- Pipeline: Not applicable
- Used to populate other related associations
Notes
-----
For first loop, simply send acquisitions and confirms back.
"""
def __init__(self, *args, **kwargs):
# Setup for checking.
self.constraints = Constraint(
[
Constraint_TargetAcq(),
SimpleConstraint(
name="force_fail",
test=lambda _x, _y: False,
value="anything but None",
reprocess_on_fail=True,
work_over=ListCategory.NONSCIENCE,
reprocess_rules=[],
),
]
)
super(Asn_Lv3ACQ_Reprocess, self).__init__(*args, **kwargs)
[docs]
@RegistryMarker.rule
class Asn_Lv3AMI(AsnMixin_Science):
"""
Level 3 Aperture Mask Interferometry Association.
Characteristics:
- Association type: ``ami3``
- Pipeline: ``calwebb_ami3``
- Gather science and related PSF exposures
Notes
-----
AMI is nearly completely defined by the association candidates
produced by APT.
Tracking Issues:
- `github #310 <https://github.com/spacetelescope/jwst/issues/310>`_
"""
def __init__(self, *args, **kwargs):
# Setup for checking.
self.constraints = Constraint(
[
Constraint_Optical_Path(),
DMSAttrConstraint(
name="exp_type",
sources=["exp_type"],
value=("nis_ami"),
),
DMSAttrConstraint(
name="target",
sources=["targetid"],
onlyif=lambda item: self.get_exposure_type(item) == "science",
force_reprocess=ListCategory.EXISTING,
only_on_match=True,
),
]
)
# PSF is required
self.validity.update(
{"has_psf": {"validated": False, "check": lambda entry: entry["exptype"] == "psf"}}
)
# Check and continue initialization.
super(Asn_Lv3AMI, self).__init__(*args, **kwargs)
def _init_hook(self, item):
"""Post-check and pre-add initialization."""
self.data["asn_type"] = "ami3"
super(Asn_Lv3AMI, self)._init_hook(item)
[docs]
@RegistryMarker.rule
class Asn_Lv3Image(AsnMixin_Science):
"""
Level 3 Science Image Association.
Characteristics:
- Association type: ``image3``
- Pipeline: ``calwebb_image3``
- Non-TSO
- Non-WFS&C
"""
def __init__(self, *args, **kwargs):
# Setup constraints
self.constraints = Constraint(
[
Constraint_Optical_Path(),
Constraint_Target(association=self),
Constraint_Image(),
DMSAttrConstraint(
name="wfsvisit", sources=["visitype"], value="((?!wfsc).)*", required=False
),
Constraint(
[
DMSAttrConstraint(
name="bkgdtarg",
sources=["bkgdtarg"],
),
Constraint_TSO(),
],
reduce=Constraint.notany,
),
]
)
# Now check and continue initialization.
super(Asn_Lv3Image, self).__init__(*args, **kwargs)
def _init_hook(self, item):
"""Post-check and pre-add initialization."""
self.data["asn_type"] = "image3"
super(Asn_Lv3Image, self)._init_hook(item)
[docs]
@RegistryMarker.rule
class Asn_Lv3ImageBackground(AsnMixin_AuxData, AsnMixin_Science):
"""
Level 3 Background Image Association.
Characteristics:
- Association type: ``image3``
- Pipeline: ``calwebb_image3``
- Non-TSO
- Non-WFS&C
"""
def __init__(self, *args, **kwargs):
# Setup constraints
self.constraints = Constraint(
[
Constraint_Optical_Path(),
Constraint_Target(association=self),
Constraint_Image(),
DMSAttrConstraint(
name="wfsvisit", sources=["visitype"], value="((?!wfsc).)*", required=False
),
DMSAttrConstraint(
name="bkgdtarg",
sources=["bkgdtarg"],
),
Constraint([Constraint_TSO()], reduce=Constraint.notany),
]
)
# Now check and continue initialization.
super(Asn_Lv3ImageBackground, self).__init__(*args, **kwargs)
def _init_hook(self, item):
"""Post-check and pre-add initialization."""
self.data["asn_type"] = "image3"
super(Asn_Lv3ImageBackground, self)._init_hook(item)
[docs]
@RegistryMarker.rule
class Asn_Lv3MIRCoron(AsnMixin_Coronagraphy, AsnMixin_Science):
"""
Level 3 Coronagraphy Association.
Characteristics:
- Association type: ``coron3``
- Pipeline: ``calwebb_coron3``
- MIRI Coronagraphy
- Gather science and related PSF exposures
Notes
-----
Coronagraphy is nearly completely defined by the association candidates
produced by APT.
Tracking Issues:
- `github #311 <https://github.com/spacetelescope/jwst/issues/311>`_
- `JP-3219 <https://jira.stsci.edu/browse/JP-3219>`_
"""
def __init__(self, *args, **kwargs):
# Setup for checking.
self.constraints = Constraint(
[
Constraint_Optical_Path(),
DMSAttrConstraint(
name="exp_type",
sources=["exp_type"],
value="mir_lyot|mir_4qpm",
),
DMSAttrConstraint(
name="target",
sources=["targetid"],
onlyif=lambda item: self.get_exposure_type(item) == "science",
force_reprocess=ListCategory.EXISTING,
only_on_match=True,
),
Constraint(
[
DMSAttrConstraint(
name="bkgdtarg",
sources=["bkgdtarg"],
force_unique=False,
)
],
reduce=Constraint.notany,
),
],
name="asn_coron",
)
# Check and continue initialization.
super(Asn_Lv3MIRCoron, self).__init__(*args, **kwargs)
[docs]
@RegistryMarker.rule
class Asn_Lv3MIRMRS(AsnMixin_Spectrum):
"""
Level 3 MIRI MRS Association.
Characteristics:
- Association type: ``spec3``
- Pipeline: ``calwebb_spec3``
- Just MIRI MRS
- optical path determined by calibration
- Cannot be TSO
- Must have pattern type defined
"""
def __init__(self, *args, **kwargs):
# Setup for checking.
self.constraints = Constraint(
[
Constraint_Target(association=self),
DMSAttrConstraint(
name="exp_type",
sources=["exp_type"],
value="mir_mrs",
),
Constraint(
[
Constraint_TSO(),
],
reduce=Constraint.notany,
),
]
)
# Check and continue initialization.
super(Asn_Lv3MIRMRS, self).__init__(*args, **kwargs)
@property
def dms_product_name(self):
"""
Return product name with no optical element entries.
Returns
-------
str
The product name with no optical elements.
"""
return dms_product_name_noopt(self)
[docs]
@RegistryMarker.rule
class Asn_Lv3MIRMRSBackground(AsnMixin_AuxData, AsnMixin_Spectrum):
"""
Level 3 MIRI MRS Association Auxiliary data.
Characteristics:
- Association type: ``spec3``
- Pipeline: ``calwebb_spec3``
- Just MIRI MRS
- optical path determined by calibration
- Cannot be TSO
- Must have pattern type defined
"""
def __init__(self, *args, **kwargs):
# Setup for checking.
self.constraints = Constraint(
[
Constraint_Target(),
DMSAttrConstraint(
name="exp_type",
sources=["exp_type"],
value="mir_mrs",
),
Constraint(
[
Constraint_TSO(),
],
reduce=Constraint.notany,
),
DMSAttrConstraint(
name="bkgdtarg",
sources=["bkgdtarg"],
value="T",
),
]
)
# Check and continue initialization.
super(Asn_Lv3MIRMRSBackground, self).__init__(*args, **kwargs)
@property
def dms_product_name(self):
"""
Return product name with no optical element entries.
Returns
-------
str
The product name built with no optical elements.
"""
return dms_product_name_noopt(self)
[docs]
@RegistryMarker.rule
class Asn_Lv3NRCCoron(AsnMixin_Coronagraphy, AsnMixin_Science):
"""
Level 3 Coronagraphy Association.
Characteristics:
- Association type: ``coron3``
- Pipeline: ``calwebb_coron3``
- Gather science and related PSF exposures
- Exclude "extra" NIRCam detectors that don't have target on them
Notes
-----
Coronagraphy is nearly completely defined by the association candidates
produced by APT.
Tracking Issues:
- `github #311 <https://github.com/spacetelescope/jwst/issues/311>`_
- `JP-3219 <https://jira.stsci.edu/browse/JP-3219>`_
"""
def __init__(self, *args, **kwargs):
# Setup for checking.
self.constraints = Constraint(
[
Constraint_Optical_Path(),
DMSAttrConstraint(
name="exp_type",
sources=["exp_type"],
value=("nrc_coron"),
),
DMSAttrConstraint(
name="target",
sources=["targetid"],
onlyif=lambda item: self.get_exposure_type(item) == "science",
force_reprocess=ListCategory.EXISTING,
only_on_match=True,
),
Constraint(
[
DMSAttrConstraint(
name="bkgdtarg",
sources=["bkgdtarg"],
force_unique=False,
)
],
reduce=Constraint.notany,
),
SimpleConstraint(
value=True,
test=lambda _value, item: nrccoron_valid_detector(item),
force_unique=False,
),
],
name="asn_coron",
)
# Check and continue initialization.
super(Asn_Lv3NRCCoron, self).__init__(*args, **kwargs)
[docs]
@RegistryMarker.rule
class Asn_Lv3NRCCoronImage(AsnMixin_Science):
"""
Level 3 Coronagraphy Association handled as regular imaging.
Characteristics:
- Association type: ``image3``
- Pipeline: ``calwebb_image3``
- Gather science exposures only, no psf exposures
- Only include NRC SW images taken in full-frame
"""
def __init__(self, *args, **kwargs):
# Setup for checking.
self.constraints = Constraint(
[
Constraint_Optical_Path(),
DMSAttrConstraint(
name="exp_type",
sources=["exp_type"],
value=("nrc_coron"),
),
DMSAttrConstraint(
name="target",
sources=["targetid"],
onlyif=lambda item: self.get_exposure_type(item) == "science",
force_reprocess=ListCategory.EXISTING,
only_on_match=True,
),
Constraint(
[
DMSAttrConstraint(
name="bkgdtarg",
sources=["bkgdtarg"],
force_unique=False,
),
DMSAttrConstraint(name="is_psf", sources=["is_psf"], value=("T")),
],
reduce=Constraint.notany,
),
DMSAttrConstraint(
name="channel",
sources=["channel"],
value=("short"),
),
DMSAttrConstraint(
name="subarray",
sources=["subarray"],
value=("full"),
),
],
)
# Check and continue initialization.
super(Asn_Lv3NRCCoronImage, self).__init__(*args, **kwargs)
@property
def dms_product_name(self):
"""
Return product name built for coronagraphic image products.
Returns
-------
str
Product name built for coronagraphic images.
"""
return dms_product_name_coronimage(self)
def _init_hook(self, item):
"""Post-check and pre-add initialization."""
self.data["asn_type"] = "image3"
super(Asn_Lv3NRCCoronImage, self)._init_hook(item)
[docs]
def is_item_coron(self, _item):
"""
Override to ignore coronagraphic designation.
Coronagraphic data is to be processed both as coronagraphic
(by default), but also as just plain imaging. Coronagraphic
data is processed using the Asn_Lv3Coron rule. This rule
will handle the creation of the image version. It causes
the input members to be of type "cal", instead of "calints".
Parameters
----------
_item : ACID
Ignored for this override method.
Returns
-------
bool
False.
"""
return False
[docs]
@RegistryMarker.rule
class Asn_Lv3NRSFSS(AsnMixin_Spectrum):
"""
Level 3 NIRSpec Fixed-slit Science.
Characteristics:
- Association type: ``spec3``
- Pipeline: ``calwebb_spec3``
- NIRSpec Fixed-slit Science
- Non-TSO
"""
def __init__(self, *args, **kwargs):
# Setup for checking.
self.constraints = Constraint(
[
Constraint([Constraint_TSO()], reduce=Constraint.notany),
DMSAttrConstraint(
name="exp_type",
sources=["exp_type"],
value=("nrs_autoflat|nrs_autowave|nrs_fixedslit"),
force_unique=False,
),
SimpleConstraint(
value=True,
test=lambda _value, item: nrsfss_valid_detector(item),
force_unique=False,
),
Constraint_Optical_Path(),
Constraint_Target(association=self),
]
)
# Check and continue initialization.
super(Asn_Lv3NRSFSS, self).__init__(*args, **kwargs)
@property
def dms_product_name(self):
"""
Return product name built for NIRSpec fixed-slit sources.
Returns
-------
str
The product name using both target and source ids.
"""
return dms_product_name_nrsfs_sources(self)
[docs]
@RegistryMarker.rule
class Asn_Lv3NRSIFU(AsnMixin_Spectrum):
"""
Level 3 IFU gratings Association.
Characteristics:
- Association type: ``spec3``
- Pipeline: ``calwebb_spec3``
- optical path determined by calibration
"""
def __init__(self, *args, **kwargs):
# Setup for checking.
self.constraints = Constraint(
[
Constraint_Target(association=self),
DMSAttrConstraint(
name="exp_type",
sources=["exp_type"],
value=("nrs_autowave|nrs_ifu"),
force_unique=False,
),
SimpleConstraint(
value=True,
test=lambda _value, item: nrsifu_valid_detector(item),
force_unique=False,
),
DMSAttrConstraint(name="patttype", sources=["patttype"], required=True),
Constraint(
[
Constraint_TSO(),
],
reduce=Constraint.notany,
),
DMSAttrConstraint(
name="opt_elem",
sources=["grating"],
),
]
)
# Check and continue initialization.
super(Asn_Lv3NRSIFU, self).__init__(*args, **kwargs)
[docs]
@RegistryMarker.rule
class Asn_Lv3NRSIFUBackground(AsnMixin_AuxData, AsnMixin_Spectrum):
"""
Level 3 Spectral Association.
Characteristics:
- Association type: ``spec3``
- Pipeline: ``calwebb_spec3``
"""
def __init__(self, *args, **kwargs):
# Setup for checking.
self.constraints = Constraint(
[
Constraint_Target(association=self),
Constraint(
[
Constraint_TSO(),
],
reduce=Constraint.notany,
),
DMSAttrConstraint(
name="bkgdtarg",
sources=["bkgdtarg"],
value="T",
),
DMSAttrConstraint(
name="allowed_bkgdtarg",
sources=["exp_type"],
value="nrs_ifu",
),
SimpleConstraint(
value=True,
test=lambda _value, item: nrsifu_valid_detector(item),
force_unique=False,
),
DMSAttrConstraint(
name="opt_elem",
sources=["grating"],
force_unique=True,
),
]
)
# Check and continue initialization.
super(Asn_Lv3NRSIFUBackground, self).__init__(*args, **kwargs)
[docs]
@RegistryMarker.rule
class Asn_Lv3SlitlessSpectral(AsnMixin_Spectrum):
"""
Level 3 slitless, target-based or single-object spectrographic Association.
Characteristics:
- Association type: ``spec3``
- Pipeline: ``calwebb_spec3``
- Single target
- Non-TSO
"""
def __init__(self, *args, **kwargs):
# Setup for checking.
self.constraints = Constraint(
[
Constraint([Constraint_TSO()], reduce=Constraint.notany),
Constraint_Optical_Path(),
Constraint_Target(association=self),
DMSAttrConstraint(
name="exp_type", sources=["exp_type"], value=("nis_soss"), force_unique=False
),
Constraint(
[
DMSAttrConstraint(
name="patttype_spectarg",
sources=["patttype"],
),
],
reduce=Constraint.notany,
),
# Constraint to prevent calibration data from level 3 processing
Constraint(
[
DMSAttrConstraint(
name="restricted_slitless",
sources=["exp_type"],
value=("mir_lrs-slitless"),
),
DMSAttrConstraint(name="tso_obs", sources=["tso_visit"], value=("T")),
],
reduce=Constraint.notany,
),
]
)
# Check and continue initialization.
super(Asn_Lv3SlitlessSpectral, self).__init__(*args, **kwargs)
[docs]
@RegistryMarker.rule
class Asn_Lv3SpecAux(AsnMixin_AuxData, AsnMixin_Spectrum):
"""
Level 3 Spectral Association.
Characteristics:
- Association type: ``spec3``
- Pipeline: ``calwebb_spec3``
"""
def __init__(self, *args, **kwargs):
# Setup for checking.
self.constraints = Constraint(
[
Constraint_Target(association=self),
Constraint(
[
Constraint_TSO(),
],
reduce=Constraint.notany,
),
DMSAttrConstraint(
name="bkgdtarg",
sources=["bkgdtarg"],
value="T",
),
DMSAttrConstraint(
name="allowed_bkgdtarg",
sources=["exp_type"],
value="mir_lrs-fixedslit|nrs_fixedslit",
),
Constraint_Optical_Path(),
]
)
# Check and continue initialization.
super(Asn_Lv3SpecAux, self).__init__(*args, **kwargs)
[docs]
@RegistryMarker.rule
class Asn_Lv3SpectralSource(AsnMixin_Spectrum):
"""
Level 3 slit-like, multi-object spectrographic Association.
Characteristics:
- Association type: ``spec3``
- Pipeline: ``calwebb_spec3``
- Multi-object
- Non-TSO
"""
def __init__(self, *args, **kwargs):
# Setup for checking.
self.constraints = Constraint(
[
Constraint([Constraint_TSO()], reduce=Constraint.notany),
Constraint_Optical_Path(),
Constraint_Target(association=self),
Constraint(
[
DMSAttrConstraint(
name="exp_type",
sources=["exp_type"],
value=("nrs_autoflat|nrs_autowave"),
force_unique=False,
),
Constraint_MSA(),
],
reduce=Constraint.any,
),
]
)
# Check and continue initialization.
super(Asn_Lv3SpectralSource, self).__init__(*args, **kwargs)
@property
def dms_product_name(self):
"""
Return source-based product name.
Returns
-------
str
The product name using source id.
"""
return dms_product_name_sources(self)
[docs]
@RegistryMarker.rule
class Asn_Lv3SpectralTarget(AsnMixin_Spectrum):
"""
Level 3 slit-like, target-based or single-object spectrographic Association.
Characteristics:
- Association type: ``spec3``
- Pipeline: ``calwebb_spec3``
- Single target
- Non-TSO
"""
def __init__(self, *args, **kwargs):
# Setup for checking.
self.constraints = Constraint(
[
Constraint([Constraint_TSO()], reduce=Constraint.notany),
Constraint_Optical_Path(),
Constraint_Target(association=self),
DMSAttrConstraint(
name="exp_type",
sources=["exp_type"],
value=("mir_lrs-fixedslit|nis_soss"),
force_unique=False,
),
Constraint(
[
DMSAttrConstraint(
name="patttype_spectarg",
sources=["patttype"],
value="2-point-nod|4-point-nod|along-slit-nod",
),
],
reduce=Constraint.any,
),
]
)
# Check and continue initialization.
super(Asn_Lv3SpectralTarget, self).__init__(*args, **kwargs)
[docs]
def finalize(self):
"""
Finalize association.
For NRS Fixed-slit, finalization means creating new members for the
background nods.
Returns
-------
associations : [association[, ...]] or None
List of fully-qualified associations that this association
represents.
`None` if a complete association cannot be produced.
"""
if self.is_valid:
return self.make_fixedslit_bkg()
return None
[docs]
@RegistryMarker.rule
class Asn_Lv3TSO(AsnMixin_Science):
"""
Level 3 Time-Series Association.
Characteristics:
- Association type: ``tso3``
- Pipeline: ``calwebb_tso3``
"""
def __init__(self, *args, **kwargs):
# Setup for checking.
self.constraints = Constraint(
[
Constraint_Target(association=self),
Constraint_Optical_Path(),
Constraint_TSO(),
DMSAttrConstraint(
name="exp_type",
sources=["exp_type"],
),
# Don't allow IFU exposures in tso3
Constraint(
[
Constraint_IFU(),
],
reduce=Constraint.notany,
),
# Don't allow NIRCam engineering mode
# with PUPIL='CLEAR' in tso3
Constraint(
[
Constraint(
[
DMSAttrConstraint(
name="restricted_grism",
sources=["exp_type"],
value="nrc_tsgrism",
),
DMSAttrConstraint(
name="grism_clear",
sources=["pupil"],
value="clear|gdhs0|gdhs60",
),
]
),
Constraint(
[
DMSAttrConstraint(
name="restricted_ts", sources=["exp_type"], value="nrc_tsgrism"
),
DMSAttrConstraint(
name="module", sources=["detector"], value="nrcblong"
),
]
),
],
reduce=Constraint.notany,
),
# Don't allow NIRISS SOSS with NINTS=1 or uncalibrated filters
Constraint(
[
Constraint(
[
DMSAttrConstraint(
name="exp_type", sources=["exp_type"], value="nis_soss"
),
DMSAttrConstraint(name="nints", sources=["nints"], value="1"),
]
),
Constraint(
[
DMSAttrConstraint(
name="exp_type", sources=["exp_type"], value="nis_soss"
),
SimpleConstraint(
value=False,
test=lambda value, item: nissoss_calibrated_filter(item)
== value,
force_unique=False,
),
]
),
],
reduce=Constraint.notany,
),
# Don't allow NIRSpec invalid optical paths in TSO3
Constraint(
[
Constraint(
[
DMSAttrConstraint(
name="exp_type", sources=["exp_type"], value="nrs_brightobj"
),
SimpleConstraint(
value=False,
test=lambda value, item: nrsfss_valid_detector(item) == value,
force_unique=False,
),
]
),
],
reduce=Constraint.notany,
),
]
)
# Only valid if candidate type is 'observation'.
self.validity.update(
{"is_type_observation": {"validated": False, "check": self._validate_candidates}}
)
super(Asn_Lv3TSO, self).__init__(*args, **kwargs)
def _init_hook(self, item):
"""Post-check and pre-add initialization."""
self.data["asn_type"] = "tso3"
super(Asn_Lv3TSO, self)._init_hook(item)
def _validate_candidates(self, _member):
"""
Allow only observation-type candidates.
Parameters
----------
_member : member
Member being added; ignored.
Returns
-------
bool
True if candidate type is observation, false otherwise.
"""
# If a group candidate, reject.
if self.acid.type.lower() != "observation":
return False
return True
[docs]
@RegistryMarker.rule
class Asn_Lv3WFSCMB(AsnMixin_Science):
"""
Level 3 Wavefront Control & Sensing Association.
For coarse and fine phasing, dither pairs need to
be associated to be combined. The optical path
is assumed to be equivalent within an activity.
Characteristics:
- Association type: ``wfs-image3``
- Pipeline: ``calwebb_wfs-image3``
- Coarse and fine phasing dithers
"""
def __init__(self, *args, **kwargs):
# Setup constraints
self.constraints = Constraint(
[
Constraint_Optical_Path(),
Constraint_Target(association=self),
Constraint_Image(),
DMSAttrConstraint(name="patttype", sources=["patttype"], value="wfsc"),
DMSAttrConstraint(name="detector", sources=["detector"]),
DMSAttrConstraint(name="obs_id", sources=["obs_id"]),
DMSAttrConstraint(name="act_id", sources=["act_id"]),
Constraint(
[
DMSAttrConstraint(
name="dms_note", sources=["dms_note"], value="wfsc_los_jitter"
),
],
reduce=Constraint.notany,
),
]
)
# Only valid if two members exist and candidate is not a GROUP.
self.validity.update(
{
"has_pair": {"validated": False, "check": self._has_pair},
"is_not_group": {"validated": False, "check": self._validate_candidates},
}
)
super(Asn_Lv3WFSCMB, self).__init__(*args, **kwargs)
def _init_hook(self, item):
"""Post-check and pre-add initialization."""
self.data["asn_type"] = "wfs-image3"
super(Asn_Lv3WFSCMB, self)._init_hook(item)
@property
def dms_product_name(self):
"""
Define product name.
Modification is to append the `expspcin` value
after the calibration suffix.
Returns
-------
str
The product name in lowercase.
"""
product_name_format = "{existing}-{detector}_{suffix}-{expspcin}"
existing = super().dms_product_name
product_name = format_product(
product_name_format,
existing=existing,
detector=self.constraints["detector"].value,
expspcin=self.constraints["act_id"].value,
)
return product_name.lower()
def _has_pair(self, entry=None):
"""
Check if current product has two members.
If `entry` is given, it is counted as one of the
members. If not, the existing member list is only
accounted for.
Parameters
----------
entry : dict or None
New entry to add to member list.
Returns
-------
bool
True if there are two members.
"""
if entry is None:
count = 2
else:
count = 1
return len(self.current_product["members"]) == count
def _validate_candidates(self, _member):
"""
Disallow GROUP candidates.
Parameters
----------
_member : Member
Member being added; ignored.
Returns
-------
bool
False if candidate is GROUP, true otherwise.
"""
# If a group candidate, reject.
if self.acid.type.lower() == "group":
return False
return True
[docs]
@RegistryMarker.rule
class Asn_Lv3WFSSNRC(AsnMixin_Spectrum):
"""
Level 3 NIRCam WFSS/Grism Association.
Characteristics:
- Association type: ``spec3``
- Pipeline: ``calwebb_spec3``
- Gather all grism exposures
"""
def __init__(self, *args, **kwargs):
# Setup for checking.
self.constraints = Constraint(
[
Constraint_Target(association=self),
DMSAttrConstraint(
name="exp_type",
sources=["exp_type"],
value="nrc_wfss",
),
DMSAttrConstraint(
name="opt_elem",
sources=["pupil"],
value="GRISMR|GRISMC",
force_unique=True,
),
]
)
# Check and continue initialization.
super().__init__(*args, **kwargs)
@property
def dms_product_name(self):
"""
Return product name.
Returns
-------
str
The product name.
"""
return dms_product_name_wfss(self)
[docs]
@RegistryMarker.rule
class Asn_Lv3WFSSNIS(AsnMixin_Spectrum):
"""
Level 3 NIRISS WFSS/Grism Association.
Characteristics:
- Association type: ``spec3``
- Pipeline: ``calwebb_spec3``
- Gather all grism exposures
"""
def __init__(self, *args, **kwargs):
# Setup for checking.
self.constraints = Constraint(
[
Constraint_Target(association=self),
DMSAttrConstraint(
name="exp_type",
sources=["exp_type"],
value="nis_wfss",
),
DMSAttrConstraint(
name="opt_elem",
sources=["filter"],
value="gr150r|gr150c",
force_unique=True,
),
DMSAttrConstraint(
name="opt_elem2",
sources=["pupil"],
),
]
)
# Check and continue initialization.
super(Asn_Lv3WFSSNIS, self).__init__(*args, **kwargs)
@property
def dms_product_name(self):
"""
Provide product name.
Returns
-------
str
Product name.
"""
return dms_product_name_wfss(self)
[docs]
@RegistryMarker.rule
class Asn_Lv3ImageMosaic(AsnMixin_Science):
"""
Level 3 Science Image Mosaic Association.
Characteristics:
- Association type: ``image3``
- Pipeline: ``calwebb_image3``
- Non-TSO
- Non-WFS&C
- Collect separate tiles of mosaic into one product
"""
def __init__(self, *args, **kwargs):
# Setup constraints
self.constraints = Constraint(
[
Constraint_Optical_Path(),
Constraint_Image(),
DMSAttrConstraint(
name="wfsvisit", sources=["visitype"], value="((?!wfsc).)*", required=False
),
Constraint(
[
DMSAttrConstraint(
name="bkgdtarg",
sources=["bkgdtarg"],
),
Constraint_TSO(),
],
reduce=Constraint.notany,
),
]
)
# Only valid if candidate type is 'mosaic'.
self.validity.update(
{"is_type_mosaic": {"validated": False, "check": self._validate_candidates}}
)
# Now check and continue initialization.
super(Asn_Lv3ImageMosaic, self).__init__(*args, **kwargs)
def _init_hook(self, item):
"""Post-check and pre-add initialization."""
self.data["asn_type"] = "image3"
super(Asn_Lv3ImageMosaic, self)._init_hook(item)
def _validate_candidates(self, _item):
"""
Allow only mosaic asn candidates.
Parameters
----------
_item : ACID
Ignored in this method.
Returns
-------
bool
True if candidate type is mosaic, false otherwise.
"""
# If a group candidate, reject.
if self.acid.type.lower() != "mosaic":
return False
return True