"""
Serialization of models to and from astropy Tables that can be serialized to FITS.
A :class:`~ctao_datamodel.ModelBase` subclass participates in table
serialization by annotating its fields with ``fits_column_dtype`` in
:func:`~ctao_datamodel.AstroField`. Fields *without* ``fits_column_dtype``
are silently ignored by these functions, so a model can still be used as both a
FITS header and a BINTABLE schema.
"""
from __future__ import annotations
import warnings
import numpy as np
from astropy import units as u
from astropy.table import Column, Table
from pydantic import BaseModel
from ._visitor import walk_model
__all__ = [
"model_to_astropy_table",
"model_validate_astropy_table",
"TableValidationError",
"ColumnValidationIssue",
]
from ._core import ValidationError, ValidationIssue
from ._visitor import get_field_metadata, get_field_unit
[docs]
class ColumnValidationIssue(ValidationIssue):
"""A single schema mismatch found during table validation."""
[docs]
class TableValidationError(ValidationError):
"""Raised when a table's schema does not match a model."""
def _column_fields(model: type[BaseModel]) -> dict[str, dict]:
"""Return a mapping of field_name → metadata for fields with fits_column_dtype.
Only fields that carry ``fits_column_dtype`` in their ``json_schema_extra``
are included. Fields without it are header/JSON-only and are ignored.
"""
result = {}
for _, path in walk_model(model):
if len(path) == 1:
continue
name = path[-1].item_name
field_info = path[-1].field
dtype = get_field_metadata(field_info, "fits_column_dtype")
if dtype is None:
continue
result[name] = {
"dtype": dtype,
"unit": get_field_unit(field_info),
"ucd": get_field_metadata(field_info, "ucd"),
"description": field_info.description,
"fits_keyword": get_field_metadata(field_info, "fits_keyword"),
"optional": path[-1].is_optional,
}
return result
def _column_name(field_name: str, meta: dict) -> str:
"""Return the FITS column name for a field.
Uses ``fits_keyword`` if present (matching the convention used in
``instance_to_fits_header``), otherwise falls back to the field name
uppercased.
"""
return meta.get("fits_keyword") or field_name.upper()
def _dtypes_compatible(declared: str, actual: np.dtype) -> bool:
"""Return True if *actual* dtype is compatible with the declared dtype string.
Compatibility means same kind (float, int, unsigned int, string, bool) and
same itemsize. Endianness differences (e.g. ``>f4`` vs ``float32``) are
intentionally ignored because FITS files always use big-endian storage but
astropy returns big-endian dtypes on read.
"""
try:
declared_dtype = np.dtype(declared)
except TypeError:
return False
# String types: FITS stores all strings as bytes (kind='S'). astropy writes
# unicode (kind='U', itemsize=4*N bytes for N chars) as bytes (kind='S',
# itemsize=N bytes) on FITS round-trip. Accept when the *character* width
# matches, i.e. declared U8 (itemsize=32) == loaded S8 (itemsize=8).
if declared_dtype.kind in ("U", "S"):
declared_char_width = (
declared_dtype.itemsize // 4
if declared_dtype.kind == "U"
else declared_dtype.itemsize
)
actual_char_width = (
actual.itemsize // 4 if actual.kind == "U" else actual.itemsize
)
return actual.kind in ("U", "S") and actual_char_width == declared_char_width
return (
actual.kind == declared_dtype.kind
and actual.itemsize == declared_dtype.itemsize
)
def _units_compatible(declared: str | None, actual) -> bool:
"""Return True if *actual* unit is compatible (physically equivalent) with *declared*.
A declared unit of ``None`` means the field is dimensionless; the column
unit must also be dimensionless or absent. Unit *equivalence* (e.g. TeV
and GeV are both energy) is accepted; strict equality is not required.
"""
if declared is None:
# No unit declared: accept dimensionless or missing unit
if actual is None:
return True
try:
actual_unit = u.Unit(str(actual))
return actual_unit.physical_type == "dimensionless"
except Exception:
return True # can't parse, don't fail on unit
declared_unit = u.Unit(declared)
if actual is None or str(actual).strip() in ("", "None"):
return False
try:
actual_unit = u.Unit(str(actual))
return declared_unit.is_equivalent(actual_unit)
except Exception:
return False
[docs]
def model_to_astropy_table(model: type[BaseModel], n_rows: int = 0) -> Table:
"""Create an :class:`astropy.table.Table` matching the schema of *model*.
Only fields annotated with ``fits_column_dtype`` in
:func:`~ctao_datamodel.AstroField` are included. The resulting
table has the correct column names, dtypes, units, UCDs, and descriptions
but no data rows (unless *n_rows* > 0, in which case columns are
zero-filled).
Parameters
----------
model : type[BaseModel]
A :class:`~ctao_datamodel.ModelBase` subclass with at least one
field carrying ``fits_column_dtype``.
n_rows : int, optional
Number of (zero-filled) rows to include. Default is 0, producing a
blank schema-only table.
Returns
-------
astropy.table.Table
Table with one column per ``fits_column_dtype`` field.
Raises
------
ValueError
If *model* has no fields with ``fits_column_dtype``.
Examples
--------
>>> table = model_to_table(EventList)
>>> table.write("events.fits", overwrite=True)
"""
col_fields = _column_fields(model)
if not col_fields:
msg = (
f"Model '{model.__name__}' has no fields with 'fits_column_dtype'. "
"Add fits_column_dtype=... to AstroField() for each column."
)
raise ValueError(msg)
columns = []
for field_name, meta in col_fields.items():
col_name = _column_name(field_name, meta)
col = Column(
name=col_name,
dtype=meta["dtype"],
length=n_rows,
unit=u.Unit(meta["unit"]) if meta["unit"] is not None else None,
description=meta["description"],
)
if meta["ucd"]:
col.meta["ucd"] = meta["ucd"]
columns.append(col)
return Table(columns)
[docs]
def model_validate_astropy_table(
table: Table,
model: type[BaseModel],
*,
strict_units: bool = True,
) -> None:
"""Validate that *table* schema matches the column fields of *model*.
Checks that every field with ``fits_column_dtype`` is present as a column,
has a compatible dtype, and (if ``strict_units`` is True) has a
physically-equivalent unit. Extra columns in the table are silently
ignored.
Parameters
----------
table : astropy.table.Table
Table to validate.
model : type[BaseModel]
Model whose ``fits_column_dtype`` fields define the expected schema.
strict_units : bool, optional
If True (default), unit mismatches are reported as errors. Set to
False to downgrade unit mismatches to warnings instead, e.g. when
loading legacy files with missing or non-standard units.
Raises
------
TableValidationError
If any required column is missing or has an incompatible dtype (or
unit, when ``strict_units=True``). All issues are collected before
raising so the full list is available in one call.
Examples
--------
>>> model_validate_table(Table.read("events.fits"), EventList)
"""
col_fields = _column_fields(model)
if not col_fields:
msg = (
f"Model '{model.__name__}' has no fields with 'fits_column_dtype'. "
"Nothing to validate."
)
raise ValueError(msg)
issues: list[ColumnValidationIssue] = []
unit_warnings: list[ColumnValidationIssue] = []
for field_name, meta in col_fields.items():
col_name = _column_name(field_name, meta)
# --- presence ---
if col_name not in table.colnames:
if not meta["optional"]:
issues.append(
ColumnValidationIssue(
column=col_name,
kind="missing",
message=(
f"Column '{col_name}' is required by model field"
f" '{field_name}' but not present in table."
),
)
)
continue # can't check dtype/unit if column absent
col = table[col_name]
# --- dtype ---
if not _dtypes_compatible(meta["dtype"], col.dtype):
issues.append(
ColumnValidationIssue(
column=col_name,
kind="dtype",
message=(
f"Expected dtype compatible with '{meta['dtype']}' "
f"(numpy kind='{np.dtype(meta['dtype']).kind}', "
f"itemsize={np.dtype(meta['dtype']).itemsize}), "
f"got '{col.dtype}' "
f"(kind='{col.dtype.kind}', itemsize={col.dtype.itemsize})."
),
)
)
# --- unit ---
col_unit = col.unit if hasattr(col, "unit") else None
if not _units_compatible(meta["unit"], col_unit):
issue = ColumnValidationIssue(
column=col_name,
kind="unit",
message=(
f"Expected unit equivalent to '{meta['unit']}', got '{col_unit}'."
),
)
if strict_units:
issues.append(issue)
else:
unit_warnings.append(issue)
for w in unit_warnings:
warnings.warn(str(w), stacklevel=2)
if issues:
raise TableValidationError(issues)
