Add support for pre-translation header correction

Corrections are applied by locating YAML files named after
the instrument/observation_id.

python/astro_metadata_translator/headers.py

@@ -11,17 +11,22 @@
 
 """Code to support header manipulation operations."""
 
-__all__ = ("merge_headers",)
+__all__ = ("merge_headers", "fix_header")
 
 import logging
 import itertools
 import copy
+import os
+import yaml
 
 from .translator import MetadataTranslator
 from .translators import FitsTranslator
 
 log = logging.getLogger(__name__)
 
+ENV_VAR_NAME = "METADATA_CORRECTIONS_PATH"
+"""Name of environment variable containing search path for header fix up."""
+
 
 def merge_headers(headers, mode="overwrite", sort=False, first=None, last=None):
     """Merge multiple headers into a single dict.
@@ -175,3 +180,114 @@ def retain_value(to_receive, to_retain, sources):
         retain_value(merged, last, tuple(reversed(all_headers)))
 
     return merged
+
+
+def fix_header(header, search_path=None, translator_class=None, filename=None):
+    """Update, in place, the supplied header with known corrections.
+
+    Parameters
+    ----------
+    header : `dict`-like
+        Header to correct.
+    search_path : `list`, optional
+        Explicit directory paths to search for correction files.
+    translator_class : `MetadataTranslator`-class, optional
+        If not `None`, the class to use to translate the supplied headers
+        into standard form. Otherwise each registered translator class will
+        be asked in turn if it knows how to translate the supplied header.
+    filename : `str`, optional
+        Name of the file whose header is being translated.  For some
+        datasets with missing header information this can sometimes
+        allow for some fixups in translations.
+
+    Returns
+    -------
+    fixed : `bool`
+        `True` if the header was updated.
+
+    Raises
+    ------
+    ValueError
+        Raised if the supplied header is not understood by any registered
+        translation classes.
+    TypeError
+        Raised if the supplied translation class is not a `MetadataTranslator`.
+
+    Notes
+    -----
+    In order to determine that a header update is required it is
+    necessary for the header to be handled by the supplied translator
+    class or else support automatic translation class determination.
+    It is also required that the ``observation_id`` and ``instrument``
+    be calculable prior to header fix up.
+
+    Correction files use names of the form ``instrument_obsid.yaml``.
+    The YAML file should have the format of:
+
+    .. code-block:: yaml
+
+        EXPTIME: 30.0
+        IMGTYPE: bias
+
+    where each key/value pair is copied directly into the supplied header,
+    overwriting any previous values.
+
+    This function searches a number of locations for such a correction file.
+    The search order is:
+
+    - Any paths explicitly supplied through ``search_path``.
+    - The contents of the PATH-like environment variable
+      ``$METADATA_CORRECTIONS_PATH``.
+    - Any search paths supplied by the matching translator class.
+
+    The first file located in the search path is used for the correction.
+    """
+
+    # PropertyList is not dict-like so force to a dict here to allow the
+    # translation classes to work. We update the original header though.
+    if hasattr(header, "toOrderedDict"):
+        header_to_translate = header.toOrderedDict()
+    else:
+        header_to_translate = header
+
+    if translator_class is None:
+        translator_class = MetadataTranslator.determine_translator(header_to_translate, filename=filename)
+    elif not issubclass(translator_class, MetadataTranslator):
+        raise TypeError(f"Translator class must be a MetadataTranslator, not {translator_class}")
+
+    # Create an instance for this header
+    translator = translator_class(header, filename=filename)
+
+    # To determine the file look up we need the observation_id and instrument
+    try:
+        obsid = translator.to_observation_id()
+        instrument = translator.to_instrument()
+    except Exception:
+        # Return without comment if these translations failed
+        return False
+
+    target_file = f"{instrument}-{obsid}.yaml"
+
+    # Work out the search path
+    paths = []
+    if search_path is not None:
+        paths.extend(search_path)
+    if ENV_VAR_NAME in os.environ and os.environ[ENV_VAR_NAME]:
+        paths.extend(os.environ[ENV_VAR_NAME].split(os.path.pathsep))
+
+    paths.extend(translator.search_paths())
+
+    for p in paths:
+        correction_file = os.path.join(p, target_file)
+        if os.path.exists(correction_file):
+            with open(correction_file) as fh:
+                log.debug("Applying header corrections from file %s", correction_file)
+                corrections = yaml.safe_load(fh)
+
+                # Apply corrections (PropertyList does not yet have update())
+                for k, v in corrections.items():
+                    header[k] = v
+
+                return True
+
+    return False

python/astro_metadata_translator/observationInfo.py

@@ -21,6 +21,7 @@
 
 from .translator import MetadataTranslator
 from .properties import PROPERTIES
+from .headers import fix_header
 
 log = logging.getLogger(__name__)
 
@@ -45,6 +46,8 @@ class ObservationInfo:
         If True the translation must succeed for all properties.  If False
         individual property translations must all be implemented but can fail
         and a warning will be issued.
+    search_path : iterable, optional
+        Override search paths to use during header fix up.
 
     Raises
     ------
@@ -58,13 +61,18 @@ class ObservationInfo:
     NotImplementedError
         Raised if the selected translator does not support a required
         property.
+
+    Notes
+    -----
+    Headers will be corrected if correction files are located.
     """
 
     _PROPERTIES = PROPERTIES
     """All the properties supported by this class with associated
     documentation."""
 
-    def __init__(self, header, filename=None, translator_class=None, pedantic=False):
+    def __init__(self, header, filename=None, translator_class=None, pedantic=False,
+                 search_path=None):
 
         # Store the supplied header for later stripping
         self._header = header
@@ -82,6 +90,10 @@ def __init__(self, header, filename=None, translator_class=None, pedantic=False)
         elif not issubclass(translator_class, MetadataTranslator):
             raise TypeError(f"Translator class must be a MetadataTranslator, not {translator_class}")
 
+        # Fix up the header (if required)
+        fix_header(header, translator_class=translator_class, filename=filename,
+                   search_path=search_path)
+
         # Create an instance for this header
         translator = translator_class(header, filename=filename)
 

python/astro_metadata_translator/translator.py

@@ -531,6 +531,18 @@ def is_keyword_defined(header, keyword):
 
         return True
 
+    def search_paths(self):
+        """Search paths to use when searching for header fix up correction
+        files.
+
+        Returns
+        -------
+        paths : `list`
+            Directory paths to search. Can be an empty list if no special
+            directories are defined.
+        """
+        return []
+
     def is_key_ok(self, keyword):
         """Return `True` if the value associated with the named keyword is
         present in this header and defined.

tests/data/SCUBA_test-20000101_00002.yaml

@@ -0,0 +1,2 @@
+TELCODE: AuxTel
+EXPID: 42

tests/test_translation.py

@@ -9,11 +9,14 @@
 # Use of this source code is governed by a 3-clause BSD-style
 # license that can be found in the LICENSE file.
 
+import os.path
 import unittest
 from astropy.time import Time
 
 from astro_metadata_translator import FitsTranslator, StubTranslator, ObservationInfo
 
+TESTDIR = os.path.abspath(os.path.dirname(__file__))
+
 
 class InstrumentTestTranslator(FitsTranslator, StubTranslator):
     """Simple FITS-like translator to test the infrastructure"""
@@ -132,6 +135,27 @@ def test_translator(self):
         summary = str(v1)
         self.assertIn("datetime_begin", summary)
 
+    def test_corrections(self):
+        """Apply corrections before translation."""
+        header = self.header
+
+        # Specify a translation class
+        with self.assertWarns(UserWarning):
+            # Since the translator is incomplete it should issue warnings
+            v1 = ObservationInfo(header, translator_class=InstrumentTestTranslator,
+                                 search_path=[os.path.join(TESTDIR, "data")])
+
+        # These values should match the expected translation
+        self.assertEqual(v1.instrument, "SCUBA_test")
+        self.assertEqual(v1.detector_name, "76")
+        self.assertEqual(v1.relative_humidity, 55.0)
+        self.assertIsInstance(v1.relative_humidity, float)
+        self.assertEqual(v1.physical_filter, "76_55")
+
+        # These two should be the "corrected" values
+        self.assertEqual(v1.telescope, "AuxTel")
+        self.assertEqual(v1.exposure_id, 42)
+
     def test_failures(self):
         header = {}