API#

Example module of the python package ctao_datamodel.

AstroField(description: str | None = None, *, fits_keyword: str | None = None, ivoa_keyword: str | None = None, unit: str | None = None, ucd: str | None = None, fits_column_dtype: str | None = None, examples: list[str] | None = None, deprecated: str | None = None, **kwargs) Field[source]#

Return a Field with extra astronomy-related metadata.

The extra info is not for validation, but used for documentation or serialization. This is just a helper to avoid having to add a json_schema_extra dict manually.

Parameters:
descriptionstr, optional

Human-readable description of the field.

fits_keywordstr, optional

FITS header keyword (max 8 characters).

ivoa_keywordstr, optional

IVOA keyword name.

unitstr, optional

Physical unit string. Deprecated — use Quantity[unit] instead.

ucdstr, optional

IVOA Unified Content Descriptor.

fits_column_dtypestr, optional

NumPy/astropy dtype string for use when this field is serialized as a FITS BINTABLE column (e.g. "float32", "int64", "U16"). When present, the field participates in model_to_astropy_table and model_validate_astropy_table. When absent, the field is treated as a scalar header/JSON-only field. Accepted values are any dtype string that astropy.table.Column accepts, which covers all standard NumPy dtype strings plus fixed-width string types such as "U8".

exampleslist of str, optional

Example values shown in documentation.

**kwargs

Forwarded to pydantic.Field().

class ColumnValidationIssue(column: str, kind: str, message: str)[source]#

A single schema mismatch found during table validation.

class Migration[source]#

Abstract class for migrations.

abstractmethod add(path: str)[source]#

Add a new field or set of fields to the model.

abstractmethod apply_transform(path: str, func: Callable, description: str)[source]#

Apply a more complex user-defined transformation.

This can be used for migrations that require more than just a rename/drop of a field, e.g. merging multiple fields into one, or changing a field format.

abstractmethod drop(path: str)[source]#

Drop field.

abstractmethod rename(old: str, new: str)[source]#

Rename a field from old to new.

abstractmethod set_default(path: str, value)[source]#

Change a field’s default.

abstractmethod update_metadata(path: str, metadata_field: str, old: Any, new: Any)[source]#

Change a field’s metadata (e.g. fits_keyword) value.

pydantic model ModelBase[source]#

Base class for all CTAO models.

Should just set the common model_config here.

Show JSON schema
{
   "title": "ModelBase",
   "description": "Base class for all CTAO models.\n\nShould just set the common model_config here.",
   "type": "object",
   "properties": {},
   "additionalProperties": false
}

Config:
  • extra: str = forbid

  • arbitrary_types_allowed: bool = True

  • validate_default: bool = True

class ParentInfo(item_name: str, item_type: type, is_optional: bool, parent_relation: Relation, field: FieldInfo)[source]#

Info about parent of the current element.

class PlantUMLDiagram(model_or_text: str | type[BaseModel | Enum], relations: bool = True, inheritance: bool = False, details: bool = False, max_depth: int = 1000, cwd: Path | str | None = None)[source]#

Render a PlantUML diagram in a jupyter notebook.

Diagrams can be composed using the + operator.

classmethod from_path(path) Self[source]#

Return diagram from an existing file at path.

class QuantityFormat(*values)[source]#

Format used to serialize Quantities.

DICT = 'dict'#

dict with value and unit keys

FLOAT = 'float'#

float-representation, units removed

STRING = 'string'#

string representation

class Relation(*values)[source]#

Parent relationship.

class StrFlag(new_class_name, /, names, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]#

Like enum.Flag, but allowing string input, like A|B.

to_string() str[source]#

Turn the flag into a string representation.

exception TableValidationError(issues: list[ValidationIssue])[source]#

Raised when a table’s schema does not match a model.

exception ValidationError(issues: list[ValidationIssue])[source]#

Raised when a table’s schema does not match a model.

Attributes:
issueslist of ValidationIssue

All issues found. Validation collects all problems before raising, so a single call reveals the complete list of mismatches.

class ValidationIssue(column: str, kind: str, message: str)[source]#

A single schema mismatch found during validation.

Attributes:
fieldstr

Name of the field where the issue was found.

kindstr

Short category: "missing", "dtype", or "unit".

messagestr

Human-readable description of the mismatch.

pydantic model VersionedModel[source]#

Base class for ModelBases that are versioned.

Show JSON schema
{
   "title": "VersionedModel",
   "description": "Base class for ModelBases that are versioned.",
   "type": "object",
   "properties": {},
   "additionalProperties": false
}

Config:
  • extra: str = forbid

  • arbitrary_types_allowed: bool = True

  • validate_default: bool = True

classmethod migration_history()[source]#

Return the version migration history for this model.

classmethod model_validate_versioned(data: dict)[source]#

Like model_validate, but applies version migrations.

Migrations are applied to the dict-like object to validate if needed, before passing to pydantic for final validation.

classmethod print_history(printer=<built-in function print>)[source]#

Print a human-readable version of the history.

all_classes_in_model(model: type[BaseModel], allowed_types=pydantic.main.BaseModel | enum.Enum)[source]#

Return all classes and related classes in the model.

classes_in_model(model: type[BaseModel], allowed_types=pydantic.main.BaseModel | enum.Enum, max_depth: int = 1000)[source]#

Return list of unique classes used in the given model.

Parameters:
modelBaseModel

model to search

allowed_typestype

Types to find. Default is BaseModel, but you can also specify e.g. BaseModel|Enum

doc(obj)[source]#

Return cleaned up docstring for the object, suitable for a description field.

enum_doc(enum: type[StrEnum]) str[source]#

Return nicer documentation for an Enum.

fits_header_to_flat_dict(header: Header, model: type[BaseModel], ignore_extra_keys: bool = True) dict[str, str][source]#

Turn a FITS header back into a flat dict of CTAO keywords.

Parameters:
header: Header

FITS header with keys and values to extract

model: type[BaseModel]

Model to use for schema and key mapping

ignore_extra_keys: bool

If False, issue warnings for keys in header that do not map to model.

Returns:
dict[str,str]:

mapping of CTAO keyword to string value.

fits_header_to_instance(header: Header, model: type[BaseModel], ignore_extra_keys: bool = True) BaseModel[source]#

Turn a FITS header back into an instance of a model.

Parameters:
header: Header

FITS header with keys and values to extract

model: type[BaseModel]

Model to use for schema and key mapping

ignore_extra_keys: bool

If False, issue warnings for keys in header that do not map to model.

Returns:
BaseModel:

instance of model provided

flatten_model_instance(model_instance: BaseModel, parent_key: str = '', separator: str = '.', to_string: bool = True, expand_lists: bool = True, quantity_format: QuantityFormat = QuantityFormat.FLOAT)[source]#

Return the flattened model instance.

Parameters:
model_instance: BaseModel

model instance to flatten (not the class!)

parent_key: str

starting key

separator: str

separator between keys

to_string: bool

if True, turn the leaf values into strings

expand_lists: bool

if True, split lists into key.N, where N is the index If this is not True, and to_string is True, the string rep of the list will be used, which might cause issues for round-tripping back to a Pydantic model.

quantity_format: QuantityFormat

Format to use when serializing Quantity fields that have units.

generate_latex_table_includes(models: list[type[BaseModel]], output_dir: Path | str) None[source]#

Write descriptive LaTeX tables for given models to the output_dir.

These can then be included in latex documents.

Parameters:
models: list[BaseModel]

models to write

output_dir: Path|str

output directory

generate_plantuml_diagrams(models: list[type[BaseModel]], output_dir: Path | str, show_inheritance: bool = True, imports: bool = True, max_depth: int = 1000) None[source]#

Generate plantuml diagrams for the given models.

Parameters:
models: list[type[BaseModel]]

list of Models to generte diagrams for

output_dir: Path | str

where to write the plantuml code

show_inheritance: bool

If True, also include parent/child relations and classes

max_depth: int

maximum depth to include in the diagrams.

instance_to_fits_header(model_instance: BaseModel, use_short: bool = True, hierarch_namespace: str = 'CTAO') Header[source]#

Convert a model instance to a FITS header.

The resulting header will have a Card for each keyword. Long keywords will use the FITS HIERARCH standard in the given namespace

Parameters:
model_instance: BaseModel

model instance to serialize

use_short: bool

if True, replace any log keywords that have fits_keyword mapping with their short form.

hierarch_namespace: str

starting string for the HIERARCH keyword

Returns:
Header:

FITS header suitable for writing to a file.

keyword_mapping_latex_table(model: type[BaseModel], separator: str = '.', return_table: bool = False)[source]#

Return LaTeX table with mappings to FITS and IVOA.

Parameters:
modeltype[BaseModel]

Model class to use.

sepstr

Separator to use when generating the CTAO key

return_tablebool

if True, return the Astropy table instead of the LaTeX representation.

migration(from_version: str, to_version: str)[source]#

Migration decorator.

model_to_astropy_table(model: type[BaseModel], n_rows: int = 0) Table[source]#

Create an astropy.table.Table matching the schema of model.

Only fields annotated with fits_column_dtype in 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:
modeltype[BaseModel]

A ModelBase subclass with at least one field carrying fits_column_dtype.

n_rowsint, 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)
model_to_latex_table(model: type[BaseModel]) str[source]#

Return LaTeX table in format used by Data Model documents.

To use this in a document, you will need the following in your preamble:

\usepackage{tocbasic}
\DeclareNewTOC[
  type=classdef,
  types=classdefs,
  name=Class Definition,
  listname={List of Class Definitions},
]{cls}

\NewTblrEnviron{classdef}
\SetTblrOuter[classdef]{long}
\SetTblrInner[classdef]{
  colspec={Q[wd=3.3cm] Q[wd=3cm] X[l]},
  row{odd}={bg=moongray},
  column{1-2} = {font=\scriptsize},
  column{1} = {font=\ttfamily\bfseries\tiny},
  column{3} = {font=\scriptsize},
  row{1} = {bg=galaxyblue, font=\normalfont\bfseries, fg=white}
}
\NewTblrTheme{classdef}{
  \DeclareTblrTemplate{caption-tag}{default}{\textbf{Class Definition\hspace{0.25em}\thetable}}
  \DeclareTblrTemplate{caption-lot}{default}{%
    \addcontentsline{cls}{classdef}{\protect\numberline{\thetable}\InsertTblrText{entry}}%
  }
  \SetTblrStyle{firsthead}{font=\normalfont\captionsize}
  \SetTblrStyle{middlehead}{font=\normalfont\captionsize}
  \SetTblrStyle{lasthead}{font=\normalfont\captionsize}
}
model_to_plantuml_class(model: type[BaseModel | Enum]) str[source]#

Return PlantUML class string.

model_to_plantuml_relations(model: type[BaseModel], *, max_depth=1000, import_definitions: bool = False, show_inheritance: bool = False) str[source]#

Output PlantUML relationships for a model.

By default containment relations are generated, and optionally inheritance relations.

Parameters:
model: type[BaseModel]

the model

max_depth: int

how deep to go

import_defintions: bool

include plantuml imports of the class definitions

show_inheritance: bool

include parent/child relations

Returns:
str:

PlantUML code

model_to_table(model: type[BaseModel], optional_symbol='${^\\oslash}$', type_sep: str = ' | ', unit_string_type='latex', include_fits=False) Table[source]#

Turn a model into an Astropy table.

model_validate_astropy_table(table: Table, model: type[BaseModel], *, strict_units: bool = True) None[source]#

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:
tableastropy.table.Table

Table to validate.

modeltype[BaseModel]

Model whose fits_column_dtype fields define the expected schema.

strict_unitsbool, 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)
print_model(model, max_depth: int = 1000) None[source]#

Print out the model, for debugging.

related_classes_in_model(model: type[BaseModel], allowed_types=pydantic.main.BaseModel | enum.Enum) tuple[list, list][source]#

Return all parent and child classes of those in the given model.

Parameters:
modeltype[BaseModel]

Root model or module containing Pydantic/BaseModel classes.

allowed_typestype | tuple[type], optional

Only include classes that are subclasses of these types.

unflatten_model_instance(flat_dict: dict, model: type[BaseModel], parent_key: str = '', separator: str = '.') BaseModel[source]#

Return the flattened model instance.

Parameters:
flat_dict: dict

flattened dictionary of keywords, e.g. as loaded from json

model_instance: BaseModel

model instance to flatten (not the class!)

parent_key: str

starting key

separator: str

separator between keys

validate_fits_bintable_hdu(fits_file: Path | str, hdu: int | str, column_model: type[BaseModel], header_model: type[BaseModel]) tuple[Table, BaseModel][source]#

Validate a FITS Bintable HDU.

Parameters:
fits_file: Path | str

the file to validate

hdu: int | str

name or number of the HDU to validate in the file

column_model: type[BaseModel]

The model defining the schema for the table columns. It must have fields that have the fits_column_dtype set.

header_model: type[BaseModel]

The model defining the schema for the header metadata to validate.

Returns:
tuple[Table,BaseModel]:

The table in Astropy Table format, and the metadata as a deserialized model.

walk_model(model: type[BaseModel], parent_key: str = '') Generator[tuple[type, list[ParentInfo]]][source]#

Recursively walk the pydantic model.

Parameters:
model: Type[BaseModel]

Model class (not instance) to visit

parent_key: str

Key to use for first element in the model If not specified, the model name is used.

Returns:
Generator[tuple[BaseModel, list[Element]]]:

Element class and path at each step. The path is a list of ParentInfo describing the parents of the current element in order of ancestry, e.g. the current model element’s parent is path[-1], the grandparent is path[-2], …

Examples

To generate a hierarchical list of the elements and sub-elements:

>>> for model_element, path in walk_model(some_model):
>>>     name = '.'.join(p.item_name for p in path)
>>>     print(f"{'*' * (len(path)+1)} {name} : {model_element.__name__}")
*  Rererence : Reference
** process : Process
*** process.type : ObservatoryProcess
** data : Product
*** data.category : DataCategory
*** data.level : DataLevel
*** data.division : DataDivision
*** data.association : DataAssociation
*** data.type : DataType
...
walk_model_to_depth(model: type[BaseModel], max_depth: int = 1000, parent_key: str = '')[source]#

Walk the model as in walk_model, but stop at depth.