"""
DataProducts.Reference metadata model.
The primary class is `Product`, which defines the common metadata for a data
product.
"""
import uuid
from enum import StrEnum, auto, nonmember
from typing import ClassVar, Self
from astropy.time import Time
from astropydantic import AstroPydanticTime
from pydantic import AnyUrl, EmailStr, model_validator
from ..._core import AstroField, ModelBase, doc, enum_doc
from ..._versioning import Migration, VersionedModel, migration
from ..common import TimeReference
from . import ident
from .acquisition import Acquisition
from .observation import Observation
from .sdc import ScienceDataChallenge
from .simulation import Simulation
PRODUCT_MODEL_VERSION = "2.0.0"
_PRODUCT_NAMESPACE = "CTAO.DataProducts"
_NAMESPACE = "CTAO.DataProducts.Reference"
EXAMPLE_TIMES = ["2023-10-15 05:45:12.21", "2023-10-15T05:45:12"]
__all__ = [
"Product",
"Contact",
"DataModelName",
"DataModel",
"Software",
"ObservatoryProcess",
"Activity",
"ExternalDataProduct",
"Curation",
"PRODUCT_MODEL_VERSION",
]
class Contact(ModelBase):
"""Contact information for this data product."""
_namespace: ClassVar[str] = _NAMESPACE
name: str = AstroField(
description="Contact name for this data product.", fits_keyword="AUTHOR"
)
organization: str = AstroField(
description="Contact organization name of this data product.",
fits_keyword="ORIGIN", # fits v4
)
email: EmailStr = AstroField(
description="Contact's email address",
fits_keyword="EMAIL", # nonstandard
ucd="meta.curation;meta.email",
)
class DataModelName(StrEnum):
"""Allowed data models/format specifications."""
_namespace = nonmember(_NAMESPACE)
GADF = auto() #: GADF
VODF = auto() #: VODF
CTAO = auto() #: CTAO internal model
class DataModel(ModelBase):
"""Describes the data model and/or format of the data product."""
_namespace: ClassVar[str] = _NAMESPACE
name: DataModelName | str = AstroField(
description=(
"Name of the overall data model, which may contain multiple data product"
" specifications. For CTAO models/formats, this should be the fully"
" qualified name of the data model as defined by the top-level model."
),
fits_keyword="MODEL",
)
version: str = AstroField(
description="Version number of the data model",
fits_keyword="MODELVER",
)
url: AnyUrl | None = AstroField(
description="URL or DOI linking to more detail.",
fits_keyword="MODELURL",
)
class Software(ModelBase):
"""Describes the software package that created the data product."""
_namespace: ClassVar[str] = _NAMESPACE
name: str = AstroField(
description="Descriptive name of the software.",
examples=["datapipe"],
fits_keyword="SOFTWARE",
ucd="meta.id;meta.software",
)
version: str = AstroField(
description="Version number",
examples=["v1.0.1"],
fits_keyword="SOFTVER",
ucd="meta.version;meta.software",
)
url: AnyUrl | None = AstroField(
description="URL or DOI linking to more detail.",
fits_keyword="SOFTURL",
ucd="meta.ref.url;meta.software",
)
class ObservatoryProcess(StrEnum):
"""Observatory operational Process.
These are the top-level processes
"""
_namespace = nonmember(_NAMESPACE)
PROPOSAL_HANDLING = auto() #: Proposal handling
OBSERVATION_PLANNING = auto() #: Observation planning
ACQUISITION = auto() #: Data Acquistion
SIMULATION = auto() #: Simulation
DATA_PROCESSING = auto() #: Data Processing
SCIENCE_DATA_CHALLENGE = auto() #: Science Data Challenge
SCIENCE_ANALYSIS = auto() #: High-level Science Analysis (DL4-DL6)
INSTRUMENT = auto() #: Instrumental engineering or maintenance related activity
USER = auto() #: Process was not an observatory activity
class ExternalDataProduct(ModelBase):
"""Another data product used as input to the Activity."""
_namespace: ClassVar[str] = _NAMESPACE
uri: AnyUrl = AstroField(
description=(
"URI of the data product, can be a https address, DOI, "
"Bulk Archive LFN, or a file:// if only known locally."
),
examples=[
"https://doi.org/10.5281/zenodo.6218687",
"file://myfile.fits.gz",
"lfn://ctao/dpps/",
],
ucd="meta.ref.uri;meta.file",
)
role: str = AstroField(
description="context of the data product",
examples=["reconstruction-model", "configuration", "camera-calibration"],
)
id: uuid.UUID = AstroField(
default=None,
description="Unique ``instance.id`` of the data product, if known.",
ucd="meta.id;meta.file",
)
class Activity(ModelBase):
"""The activity (software or human) that produced this data product."""
_namespace: ClassVar[str] = _NAMESPACE
process: ObservatoryProcess | None = AstroField(
doc(ObservatoryProcess),
default=ObservatoryProcess.USER,
fits_keyword="ACTPROC",
)
name: str = AstroField(
description="Name of the activity that produced this data product.",
examples=["observation data processing", "lab calibration"],
fits_keyword="ACTIVITY",
)
description: str | None = AstroField(
description=(
"Human-readable details of the activity, if more specificity than the "
"`name` is required."
),
default=None,
fits_keyword="ACTDESC",
)
id: uuid.UUID = AstroField(
description=(
"Unique identifier of this process, which is used to link multiple "
"data products produced together. If this is a human process, this can "
"be generated with the `uuidgen` linux command for example."
),
fits_keyword="ACTID",
ucd="meta.id.parent",
)
start: AstroPydanticTime = AstroField(
description="Start time of the activity", fits_keyword="ACTSTART"
)
end: AstroPydanticTime | None = AstroField(
description=(
"End time of the activity, which is generally not known when the data"
" product is being written and is therefore optional. "
),
default=None,
fits_keyword="ACTSTOP",
)
software: Software | None = AstroField(
description="if the activity involved software, describes it.",
default=None,
)
configuration_id: str = AstroField(
description=(
"Identifier for the configuration for the software used to produce this"
" data product. This is normally a unique identifier of the configuration"
" used to produce this data product, It should be possible for a user"
" knowing this identifier and details of how and where configurations are"
" stored, to find the full original configuration. It is recommended to use"
" a URI when possible, particularly for REST API endpoints or git"
" repositories."
),
fits_keyword="ACTCONF",
examples=[
"dl3/hillas-standard/v1",
"dl5/quicklook-followup/v1",
"https://myobservatory.org/API/v1/config/data-processing/standard-v1.2",
],
)
inputs: list[ExternalDataProduct] | None = AstroField(
default=None,
description=(
"Describes the input data products to the activity that produced this data"
" product."
),
)
class DataRights(StrEnum):
"""Availability of the data product, if known at time of creation."""
_namespace = nonmember(_NAMESPACE)
PUBLIC = auto()
SECURE = auto()
PROPRIETARY = auto()
class Curation(ModelBase):
"""Information related to the release and usage of data products."""
_namespace: ClassVar[str] = _NAMESPACE
release: str | None = AstroField(
description=(
"Name of the data release (data collection) that this data product"
" belongs to."
),
default=None,
fits_keyword="RELEASE",
ivoa_keyword="obs_collection",
examples=["CTAO/DL3-DR1", "CTAO/DL3-SDC1"],
ucd="meta.dataset;meta.curation",
)
reference: AnyUrl | None = AstroField(
description=(
"a reference to where the data product is published. It is recommended that"
" either the 19-digit bibliographic identifier used in the Astrophysics"
" Data System bibliographic databases (http://adswww.harvard.edu/) or the"
" Digital Object Identifier (http://doi.org) be included in the value"
" string, when available."
),
default=None,
examples=["1994AAS..103..135A", "doi:10.1006/jmbi.1998.2354"],
fits_keyword="REFERENC",
ucd="meta.ref;meta.curation",
)
license: str = AstroField(
description="License for this data product",
default="CC-BY-SA-4.0",
examples=["CC-BY-SA-4.0"],
fits_keyword="LICENSE",
)
license_url: str = AstroField(
description=(
"URL pointing to the details of the license described by ``license``"
),
default="https://creativecommons.org/licenses/by-sa/4.0/",
examples=["https://creativecommons.org/licenses/by-sa/4.0/"],
fits_keyword="LICENURL",
)
copyright: str | None = AstroField(
description="Copyright holder(s) of this data product",
examples=["CTAO 2025"],
default="CTAO 2025",
fits_keyword="COPYRIGH",
)
rights: DataRights | None = AstroField(
enum_doc(DataRights),
default=None,
ivoa_keyword="data_rights",
fits_keyword="RIGHTS",
)
release_date: AstroPydanticTime | None = AstroField(
"Date of the public release of this data product, if different from the"
" creation_time, e.g. if there is a proprietary period this may be set to the"
" date when the data product is no longer private.",
default=None,
fits_keyword="PUBDATE",
ivoa_keyword="obs_release_date",
ucd="time.release",
)
valid_from: AstroPydanticTime | None = AstroField(
description=(
"Starting UTC date-time where the contents of this data product are valid,"
" if known. This is meant to be able to time-order data products by"
" scientific validity, rather than creation_date. "
),
default=None,
examples=EXAMPLE_TIMES,
)
valid_to: AstroPydanticTime | None = AstroField(
description=(
"Ending UTC date-time where the contents of this data product are valid,"
" if known. This is meant to be able to time-order data products by"
" scientific validity, rather than creation_date. "
),
default=None,
examples=EXAMPLE_TIMES,
)
def transform_met_time_to_iso(obj: dict, key: str, reference_key: str = "reference"):
"""
Convert MET time assuming there is a TimeReference Object.
In older DataProducts model, the time coverage was allowed to be in MET or
ISO format. Here we treat data using the old format by converting the
TimeReference and MET values to ISO strings.
"""
# check if it's already ISO:
if isinstance(obj.get(key, None), str):
return
reference = 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 Product(VersionedModel):
"""Common metadata for all CTAO data products."""
_namespace: ClassVar[str] = _PRODUCT_NAMESPACE
ctao_metadata_version: str = AstroField(
description="CTAO DataProducts Metadata Version",
default=PRODUCT_MODEL_VERSION,
fits_keyword="CTAOMETA",
)
description: str = AstroField(
description="Human-readable description of this data product",
ivoa_keyword="obs_title",
fits_keyword="TITLE",
)
data: ident.ProductType = AstroField(doc(ident.ProductType))
instance: ident.InstanceIdentifier = AstroField(doc(ident.InstanceIdentifier))
curation: Curation = AstroField(doc(Curation))
model: DataModel = AstroField(doc(DataModel))
disclaimer: str | None = AstroField(
description=(
"Text describing e.g. the allowed usage of this data product, if specifics"
" are required."
),
default=None,
fits_keyword="COMMENT",
)
# note `DATE` (FITS v4, p32) is HDU creation date, IVOA recommends CREATED
# for the product
creation_time: AstroPydanticTime = AstroField(
description="UTC Date-time the data product was created",
examples=EXAMPLE_TIMES,
fits_keyword="CREATED",
ivoa_keyword="obs_creation_date",
ucd="time.creation",
)
contact: Contact = AstroField(description=doc(Contact))
# provenance
activity: Activity | None = AstroField(doc(Activity), default=None)
observation: Observation | None = AstroField(doc(Observation), default=None)
acquisition: Acquisition | None = AstroField(doc(Acquisition), default=None)
sdc: ScienceDataChallenge | None = AstroField(
doc(ScienceDataChallenge), default=None
)
simulation: Simulation | None = AstroField(doc(Simulation), default=None)
@model_validator(mode="after")
def _check_simulation(self) -> Self:
"""Check some fields to be consistent with simulated data."""
# Check that if this has SDC information, that we use the right facility name.
if self.sdc:
if self.instance.facility_name != ident.FacilityName.SIMULATED_CTAO:
raise ValueError(
"This data product is simulated (has SDC context info), so the"
" instance.facility_name should be set to SIMULATED_CTAO."
)
# check that the simulation grid info is consistent
if self.simulation and self.simulation.node:
if self.observation and self.observation.location:
location = self.observation.location.to_earthlocation()
self.simulation.node.validate_coordinates(location)
elif self.instance.site_id:
pass # TODO: implement checks with standard locations
else:
raise ValueError(
"Simulation node info was specified, but no observatory location or"
" name was given. Please set either `observation.location` or"
" `instance.site_id` or both, so that the node info can be"
" validated."
)
return self
[docs]
@migration("1.0.0", "1.1.0")
def v100_110(m: Migration):
"""Migrate version."""
m.update_metadata(
"activity.configuration_id", "fits_keyword", "ANAMODE", "ACTCONF"
)
[docs]
@migration("1.1.0", "2.0.0")
def v110_200(m: Migration):
"""Migrate version."""
m.apply_transform(
"observation.coverage.time.t_min", transform_met_time_to_iso, "MET to ISO"
)
m.apply_transform(
"observation.coverage.time.t_max", transform_met_time_to_iso, "MET to ISO"
)
m.drop("observation.coverage.time.reference")
m.add("simulation")
m.add("instance.messenger")
