# Licensed under a 3-clause BSD style license - see LICENSE.rst
"""Fermi catalog and source classes."""
import warnings
import numpy as np
import astropy.units as u
from astropy.table import Column, Table
from astropy.time import Time
from gammapy.maps import Map
from gammapy.modeling.models import (
DiskSpatialModel,
ExpCutoffPowerLaw3FGLSpectralModel,
GaussianSpatialModel,
LogParabolaSpectralModel,
PointSpatialModel,
PowerLaw2SpectralModel,
PowerLawSpectralModel,
SkyModel,
SuperExpCutoffPowerLaw3FGLSpectralModel,
SuperExpCutoffPowerLaw4FGLSpectralModel,
TemplateSpatialModel,
)
from gammapy.spectrum import FluxPoints
from gammapy.time import LightCurve
from gammapy.utils.scripts import make_path
from gammapy.utils.table import table_standardise_units_inplace
from .core import SourceCatalog, SourceCatalogObject
__all__ = [
"SourceCatalogObject4FGL",
"SourceCatalogObject3FGL",
"SourceCatalogObject1FHL",
"SourceCatalogObject2FHL",
"SourceCatalogObject3FHL",
"SourceCatalog4FGL",
"SourceCatalog3FGL",
"SourceCatalog1FHL",
"SourceCatalog2FHL",
"SourceCatalog3FHL",
]
def compute_flux_points_ul(quantity, quantity_errp):
"""Compute UL value for fermi flux points.
See https://arxiv.org/pdf/1501.02003.pdf (page 30)
"""
return 2 * quantity_errp + quantity
[docs]class SourceCatalogObject4FGL(SourceCatalogObject):
"""One source from the Fermi-LAT 4FGL catalog.
Catalog is represented by `~gammapy.catalog.SourceCatalog4FGL`.
"""
@property
def spectral_model(self):
"""Best fit spectral model (`~gammapy.modeling.models.SpectralModel`)."""
spec_type = self.data["SpectrumType"].strip()
pars, errs = {}, {}
pars["reference"] = self.data["Pivot_Energy"]
if spec_type == "PowerLaw":
pars["amplitude"] = self.data["PL_Flux_Density"]
pars["index"] = self.data["PL_Index"]
errs["amplitude"] = self.data["Unc_PL_Flux_Density"]
errs["index"] = self.data["Unc_PL_Index"]
model = PowerLawSpectralModel(**pars)
elif spec_type == "LogParabola":
pars["amplitude"] = self.data["LP_Flux_Density"]
pars["alpha"] = self.data["LP_Index"]
pars["beta"] = self.data["LP_beta"]
errs["amplitude"] = self.data["Unc_LP_Flux_Density"]
errs["alpha"] = self.data["Unc_LP_Index"]
errs["beta"] = self.data["Unc_LP_beta"]
model = LogParabolaSpectralModel(**pars)
elif spec_type == "PLSuperExpCutoff":
pars["amplitude"] = self.data["PLEC_Flux_Density"]
pars["index_1"] = self.data["PLEC_Index"]
pars["index_2"] = self.data["PLEC_Exp_Index"]
pars["expfactor"] = self.data["PLEC_Expfactor"]
errs["amplitude"] = self.data["Unc_PLEC_Flux_Density"]
errs["index_1"] = self.data["Unc_PLEC_Index"]
errs["index_2"] = np.nan_to_num(self.data["Unc_PLEC_Exp_Index"])
errs["expfactor"] = self.data["Unc_PLEC_Expfactor"]
model = SuperExpCutoffPowerLaw4FGLSpectralModel(**pars)
else:
raise ValueError(f"Invalid spec_type: {spec_type!r}")
model.parameters.set_parameter_errors(errs)
return model
[docs]class SourceCatalogObject3FGL(SourceCatalogObject):
"""One source from the Fermi-LAT 3FGL catalog.
Catalog is represented by `~gammapy.catalog.SourceCatalog3FGL`.
"""
_ebounds = u.Quantity([100, 300, 1000, 3000, 10000, 100000], "MeV")
_ebounds_suffix = ["100_300", "300_1000", "1000_3000", "3000_10000", "10000_100000"]
energy_range = u.Quantity([100, 100000], "MeV")
"""Energy range of the catalog.
Paper says that analysis uses data up to 300 GeV,
but results are all quoted up to 100 GeV only to
be consistent with previous catalogs.
"""
def __str__(self):
return self.info()
[docs] def info(self, info="all"):
"""Summary info string.
Parameters
----------
info : {'all', 'basic', 'position', 'spectral', 'lightcurve'}
Comma separated list of options
"""
if info == "all":
info = "basic,position,spectral,lightcurve"
ss = ""
ops = info.split(",")
if "basic" in ops:
ss += self._info_basic()
if "position" in ops:
ss += self._info_position()
if "spectral" in ops:
ss += self._info_spectral_fit()
ss += self._info_spectral_points()
if "lightcurve" in ops:
ss += self._info_lightcurve()
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 += "{:<20s} : {}\n".format("Source name", d["Source_Name"])
ss += "{:<20s} : {}\n".format("Extended name", d["Extended_Source_Name"])
def get_nonentry_keys(keys):
vals = [d[_].strip() for _ in keys]
return ", ".join([_ for _ in vals if _ != ""])
keys = [
"ASSOC1",
"ASSOC2",
"ASSOC_TEV",
"ASSOC_GAM1",
"ASSOC_GAM2",
"ASSOC_GAM3",
]
associations = get_nonentry_keys(keys)
ss += "{:<20s} : {}\n".format("Associations", associations)
keys = ["0FGL_Name", "1FGL_Name", "2FGL_Name", "1FHL_Name"]
other_names = get_nonentry_keys(keys)
ss += "{:<20s} : {}\n".format("Other names", other_names)
ss += "{:<20s} : {}\n".format("Class", d["CLASS1"])
tevcat_flag = d["TEVCAT_FLAG"]
if tevcat_flag == "N":
tevcat_message = "No TeV association"
elif tevcat_flag == "P":
tevcat_message = "Small TeV source"
elif tevcat_flag == "E":
tevcat_message = "Extended TeV source (diameter > 40 arcmins)"
else:
tevcat_message = "N/A"
ss += "{:<20s} : {}\n".format("TeVCat flag", tevcat_message)
flag_message = {
0: "None",
1: "Source with TS > 35 which went to TS < 25 when changing the diffuse model. Note that sources with TS < "
"35 are not flagged with this bit because normal statistical fluctuations can push them to TS < 25.",
3: "Flux (> 1 GeV) or energy flux (> 100 MeV) changed by more than 3 sigma when changing the diffuse model."
" Requires also that the flux change by more than 35% (to not flag strong sources).",
4: "Source-to-background ratio less than 10% in highest band in which TS > 25. Background is integrated "
"over the 68%-confidence area (pi*r_682) or 1 square degree, whichever is smaller.",
5: "Closer than theta_ref from a brighter neighbor, where theta_ref is defined in the highest band in which"
" source TS > 25, or the band with highest TS if all are < 25. theta_ref is set to 2.17 degrees (FWHM)"
" below 300 MeV, 1.38 degrees between 300 MeV and 1 GeV, 0.87 degrees between 1 GeV and 3 GeV, 0.67"
" degrees between 3 and 10 GeV and 0.45 degrees about 10 GeV (2*r_68).",
6: "On top of an interstellar gas clump or small-scale defect in the model of diffuse emission. This flag "
'is equivalent to the "c" suffix in the source name.',
7: "Unstable position determination; result from gtfindsrc outside the 95% ellipse from pointlike.",
9: "Localization Quality > 8 in pointlike (see Section 3.1 in catalog paper) or long axis of 95% ellipse >"
" 0.25.",
10: "Spectral Fit Quality > 16.3 (see Equation 3 in 2FGL catalog paper).",
11: "Possibly due to the Sun (see Section 3.6 in catalog paper).",
12: "Highly curved spectrum; LogParabolaSpectralModel beta fixed to 1 or PLExpCutoff Spectral Index fixed to 0 (see "
"Section 3.3 in catalog paper).",
}
ss += "{:<20s} : {}\n".format(
"Other flags", flag_message.get(d["Flags"], "N/A")
)
return ss
def _info_position(self):
"""Print position info."""
d = self.data
ss = "\n*** Position info ***\n\n"
ss += "{:<20s} : {:.3f}\n".format("RA", d["RAJ2000"])
ss += "{:<20s} : {:.3f}\n".format("DEC", d["DEJ2000"])
ss += "{:<20s} : {:.3f}\n".format("GLON", d["GLON"])
ss += "{:<20s} : {:.3f}\n".format("GLAT", d["GLAT"])
ss += "\n"
ss += "{:<20s} : {:.4f}\n".format("Semimajor (68%)", d["Conf_68_SemiMajor"])
ss += "{:<20s} : {:.4f}\n".format("Semiminor (68%)", d["Conf_68_SemiMinor"])
ss += "{:<20s} : {:.2f}\n".format("Position angle (68%)", d["Conf_68_PosAng"])
ss += "{:<20s} : {:.4f}\n".format("Semimajor (95%)", d["Conf_95_SemiMajor"])
ss += "{:<20s} : {:.4f}\n".format("Semiminor (95%)", d["Conf_95_SemiMinor"])
ss += "{:<20s} : {:.2f}\n".format("Position angle (95%)", d["Conf_95_PosAng"])
ss += "{:<20s} : {:.0f}\n".format("ROI number", d["ROI_num"])
return ss
def _info_spectral_fit(self):
"""Print spectral info."""
d = self.data
spec_type = d["SpectrumType"].strip()
ss = "\n*** Spectral info ***\n\n"
ss += "{:<45s} : {}\n".format("Spectrum type", d["SpectrumType"])
fmt = "{:<45s} : {:.3f}\n"
ss += fmt.format("Detection significance (100 MeV - 300 GeV)", d["Signif_Avg"])
ss += "{:<45s} : {:.1f}\n".format("Significance curvature", d["Signif_Curve"])
if spec_type == "PowerLaw":
pass
elif spec_type == "LogParabola":
ss += "{:<45s} : {} +- {}\n".format("beta", d["beta"], d["Unc_beta"])
elif spec_type in ["PLExpCutoff", "PlSuperExpCutoff"]:
fmt = "{:<45s} : {:.0f} +- {:.0f} {}\n"
ss += fmt.format(
"Cutoff energy",
d["Cutoff"].value,
d["Unc_Cutoff"].value,
d["Cutoff"].unit,
)
elif spec_type == "PLSuperExpCutoff":
ss += "{:<45s} : {} +- {}\n".format(
"Super-exponential cutoff index", d["Exp_Index"], d["Unc_Exp_Index"]
)
else:
raise ValueError(f"Invalid spec_type: {spec_type!r}")
ss += "{:<45s} : {:.0f} {}\n".format(
"Pivot energy", d["Pivot_Energy"].value, d["Pivot_Energy"].unit
)
ss += "{:<45s} : {:.3f}\n".format(
"Power law spectral index", d["PowerLaw_Index"]
)
fmt = "{:<45s} : {:.3f} +- {:.3f}\n"
ss += fmt.format("Spectral index", d["Spectral_Index"], d["Unc_Spectral_Index"])
fmt = "{:<45s} : {:.3} +- {:.3} {}\n"
ss += fmt.format(
"Flux Density at pivot energy",
d["Flux_Density"].value,
d["Unc_Flux_Density"].value,
"cm-2 MeV-1 s-1",
)
fmt = "{:<45s} : {:.3} +- {:.3} {}\n"
ss += fmt.format(
"Integral flux (1 - 100 GeV)",
d["Flux1000"].value,
d["Unc_Flux1000"].value,
"cm-2 s-1",
)
fmt = "{:<45s} : {:.3} +- {:.3} {}\n"
ss += fmt.format(
"Energy flux (100 MeV - 100 GeV)",
d["Energy_Flux100"].value,
d["Unc_Energy_Flux100"].value,
"erg cm-2 s-1",
)
return ss
def _info_spectral_points(self):
"""Print spectral points."""
ss = "\n*** Spectral points ***\n\n"
lines = self.flux_points.table_formatted.pformat(max_width=-1, max_lines=-1)
ss += "\n".join(lines)
return ss + "\n"
def _info_lightcurve(self):
"""Print lightcurve info."""
d = self.data
ss = "\n*** Lightcurve info ***\n\n"
ss += "Lightcurve measured in the energy band: 100 MeV - 100 GeV\n\n"
ss += "{:<15s} : {:.3f}\n".format("Variability index", d["Variability_Index"])
if d["Signif_Peak"] == np.nan:
ss += "{:<40s} : {:.3f}\n".format(
"Significance peak (100 MeV - 100 GeV)", d["Signif_Peak"]
)
fmt = "{:<40s} : {:.3} +- {:.3} cm^-2 s^-1\n"
ss += fmt.format(
"Integral flux peak (100 MeV - 100 GeV)",
d["Flux_Peak"],
d["Unc_Flux_Peak"],
)
# TODO: give time as UTC string, not MET
ss += "{:<40s} : {:.3} s (Mission elapsed time)\n".format(
"Time peak", d["Time_Peak"]
)
peak_interval = d["Peak_Interval"].to_value("day")
ss += "{:<40s} : {:.3} day\n".format("Peak interval", peak_interval)
else:
ss += "\nNo peak measured for this source.\n"
# TODO: Add a lightcurve table with d['Flux_History'] and d['Unc_Flux_History']
return ss
@property
def spectral_model(self):
"""Best fit spectral model (`~gammapy.modeling.models.SpectralModel`)."""
spec_type = self.data["SpectrumType"].strip()
pars, errs = {}, {}
pars["amplitude"] = self.data["Flux_Density"]
errs["amplitude"] = self.data["Unc_Flux_Density"]
pars["reference"] = self.data["Pivot_Energy"]
if spec_type == "PowerLaw":
pars["index"] = self.data["Spectral_Index"]
errs["index"] = self.data["Unc_Spectral_Index"]
model = PowerLawSpectralModel(**pars)
elif spec_type == "PLExpCutoff":
pars["index"] = self.data["Spectral_Index"]
pars["ecut"] = self.data["Cutoff"]
errs["index"] = self.data["Unc_Spectral_Index"]
errs["ecut"] = self.data["Unc_Cutoff"]
model = ExpCutoffPowerLaw3FGLSpectralModel(**pars)
elif spec_type == "LogParabola":
pars["alpha"] = self.data["Spectral_Index"]
pars["beta"] = self.data["beta"]
errs["alpha"] = self.data["Unc_Spectral_Index"]
errs["beta"] = self.data["Unc_beta"]
model = LogParabolaSpectralModel(**pars)
elif spec_type == "PLSuperExpCutoff":
# TODO: why convert to GeV here? Remove?
pars["reference"] = pars["reference"].to("GeV")
pars["index_1"] = self.data["Spectral_Index"]
pars["index_2"] = self.data["Exp_Index"]
pars["ecut"] = self.data["Cutoff"].to("GeV")
errs["index_1"] = self.data["Unc_Spectral_Index"]
errs["index_2"] = self.data["Unc_Exp_Index"]
errs["ecut"] = self.data["Unc_Cutoff"].to("GeV")
model = SuperExpCutoffPowerLaw3FGLSpectralModel(**pars)
else:
raise ValueError(f"Invalid spec_type: {spec_type!r}")
model.parameters.set_parameter_errors(errs)
return model
@property
def spatial_model(self):
"""
Source spatial model (`~gammapy.modeling.models.SpatialModel`).
"""
d = self.data
pars = {}
glon = d["GLON"]
glat = d["GLAT"]
if self.is_pointlike:
pars["lon_0"] = glon
pars["lat_0"] = glat
return PointSpatialModel(lon_0=glon, lat_0=glat, frame="galactic")
else:
de = self.data_extended
morph_type = de["Model_Form"].strip()
if morph_type == "Disk":
r_0 = de["Model_SemiMajor"].to("deg")
return DiskSpatialModel(
lon_0=glon, lat_0=glat, r_0=r_0, frame="galactic"
)
elif morph_type in ["Map", "Ring", "2D Gaussian x2"]:
filename = de["Spatial_Filename"].strip()
path = make_path(
"$GAMMAPY_DATA/catalogs/fermi/Extended_archive_v15/Templates/"
)
return TemplateSpatialModel.read(path / filename)
elif morph_type == "2D Gaussian":
# TODO: fill elongation info as soon as model supports it
sigma = de["Model_SemiMajor"].to("deg")
return GaussianSpatialModel(
lon_0=glon, lat_0=glat, sigma=sigma, frame="galactic"
)
else:
raise ValueError(f"Invalid 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)
@property
def is_pointlike(self):
return self.data["Extended_Source_Name"].strip() == ""
@property
def flux_points(self):
"""Flux points (`~gammapy.spectrum.FluxPoints`)."""
table = Table()
table.meta["SED_TYPE"] = "flux"
table["e_min"] = self._ebounds[:-1]
table["e_max"] = self._ebounds[1:]
flux = self._get_flux_values("Flux")
flux_err = self._get_flux_values("Unc_Flux")
table["flux"] = flux
table["flux_errn"] = np.abs(flux_err[:, 0])
table["flux_errp"] = flux_err[:, 1]
nuFnu = self._get_flux_values("nuFnu", "erg cm-2 s-1")
table["e2dnde"] = nuFnu
table["e2dnde_errn"] = np.abs(nuFnu * flux_err[:, 0] / flux)
table["e2dnde_errp"] = nuFnu * flux_err[:, 1] / flux
is_ul = np.isnan(table["flux_errn"])
table["is_ul"] = is_ul
# handle upper limits
table["flux_ul"] = np.nan * flux_err.unit
flux_ul = compute_flux_points_ul(table["flux"], table["flux_errp"])
table["flux_ul"][is_ul] = flux_ul[is_ul]
# handle upper limits
table["e2dnde_ul"] = np.nan * nuFnu.unit
e2dnde_ul = compute_flux_points_ul(table["e2dnde"], table["e2dnde_errp"])
table["e2dnde_ul"][is_ul] = e2dnde_ul[is_ul]
# Square root of test statistic
table["sqrt_TS"] = [self.data["Sqrt_TS" + _] for _ in self._ebounds_suffix]
return FluxPoints(table)
def _get_flux_values(self, prefix, unit="cm-2 s-1"):
values = [self.data[prefix + _] for _ in self._ebounds_suffix]
return u.Quantity(values, unit)
@property
def lightcurve(self):
"""Lightcurve (`~gammapy.time.LightCurve`).
Examples
--------
>>> from gammapy.catalog import source_catalogs
>>> source = source_catalogs['3fgl']['3FGL J0349.9-2102']
>>> lc = source.lightcurve
>>> lc.plot()
"""
flux = self.data["Flux_History"]
# Flux error is given as asymmetric high/low
flux_errn = -self.data["Unc_Flux_History"][:, 0]
flux_errp = self.data["Unc_Flux_History"][:, 1]
# Really the time binning is stored in a separate HDU in the FITS
# catalog file called `Hist_Start`, with a single column `Hist_Start`
# giving the time binning in MET (mission elapsed time)
# This is not available here for now.
# TODO: read that info in `SourceCatalog3FGL` and pass it down to the
# `SourceCatalogObject3FGL` object somehow.
# For now, we just hard-code the start and stop time and assume
# equally-spaced time intervals. This is roughly correct,
# for plotting the difference doesn't matter, only for analysis
time_start = Time("2008-08-02T00:33:19")
time_end = Time("2012-07-31T22:45:47")
n_points = len(flux)
time_step = (time_end - time_start) / n_points
time_bounds = time_start + np.arange(n_points + 1) * time_step
table = Table(
[
Column(time_bounds[:-1].utc.mjd, "time_min"),
Column(time_bounds[1:].utc.mjd, "time_max"),
Column(flux, "flux"),
Column(flux_errp, "flux_errp"),
Column(flux_errn, "flux_errn"),
]
)
return LightCurve(table)
[docs]class SourceCatalogObject1FHL(SourceCatalogObject):
"""One source from the Fermi-LAT 1FHL catalog.
Catalog is represented by `~gammapy.catalog.SourceCatalog1FHL`.
"""
_ebounds = u.Quantity([10, 30, 100, 500], "GeV")
_ebounds_suffix = ["10_30", "30_100", "100_500"]
energy_range = u.Quantity([0.01, 0.5], "TeV")
"""Energy range of the Fermi 1FHL source catalog"""
def __str__(self):
return self.info()
[docs] def info(self):
"""Print summary info."""
# TODO: can we share code with 3FGL summary function?
d = self.data
ss = "Source: {}\n".format(d["Source_Name"])
ss += "\n"
ss += "RA (J2000) : {}\n".format(d["RAJ2000"])
ss += "Dec (J2000) : {}\n".format(d["DEJ2000"])
ss += "GLON : {}\n".format(d["GLON"])
ss += "GLAT : {}\n".format(d["GLAT"])
ss += "\n"
return ss
def _get_flux_values(self, prefix, unit="cm-2 s-1"):
values = [self.data[prefix + _ + "GeV"] for _ in self._ebounds_suffix]
return u.Quantity(values, unit)
@property
def flux_points(self):
"""Integral flux points (`~gammapy.spectrum.FluxPoints`)."""
table = Table()
table.meta["SED_TYPE"] = "flux"
table["e_min"] = self._ebounds[:-1]
table["e_max"] = self._ebounds[1:]
table["flux"] = self._get_flux_values("Flux")
flux_err = self._get_flux_values("Unc_Flux")
table["flux_errn"] = np.abs(flux_err[:, 0])
table["flux_errp"] = flux_err[:, 1]
# handle upper limits
is_ul = np.isnan(table["flux_errn"])
table["is_ul"] = is_ul
table["flux_ul"] = np.nan * flux_err.unit
flux_ul = compute_flux_points_ul(table["flux"], table["flux_errp"])
table["flux_ul"][is_ul] = flux_ul[is_ul]
return FluxPoints(table)
@property
def spectral_model(self):
"""Best fit spectral model `~gammapy.modeling.models.SpectralModel`."""
pars, errs = {}, {}
pars["amplitude"] = self.data["Flux"]
pars["emin"], pars["emax"] = self.energy_range
pars["index"] = self.data["Spectral_Index"]
errs["amplitude"] = self.data["Unc_Flux"]
errs["index"] = self.data["Unc_Spectral_Index"]
model = PowerLaw2SpectralModel(**pars)
model.parameters.set_parameter_errors(errs)
return model
[docs]class SourceCatalogObject2FHL(SourceCatalogObject):
"""One source from the Fermi-LAT 2FHL catalog.
Catalog is represented by `~gammapy.catalog.SourceCatalog2FHL`.
"""
_ebounds = u.Quantity([50, 171, 585, 2000], "GeV")
_ebounds_suffix = ["50_171", "171_585", "585_2000"]
energy_range = u.Quantity([0.05, 2], "TeV")
"""Energy range of the Fermi 2FHL source catalog"""
def __str__(self):
return self.info()
[docs] def info(self):
"""Print summary info."""
# TODO: can we share code with 3FGL summary funtion?
d = self.data
ss = "Source: {}\n".format(d["Source_Name"])
ss += "\n"
ss += "RA (J2000) : {}\n".format(d["RAJ2000"])
ss += "Dec (J2000) : {}\n".format(d["DEJ2000"])
ss += "GLON : {}\n".format(d["GLON"])
ss += "GLAT : {}\n".format(d["GLAT"])
ss += "\n"
# val, err = d['Energy_Flux100'], d['Unc_Energy_Flux100']
# ss += 'Energy flux (100 MeV - 100 GeV) : {} +- {} erg cm^-2 s^-1\n'.format(val, err)
# ss += 'Detection significance : {}\n'.format(d['Signif_Avg'])
return ss
def _get_flux_values(self, prefix, unit="cm-2 s-1"):
values = [self.data[prefix + _ + "GeV"] for _ in self._ebounds_suffix]
return u.Quantity(values, unit)
@property
def flux_points(self):
"""Integral flux points (`~gammapy.spectrum.FluxPoints`)."""
table = Table()
table.meta["SED_TYPE"] = "flux"
table["e_min"] = self._ebounds[:-1]
table["e_max"] = self._ebounds[1:]
table["flux"] = self._get_flux_values("Flux")
flux_err = self._get_flux_values("Unc_Flux")
table["flux_errn"] = np.abs(flux_err[:, 0])
table["flux_errp"] = flux_err[:, 1]
# handle upper limits
is_ul = np.isnan(table["flux_errn"])
table["is_ul"] = is_ul
table["flux_ul"] = np.nan * flux_err.unit
flux_ul = compute_flux_points_ul(table["flux"], table["flux_errp"])
table["flux_ul"][is_ul] = flux_ul[is_ul]
return FluxPoints(table)
@property
def spectral_model(self):
"""Best fit spectral model (`~gammapy.modeling.models.SpectralModel`)."""
pars, errs = {}, {}
pars["amplitude"] = self.data["Flux50"]
pars["emin"], pars["emax"] = self.energy_range
pars["index"] = self.data["Spectral_Index"]
errs["amplitude"] = self.data["Unc_Flux50"]
errs["index"] = self.data["Unc_Spectral_Index"]
model = PowerLaw2SpectralModel(**pars)
model.parameters.set_parameter_errors(errs)
return model
[docs]class SourceCatalogObject3FHL(SourceCatalogObject):
"""One source from the Fermi-LAT 3FHL catalog.
Catalog is represented by `~gammapy.catalog.SourceCatalog3FHL`.
"""
energy_range = u.Quantity([0.01, 2], "TeV")
"""Energy range of the Fermi 1FHL source catalog"""
_ebounds = u.Quantity([10, 20, 50, 150, 500, 2000], "GeV")
def __str__(self):
return self.info()
[docs] def info(self, info="all"):
"""Summary info string.
Parameters
----------
info : {'all', 'basic', 'position', 'spectral'}
Comma separated list of options
"""
if info == "all":
info = "basic,position,spectral,other"
ss = ""
ops = info.split(",")
if "basic" in ops:
ss += self._info_basic()
if "position" in ops:
ss += self._info_position()
if not self.is_pointlike:
ss += self._info_morphology()
if "spectral" in ops:
ss += self._info_spectral_fit()
ss += self._info_spectral_points()
if "other" in ops:
ss += self._info_other()
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 += "{:<20s} : {}\n".format("Source name", d["Source_Name"])
ss += "{:<20s} : {}\n".format("Extended name", d["Extended_Source_Name"])
def get_nonentry_keys(keys):
vals = [d[_].strip() for _ in keys]
return ", ".join([_ for _ in vals if _ != ""])
keys = ["ASSOC1", "ASSOC2", "ASSOC_TEV", "ASSOC_GAM"]
associations = get_nonentry_keys(keys)
ss += "{:<16s} : {}\n".format("Associations", associations)
ss += "{:<16s} : {:.3f}\n".format("ASSOC_PROB_BAY", d["ASSOC_PROB_BAY"])
ss += "{:<16s} : {:.3f}\n".format("ASSOC_PROB_LR", d["ASSOC_PROB_LR"])
ss += "{:<16s} : {}\n".format("Class", d["CLASS"])
tevcat_flag = d["TEVCAT_FLAG"]
if tevcat_flag == "N":
tevcat_message = "No TeV association"
elif tevcat_flag == "P":
tevcat_message = "Small TeV source"
elif tevcat_flag == "E":
tevcat_message = "Extended TeV source (diameter > 40 arcmins)"
else:
tevcat_message = "N/A"
ss += "{:<16s} : {}\n".format("TeVCat flag", tevcat_message)
fmt = "\n{:<32s} : {:.3f}\n"
ss += fmt.format("Significance (10 GeV - 2 TeV)", d["Signif_Avg"])
ss += "{:<32s} : {:.1f}\n".format("Npred", d["Npred"])
return ss
def _info_position(self):
"""Print position info."""
d = self.data
ss = "\n*** Position info ***\n\n"
ss += "{:<20s} : {:.3f}\n".format("RA", d["RAJ2000"])
ss += "{:<20s} : {:.3f}\n".format("DEC", d["DEJ2000"])
ss += "{:<20s} : {:.3f}\n".format("GLON", d["GLON"])
ss += "{:<20s} : {:.3f}\n".format("GLAT", d["GLAT"])
# TODO: All sources are non-elliptical; just give one number for radius?
ss += "\n"
ss += "{:<20s} : {:.4f}\n".format("Semimajor (95%)", d["Conf_95_SemiMajor"])
ss += "{:<20s} : {:.4f}\n".format("Semiminor (95%)", d["Conf_95_SemiMinor"])
ss += "{:<20s} : {:.2f}\n".format("Position angle (95%)", d["Conf_95_PosAng"])
ss += "{:<20s} : {:.0f}\n".format("ROI number", d["ROI_num"])
return ss
def _info_morphology(self):
e = self.data_extended
ss = "*** Extended source information ***\n"
ss += "{:<16s} : {}\n".format("Model form", e["Model_Form"])
ss += "{:<16s} : {:.4f}\n".format("Model semimajor", e["Model_SemiMajor"])
ss += "{:<16s} : {:.4f}\n".format("Model semiminor", e["Model_SemiMinor"])
ss += "{:<16s} : {:.4f}\n".format("Position angle", e["Model_PosAng"])
ss += "{:<16s} : {}\n".format("Spatial function", e["Spatial_Function"])
ss += "{:<16s} : {}\n\n".format("Spatial filename", e["Spatial_Filename"])
return ss
def _info_spectral_fit(self):
"""Print model data."""
d = self.data
spec_type = d["SpectrumType"].strip()
ss = "\n*** Spectral fit info ***\n\n"
ss += "{:<32s} : {}\n".format("Spectrum type", d["SpectrumType"])
ss += "{:<32s} : {:.1f}\n".format("Significance curvature", d["Signif_Curve"])
# Power-law parameters are always given; give in any case
fmt = "{:<32s} : {:.3f} +- {:.3f}\n"
ss += fmt.format(
"Power-law spectral index", d["PowerLaw_Index"], d["Unc_PowerLaw_Index"]
)
if spec_type == "PowerLaw":
pass
elif spec_type == "LogParabola":
fmt = "{:<32s} : {:.3f} +- {:.3f}\n"
ss += fmt.format(
"LogParabolaSpectralModel spectral index",
d["Spectral_Index"],
d["Unc_Spectral_Index"],
)
ss += "{:<32s} : {:.3f} +- {:.3f}\n".format(
"LogParabolaSpectralModel beta", d["beta"], d["Unc_beta"]
)
else:
raise ValueError(f"Invalid spec_type: {spec_type!r}")
ss += "{:<32s} : {:.1f} {}\n".format(
"Pivot energy", d["Pivot_Energy"].value, d["Pivot_Energy"].unit
)
ss += "{:<32s} : {:.3} +- {:.3} {}\n".format(
"Flux Density at pivot energy",
d["Flux_Density"].value,
d["Unc_Flux_Density"].value,
"cm-2 GeV-1 s-1",
)
ss += "{:<32s} : {:.3} +- {:.3} {}\n".format(
"Integral flux (10 GeV - 1 TeV)",
d["Flux"].value,
d["Unc_Flux"].value,
"cm-2 s-1",
)
ss += "{:<32s} : {:.3} +- {:.3} {}\n".format(
"Energy flux (10 GeV - TeV)",
d["Energy_Flux"].value,
d["Unc_Energy_Flux"].value,
"erg cm-2 s-1",
)
return ss
def _info_spectral_points(self):
"""Print spectral points."""
ss = "\n*** Spectral points ***\n\n"
lines = self.flux_points.table_formatted.pformat(max_width=-1, max_lines=-1)
ss += "\n".join(lines)
return ss + "\n"
def _info_other(self):
"""Print other info."""
d = self.data
ss = "\n*** Other info ***\n\n"
ss += "{:<16s} : {:.3f} {}\n".format(
"HEP Energy", d["HEP_Energy"].value, d["HEP_Energy"].unit
)
ss += "{:<16s} : {:.3f}\n".format("HEP Probability", d["HEP_Prob"])
# This is the number of Bayesian blocks for most sources,
# except -1 means "could not be tested"
msg = d["Variability_BayesBlocks"]
if msg == 1:
msg = "1 (not variable)"
elif msg == -1:
msg = "Could not be tested"
ss += "{:<16s} : {}\n".format("Bayesian Blocks", msg)
ss += "{:<16s} : {:.3f}\n".format("Redshift", d["Redshift"])
ss += "{:<16s} : {:.3} {}\n".format(
"NuPeak_obs", d["NuPeak_obs"].value, d["NuPeak_obs"].unit
)
return ss
@property
def spectral_model(self):
"""Best fit spectral model (`~gammapy.modeling.models.SpectralModel`)."""
d = self.data
spec_type = self.data["SpectrumType"].strip()
pars, errs = {}, {}
pars["amplitude"] = d["Flux_Density"]
errs["amplitude"] = d["Unc_Flux_Density"]
pars["reference"] = d["Pivot_Energy"]
if spec_type == "PowerLaw":
pars["index"] = d["PowerLaw_Index"]
errs["index"] = d["Unc_PowerLaw_Index"]
model = PowerLawSpectralModel(**pars)
elif spec_type == "LogParabola":
pars["alpha"] = d["Spectral_Index"]
pars["beta"] = d["beta"]
errs["alpha"] = d["Unc_Spectral_Index"]
errs["beta"] = d["Unc_beta"]
model = LogParabolaSpectralModel(**pars)
else:
raise ValueError(f"Invalid spec_type: {spec_type!r}")
model.parameters.set_parameter_errors(errs)
return model
@property
def flux_points(self):
"""Flux points (`~gammapy.spectrum.FluxPoints`)."""
table = Table()
table.meta["SED_TYPE"] = "flux"
table["e_min"] = self._ebounds[:-1]
table["e_max"] = self._ebounds[1:]
flux = self.data["Flux_Band"]
flux_err = self.data["Unc_Flux_Band"]
e2dnde = self.data["nuFnu"]
table["flux"] = flux
table["flux_errn"] = np.abs(flux_err[:, 0])
table["flux_errp"] = flux_err[:, 1]
table["e2dnde"] = e2dnde
table["e2dnde_errn"] = np.abs(e2dnde * flux_err[:, 0] / flux)
table["e2dnde_errp"] = e2dnde * flux_err[:, 1] / flux
is_ul = np.isnan(table["flux_errn"])
table["is_ul"] = is_ul
# handle upper limits
table["flux_ul"] = np.nan * flux_err.unit
flux_ul = compute_flux_points_ul(table["flux"], table["flux_errp"])
table["flux_ul"][is_ul] = flux_ul[is_ul]
table["e2dnde_ul"] = np.nan * e2dnde.unit
e2dnde_ul = compute_flux_points_ul(table["e2dnde"], table["e2dnde_errp"])
table["e2dnde_ul"][is_ul] = e2dnde_ul[is_ul]
# Square root of test statistic
table["sqrt_ts"] = self.data["Sqrt_TS_Band"]
return FluxPoints(table)
@property
def spatial_model(self):
"""Source spatial model (`~gammapy.modeling.models.SpatialModel`)."""
d = self.data
pars = {}
glon = d["GLON"]
glat = d["GLAT"]
if self.is_pointlike:
pars["lon_0"] = glon
pars["lat_0"] = glat
return PointSpatialModel(lon_0=glon, lat_0=glat, frame="galactic")
else:
de = self.data_extended
morph_type = de["Spatial_Function"].strip()
if morph_type == "RadialDisk":
r_0 = de["Model_SemiMajor"].to("deg")
return DiskSpatialModel(
lon_0=glon, lat_0=glat, r_0=r_0, frame="galactic"
)
elif morph_type in ["SpatialMap"]:
filename = de["Spatial_Filename"].strip()
path = make_path(
"$GAMMAPY_DATA/catalogs/fermi/Extended_archive_v18/Templates/"
)
return TemplateSpatialModel.read(path / filename)
elif morph_type == "RadialGauss":
# TODO: fill elongation info as soon as model supports it
sigma = de["Model_SemiMajor"].to("deg")
return GaussianSpatialModel(
lon_0=glon, lat_0=glat, sigma=sigma, frame="galactic"
)
else:
raise ValueError(f"Invalid morph_type: {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)
@property
def is_pointlike(self):
return self.data["Extended_Source_Name"].strip() == ""
[docs]class SourceCatalog3FGL(SourceCatalog):
"""Fermi-LAT 3FGL source catalog.
Reference: https://ui.adsabs.harvard.edu/#abs/2015ApJS..218...23A
One source is represented by `~gammapy.catalog.SourceCatalogObject3FGL`.
"""
name = "3fgl"
description = "LAT 4-year point source catalog"
source_object_class = SourceCatalogObject3FGL
source_categories = {
"galactic": ["psr", "pwn", "snr", "spp", "glc"],
"extra-galactic": [
"css",
"bll",
"fsrq",
"agn",
"nlsy1",
"rdg",
"sey",
"bcu",
"gal",
"sbg",
"ssrq",
],
"GALACTIC": ["PSR", "PWN", "SNR", "HMB", "BIN", "NOV", "SFR"],
"EXTRA-GALACTIC": [
"CSS",
"BLL",
"FSRQ",
"AGN",
"NLSY1",
"RDG",
"SEY",
"BCU",
"GAL",
"SBG",
"SSRQ",
],
"unassociated": [""],
}
def __init__(self, filename="$GAMMAPY_DATA/catalogs/fermi/gll_psc_v16.fit.gz"):
filename = str(make_path(filename))
with warnings.catch_warnings(): # ignore FITS units warnings
warnings.simplefilter("ignore", u.UnitsWarning)
table = Table.read(filename, hdu="LAT_Point_Source_Catalog")
table_standardise_units_inplace(table)
source_name_key = "Source_Name"
source_name_alias = (
"Extended_Source_Name",
"0FGL_Name",
"1FGL_Name",
"2FGL_Name",
"1FHL_Name",
"ASSOC_TEV",
"ASSOC1",
"ASSOC2",
)
super().__init__(
table=table,
source_name_key=source_name_key,
source_name_alias=source_name_alias,
)
self.extended_sources_table = Table.read(filename, hdu="ExtendedSources")
[docs] def is_source_class(self, source_class):
"""
Check if source belongs to a given source class.
The classes are described in Table 3 of the 3FGL paper:
https://ui.adsabs.harvard.edu/abs/2015ApJS..218...23A
Parameters
----------
source_class : str
Source class designator as defined in Table 3. There are a few extra
selections available:
- 'ALL': all identified objects
- 'all': all objects with associations
- 'galactic': all sources with an associated galactic object
- 'GALACTIC': all identified galactic sources
- 'extra-galactic': all sources with an associated extra-galactic object
- 'EXTRA-GALACTIC': all identified extra-galactic sources
- 'unassociated': all unassociated objects
Returns
-------
selection : `~numpy.ndarray`
Selection mask.
"""
source_class_info = np.array([_.strip() for _ in self.table["CLASS1"]])
cats = self.source_categories
if source_class in cats:
category = set(cats[source_class])
elif source_class == "ALL":
category = set(cats["EXTRA-GALACTIC"] + cats["GALACTIC"])
elif source_class == "all":
category = set(cats["extra-galactic"] + cats["galactic"])
elif source_class in np.unique(source_class_info):
category = {source_class}
else:
raise ValueError(f"Invalid source_class: {source_class!r}")
return np.array([_ in category for _ in source_class_info])
[docs] def select_source_class(self, source_class):
"""
Select all sources of a given source class.
See `SourceCatalog3FHL.is_source_class` for further documentation
Parameters
----------
source_class : str
Source class designator.
Returns
-------
selection : `SourceCatalog3FHL`
Subset of the 3FHL catalog containing only the selected source class.
"""
catalog = self.copy()
selection = self.is_source_class(source_class)
catalog.table = catalog.table[selection]
return catalog
[docs]class SourceCatalog4FGL(SourceCatalog):
"""Fermi-LAT 4FGL source catalog.
References:
- https://arxiv.org/abs/1902.10045
- https://fermi.gsfc.nasa.gov/ssc/data/access/lat/8yr_catalog/
One source is represented by `~gammapy.catalog.SourceCatalogObject4FGL`.
"""
name = "4fgl"
description = "LAT 8-year point source catalog"
source_object_class = SourceCatalogObject4FGL
def __init__(self, filename="$GAMMAPY_DATA/catalogs/fermi/gll_psc_v19.fit.gz"):
filename = str(make_path(filename))
table = Table.read(filename, hdu="LAT_Point_Source_Catalog")
table_standardise_units_inplace(table)
source_name_key = "Source_Name"
source_name_alias = (
"Extended_Source_Name",
"ASSOC_FGL",
"ASSOC_FHL",
"ASSOC_GAM1",
"ASSOC_GAM2",
"ASSOC_GAM3",
"ASSOC_TEV",
"ASSOC1",
"ASSOC2",
)
super().__init__(
table=table,
source_name_key=source_name_key,
source_name_alias=source_name_alias,
)
self.extended_sources_table = Table.read(filename, hdu="ExtendedSources")
[docs]class SourceCatalog1FHL(SourceCatalog):
"""Fermi-LAT 1FHL source catalog.
Reference: https://ui.adsabs.harvard.edu/abs/2013ApJS..209...34A
One source is represented by `~gammapy.catalog.SourceCatalogObject1FHL`.
"""
name = "1fhl"
description = "First Fermi-LAT Catalog of Sources above 10 GeV"
source_object_class = SourceCatalogObject1FHL
def __init__(self, filename="$GAMMAPY_DATA/catalogs/fermi/gll_psch_v07.fit.gz"):
filename = str(make_path(filename))
with warnings.catch_warnings(): # ignore FITS units warnings
warnings.simplefilter("ignore", u.UnitsWarning)
table = Table.read(filename, hdu="LAT_Point_Source_Catalog")
table_standardise_units_inplace(table)
source_name_key = "Source_Name"
source_name_alias = ("ASSOC1", "ASSOC2", "ASSOC_TEV", "ASSOC_GAM")
super().__init__(
table=table,
source_name_key=source_name_key,
source_name_alias=source_name_alias,
)
self.extended_sources_table = Table.read(filename, hdu="ExtendedSources")
[docs]class SourceCatalog2FHL(SourceCatalog):
"""Fermi-LAT 2FHL source catalog.
Reference: https://ui.adsabs.harvard.edu/abs/2016ApJS..222....5A
One source is represented by `~gammapy.catalog.SourceCatalogObject2FHL`.
"""
name = "2fhl"
description = "LAT second high-energy source catalog"
source_object_class = SourceCatalogObject2FHL
def __init__(self, filename="$GAMMAPY_DATA/catalogs/fermi/gll_psch_v08.fit.gz"):
filename = str(make_path(filename))
with warnings.catch_warnings(): # ignore FITS units warnings
warnings.simplefilter("ignore", u.UnitsWarning)
table = Table.read(filename, hdu="2FHL Source Catalog")
table_standardise_units_inplace(table)
source_name_key = "Source_Name"
source_name_alias = ("ASSOC", "3FGL_Name", "1FHL_Name", "TeVCat_Name")
super().__init__(
table=table,
source_name_key=source_name_key,
source_name_alias=source_name_alias,
)
self.counts_image = Map.read(filename, hdu="Count Map")
self.extended_sources_table = Table.read(filename, hdu="Extended Sources")
self.rois = Table.read(filename, hdu="ROIs")
[docs]class SourceCatalog3FHL(SourceCatalog):
"""Fermi-LAT 3FHL source catalog.
Reference: https://ui.adsabs.harvard.edu/abs/2017ApJS..232...18A
One source is represented by `~gammapy.catalog.SourceCatalogObject3FHL`.
"""
name = "3fhl"
description = "LAT third high-energy source catalog"
source_object_class = SourceCatalogObject3FHL
source_categories = {
"galactic": ["glc", "hmb", "psr", "pwn", "sfr", "snr", "spp"],
"extra-galactic": ["agn", "bcu", "bll", "fsrq", "rdg", "sbg"],
"GALACTIC": ["BIN", "HMB", "PSR", "PWN", "SFR", "SNR"],
"EXTRA-GALACTIC": ["BLL", "FSRQ", "NLSY1", "RDG"],
"unassociated": [""],
}
def __init__(self, filename="$GAMMAPY_DATA/catalogs/fermi/gll_psch_v13.fit.gz"):
filename = str(make_path(filename))
with warnings.catch_warnings(): # ignore FITS units warnings
warnings.simplefilter("ignore", u.UnitsWarning)
table = Table.read(filename, hdu="LAT_Point_Source_Catalog")
table_standardise_units_inplace(table)
source_name_key = "Source_Name"
source_name_alias = ("ASSOC1", "ASSOC2", "ASSOC_TEV", "ASSOC_GAM")
super().__init__(
table=table,
source_name_key=source_name_key,
source_name_alias=source_name_alias,
)
self.extended_sources_table = Table.read(filename, hdu="ExtendedSources")
self.rois = Table.read(filename, hdu="ROIs")
self.energy_bounds_table = Table.read(filename, hdu="EnergyBounds")
[docs] def is_source_class(self, source_class):
"""
Check if source belongs to a given source class.
The classes are described in Table 3 of the 3FGL paper:
https://ui.adsabs.harvard.edu/abs/2015ApJS..218...23A
Parameters
----------
source_class : str
Source class designator as defined in Table 3. There are a few extra
selections available:
- 'ALL': all identified objects
- 'all': all objects with associations
- 'galactic': all sources with an associated galactic object
- 'GALACTIC': all identified galactic sources
- 'extra-galactic': all sources with an associated extra-galactic object
- 'EXTRA-GALACTIC': all identified extra-galactic sources
- 'unassociated': all unassociated objects
Returns
-------
selection : `~numpy.ndarray`
Selection mask.
"""
source_class_info = np.array([_.strip() for _ in self.table["CLASS"]])
cats = self.source_categories
if source_class in cats:
category = set(cats[source_class])
elif source_class == "ALL":
category = set(cats["EXTRA-GALACTIC"] + cats["GALACTIC"])
elif source_class == "all":
category = set(cats["extra-galactic"] + cats["galactic"])
elif source_class in np.unique(source_class_info):
category = {source_class}
else:
raise ValueError(f"Invalid source_class: {source_class!r}")
return np.array([_ in category for _ in source_class_info])
[docs] def select_source_class(self, source_class):
"""
Select all sources of a given source class.
See `SourceCatalog3FHL.is_source_class` for further documentation
Parameters
----------
source_class : str
Source class designator.
Returns
-------
selection : `SourceCatalog3FHL`
Subset of the 3FHL catalog containing only the selected source class.
"""
catalog = self.copy()
selection = self.is_source_class(source_class)
catalog.table = catalog.table[selection]
return catalog