Source code for gammapy.data.hdu_index_table

# Licensed under a 3-clause BSD style license - see LICENSE.rst
import logging
import numpy as np
from astropy.table import Table
from astropy.utils import lazyproperty
from gammapy.utils.fits import HDULocation
from gammapy.utils.scripts import make_path

__all__ = ["HDUIndexTable"]

log = logging.getLogger(__name__)


[docs]class HDUIndexTable(Table): """HDU index table. See :ref:`gadf:hdu-index`. """ VALID_HDU_TYPE = [ "events", "gti", "aeff", "edisp", "psf", "bkg", "rad_max", "pointing", ] """Valid values for `HDU_TYPE`.""" VALID_HDU_CLASS = [ "events", "gti", "aeff_2d", "edisp_2d", "psf_table", "psf_3gauss", "psf_king", "bkg_2d", "bkg_3d", "rad_max_2d", "pointing", ] """Valid values for `HDU_CLASS`."""
[docs] @classmethod def read(cls, filename, **kwargs): """Read :ref:`gadf:hdu-index`. Parameters ---------- filename : `pathlib.Path` or str Filename. **kwargs : dict, optional Keyword arguments passed to `~astropy.table.Table.read`. """ filename = make_path(filename) table = super().read(filename, **kwargs) table.meta["BASE_DIR"] = filename.parent.as_posix() # TODO: this is a workaround for the joint-crab validation with astropy>4.0. # TODO: Remove when handling of empty columns is clarified table["FILE_DIR"].fill_value = "" return table.filled()
@property def base_dir(self): """Base directory.""" return make_path(self.meta.get("BASE_DIR", ""))
[docs] def hdu_location(self, obs_id, hdu_type=None, hdu_class=None, warn_missing=True): """Create `HDULocation` for a given selection. Parameters ---------- obs_id : int Observation ID. hdu_type : str, optional HDU type (see `~gammapy.data.HDUIndexTable.VALID_HDU_TYPE`). Default is None. hdu_class : str, optional HDU class (see `~gammapy.data.HDUIndexTable.VALID_HDU_CLASS`). Default is None. warn_missing : bool, optional Warn if no HDU is found matching the selection. Default is True. Returns ------- location : `~gammapy.data.HDULocation` HDU location. """ self._validate_selection(obs_id=obs_id, hdu_type=hdu_type, hdu_class=hdu_class) idx = self.row_idx(obs_id=obs_id, hdu_type=hdu_type, hdu_class=hdu_class) if len(idx) == 1: idx = idx[0] elif len(idx) == 0: if warn_missing: log.warning( f"No HDU found matching: OBS_ID = {obs_id}, HDU_TYPE = {hdu_type}," " HDU_CLASS = {hdu_class}" ) return None else: idx = idx[0] log.warning( f"Found multiple HDU matching: OBS_ID = {obs_id}, HDU_TYPE = {hdu_type}," " HDU_CLASS = {hdu_class}." f" Returning the first entry, which has " f"HDU_TYPE = {self[idx]['HDU_TYPE']} and HDU_CLASS = {self[idx]['HDU_CLASS']}" ) return self.location_info(idx)
def _validate_selection(self, obs_id, hdu_type, hdu_class): """Validate HDU selection. The goal is to give helpful error messages to the user. """ if hdu_type is None and hdu_class is None: raise ValueError("You have to specify `hdu_type` or `hdu_class`.") if hdu_type and hdu_type not in self.VALID_HDU_TYPE: valid = [str(_) for _ in self.VALID_HDU_TYPE] raise ValueError(f"Invalid hdu_type: {hdu_type}. Valid values are: {valid}") if hdu_class and hdu_class not in self.VALID_HDU_CLASS: valid = [str(_) for _ in self.VALID_HDU_CLASS] raise ValueError( f"Invalid hdu_class: {hdu_class}. Valid values are: {valid}" ) if obs_id not in self["OBS_ID"]: raise IndexError(f"No entry available with OBS_ID = {obs_id}")
[docs] def row_idx(self, obs_id, hdu_type=None, hdu_class=None): """Table row indices for a given selection. Parameters ---------- obs_id : int Observation ID. hdu_type : str, optional HDU type (see `~gammapy.data.HDUIndexTable.VALID_HDU_TYPE`). Default is None. hdu_class : str, optional HDU class (see `~gammapy.data.HDUIndexTable.VALID_HDU_CLASS`). Default is None. Returns ------- idx : list of int List of row indices matching the selection. """ selection = self["OBS_ID"] == obs_id if hdu_class: is_hdu_class = self._hdu_class_stripped == hdu_class selection &= is_hdu_class if hdu_type: is_hdu_type = self._hdu_type_stripped == hdu_type selection &= is_hdu_type idx = np.where(selection)[0] return list(idx)
[docs] def location_info(self, idx): """Create `HDULocation` for a given row index.""" row = self[idx] return HDULocation( hdu_class=row["HDU_CLASS"].strip(), base_dir=self.base_dir.as_posix(), file_dir=row["FILE_DIR"].strip(), file_name=row["FILE_NAME"].strip(), hdu_name=row["HDU_NAME"].strip(), )
@lazyproperty def _hdu_class_stripped(self): return np.array([_.strip() for _ in self["HDU_CLASS"]]) @lazyproperty def _hdu_type_stripped(self): return np.array([_.strip() for _ in self["HDU_TYPE"]]) @lazyproperty def obs_id_unique(self): """Observation IDs (unique).""" return np.unique(np.sort(self["OBS_ID"])) @lazyproperty def hdu_type_unique(self): """HDU types (unique).""" return list(np.unique(np.sort([_.strip() for _ in self["HDU_TYPE"]]))) @lazyproperty def hdu_class_unique(self): """HDU classes (unique).""" return list(np.unique(np.sort([_.strip() for _ in self["HDU_CLASS"]])))
[docs] def summary(self): """Summary report as a string.""" obs_id = self.obs_id_unique return ( "HDU index table:\n" f"BASE_DIR: {self.base_dir}\n" f"Rows: {len(self)}\n" f"OBS_ID: {obs_id[0]} -- {obs_id[-1]}\n" f"HDU_TYPE: {self.hdu_type_unique}\n" f"HDU_CLASS: {self.hdu_class_unique}\n" )