Astropy Table and FITS BINTABLE Schemas#

In this example, we will show how to use model classes to validate a FITS (or any Astropy table)

from typing import ClassVar

import ctao_datamodel as dm
import numpy as np
from astropy import units as u
from astropy.io import fits
from astropy.table import Table
from ctao_datamodel import AstroField, ModelBase, PlantUMLDiagram, Quantity

Define a table schema#

Here, we will implement a simple “event-list” table schema:

class EventListTable(ModelBase):
    """The column schema for an EventList"""

    _namespace: ClassVar[str] = "Example.DL3"

    event_id: int = AstroField(
        "Unique event identifier", fits_column_dtype="int64", fits_keyword="EVENT_ID"
    )
    energy: Quantity["TeV"] = AstroField(
        "Reconstructed energy", fits_column_dtype="float32", ucd="em.energy"
    )
    ra: Quantity["deg"] = AstroField(
        "Right ascension", fits_column_dtype="float64", ucd="pos.eq.ra"
    )
    dec: Quantity["deg"] = AstroField(
        "Declination", fits_column_dtype="float64", ucd="pos.eq.dec"
    )
    gammaness: float | None = AstroField(
        "Gammaness", default=None, fits_column_dtype="float64"
    )

PlantUMLDiagram(EventListTable)

dm.model_to_table(EventListTable)

Table length=5

Name	Unit	Description	Type
str21	str21	str64	str8
event_id		Unique event identifier	int
energy	$\mathrm{TeV}$	Reconstructed energy \newline(\textbf{UCD}: \texttt{em.energy})	Quantity
ra	$\mathrm{{}^{\circ}}$	Right ascension \newline(\textbf{UCD}: \texttt{pos.eq.ra})	Quantity
dec	$\mathrm{{}^{\circ}}$	Declination \newline(\textbf{UCD}: \texttt{pos.eq.dec})	Quantity
gammaness${^\oslash}$		Gammaness	float

Create an Empty Table#

First, we can create an empty table from the schema:

table = dm.model_to_astropy_table(EventListTable, n_rows=2)
table

Table length=2

EVENT_ID	ENERGY	RA	DEC	GAMMANESS
	TeV	deg	deg
int64	float32	float64	float64	float64
0	0.0	0.0	0.0	0.0
0	0.0	0.0	0.0	0.0

And convert it to a FITS file, to see the resulting headers:

table.write("example_table.fits", overwrite=True)
fits.getheader("example_table.fits", ext=1)

XTENSION= 'BINTABLE'           / binary table extension                         
BITPIX  =                    8 / array data type                                
NAXIS   =                    2 / number of array dimensions                     
NAXIS1  =                   36 / length of dimension 1                          
NAXIS2  =                    2 / length of dimension 2                          
PCOUNT  =                    0 / number of group parameters                     
GCOUNT  =                    1 / number of groups                               
TFIELDS =                    5 / number of table fields                         
TTYPE1  = 'EVENT_ID'                                                            
TFORM1  = 'K       '                                                            
TTYPE2  = 'ENERGY  '                                                            
TFORM2  = 'E       '                                                            
TUNIT2  = 'TeV     '                                                            
TTYPE3  = 'RA      '                                                            
TFORM3  = 'D       '                                                            
TUNIT3  = 'deg     '                                                            
TTYPE4  = 'DEC     '                                                            
TFORM4  = 'D       '                                                            
TUNIT4  = 'deg     '                                                            
TTYPE5  = 'GAMMANESS'                                                           
TFORM5  = 'D       '                                                            
COMMENT --BEGIN-ASTROPY-SERIALIZED-COLUMNS--                                    
COMMENT datatype:                                                               
COMMENT - {name: EVENT_ID, datatype: int64, description: Unique event identifi\ 
COMMENT er}                                                                     
COMMENT - name: ENERGY                                                          
COMMENT   unit: TeV                                                             
COMMENT   datatype: float32                                                     
COMMENT   description: Reconstructed energy                                     
COMMENT   meta: !!omap                                                          
COMMENT   - {ucd: em.energy}                                                    
COMMENT - name: RA                                                              
COMMENT   unit: deg                                                             
COMMENT   datatype: float64                                                     
COMMENT   description: Right ascension                                          
COMMENT   meta: !!omap                                                          
COMMENT   - {ucd: pos.eq.ra}                                                    
COMMENT - name: DEC                                                             
COMMENT   unit: deg                                                             
COMMENT   datatype: float64                                                     
COMMENT   description: Declination                                              
COMMENT   meta: !!omap                                                          
COMMENT   - {ucd: pos.eq.dec}                                                   
COMMENT - {name: GAMMANESS, datatype: float64, description: Gammaness}          
COMMENT meta: !!omap                                                            
COMMENT - __serialized_columns__: {}                                            
COMMENT --END-ASTROPY-SERIALIZED-COLUMNS--                                      

Validate Existing Table#

Here, we take an existing table. Note that we don’t have a GAMMANESS column, which shouldn’t be a problem, since it is optional. We also use different (but compatible) units to what is in the schema

existing_table = Table(
    dict(
        EVENT_ID=[12345, 12346],
        ENERGY=([1.0, 3.0] * u.TeV).astype(np.float32),
        RA=[15.0, 17.4] * u.hourangle,
        DEC=[-2.3, -2.4] * u.deg,
    )
)
existing_table

Table length=2

EVENT_ID	ENERGY	RA	DEC
	TeV	hourangle	deg
int64	float32	float64	float64
12345	1.0	15.0	-2.3
12346	3.0	17.4	-2.4

dm.model_validate_astropy_table(existing_table, EventListTable)

and, indeed this validates! (no errors)

But if we have a wrong column unit for example, it should fail:

bad_table = Table(
    dict(
        EVENT_ID=[12345, 12346],
        ENERGY=([1.0, 3.0] * u.TeV).astype(np.float32),
        # missing RA column
        DEC=[-2.3, -2.4] * u.m,  # wrong unit
    )
)
bad_table

Table length=2

EVENT_ID	ENERGY	DEC
	TeV	m
int64	float32	float64
12345	1.0	-2.3
12346	3.0	-2.4

try:
    dm.model_validate_astropy_table(bad_table, EventListTable)
except dm.TableValidationError as err:
    print(err)

TableValidationError: 
  [missing] RA: Column 'RA' is required by model field 'ra' but not present in table.
  [unit] DEC: Expected unit equivalent to 'deg', got 'm'.

Validating A full FITS BINTABLE with Header metadata#

Table metadata can be validated in different ways dependending on its format.

If the metadata is stored in a hierarchical JSON representation, the usual pydantic validation can be used directly, e.g. just using SomeModel.model_validate(table.meta)
If the metadata have been flattened, the same still works but they must be unflattened first, e.g. SomeModel.model_validate(dm.unflatten_model_instance(table.meta))
if the metadata is in FITS keyword format, ~ctao_datamodel.fits_header_to_instance method can be used to validate and transform back into the hierarchical representation.

Here is an example:

class EventListMetadata(ModelBase):
    """Schema for some metadata keywords, can be hierarchical."""

    object: str = AstroField("object name", fits_keyword="OBJECT")
    contact: dm.models.dataproducts.Contact = AstroField("contact info.")


dm.PlantUMLDiagram(EventListMetadata, details=True)

meta = EventListMetadata(
    object="Crab Nebula",
    contact=dm.models.dataproducts.Contact(
        name="Jane User", organization="none", email="jane@none.org"
    ),
)
header = dm.instance_to_fits_header(meta)
header

OBJECT  = 'Crab Nebula'        / object name                                    
AUTHOR  = 'Jane User'          / Contact name for this data product.            
ORIGIN  = 'none    '           / Contact organization name of this data product.
EMAIL   = 'jane@none.org'      / Contact's email address                        

new_table = Table(
    dict(
        EVENT_ID=[12345, 12346],
        ENERGY=([1.0, 3.0] * u.TeV).astype(np.float32),
        RA=[15.0, 17.4] * u.deg,
        DEC=[-2.3, -2.4] * u.deg,
    )
)

hdu = fits.BinTableHDU(data=new_table, header=header, name="EVENTS")
hdu.writeto("test_with_header.fits", overwrite=True)
fits.getheader("test_with_header.fits", "EVENTS")

XTENSION= 'BINTABLE'           / binary table extension                         
BITPIX  =                    8 / array data type                                
NAXIS   =                    2 / number of array dimensions                     
NAXIS1  =                   28 / length of dimension 1                          
NAXIS2  =                    2 / length of dimension 2                          
PCOUNT  =                    0 / number of group parameters                     
GCOUNT  =                    1 / number of groups                               
TFIELDS =                    4 / number of table fields                         
TTYPE1  = 'EVENT_ID'                                                            
TFORM1  = 'K       '                                                            
TTYPE2  = 'ENERGY  '                                                            
TFORM2  = 'E       '                                                            
TUNIT2  = 'TeV     '                                                            
TTYPE3  = 'RA      '                                                            
TFORM3  = 'D       '                                                            
TUNIT3  = 'deg     '                                                            
TTYPE4  = 'DEC     '                                                            
TFORM4  = 'D       '                                                            
TUNIT4  = 'deg     '                                                            
OBJECT  = 'Crab Nebula'        / object name                                    
AUTHOR  = 'Jane User'          / Contact name for this data product.            
ORIGIN  = 'none    '           / Contact organization name of this data product.
EMAIL   = 'jane@none.org'      / Contact's email address                        
EXTNAME = 'EVENTS  '           / extension name                                 

Now, let’s try validating the FITS file we wrote using a helper function that validates both the column schema and the metadata at once, and returns deserialized metadata:

table, metadata = dm.validate_fits_bintable_hdu(
    fits_file="test_with_header.fits",
    hdu="EVENTS",
    column_model=EventListTable,
    header_model=EventListMetadata,
)

table

Table length=2

EVENT_ID	ENERGY	RA	DEC
	TeV	deg	deg
int64	float32	float64	float64
12345	1.0	15.0	-2.3
12346	3.0	17.4	-2.4

table.meta

{'OBJECT': 'Crab Nebula',
 'AUTHOR': 'Jane User',
 'ORIGIN': 'none',
 'EMAIL': 'jane@none.org',
 'EXTNAME': 'EVENTS'}

metadata

EventListMetadata(object='Crab Nebula', contact=Contact(name='Jane User', organization='none', email='jane@none.org'))

metadata.contact.email

'jane@none.org'

And then you can see the table is valid (no errors) and the metadata has been checked and converted to a model instance. If there had been a problem, you would get an exception with the details.

Other table formats#

Since the underlying implementation uses astropy ~astropy.table.Table

table_2 = dm.model_to_astropy_table(EventListTable)
table_2.add_row([12345, 0.2, 10, -20.2, 0.2])
table_2.add_row([12346, 100.2, 10.2, -20.1, 0.9])
table_2.meta = dm.flatten_model_instance(metadata) 
table_2.meta

{'object': 'Crab Nebula',
 'contact.name': 'Jane User',
 'contact.organization': 'none',
 'contact.email': 'jane@none.org'}

table_2["ENERGY"].meta

{'ucd': 'em.energy'}

For example, one can write the table to a VOTable as follows. However, the user should be careful here, as currently astropy does not serialize table metadata to this format, so information is lost.

table_2.write("test.votable", format="votable", overwrite=True)

!cat test.votable

<?xml version="1.0" encoding="utf-8"?>
<!-- Produced with astropy.io.votable version 8.0.1
     http://www.astropy.org/ -->
<VOTABLE version="1.4" xmlns="http://www.ivoa.net/xml/VOTable/v1.3" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.ivoa.net/xml/VOTable/v1.3 http://www.ivoa.net/xml/VOTable/VOTable-1.4.xsd">
 <RESOURCE type="results">
  <TABLE>
   <FIELD ID="EVENT_ID" datatype="long" name="EVENT_ID">
    <DESCRIPTION>
     Unique event identifier
    </DESCRIPTION>
   </FIELD>
   <FIELD ID="ENERGY" datatype="float" name="ENERGY" ucd="em.energy" unit="TeV">
    <DESCRIPTION>
     Reconstructed energy
    </DESCRIPTION>
   </FIELD>
   <FIELD ID="RA" datatype="double" name="RA" ucd="pos.eq.ra" unit="deg">
    <DESCRIPTION>
     Right ascension
    </DESCRIPTION>
   </FIELD>
   <FIELD ID="DEC" datatype="double" name="DEC" ucd="pos.eq.dec" unit="deg">
    <DESCRIPTION>
     Declination
    </DESCRIPTION>
   </FIELD>
   <FIELD ID="GAMMANESS" datatype="double" name="GAMMANESS">
    <DESCRIPTION>
     Gammaness
    </DESCRIPTION>
   </FIELD>
   <DATA>
    <TABLEDATA>
     <TR>
      <TD>12345</TD>
      <TD>0.2</TD>
      <TD>10</TD>
      <TD>-20.2</TD>
      <TD>0.2</TD>
     </TR>
     <TR>
      <TD>12346</TD>
      <TD>100.2</TD>
      <TD>10.2</TD>
      <TD>-20.1</TD>
      <TD>0.9</TD>
     </TR>
    </TABLEDATA>
   </DATA>
  </TABLE>
 </RESOURCE>
</VOTABLE>

And here it is in ECSV format, which does serialize metadata, and has full round-tripping?

table_2.write("test.ecsv", overwrite=True)

!cat test.ecsv

# %ECSV 1.0
# ---
# datatype:
# - {name: EVENT_ID, datatype: int64, description: Unique event identifier}
# - name: ENERGY
#   unit: TeV
#   datatype: float32
#   description: Reconstructed energy
#   meta: !!omap
#   - {ucd: em.energy}
# - name: RA
#   unit: deg
#   datatype: float64
#   description: Right ascension
#   meta: !!omap
#   - {ucd: pos.eq.ra}
# - name: DEC
#   unit: deg
#   datatype: float64
#   description: Declination
#   meta: !!omap
#   - {ucd: pos.eq.dec}
# - {name: GAMMANESS, datatype: float64, description: Gammaness}
# meta: !!omap
# - {object: Crab Nebula}
# - {contact.name: Jane User}
# - {contact.organization: none}
# - {contact.email: jane@none.org}
# schema: astropy-2.0
EVENT_ID ENERGY RA DEC GAMMANESS
12345 0.2 10.0 -20.2 0.2
12346 100.2 10.2 -20.1 0.9

Astropy Table and FITS BINTABLE Schemas#

Define a table schema#

Create an Empty Table#

Validate Existing Table#

Validating A full FITS BINTABLE with Header metadata#

Other table formats#

This Page