Source code for gammapy.catalog.gammacat

# Licensed under a 3-clause BSD style license - see LICENSE.rst
"""Gammacat open TeV source catalog.

https://github.com/gammapy/gamma-cat
"""
import collections
import functools
import json
import logging
import numpy as np
from astropy import units as u
from astropy.table import Table
from gammapy.modeling.models import (
    ExpCutoffPowerLawSpectralModel,
    GaussianSpatialModel,
    PointSpatialModel,
    PowerLaw2SpectralModel,
    PowerLawSpectralModel,
    ShellSpatialModel,
    SkyModel,
    SkyModels,
)
from gammapy.spectrum import FluxPoints
from gammapy.utils.scripts import make_path
from .core import SourceCatalog, SourceCatalogObject

__all__ = [
    "SourceCatalogGammaCat",
    "SourceCatalogObjectGammaCat",
    "GammaCatDataCollection",
    "GammaCatResource",
    "GammaCatResourceIndex",
]

log = logging.getLogger(__name__)


class NoDataAvailableError(LookupError):
    """Generic error used in Gammapy, when some data isn't available."""


class GammaCatNotFoundError(OSError):
    """The gamma-cat repo is not available."""


[docs]class SourceCatalogObjectGammaCat(SourceCatalogObject): """One object from the gamma-cat source catalog. Catalog is represented by `~gammapy.catalog.SourceCatalogGammaCat`. """ _source_name_key = "common_name" _source_index_key = "catalog_row_index" def __str__(self): return self.info()
[docs] def info(self, info="all"): """Info string. Parameters ---------- info : {'all', 'basic', 'position, 'model'} Comma separated list of options """ if info == "all": info = "basic,position,model" ss = "" ops = info.split(",") if "basic" in ops: ss += self._info_basic() if "position" in ops: ss += self._info_position() if "model" in ops: ss += self._info_morph() ss += self._info_spectral_fit() ss += self._info_spectral_points() return ss
def _info_basic(self): """Print basic info.""" d = self.data ss = "\n*** Basic info ***\n\n" ss += "Catalog row index (zero-based) : {}\n".format(d["catalog_row_index"]) ss += "{:<15s} : {}\n".format("Common name", d["common_name"]) # ss += '{:<15s} : {}\n'.format('Gamma names', d['gamma_names']) # ss += '{:<15s} : {}\n'.format('Fermi names', d['fermi_names']) # ss += '{:<15s} : {}\n'.format('Other names', d['other_names']) def get_nonentry_keys(keys): vals = [d[_].strip() for _ in keys] return ",".join([_ for _ in vals if _ != ""]) keys = ["gamma_names", "fermi_names", "other_names"] other_names = get_nonentry_keys(keys) ss += "{:<15s} : {}\n".format("Other names", other_names) ss += "{:<15s} : {}\n".format("Location", d["where"]) ss += "{:<15s} : {}\n".format("Class", d["classes"]) ss += "\n{:<15s} : {}\n".format("TeVCat ID", d["tevcat_id"]) ss += "{:<15s} : {}\n".format("TeVCat 2 ID", d["tevcat2_id"]) ss += "{:<15s} : {}\n".format("TeVCat name", d["tevcat_name"]) ss += "\n{:<15s} : {}\n".format("TGeVCat ID", d["tgevcat_id"]) ss += "{:<15s} : {}\n".format("TGeVCat name", d["tgevcat_name"]) ss += "\n{:<15s} : {}\n".format("Discoverer", d["discoverer"]) ss += "{:<15s} : {}\n".format("Discovery date", d["discovery_date"]) ss += "{:<15s} : {}\n".format("Seen by", d["seen_by"]) ss += "{:<15s} : {}\n".format("Reference", d["reference_id"]) return ss def _info_position(self): """Print position info.""" d = self.data ss = "\n*** Position info ***\n\n" ss += "SIMBAD:\n" ss += "{:<20s} : {:.3f}\n".format("RA", d["ra"]) ss += "{:<20s} : {:.3f}\n".format("DEC", d["dec"]) ss += "{:<20s} : {:.3f}\n".format("GLON", d["glon"]) ss += "{:<20s} : {:.3f}\n".format("GLAT", d["glat"]) ss += "\nMeasurement:\n" ss += "{:<20s} : {:.3f}\n".format("RA", d["pos_ra"]) ss += "{:<20s} : {:.3f}\n".format("DEC", d["pos_dec"]) ss += "{:<20s} : {:.3f}\n".format("GLON", d["pos_glon"]) ss += "{:<20s} : {:.3f}\n".format("GLAT", d["pos_glat"]) ss += "{:<20s} : {:.3f}\n".format("Position error", d["pos_err"]) return ss def _info_morph(self): """Print morphology info.""" ss = "\n*** Morphology info ***\n\n" d = self.data ss += "{:<25s} : {}\n".format("Morphology model type", d["morph_type"]) # TODO: change to morphology model dependent printout # (see spectra printout and `spatial_model` property) ss += "{:<25s} : {:.3f}\n".format("Sigma", d["morph_sigma"]) ss += "{:<25s} : {:.3f}\n".format("Sigma error", d["morph_sigma_err"]) ss += "{:<25s} : {:.3f}\n".format("Sigma2", d["morph_sigma2"]) ss += "{:<25s} : {:.3f}\n".format("Sigma2 error", d["morph_sigma2_err"]) ss += "{:<25s} : {:.3f}\n".format("Position angle", d["morph_pa"]) ss += "{:<25s} : {:.3f}\n".format("Position angle error", d["morph_pa_err"]) ss += "{:<25s} : {}\n".format("Position angle frame", d["morph_pa_frame"]) return ss def _info_spectral_fit(self): """Print spectral info.""" d = self.data ss = "\n*** Spectral info ***\n\n" ss += "{:<15s} : {:.3f}\n".format("Significance", d["significance"]) ss += "{:<15s} : {:.3f}\n".format("Livetime", d["livetime"]) spec_type = d["spec_type"] ss += "\n{:<15s} : {}\n".format("Spectrum type", spec_type) # Spectral model parameters if spec_type == "pl": ss += "{:<15s} : {:.3} +- {:.3} (stat) +- {:.3} (sys) {}\n".format( "norm", d["spec_pl_norm"].value, d["spec_pl_norm_err"].value, d["spec_pl_norm_err_sys"].value, "cm-2 s-1 TeV-1", ) ss += "{:<15s} : {:.3} +- {:.3} (stat) +- {:.3} (sys)\n".format( "index", d["spec_pl_index"], d["spec_pl_index_err"], d["spec_pl_index_err_sys"], ) ss += "{:<15s} : {:.3}\n".format("reference", d["spec_pl_e_ref"]) elif spec_type == "pl2": ss += "{:<15s} : {:.3} +- {:.3} (stat) +- {:.3} (sys) {}\n".format( "flux", d["spec_pl2_flux"].value, d["spec_pl2_flux_err"].value, d["spec_pl2_flux_err_sys"].value, "cm-2 s-1", ) ss += "{:<15s} : {:.3} +- {:.3} (stat) +- {:.3} (sys)\n".format( "index", d["spec_pl2_index"], d["spec_pl2_index_err"], d["spec_pl2_index_err_sys"], ) ss += "{:<15s} : {:.3}\n".format("e_min", d["spec_pl2_e_min"]) ss += "{:<15s} : {:.3}\n".format("e_max", d["spec_pl2_e_max"]) elif spec_type == "ecpl": ss += "{:<15s} : {:.3g} +- {:.3g} (stat) +- {:.03g} (sys) {}\n".format( "norm", d["spec_ecpl_norm"].value, d["spec_ecpl_norm_err"].value, d["spec_ecpl_norm_err_sys"].value, "cm-2 s-1 TeV-1", ) ss += "{:<15s} : {:.3} +- {:.3} (stat) +- {:.3} (sys)\n".format( "index", d["spec_ecpl_index"], d["spec_ecpl_index_err"], d["spec_ecpl_index_err_sys"], ) ss += "{:<15s} : {:.3} +- {:.3} (stat) +- {:.3} (stat) {}\n".format( "e_cut", d["spec_ecpl_e_cut"].value, d["spec_ecpl_e_cut_err"].value, d["spec_ecpl_e_cut_err_sys"].value, "TeV", ) ss += "{:<15s} : {:.3}\n".format("reference", d["spec_ecpl_e_ref"]) else: # raise ValueError('Spectral model printout not implemented: {}'.format(spec)) ss += "\nSpectral model printout not yet implemented.\n" ss += "\n{:<20s} : ({:.3}, {:.3}) TeV\n".format( "Energy range", d["spec_erange_min"].value, d["spec_erange_max"].value ) ss += "{:<20s} : {:.3}\n".format("theta", d["spec_theta"]) ss += "\n\nDerived fluxes:\n" ss += "{:<30s} : {:.3} +- {:.3} (stat) {}\n".format( "Spectral model norm (1 TeV)", d["spec_dnde_1TeV"].value, d["spec_dnde_1TeV_err"].value, "cm-2 s-1 TeV-1", ) ss += "{:<30s} : {:.3} +- {:.3} (stat) {}\n".format( "Integrated flux (>1 TeV)", d["spec_flux_1TeV"].value, d["spec_flux_1TeV_err"].value, "cm-2 s-1", ) ss += "{:<30s} : {:.3f} +- {:.3f} {}\n".format( "Integrated flux (>1 TeV)", d["spec_flux_1TeV_crab"], d["spec_flux_1TeV_crab_err"], "(% Crab)", ) ss += "{:<30s} : {:.3} +- {:.3} (stat) {}\n".format( "Integrated flux (1-10 TeV)", d["spec_eflux_1TeV_10TeV"].value, d["spec_eflux_1TeV_10TeV_err"].value, "erg cm-2 s-1", ) return ss def _info_spectral_points(self): """Print spectral points info.""" d = self.data ss = "\n*** Spectral points ***\n\n" ss += "{:<25s} : {}\n".format("SED reference id", d["sed_reference_id"]) ss += "{:<25s} : {}\n".format("Number of spectral points", d["sed_n_points"]) ss += "{:<25s} : {}\n\n".format("Number of upper limits", d["sed_n_ul"]) try: lines = self.flux_points.table_formatted.pformat(max_width=-1, max_lines=-1) ss += "\n".join(lines) except NoDataAvailableError: ss += "\nNo spectral points available for this source." return ss + "\n" @property def spectral_model(self): """Source spectral model (`~gammapy.modeling.models.SpectralModel`). Parameter errors are statistical errors only. """ data = self.data spec_type = data["spec_type"] pars, errs = {}, {} if spec_type == "pl": model_class = PowerLawSpectralModel pars["amplitude"] = data["spec_pl_norm"] errs["amplitude"] = data["spec_pl_norm_err"] pars["index"] = data["spec_pl_index"] errs["index"] = data["spec_pl_index_err"] pars["reference"] = data["spec_pl_e_ref"] elif spec_type == "pl2": model_class = PowerLaw2SpectralModel pars["amplitude"] = data["spec_pl2_flux"] errs["amplitude"] = data["spec_pl2_flux_err"] pars["index"] = data["spec_pl2_index"] errs["index"] = data["spec_pl2_index_err"] pars["emin"] = data["spec_pl2_e_min"] e_max = data["spec_pl2_e_max"] DEFAULT_E_MAX = u.Quantity(1e5, "TeV") if np.isnan(e_max.value): e_max = DEFAULT_E_MAX pars["emax"] = e_max elif spec_type == "ecpl": model_class = ExpCutoffPowerLawSpectralModel pars["amplitude"] = data["spec_ecpl_norm"] errs["amplitude"] = data["spec_ecpl_norm_err"] pars["index"] = data["spec_ecpl_index"] errs["index"] = data["spec_ecpl_index_err"] pars["lambda_"] = 1.0 / data["spec_ecpl_e_cut"] errs["lambda_"] = data["spec_ecpl_e_cut_err"] / data["spec_ecpl_e_cut"] ** 2 pars["reference"] = data["spec_ecpl_e_ref"] else: raise ValueError(f"Invalid spec_type: {spec_type}") model = model_class(**pars) model.parameters.set_parameter_errors(errs) return model @property def spatial_model(self): """Source spatial model (`~gammapy.modeling.models.SpatialModel`). TODO: add parameter errors! """ d = self.data morph_type = d["morph_type"] glon = d["glon"] glat = d["glat"] if morph_type == "point": return PointSpatialModel(lon_0=glon, lat_0=glat, frame="galactic") elif morph_type == "gauss": # TODO: add infos back once model support elongation # pars['x_stddev'] = d['morph_sigma'] # pars['y_stddev'] = d['morph_sigma'] # if not np.isnan(d['morph_sigma2']): # pars['y_stddev'] = d['morph_sigma2'] # if not np.isnan(d['morph_pa']): # # TODO: handle reference frame for rotation angle # pars['theta'] = Angle(d['morph_pa'], 'deg').rad return GaussianSpatialModel( lon_0=glon, lat_0=glat, sigma=d["morph_sigma"], frame="galactic" ) elif morph_type == "shell": return ShellSpatialModel( lon_0=glon, lat_0=glat, # TODO: probably we shouldn't guess a shell width here! radius=0.8 * d["morph_sigma"], width=0.2 * d["morph_sigma"], frame="galactic", ) elif morph_type == "none": raise NoDataAvailableError(f"No spatial model available: {self.name}") else: raise NotImplementedError(f"Unknown spatial model: {morph_type!r}") @property def sky_model(self): """Source sky model (`~gammapy.modeling.models.SkyModel`).""" spatial_model = self.spatial_model spectral_model = self.spectral_model return SkyModel(spatial_model, spectral_model, name=self.name) def _add_source_meta(self, table): """Copy over some info to table.meta""" d = self.data m = table.meta m["origin"] = "Data from gamma-cat" m["source_id"] = d["source_id"] m["common_name"] = d["common_name"] m["reference_id"] = d["reference_id"] @property def flux_points(self): """Differential flux points (`~gammapy.spectrum.FluxPoints`).""" d = self.data table = Table() table.meta["SED_TYPE"] = "dnde" self._add_source_meta(table) valid = np.isfinite(d["sed_e_ref"].value) if valid.sum() == 0: raise NoDataAvailableError(f"No flux points available: {self.name}") table["e_ref"] = d["sed_e_ref"] table["e_min"] = d["sed_e_min"] table["e_max"] = d["sed_e_max"] table["dnde"] = d["sed_dnde"] table["dnde_err"] = d["sed_dnde_err"] table["dnde_errn"] = d["sed_dnde_errn"] table["dnde_errp"] = d["sed_dnde_errp"] table["dnde_ul"] = d["sed_dnde_ul"] # Only keep rows that actually contain information table = table[valid] # Only keep columns that actually contain information def _del_nan_col(table, colname): if np.isfinite(table[colname]).sum() == 0: del table[colname] for colname in table.colnames: _del_nan_col(table, colname) return FluxPoints(table) @property def is_pointlike(self): """ Source is pointlike. """ return self.data["morph_type"] == "point"
[docs]class SourceCatalogGammaCat(SourceCatalog): """Gammacat open TeV source catalog. See: https://github.com/gammapy/gamma-cat One source is represented by `~gammapy.catalog.SourceCatalogObjectGammaCat`. Parameters ---------- filename : str Path to the gamma-cat fits file. Examples -------- Load the catalog data: >>> from gammapy.catalog import SourceCatalogGammaCat >>> cat = SourceCatalogGammaCat() Access a source by name: >>> source = cat['Vela Junior'] Access source spectral data and plot it: >>> source.spectral_model.plot() >>> source.spectral_model.plot_error() >>> source.flux_points.plot() """ name = "gamma-cat" description = "An open catalog of gamma-ray sources" source_object_class = SourceCatalogObjectGammaCat def __init__(self, filename="$GAMMAPY_DATA/catalogs/gammacat/gammacat.fits.gz"): filename = str(make_path(filename)) table = Table.read(filename, hdu=1) self.filename = filename source_name_key = "common_name" source_name_alias = ("other_names", "gamma_names") super().__init__( table=table, source_name_key=source_name_key, source_name_alias=source_name_alias, )
[docs] def to_sky_models(self): """Convert to a `~gammapy.modeling.models.SkyModels`. TODO: add an option whether to skip or raise on missing models or data. """ source_list = [] for source_idx in range(len(self.table)): source = self[source_idx] try: source_list.append(source.sky_model) except NoDataAvailableError: log.warning( f"Skipping source {source.name} (missing data in gamma-cat)" ) continue return SkyModels(source_list)
[docs]class GammaCatDataCollection: """Data store for gamma-cat. Gives access to all data from https://github.com/gammapy/gamma-cat . Holds a `GammaCatResourceIndex` to locate resources, but also more info about gamma-cat, as well as methods to create Gammapy objects (spectral models, flux points, lightcurves) from the datasets. """ def __init__(self, data_index): self.data_index = data_index
[docs] @classmethod def from_index_file( cls, filename="$GAMMAPY_DATA/catalogs/gammacat/gammacat-datasets.json" ): """Create from index file.""" path = make_path(filename) # TODO: make a list of `GammaCatResource`, as well as a dict by ``resource_id`` for lookup! data_index = json.load(path.read_text()) return cls(data_index=data_index)
def __str__(self): ss = "version = {}".format(self.data_index["info"]["version"]) return ss
[docs]@functools.total_ordering class GammaCatResource: """Reference for a single resource in gamma-cat. This can be considered an implementation detail, used to assign ``global_id`` and to load resources. TODO: explain how ``global_id``, ``type`` and ``location`` work. Uses the Python ``hash`` function on the tuple ``(source_id, reference_id, file_id)`` Parameters ---------- source_id : int Gamma-cat source ID reference_id : str Gamma-cat reference ID (usually the ADS paper bibcode) file_id : int File ID (a counter for cases with multiple measurements per reference / source) (use integer -1 if missing) type : str Resource type (use string 'none' if missing) location : str Resource location (use string 'none' if missing) Examples -------- >>> from gammapy.catalog.gammacat import GammaCatResource >>> resource = GammaCatResource(source_id=42, reference_id='2010A&A...516A..62A', file_id=2) >>> resource GammaCatResource(source_id=42, reference_id='2010A&A...516A..62A', file_id=2, type='none', location='none') """ _NA_FILL = dict(file_id=-1, type="none", location="none") def __init__( self, source_id, reference_id, file_id=-1, type="none", location="none" ): self.source_id = int(source_id) self.reference_id = str(reference_id) self.file_id = int(file_id) self.type = str(type) self.location = str(location) @property def global_id(self): """Globally unique (within gamma-cat) resource ID (str). (see class docstring for explanation and example). """ return "|".join( (str(self.source_id), self.reference_id, str(self.file_id), self.type) ) def __repr__(self): return ( f"{self.__class__.__name__}(" f"source_id={self.source_id!r}, " f"reference_id={self.reference_id!r}, " f"file_id={self.file_id!r}, " f"type={self.type!r}, " f"location={self.location!r})" ) def __eq__(self, other): return self.to_namedtuple() == other.to_namedtuple() def __lt__(self, other): return self.to_namedtuple() < other.to_namedtuple()
[docs] def to_namedtuple(self): """Convert to `collections.namedtuple`.""" d = self.to_dict() return collections.namedtuple("GammaCatResourceNamedTuple", d.keys())(**d)
[docs] def to_dict(self): """Convert to `dict`.""" return { "source_id": self.source_id, "reference_id": self.reference_id, "file_id": self.file_id, "type": self.type, "location": self.location, }
[docs] @classmethod def from_dict(cls, data): """Create from dict.""" return cls( source_id=data["source_id"], reference_id=data["reference_id"], file_id=data.get("file_id", cls._NA_FILL["file_id"]), type=data.get("type", cls._NA_FILL["type"]), location=data.get("location", cls._NA_FILL["location"]), )
[docs]class GammaCatResourceIndex: """Resource index for gamma-cat. Parameters ---------- resources : list List of `GammaCatResource` objects """ def __init__(self, resources): self.resources = resources def __repr__(self): return f"{self.__class__.__name__}(n_resources={len(self.resources)})" def __eq__(self, other): if len(self.resources) != len(other.resources): return False return all(a == b for (a, b) in zip(self.resources, other.resources)) @property def unique_source_ids(self): """Sorted list of unique source IDs (list of int).""" return sorted({resource.source_id for resource in self.resources}) @property def unique_reference_ids(self): """Sorted list of unique source IDs (list of str).""" return sorted({resource.reference_id for resource in self.resources}) @property def global_ids(self): """List of global resource IDs (list of str). In original order, not sorted. """ return [resource.global_id for resource in self.resources]
[docs] def sort(self): """Return a sorted copy (leave self unchanged).""" return self.__class__(sorted(self.resources))
[docs] def to_list(self): """Convert to list of dict.""" return [resource.to_dict() for resource in self.resources]
[docs] @classmethod def from_list(cls, data): """Create from list of dicts.""" return cls([GammaCatResource.from_dict(_) for _ in data])
[docs] def to_table(self): """Convert to `~astropy.table.Table`.""" rows = self.to_list() return Table(rows=rows, names=list(rows[0].keys()))
[docs] @classmethod def from_table(cls, table): """Create from `~astropy.table.Table`.""" resources = [] for row in table: data = {k: row[k] for k in table.colnames} resources.append(GammaCatResource.from_dict(data)) return cls(resources=resources)
[docs] def to_pandas(self): """Convert to `pandas.DataFrame`.""" # This is inefficient. Could implement direct conversion if needed. table = self.to_table() return table.to_pandas()
[docs] @classmethod def from_pandas(cls, dataframe): """Create from `pandas.DataFrame`.""" table = Table.from_pandas(dataframe) return cls.from_table(table)
[docs] def query(self, *args, **kwargs): """Query to select subset of resources. Calls `pandas.DataFrame.query` and passes arguments to that method. Examples -------- >>> resource_index = GammaCatResourceIndex(...) >>> resource_index2 = resource_index.query('type == "sed" and source_id == 42') """ df = self.to_pandas() df2 = df.query(*args, **kwargs) return self.from_pandas(df2)