import json
import logging
import re
import warnings
from collections.abc import MutableMapping
from datetime import UTC, datetime
from pathlib import Path
import jsonschema
from astropy.utils.decorators import deprecated_renamed_argument
from astropy.utils.exceptions import AstropyDeprecationWarning
from jwst import __version__
from jwst.associations.association_io import json_asn_dump, json_asn_load
from jwst.associations.exceptions import AssociationNotValidError
from jwst.associations.format_template import FormatTemplate
from jwst.associations.lib.constraint import Constraint, meets_conditions
__all__ = ["Association"]
# Configure logging
logger = logging.getLogger(__name__)
# Timestamp template
_TIMESTAMP_TEMPLATE = "%Y%m%dt%H%M%S"
[docs]
class Association(MutableMapping):
"""
Association base class.
Parameters
----------
version_id : str or None
Version ID to use in the name of this association.
If None, nothing is added.
Attributes
----------
instance : dict-like
The instance is the association data structure.
See ``data`` below.
meta : dict
Information about the association.
data : dict
The association. The format of this data structure
is determined by the individual associations and, if
defined, validated against their specified schema.
schema_file : str
The name of the output schema that an association
must adhere to.
Raises
------
jwst.associations.exceptions.AssociationError
If an item doesn't match.
"""
registry = None
"""Registry this rule has been placed in."""
DEFAULT_FORCE_UNIQUE = False
"""Default whether to force constraints to use unique values."""
DEFAULT_REQUIRE_CONSTRAINT = True
"""Default require that the constraint exists or otherwise
can be explicitly checked.
"""
DEFAULT_EVALUATE = False
"""Default do not evaluate input values."""
GLOBAL_CONSTRAINT = None
"""Global constraints."""
INVALID_VALUES: tuple | None = None
"""Attribute values that indicate the attribute is not specified."""
def __init__(self, version_id=None):
self.data = {}
self.run_init_hook = True
self.meta = {}
self.version_id = version_id
self.data.update(
{
"asn_type": "None",
"asn_rule": self.asn_rule,
"version_id": self.version_id,
"code_version": __version__,
}
)
# Setup constraints
# These may be predefined by a rule.
try:
constraints = self.constraints
except AttributeError:
constraints = Constraint()
if self.GLOBAL_CONSTRAINT is not None:
constraints.append(self.GLOBAL_CONSTRAINT.copy())
self.constraints = constraints
[docs]
@classmethod
def create(cls, item, version_id=None):
"""
Create association if item belongs.
Parameters
----------
item : dict
The item to initialize the association with.
version_id : str or None
Version ID to use in the name of this association.
If None, nothing is added.
Returns
-------
asn : `~jwst.associations.association.Association` or None
The association or, if the item does not match this rule, None.
reprocess : list of `~jwst.associations.lib.process_list.ProcessList`
List of items to process again.
"""
asn = cls(version_id=version_id)
matches, reprocess = asn.add(item)
if not matches:
return None, reprocess
return asn, reprocess
@property
def asn_name(self):
"""
Suggest filename for the association.
Returns
-------
str
Default asn name of 'unnamed_association'.
"""
return "unnamed_association"
[docs]
@classmethod
def rule_name(cls):
"""
Return rule name.
Returns
-------
str
The name of the rule class.
"""
return cls.__name__
@property
def asn_rule(self): # numpydoc ignore=RT01
"""Same as :meth:`rule_name`."""
return self.rule_name()
[docs]
@classmethod
def validate(cls, asn, error_on_fail=True):
"""
Validate an association against this rule.
Parameters
----------
asn : `~jwst.associations.association.Association`
The association structure to examine.
error_on_fail : bool
On validation error, throw exception instead of
changing return status.
Returns
-------
valid : bool
`True` if valid. When invalid, an exception is raised
if ``error_on_fail`` is `True`, otherwise `False`.
Raises
------
jwst.associations.exceptions.AssociationNotValidError
Validation failed and ``error_on_fail`` is `True`.
Notes
-----
The base method checks against the rule class' schema (``schema_file``).
If the rule class does not define a schema, a warning is issued
in logger but the routine will still return `True`.
"""
if not hasattr(cls, "schema_file"):
logger.warning("Cannot validate: %s has no schema. Presuming OK.", str(cls))
return True
if isinstance(asn, cls):
asn_data = asn.data
else:
asn_data = asn
with Path(cls.schema_file).open("r") as schema_file:
asn_schema = json.load(schema_file)
try:
jsonschema.validate(asn_data, asn_schema)
except (AttributeError, jsonschema.ValidationError) as err:
logger.debug("Validation failed:\n%s", str(err))
if error_on_fail:
raise AssociationNotValidError("Validation failed") from err
else:
return False
# Warn if path data found for expnames
no_path = Path()
for product in asn_data["products"]:
members = product["members"]
for member in members:
fpath = Path(member["expname"]).parent
if fpath != no_path:
err_str = (
"Input association file contains path information; "
"note that this can complicate usage and/or sharing "
"of such files."
)
logger.debug(err_str)
warnings.warn(err_str, UserWarning, stacklevel=1)
return True
[docs]
@deprecated_renamed_argument("fmt", None, since="2.1")
def dump(
self,
fmt=None, # noqa: ARG002
**kwargs,
):
"""
Serialize the association.
Parameters
----------
fmt : str
The format to use to dump the association into.
.. version-deprecated:: 2.1
Only JSON format is supported now.
**kwargs
List of arguments to pass to the registered
routines for the current association type.
.. version-deprecated:: 2.1
This is completely ignored.
Returns
-------
asn_filename : str
Suggested base name for the JSON file.
This is taken from the ``asn_name`` attribute.
serialized : str
JSON serialization of this association.
Raises
------
jwst.associations.exceptions.AssociationNotValidError
If the association does not validate.
"""
if kwargs:
warnings.warn(
"Usage of kwargs was deprecated in version 2.1 and "
"will be removed in a future version; it currently "
"is completely ignored.",
AstropyDeprecationWarning,
stacklevel=2,
)
if self.is_valid:
return json_asn_dump(self)
raise AssociationNotValidError(f"Association {self} is not valid")
[docs]
@classmethod
@deprecated_renamed_argument("fmt", None, since="2.1")
def load(
cls,
serialized,
fmt=None, # noqa: ARG003
validate=True,
**kwargs,
):
"""
Load a serialized association.
Parameters
----------
serialized : str, dict, or file-like
The serialized form of the association.
fmt : str or None
The format to force. If None, try all available.
.. version-deprecated:: 2.1
Only JSON format is supported now.
validate : bool
Validate against the class' defined schema, if any.
**kwargs : dict
Other arguments to pass to the ``load`` method.
.. version-deprecated:: 2.1
This is completely ignored.
Returns
-------
association : `~jwst.associations.association.Association`
The association.
Raises
------
jwst.associations.exceptions.AssociationNotValidError
Cannot create or validate the association.
"""
if kwargs:
warnings.warn(
"Usage of kwargs was deprecated in version 2.1 and "
"will be removed in a future version; it currently "
"is completely ignored.",
AstropyDeprecationWarning,
stacklevel=2,
)
asn = json_asn_load(serialized)
if validate:
cls.validate(asn)
return asn
@property
def is_valid(self):
"""
Check if association is valid.
Returns
-------
bool
`True` if association is valid, otherwise `False`.
"""
return self.__class__.validate(self, error_on_fail=False)
[docs]
def add(self, item, check_constraints=True):
"""
Add the item to the association.
Parameters
----------
item : dict
The item to add.
check_constraints : bool
If `True`, see if the item should belong to this association.
If `False`, just add it.
Returns
-------
match : bool
`True` if the all constraints are satisfied or skipped.
This could also be the value of ``self.constraints['force_match']``
when it is set.
reprocess : list of `~jwst.associations.lib.process_list.ProcessList`
List of items to process again.
"""
if self.is_item_member(item):
return True, []
match = not check_constraints
if check_constraints:
match, reprocess = self.check_and_set_constraints(item)
if match:
if self.run_init_hook:
self._init_hook(item)
self.run_init_hook = False
self._add(item)
# If a constraint `force_match` exists, set the `match`
# result to the value of the constraint.
try:
force_match = self.constraints["force_match"].value
except (KeyError, TypeError):
pass
else:
if force_match is not None:
match = force_match
return match, reprocess
[docs]
def check_and_set_constraints(self, item):
"""
Check whether the given dictionaries match parameters for this association.
Parameters
----------
item : dict
The parameters to check/set for this association.
This can be a list of dictionaries.
Returns
-------
match : bool
`True` if the all constraints are satisfied.
reprocess : list of `~jwst.associations.lib.process_list.ProcessList`
List of items to process again.
"""
self.constraints.preserve()
match, reprocess = self.constraints.check_and_set(item)
if not match:
self.constraints.restore()
# Set the association type for all reprocessed items.
for process_list in reprocess:
process_list.trigger_rules.update([type(self)])
if process_list.rules is None:
process_list.rules = [type(self)]
return match, reprocess
[docs]
def match_constraint(self, item, conditions):
"""
Match against constraints.
Parameters
----------
item : dict
The item to retrieve the values from.
conditions : dict
The conditions structure.
Returns
-------
match : bool
`True` if the all constraints are satisfied.
reprocess : list of `~jwst.associations.lib.process_list.ProcessList`
List of items to process again.
"""
reprocess = []
evaled_str = conditions["inputs"](item)
if conditions["value"] is not None:
if not meets_conditions(evaled_str, conditions["value"]):
return False, reprocess
# At this point, the constraint has passed.
# Fix the conditions.
escaped_value = re.escape(evaled_str)
conditions["found_values"].add(escaped_value)
if conditions["value"] is None or conditions.get("force_unique", self.DEFAULT_FORCE_UNIQUE):
conditions["value"] = escaped_value
conditions["force_unique"] = False
return True, reprocess
[docs]
def finalize(self):
"""
Finalize association.
Finalize or close off this association. Perform validations,
modifications, etc. to ensure that the association is
complete.
Returns
-------
associations : list of `~jwst.associations.association.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]
else:
return None
[docs]
def is_item_member(self, item):
"""
Check if item is already a member of this association.
Parameters
----------
item : dict
The item to check.
Returns
-------
is_item_member : bool
`True` if item is a member.
"""
raise NotImplementedError(
"Association.is_item_member must be implemented by a specific association rule."
)
def _init_hook(self, item):
"""Post-check and pre-item-adding initialization."""
pass
def _add(self, item):
"""Add an item, association-specific."""
raise NotImplementedError(
"Association._add must be implemented by a specific association rule."
)
def _add_items(self, items, **kwargs): # noqa: ARG002
"""
Force adding items to the association.
Parameters
----------
items : list
A list of items to make members of the association.
**kwargs
Added to signature for potentional compatibility elsewhere
but not used in base implementation.
Notes
-----
This is a low-level shortcut into adding members, such as file names,
to an association. All defined shortcuts and other initializations are
bypassed, resulting in a potentially unusable association.
"""
try:
self["members"].update(items)
except KeyError:
self["members"] = items
# Methods required for implementing MutableMapping
def __getitem__(self, key):
return self.data[self.__keytransform__(key)]
def __setitem__(self, key, value):
self.data[self.__keytransform__(key)] = value
def __delitem__(self, key):
del self.data[self.__keytransform__(key)]
def __iter__(self):
return iter(self.data)
def __len__(self):
return len(self.data)
def __keytransform__(self, key):
return key
[docs]
def keys(self):
"""
Provide keys of data dictionary.
Returns
-------
dict_keys : iter
The keys of the data dictionary.
"""
return self.data.keys()
[docs]
def items(self):
"""
Provide items of data dictionary.
Returns
-------
dict_items : iter
The items of the data dictionary.
"""
return self.data.items()
[docs]
def values(self):
"""
Provide values of data dictionary.
Returns
-------
dict_values : iter
The values of the data dictionary.
"""
return self.data.values()
# Utilities
def finalize(asns):
"""
Finalize associations by calling their ``finalize_hook`` method.
Parameters
----------
asns : list of `~jwst.associations.association.Association`
The list of associations to be finalized.
Returns
-------
list of `~jwst.associations.association.Association`
The finalized list of associations.
Notes
-----
This is a functioning example of a finalize callback, and can be used
as the generic callback. Suggested usage is as follows:
.. code-block:: python
from jwst.associations.association import finalize as generic_finalize
from jwst.associations.registry import RegistryMarker
RegistryMarker.callback("finalize")(generic_finalize)
"""
finalized_asns = list(filter(lambda asn: asn is not None, [asn.finalize() for asn in asns]))
return finalized_asns
def make_timestamp():
"""
Timestamp of current time in UTC.
Returns
-------
timestamp : str
UTC time in pre-determined format.
"""
timestamp = datetime.now(UTC).strftime(_TIMESTAMP_TEMPLATE)
return timestamp
format_product = FormatTemplate()
"""Default product name filling."""