#
# Copyright 2026 by University College Dublin. All rights reserved.
#
# Developed by: Derek O'Callaghan
# University College Dublin
# https://www.ucd.ie/
#
# Builds on:
# Gamma-ray Data Tools - Core Components (https://github.com/USRA-STI/gdt-core)
# Gamma-ray Data Tools - Fermi mission components (https://github.com/USRA-STI/gdt-fermi/)
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
# in compliance with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software distributed under the License
# is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied. See the License for the specific language governing permissions and limitations under the
# License.
#
import copy
import operator
import re
from typing import List, Tuple, Type
from gdt.core.headers import Header, FileHeaders
from .time import Time
from .detectors import GiftsDetectors
from gdt.missions.fermi.gbm.detectors import GbmDetectors
from gdt.missions.fermi.gbm.headers import (
_telescope as _gbm_telescope,
_instrument as _gbm_instrument,
_observer as _gbm_observer,
_origin as _gbm_origin,
_datatype_card as _gbm_datatype_card,
_instrument_card as _gbm_instrument_card,
_err_rad_card as _gbm_err_rad_card,
_filetype_card as _gbm_filetype_card,
_filever_card as _gbm_filever_card,
_mjdrefi_card as _gbm_mjdrefi_card,
_mjdreff_card as _gbm_mjdreff_card,
_observer_card as _gbm_observer_card,
_origin_card as _gbm_origin_card,
_telescope_card as _gbm_telescope_card,
_tstart_card as _gbm_tstart_card,
_tstop_card as _gbm_tstop_card,
GbmHeader,
DataPrimaryHeader as GbmDataPrimaryHeader,
DataTriggerPrimaryHeader as GbmDataTriggerPrimaryHeader,
EboundsHeader as GbmEboundsHeader,
EboundsTriggerHeader as GbmEboundsTriggerHeader,
EventsHeader as GbmEventsHeader,
EventsTriggerHeader as GbmEventsTriggerHeader,
HealpixPrimaryHeader as GbmHealpixPrimaryHeader,
HealpixImageHeader as GbmHealpixImageHeader,
GtiHeader as GbmGtiHeader,
GtiTriggerHeader as GbmGtiTriggerHeader,
PosHistPrimaryHeader as GbmPosHistPrimaryHeader,
PosHistDataHeader as GbmPosHistDataHeader,
SpectrumHeader as GbmSpectrumHeader,
SpectrumTriggerHeader as GbmSpectrumTriggerHeader,
TcatHeader as GbmTcatHeader,
TrigdatPrimaryHeader as GbmTrigdatPrimaryHeader,
TrigdatSecondaryHeader as GbmTrigdatSecondaryHeader,
TrigdatTrigrateHeader as GbmTrigdatTrigrateHeader,
TrigdatBackratesHeader as GbmTrigdatBackratesHeader,
TrigdatObCalcHeader as GbmTrigdatObCalcHeader,
TrigdatMaxratesHeader as GbmTrigdatMaxratesHeader,
TrigdatEventrateHeader as GbmTrigdatEventrateHeader,
)
__all__ = ["HealpixHeaders", "PhaiiHeaders", "PhaiiTriggerHeaders", "PosHistHeaders", "TcatHeaders", "TrigdatHeaders", "TteHeaders", "TteTriggerHeaders"]
# GIFTS common keyword cards
_gifts_telescope = "GIFTS (P.I.: McBreen)"
_gifts_instrument = "GIFTS"
_gifts_observer = "O'Callaghan"
_gifts_origin = "GIFTS Team"
_gifts_filever = "0.1.0" # TODO: configurable using gdt-gifts version
HeaderCard = Tuple[str|int]
def _gbm_to_gifts_card(gbm_card: HeaderCard, gbm_update: str, gifts_update: str, update_index: int) -> HeaderCard:
gifts_card = list(gbm_card)
gifts_card[update_index] = re.sub(gbm_update, gifts_update, gifts_card[update_index]) if gbm_update else gifts_update
return tuple(gifts_card)
def _find_header_card(keywords: List[HeaderCard], keyword: str) -> HeaderCard:
card = [x for x in keywords if x[0] == keyword]
return card[0] if card and len(card) == 1 else None
_gbm_tcat_loc_src_card = _find_header_card(GbmTcatHeader.keywords, "LOC_SRC")
_gbm_tcat_det_mask_card = _find_header_card(GbmTcatHeader.keywords, "DET_MASK")
# Used for both Tcat and Trigdat
# TODO: might want to add this to other headers
_gbm_infile01_card = _find_header_card(GbmTcatHeader.keywords, "INFILE01")
_gifts_infile01_card = _gbm_to_gifts_card(_gbm_infile01_card, r"Level [0-9]", "GBM catalog", 2)
class GiftsHeader(GbmHeader):
# Will override corresponding GBM header keywords
# Common to multiple GIFTS headers
_common_gbm_gifts_keywords: List[Tuple[HeaderCard]] = [
(_gbm_instrument_card, _gbm_to_gifts_card(_gbm_instrument_card, _gbm_instrument, _gifts_instrument, 1)),
(_gbm_mjdrefi_card, _gbm_to_gifts_card(_gbm_mjdrefi_card, _gbm_telescope, _gifts_instrument, 2)),
(_gbm_mjdreff_card, _gbm_to_gifts_card(_gbm_mjdreff_card, _gbm_telescope, _gifts_instrument, 2)),
(_gbm_observer_card, _gbm_to_gifts_card(_gbm_to_gifts_card(_gbm_observer_card, _gbm_observer, _gifts_observer, 1), None, f"{_gifts_instrument} Pipeline Development", 2)),
(_gbm_origin_card, _gbm_to_gifts_card(_gbm_origin_card, _gbm_origin, _gifts_origin, 1)),
(_gbm_telescope_card, _gbm_to_gifts_card(_gbm_telescope_card, _gbm_telescope, _gifts_telescope, 1)),
(_gbm_tstart_card, _gbm_to_gifts_card(_gbm_tstart_card, _gbm_telescope, _gifts_instrument, 2)),
(_gbm_tstop_card, _gbm_to_gifts_card(_gbm_tstop_card, _gbm_telescope, _gifts_instrument, 2)),
(_gbm_tcat_loc_src_card, _gbm_to_gifts_card(_gbm_tcat_loc_src_card, None, _gifts_instrument, 1)),
(_gbm_tcat_det_mask_card, _gbm_to_gifts_card(_gbm_tcat_det_mask_card, f"{len(GbmDetectors) - 1}", f"{len(GiftsDetectors) - 1}", 2)),
(_gbm_infile01_card, _gifts_infile01_card),
(_gbm_filever_card, _gbm_to_gifts_card(_gbm_filever_card, None, _gifts_filever, 1)),
(_gbm_datatype_card, _gbm_to_gifts_card(_gbm_datatype_card, _gbm_instrument, _gifts_instrument, 2)),
# (_gbm_err_rad_card, _gbm_to_gifts_card(_gbm_to_gifts_card(_gbm_err_rad_card, None, "LOC_ERR", 0), " Radius", "", 2)),
]
# Will be set by subclass for specific headers
_update_gbm_gifts_keywords: List[HeaderCard] = [] # GBM keyword cards to be updated for GIFTS
_add_gifts_keywords: List[HeaderCard] = [] # GBM keyword cards to be added for GIFTS
_remove_gbm_keywords: List[str] = [] # GBM keyword cards to be removed
def __init__(self, *args, **kwargs):
# Replace any GBM headers with GIFTS equivalents,
# prior to initialisation
self.keywords = self.keywords.copy()
all_gbm_gifts_keywords = self._update_gbm_gifts_keywords if self._update_gbm_gifts_keywords else []
all_gbm_gifts_keywords += self._common_gbm_gifts_keywords
for gbm_keyword, gifts_keyword in all_gbm_gifts_keywords:
try:
i = self.keywords.index(gbm_keyword)
self.keywords[i] = gifts_keyword
except Exception as e:
pass
# Add any keywords specified by the subclass
if self._add_gifts_keywords:
for gifts_keyword in self._add_gifts_keywords:
try:
self.keywords.index(gifts_keyword)
except:
self.keywords.append(gifts_keyword)
# Remove any keywords as specified by the subclass
if self._remove_gbm_keywords:
for gbm_keyword in self._remove_gbm_keywords:
gbm_keyword_card = _find_header_card(self.keywords, gbm_keyword)
if gbm_keyword_card:
self.keywords.remove(gbm_keyword_card)
super().__init__(args, kwargs)
def __setitem__(self, key, val):
# Only GIFTS-specific logic here is the time format,
# currently unable to specify it to the parent GbmHeader
# so have to duplicate the logic here
if not isinstance(key, tuple) and not isinstance(val, tuple):
if key.upper() == 'TSTART':
self['DATE-OBS'] = Time(val, format='gifts').iso
elif key.upper() == 'TSTOP':
self['DATE-END'] = Time(val, format='gifts').iso
else:
pass
# if 'INFILE' in key.upper():
# super(Header, self).__setitem__(key, val)
# return
super().__setitem__(key, val)
class DataPrimaryHeader(GiftsHeader, GbmDataPrimaryHeader): pass
class DataTriggerPrimaryHeader(GiftsHeader, GbmDataTriggerPrimaryHeader): pass
class EboundsHeader(GiftsHeader, GbmEboundsHeader): pass
class EboundsTriggerHeader(GiftsHeader, GbmEboundsTriggerHeader): pass
class SpectrumHeader(GiftsHeader, GbmSpectrumHeader): pass
class SpectrumTriggerHeader(GiftsHeader, GbmSpectrumTriggerHeader): pass
class EventsHeader(GiftsHeader, GbmEventsHeader):
# TODO Deadtime comment, no duplication
# ('EVT_DEAD', 2.6e-6, '[s] Deadtime per event'), # TODO KEEP
# ('EVTDEDHI', 1.0417e-5, '[s] Deadtime per overflow channel event'), # TODO KEEP
pass
class EventsTriggerHeader(GiftsHeader, GbmEventsTriggerHeader):
# TODO Deadtime comment, no duplication
# ('EVT_DEAD', 2.6e-6, 'Deadtime per event (s)'),
pass
class GtiHeader(GiftsHeader, GbmGtiHeader): pass
class GtiTriggerHeader(GiftsHeader, GbmGtiTriggerHeader): pass
class HealpixPrimaryHeader(GiftsHeader, GbmHealpixPrimaryHeader):
# If the card name is renamed, it can't be read during open() in a parent class
# For now, only update the description and retain ERR_RAD consistent with GBM
_update_gbm_gifts_keywords = [(_gbm_err_rad_card, _gbm_to_gifts_card(_gbm_err_rad_card, " Radius", "", 2))]
# TODO: this should be accessible
CLOSEST_DETECTORS = {
"G0": "n7",
"G1": "n6",
"G2": "n9",
"G3": "n1",
"G4": "n0",
"G5": "n3",
}
class HealpixImageHeader(GiftsHeader, GbmHealpixImageHeader):
def __init__(self, *args, **kwargs):
# TODO: get the list of GBM detectors not in CLOSEST
remove_detectors = [f"N{i}" for i in [2, 4, 5, 8, "A", "B"]] + [f"B{i}" for i in [0, 1]]
self._remove_gbm_keywords = [f"{d}_RA" for d in remove_detectors]
self._remove_gbm_keywords += [f"{d}_DEC" for d in remove_detectors]
self._add_gifts_keywords = []
for gifts_detector, gbm_detector in CLOSEST_DETECTORS.items():
for c in ["RA", "DEC"]:
gbm_keyword_card = _find_header_card(self.keywords, f"{gbm_detector.upper()}_{c}")
if gbm_keyword_card:
gifts_keyword_card = list(_gbm_to_gifts_card(gbm_keyword_card, gbm_detector, gifts_detector, 2))
gifts_keyword_card[0] = f"{gifts_detector}_{c}"
# TODO: this results in the GIFTS detector headers being added
# in order of closest GBM detectors
# self._update_gbm_gifts_keywords.append((gbm_keyword_card, tuple(gifts_keyword_card)))
self._add_gifts_keywords.append(tuple(gifts_keyword_card))
self._remove_gbm_keywords.append(gbm_keyword_card[0])
self._update_gbm_gifts_keywords = []
for c in ["RA", "DEC"]:
gbm_keyword_card = _find_header_card(self.keywords, f"GEO_{c}")
if gbm_keyword_card:
# 'Fermi' is hardcoded in GBM headers, nothing to import atm
self._update_gbm_gifts_keywords.append((gbm_keyword_card,
_gbm_to_gifts_card(gbm_keyword_card, "Fermi", _gifts_instrument, 2)))
super().__init__(args, kwargs)
class TcatHeader(GiftsHeader, GbmTcatHeader): pass
class TrigdatCommonHeader(GiftsHeader):
_remove_gbm_keywords = ["DETTYPE", "DATATYPE"]
class TrigdatPrimaryHeader(TrigdatCommonHeader, GbmTrigdatPrimaryHeader): pass
class TrigdatSecondaryHeader(TrigdatCommonHeader, GbmTrigdatSecondaryHeader): pass
class TrigdatTrigrateHeader(TrigdatCommonHeader, GbmTrigdatTrigrateHeader): pass
class TrigdatBackratesHeader(TrigdatCommonHeader, GbmTrigdatBackratesHeader): pass
class TrigdatObCalcHeader(TrigdatCommonHeader, GbmTrigdatObCalcHeader): pass
class TrigdatMaxratesHeader(TrigdatCommonHeader, GbmTrigdatMaxratesHeader): pass
class TrigdatEventrateHeader(TrigdatCommonHeader, GbmTrigdatEventrateHeader): pass
class PosHistPrimaryHeader(GiftsHeader, GbmPosHistPrimaryHeader): pass
class PosHistDataHeader(GiftsHeader, GbmPosHistDataHeader):
name = GbmPosHistDataHeader.name.replace(_gbm_telescope, _gifts_instrument)
class GiftsFiletypePrimaryHeader(GiftsHeader):
"""Enable file types to be specified"""
_add_gifts_keywords = [_gifts_infile01_card]
class GiftsDataPrimaryHeader(GiftsFiletypePrimaryHeader, DataPrimaryHeader): pass
class GiftsDataTriggerPrimaryHeader(GiftsFiletypePrimaryHeader, DataTriggerPrimaryHeader): pass
#-------------------------------------
class GiftsFileHeaders(FileHeaders):
# Associated FITS file type
filetype = None
# Define keywords to be excluded when copying from source headers
_exclude_file_keywords = set([])
@classmethod
def from_file_headers(cls, input_headers: FileHeaders) -> FileHeaders:
"""
Initialises a GIFTS headers set using a corresponding set of file (GBM/GIFTS) headers
"""
gifts_keywords = set(["CREATOR", "TELESCOP", "INSTRUME", 'OBSERVER', "ORIGIN", "LOC_SRC", "DET_MASK", "FILETYPE", "FILE-VER"])
commentary_keywords = set(["COMMENT", "HISTORY"])
output_headers = cls()
#for header_name in input_headers.keys():
# Using index enables headers with different names to be copied to each other
# E.g. TTE headers -> PHAII headers
for i in range(input_headers.num_headers):
#input_header = input_headers[header_name]
#output_header = output_headers[header_name]
try:
input_header = input_headers[i]
output_header = output_headers[i]
except:
continue
# TODO: log warning
for commentary_keyword in commentary_keywords:
commentary_indices = [i for i, c in enumerate(output_header._cards) if c.keyword == commentary_keyword]
if len(commentary_indices) > 1:
# Have to delete directly from the cards
# Assumes they'll be co-located
del output_header._cards[commentary_indices[0]:commentary_indices[-1] + 1]
for k, v in [(k, v) for k, v in input_header.items() if k not in (gifts_keywords | commentary_keywords | set(output_headers._exclude_file_keywords))]:
try:
output_header[k] = v
except:
# Header key is present in input header but not in output
pass
for commentary_keyword in commentary_keywords:
for c in [c for c in input_header._cards if c.keyword == commentary_keyword]:
# Have to add directly to the cards, as astropy.io.fits.header.Header.keys() checks _cards.
# These will have previously been deleted above.
# Can't use add_comment() or add_history() as at least one of these is required initially.
output_header._cards.append(c)
return output_headers
def update(self):
"""Update the 'FILETYPE' keywords to the instance type.
"""
if self.filetype:
self._update_headers(headers=self._headers.values(), keyword="FILETYPE", value=self.filetype)
super().update()
def _update_headers(self, headers, keyword, value):
for hdr in headers:
try:
hdr[keyword] = value
except:
pass
class PhaiiCommonHeaders(GiftsFileHeaders):
filetype = f"{_gifts_instrument} PHAII"
_exclude_file_keywords = ['CREATOR', 'DATATYPE', 'EXTNAME', 'FILENAME', 'FILETYPE', 'HDUCLAS1', "INFILE01"]
def update(self):
"""Update the 'DATATYPE' keyword to the instance type.
"""
if self.filetype:
self._update_headers(headers=[self._headers["PRIMARY"]], keyword="DATATYPE", value=self.filetype)
super().update()
class TteCommonHeaders(GiftsFileHeaders):
filetype = f"{_gifts_instrument} PHOTON LIST"