Model Versions and Migrations#

Since models may evolve in time, it is important to be able to keep track of changes and to maintain backward compatibility when possible. ctao-datamodel provides an API for this purpose: the VersionedModel. Here is an example:

from astropy import units as u
from astropy.io import fits
from astropydantic import AstroPydanticTime
from ctao_datamodel import (
    AstroField,
    Migration,
    ModelBase,
    Quantity,
    VersionedModel,
    fits_header_to_instance,
    instance_to_fits_header,
    migration,
    models,
)

The original revision#

Imagine we had an old model that looked like this:

class TimeCov(ModelBase):
    """A sub-model example"""
    reference: TimeReference | None = AstroField(
        description=doc(TimeReference), default=None
    )
    t_start: float | AstroPydanticTime = AstroField(
        "Start of time range of the data as either an ISO string or a float in the"
        " system defined by the time reference.",
        fits_keyword="TSTART",
    )
    t_stop: float | AstroPydanticTime = AstroField(
        "End of time range of the data, as either an ISO string or a float in the"
        " system defined by the time reference.",
        fits_keyword="TSTOP",
    )


class Target(VersionedModel):
    """A simple example"""

    object_name: str = AstroField("target name", fits_keyword="OBJECT")
    ra: Quantity["deg"] = AstroField(
        "Right ascension", fits_keyword="RA2000", fits_column_dtype="float64", ucd="pos.eq.ra"
    )
    dec: Quantity["deg"] = AstroField(
        "Declination", fits_column_dtype="float64", fits_keyword="DE2000", ucd="pos.eq.dec"
    )

    coverage: TimeCov

But in the new version, we want to change some things:

rename the field object_name to name
rename the sub-field coverage.t_start to coverage.t_min, and similar for coverage.t_stop to coverage.t_max
change the FITS serialization keys of RA2000/DE2000 to RA/DEC
drop support for t_min and t_max in an arbitrary time system, and allow only ISO strings
make the time reference info deprecated

And we want the new class to be backward compatible.

The new version#

To achieve this, we created the new model version

class TimeCov(ModelBase):
    """A sub-model example"""

    reference: models.common.TimeReference | None = AstroField(
        "DEPRECATED, do not use", default=None
    )

    t_min: AstroPydanticTime = AstroField(
        "Start of time range of the data as either an ISO string or a float.",
        fits_keyword="TSTART",
    )
    t_max: AstroPydanticTime = AstroField(
        "End of time range of the data, as either an ISO string or a float.",
        fits_keyword="TSTOP",
    )


class Target(ModelBase):
    """A simple example"""

    name: str = AstroField("target name", fits_keyword="OBJECT")
    ra: Quantity[u.deg] = AstroField(
        "Right ascension",
        fits_keyword="RA",
        fits_column_dtype="float64",
        ucd="pos.eq.ra",
    )
    dec: Quantity[u.deg] = AstroField(
        "Declination", fits_column_dtype="float64", fits_keyword="DEC", ucd="pos.eq.dec"
    )

    coverage: TimeCov

Let’s set up a test case that was serialized with the old version into a FITS header:

old_dict = dict(
    object_name="Crab",
    ra=83.624 * u.deg,
    dec=22.0174 * u.deg,
    coverage=dict(
        t_start=317788715.75,
        t_stop=317788715.85,
        reference={
            "position": "TOPOCENTER",
            "time_mjd": {"value": 58119.00042824074, "unit": "d"},
            "unit": "s",
            "system": "TAI",
        },
    ),
)
old_dict

{'object_name': 'Crab',
 'ra': <Quantity 83.624 deg>,
 'dec': <Quantity 22.0174 deg>,
 'coverage': {'t_start': 317788715.75,
  't_stop': 317788715.85,
  'reference': {'position': 'TOPOCENTER',
   'time_mjd': {'value': 58119.00042824074, 'unit': 'd'},
   'unit': 's',
   'system': 'TAI'}}}

if we try to deserialize this, it fails:

try:
    Target.model_validate(old_dict)
except ValueError as err:
    print(err)

6 validation errors for Target
name
  Field required [type=missing, input_value={'object_name': 'Crab', '... 's', 'system': 'TAI'}}}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.13/v/missing
coverage.t_min
  Field required [type=missing, input_value={'t_start': 317788715.75,...: 's', 'system': 'TAI'}}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.13/v/missing
coverage.t_max
  Field required [type=missing, input_value={'t_start': 317788715.75,...: 's', 'system': 'TAI'}}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.13/v/missing
coverage.t_start
  Extra inputs are not permitted [type=extra_forbidden, input_value=317788715.75, input_type=float]
    For further information visit https://errors.pydantic.dev/2.13/v/extra_forbidden
coverage.t_stop
  Extra inputs are not permitted [type=extra_forbidden, input_value=317788715.85, input_type=float]
    For further information visit https://errors.pydantic.dev/2.13/v/extra_forbidden
object_name
  Extra inputs are not permitted [type=extra_forbidden, input_value='Crab', input_type=str]
    For further information visit https://errors.pydantic.dev/2.13/v/extra_forbidden

Adding Migration Information#

To fix this, we re-define our model and make it a subclass of VersionedModel, and we need to add the steps that change the old version into the new version by using the Migration class, which provides migrations for standard changes like renaming or dropping columns, changing metadata, and more complex user-defined transformations.

To use this, we add a function for each version using the migration() decorator.

Note however the step of disallowing times in MET with a time reference would break the conversion, so for that we need to define a more complex transformation:

def transform_met_time_to_iso(obj: dict, key: str, reference_key: str = "reference"):
    """Convert MET time assuming there is a TimeReference Object."""

    from astropy.time import Time

    # check if it's already ISO:
    if isinstance(obj.get(key, None), str):
        return

    reference = models.common.TimeReference.model_validate(obj.get(reference_key))
    ref_time = Time(reference.time_mjd, format="mjd", scale=reference.system.lower())
    new_time = Time(ref_time + (obj[key] * reference.unit))
    obj[key] = new_time.iso
    return obj

class Target(VersionedModel):
    """A simple example"""

    name: str = AstroField("target name", fits_keyword="OBJECT")
    ra: Quantity["deg"] = AstroField(
        "Right ascension",
        fits_keyword="RA",
        fits_column_dtype="float64",
        ucd="pos.eq.ra",
    )
    dec: Quantity["deg"] = AstroField(
        "Declination", fits_column_dtype="float64", fits_keyword="DEC", ucd="pos.eq.dec"
    )

    coverage: TimeCov

    @migration("1.0.0", "2.0.0")
    def _(m: Migration):
        m.rename("object_name", "name")
        m.rename("coverage.t_start", "coverage.t_min")
        m.rename("coverage.t_stop", "coverage.t_max")
        m.update_metadata("ra", "fits_keyword", "RA2000", "RA")
        m.update_metadata("dec", "fits_keyword", "DEC2000", "DEC")
        m.apply_transform(
            "coverage.t_min", transform_met_time_to_iso, "Convert MET to ISO"
        )
        m.apply_transform(
            "coverage.t_max", transform_met_time_to_iso, "Convert MET to ISO"
        )
        m.drop("coverage.reference")

Target.print_history()

Target
======
* 1.0.0 -> 2.0.0
    - Renamed 'object_name' to 'name'
    - Renamed 'coverage.t_start' to 'coverage.t_min'
    - Renamed 'coverage.t_stop' to 'coverage.t_max'
    - Updated 'fits_keyword' metadata for 'ra' from 'RA2000' to 'RA'
    - Updated 'fits_keyword' metadata for 'dec' from 'DEC2000' to 'DEC'
    - Transformed 'coverage.t_min' using transform_met_time_to_iso (Convert MET to ISO)
    - Transformed 'coverage.t_max' using transform_met_time_to_iso (Convert MET to ISO)
    - Dropped 'coverage.reference'

Now, we use VersionedModel.model_validate_versioned to do the migration:

try:
    validated = Target.model_validate_versioned(old_dict)
except ValueError as err:
    print(err)

validated

Target(name='Crab', ra=<Quantity 83.624 deg>, dec=<Quantity 22.0174 deg>, coverage=TimeCov(reference=None, t_min=<Time object: scale='utc' format='iso' value=2028-01-27 02:39:12.750>, t_max=<Time object: scale='utc' format='iso' value=2028-01-27 02:39:12.850>))

print(validated.model_dump_json(indent=2))

{
  "name": "Crab",
  "ra": {
    "value": 83.624,
    "unit": "deg"
  },
  "dec": {
    "value": 22.0174,
    "unit": "deg"
  },
  "coverage": {
    "reference": null,
    "t_min": "2028-01-27T02:39:12.750000000",
    "t_max": "2028-01-27T02:39:12.850000000"
  }
}

It works! The old format was read and transformed to the the new one.

FITS Headers#

The migration should work for FITS header representation as well, since we used the Migration.update_metadata() migration:

old_header = fits.Header(
    dict(
        OBJECT="Crab",
        RA2000=83.624,
        DEC2000=22.0174,
        TREFPOS="TOPOCENTER",
        MJDREF=58119.00042824074,
        TIMEUNIT="s       ",
        TIMESYS="TAI     ",
        TSTART=317788715.75,
        TSTOP=317788715.85,
    )
)

validated_header = fits_header_to_instance(old_header, Target)
validated_header

Target(name='Crab', ra=<Quantity 83.624 deg>, dec=<Quantity 22.0174 deg>, coverage=TimeCov(reference=None, t_min=<Time object: scale='utc' format='iso' value=2028-01-27 02:39:12.750>, t_max=<Time object: scale='utc' format='iso' value=2028-01-27 02:39:12.850>))

instance_to_fits_header(validated_header)

WARNING: VerifyWarning: Card is too long, comment will be truncated. [astropy.io.fits.card]

OBJECT  = 'Crab    '           / target name                                    
RA      =               83.624 / [deg] Right ascension                          
DEC     =              22.0174 / [deg] Declination                              
TSTART  = '2028-01-27T02:39:12.750000000' / Start of time range of the data as e
TSTOP   = '2028-01-27T02:39:12.850000000' / End of time range of the data, as ei

And you can see that even the FITS header was converted to the new format.

Model Versions and Migrations#

The original revision#

The new version#

Adding Migration Information#

FITS Headers#

This Page