Source code for jwst.model_blender.blender

from astropy.io import fits
from stdatamodels import fits_support

from .rules import make_blender
from ._schemautil import parse_schema
from ._tablebuilder import TableBuilder, table_to_schema


__all__ = ["ModelBlender"]


[docs] class ModelBlender: """ Class to "blend" metadata from several datamodels. The output "blended" model will contain: - metadata for a combined model - a table with metadata of each datamodel Input models can be added to the blender using `ModelBlender.accumulate` and the output/combined model updated using `ModelBlender.finalize_model`. All input/accumulated models must be of the same type. >>> blender = ModelBlender() >>> blender.accumulate(input_model_a) # doctest: +SKIP >>> blender.accumulate(input_model_b) # doctest: +SKIP >>> blender.finalize_model(combined_model) # doctest: +SKIP """ def __init__(self, blend_ignore_attrs=None): """ Create a new `ModelBlender`. Parameters ---------- blend_ignore_attrs : list or None A list of metadata attributes to ignore during blending. These attributes will not be set on the output/combined. These attributes must be strings containing the dotted path of each attribute (for example "meta.filename"). (Note that "meta.wcs" will always be ignored). """ self._model_type = None self._first_header_meta = None self._blenders = None self._table_builder = None self._blend_ignore_attrs = ["meta.wcs"] if blend_ignore_attrs is not None: self._blend_ignore_attrs.extend(blend_ignore_attrs)
[docs] def accumulate(self, model): """ Blend metadata for model. Process model adding its metadata to the blended metadata and the metadata table. Parameters ---------- model : `jwst.datamodels.JwstDataModel` The datamodel to blend. """ if self._first_header_meta is None: self._model_type = type(model) # search the schema for other metadata to "blend" and to add to the table attr_to_columns, attr_to_blend_rules, schema_ignores = parse_schema(model.schema) # update ignores list for items in schema that can't be blended self._blend_ignore_attrs.extend(schema_ignores) # capture the entire contents of the first model metadata self._first_header_meta = {} for attr, v in model.to_flat_dict(include_arrays=False).items(): if not attr.startswith("meta"): continue if any(attr.startswith(i) for i in self._blend_ignore_attrs): continue self._first_header_meta[attr] = v # make "blenders" for the metadata with special rules self._blenders = {} for attr, rule in attr_to_blend_rules.items(): if rule == "first": continue if any(attr.startswith(i) for i in self._blend_ignore_attrs): continue self._blenders[attr] = make_blender(rule) # make a table builder using the mapping from the schema self._table_builder = TableBuilder(attr_to_columns) else: if type(model) != self._model_type: # noqa: E721 raise ValueError( f"model of type {type(model)} " f"does not match previous type({self._model_type}). " "ModelBlender only supports blending models of the same model type." ) # convert the model to a flat header header = model.to_flat_dict(include_arrays=False) # add the header to the table self._table_builder.header_to_row(header) # and perform any special blending for attr, blender in self._blenders.items(): if attr in header: blender.accumulate(header[attr])
def _finalize_metadata(self): # start with the entire contents of the first model meta = self._first_header_meta.copy() for attr, blender in self._blenders.items(): meta[attr] = blender.finalize() return meta def _finalize_table(self): return self._table_builder.build_table()
[docs] def finalize_model(self, model): """ Update model with the blend results. Add blended metadata and the accumulated metadata table to the provided datamodel. The update process involves: - setting the model metadata to the blended metadata values - adding an "hdrtab" attribute (containing the metadata table) - updating the model schema to save "hdrtab" The provided model will be updated in-place. Parameters ---------- model : `jwst.datamodels.JwstDataModel` A datamodel that will have its metadata set to the blended metadata and have the metadata table assigned to the "hdrtab" attribute. """ # update metadata of the output model based on the results # of the "blenders" for attr, val in self._finalize_metadata().items(): try: model[attr] = val except KeyError: # Ignore keys that are in the asdf tree but not in the schema pass # patch the table into the output model and the schema table = self._finalize_table() schema = table_to_schema(table) model.add_schema_entry("hdrtab", schema) # because astropy will silently mangle boolean columns on write # we roundtrip the data through a BinTableHDU here to allow # stdatamodels to correct the data in a way that won't result in mangling model.hdrtab = fits_support.from_fits_hdu(fits.BinTableHDU.from_columns(table), schema)