# Licensed under a 3-clause BSD style license - see LICENSE.rst
from __future__ import absolute_import, division, print_function, unicode_literals
import numpy as np
from astropy.table import Table, Column
import astropy.units as u
from ..spectrum import CountsSpectrum, models
from ..utils.scripts import read_yaml, make_path
from ..utils.energy import EnergyBounds
__all__ = ["SpectrumFitResult", "SpectrumResult"]
[docs]class SpectrumFitResult(object):
"""Result of a `~gammapy.spectrum.SpectrumFit`.
All fit results should be accessed via this class.
Parameters
----------
model : `~gammapy.spectrum.models.SpectralModel`
Best-fit model
fit_range : `~astropy.units.Quantity`
Energy range of the spectral fit
statname : str, optional
Statistic used for the fit
statval : float, optional
Final fit statistic
stat_per_bin : float, optional
Fit statistic value per bin
npred : array-like, optional
Counts predicted by the fit
obs : `~gammapy.spectrum.SpectrumObservation`
Input data used for the fit
"""
__slots__ = [
"model",
"fit_range",
"statname",
"statval",
"stat_per_bin",
"npred",
"obs",
]
def __init__(
self,
model,
fit_range=None,
statname=None,
statval=None,
stat_per_bin=None,
npred=None,
obs=None,
):
self.model = model
self.fit_range = fit_range
self.statname = statname
self.statval = statval
self.stat_per_bin = stat_per_bin
self.npred = npred
self.obs = obs
[docs] @classmethod
def from_yaml(cls, filename):
"""Create from YAML file.
Parameters
----------
filename : str, Path
File to read
"""
filename = make_path(filename)
val = read_yaml(str(filename))
return cls.from_dict(val)
[docs] def to_yaml(self, filename, mode="w"):
"""Write to YAML file.
Parameters
----------
filename : str
File to write
mode : str
Write mode
"""
import yaml
d = self.to_dict()
val = yaml.safe_dump(d, default_flow_style=False)
with open(str(filename), mode) as outfile:
outfile.write(val)
[docs] def to_dict(self):
"""Convert to dict."""
val = dict()
val["model"] = self.model.to_dict()
if self.fit_range is not None:
val["fit_range"] = dict(
min=self.fit_range[0].value,
max=self.fit_range[1].value,
unit=self.fit_range.unit.to_string("fits"),
)
if self.statval is not None:
val["statval"] = float(self.statval)
if self.statname is not None:
val["statname"] = self.statname
return val
[docs] @classmethod
def from_dict(cls, val):
"""Create from dict."""
modeldict = val["model"]
model = models.SpectralModel.from_dict(modeldict)
try:
erange = val["fit_range"]
energy_range = (erange["min"], erange["max"]) * u.Unit(erange["unit"])
except KeyError:
energy_range = None
return cls(model=model, fit_range=energy_range)
# TODO: rather add this to Parameters?
[docs] def to_table(self, energy_unit="TeV", flux_unit="cm-2 s-1 TeV-1", **kwargs):
"""Convert to `~astropy.table.Table`.
Produce overview table containing the most important parameters
"""
t = Table()
t["model"] = [self.model.__class__.__name__]
for par_name, value in self.model.parameters._ufloats.items():
val = value.n
err = value.s
# Apply correction factor for units
# TODO: Refactor
current_unit = u.Unit(self.model.parameters[par_name].unit)
if current_unit.is_equivalent(energy_unit):
factor = current_unit.to(energy_unit)
col_unit = energy_unit
elif current_unit.is_equivalent(1 / u.Unit(energy_unit)):
factor = current_unit.to(1 / u.Unit(energy_unit))
col_unit = 1 / u.Unit(energy_unit)
elif current_unit.is_equivalent(flux_unit):
factor = current_unit.to(flux_unit)
col_unit = flux_unit
elif current_unit.is_equivalent(u.dimensionless_unscaled):
factor = 1
col_unit = current_unit
else:
raise ValueError(current_unit)
t[par_name] = Column(
data=np.atleast_1d(val * factor), unit=col_unit, **kwargs
)
t["{}_err".format(par_name)] = Column(
data=np.atleast_1d(err * factor), unit=col_unit, **kwargs
)
t["fit_range"] = Column(
data=[self.fit_range.to(energy_unit)], unit=energy_unit, **kwargs
)
return t
def __str__(self):
s = "\nFit result info \n"
s += "--------------- \n"
s += "Model: {} \n".format(self.model)
if self.statval is not None:
s += "\nStatistic: {0:.3f} ({1})".format(self.statval, self.statname)
if self.fit_range is not None:
s += "\nFit Range: {}".format(self.fit_range)
s += "\n"
return s
[docs] def butterfly(self, energy=None, flux_unit="TeV-1 cm-2 s-1"):
"""Compute butterfly table.
Parameters
----------
energy : `~astropy.units.Quantity`, optional
Energies at which to evaluate the butterfly.
flux_unit : str
Flux unit for the butterfly.
Returns
-------
table : `~astropy.table.Table`
Butterfly info in table (cols: 'energy', 'flux', 'flux_lo', 'flux_hi')
"""
if energy is None:
energy = EnergyBounds.equal_log_spacing(
self.fit_range[0], self.fit_range[1], 100
)
flux, flux_err = self.model.evaluate_error(energy)
table = Table()
table["energy"] = energy
table["flux"] = flux.to(flux_unit)
table["flux_lo"] = flux - flux_err.to(flux_unit)
table["flux_hi"] = flux + flux_err.to(flux_unit)
return table
@property
def expected_source_counts(self):
"""Predicted source counts (`~gammapy.spectrum.CountsSpectrum`)."""
energy = self.obs.on_vector.energy
data = self.npred
return CountsSpectrum(data=data, energy_lo=energy.lo, energy_hi=energy.hi)
# TODO: is this the quantity, and sign, we want for residuals?
@property
def residuals(self):
"""Residuals (predicted source - excess).
"""
resspec = self.expected_source_counts.copy()
resspec.data.data -= self.obs.excess_vector.data.data
return resspec
[docs] def plot(self, **kwargs):
"""Plot counts and residuals in two panels.
Calls ``plot_counts`` and ``plot_residuals``.
"""
ax0, ax1 = get_plot_axis(**kwargs)
self.plot_counts(ax0)
self.plot_residuals(ax1)
return ax0, ax1
[docs] def plot_counts(self, ax):
"""Plot predicted and detected counts."""
self.expected_source_counts.plot(ax=ax, label="mu_src")
self.obs.excess_vector.plot(ax=ax, label="excess", fmt=".", energy_unit="TeV")
ax.axvline(
self.fit_range.to("TeV").value[0],
color="black",
linestyle="dashed",
label="fit range",
)
ax.axvline(self.fit_range.to("TeV").value[1], color="black", linestyle="dashed")
ax.legend(numpoints=1)
ax.set_title("")
[docs] def plot_residuals(self, ax):
"""Plot residuals."""
self.residuals.plot(ax=ax, ecolor="black", fmt="none")
ax.axhline(color="black")
ymax = 1.4 * max(self.residuals.data.data.value)
ax.set_ylim(-ymax, ymax)
ax.set_xlabel("Energy [{}]".format("TeV"))
ax.set_ylabel("ON (Predicted - Detected)")
[docs]class SpectrumResult(object):
"""Spectrum analysis results.
Contains best fit model and flux points.
Parameters
----------
model : `~gammapy.spectrum.models.SpectralModel`
Best Fit model
points : `~gammapy.spectrum.FluxPoints`, optional
Flux points
"""
def __init__(self, model, points):
self.model = model
self.points = points
@property
def flux_point_residuals(self):
"""Residuals.
Defined as ``(points - model) / model``
Returns
-------
residuals : `numpy.ndarray`
Residuals
residuals_err : `numpy.ndarray`
Residuals error
"""
e_ref = self.points.table["e_ref"].quantity
points = self.points.table["dnde"].quantity
points_err = self.points.get_flux_err()
# Deal with asymmetric errors
if isinstance(points_err, tuple):
points_err = np.sqrt(points_err[0] * points_err[1])
model_val = self.model(e_ref)
residuals = ((points - model_val) / model_val).to("").value
residuals_err = (points_err / model_val).to("").value
# Remove residuals for upper_limits
residuals[self.points.is_ul] = np.nan
residuals_err[self.points.is_ul] = np.nan
return residuals, residuals_err
[docs] def plot(
self,
energy_range,
energy_unit="TeV",
flux_unit="cm-2 s-1 TeV-1",
energy_power=0,
fit_kwargs=dict(),
butterfly_kwargs=dict(),
point_kwargs=dict(),
fig_kwargs=dict(),
):
"""Plot spectrum.
Plot best fit model, flux points and residuals.
Parameters
----------
energy_range : `~astropy.units.Quantity`
Energy range for the plot
energy_unit : str, `~astropy.units.Unit`, optional
Unit of the energy axis
flux_unit : str, `~astropy.units.Unit`, optional
Unit of the flux axis
energy_power : int
Power of energy to multiply flux axis with
fit_kwargs : dict, optional
forwarded to :func:`gammapy.spectrum.models.SpectralModel.plot`
butterfly_kwargs : dict, optional
forwarded to :func:`gammapy.spectrum.models.SpectralModel.plot_error`
point_kwargs : dict, optional
forwarded to :func:`gammapy.spectrum.FluxPoints.plot`
fig_kwargs : dict, optional
forwarded to :func:`matplotlib.pyplot.figure`
Returns
-------
ax0 : `~matplotlib.axes.Axes`
Spectrum plot axis
ax1 : `~matplotlib.axes.Axes`
Residuals plot axis
"""
ax0, ax1 = get_plot_axis(**fig_kwargs)
ax0.set_yscale("log")
common_kwargs = dict(
energy_unit=energy_unit, flux_unit=flux_unit, energy_power=energy_power
)
fit_kwargs.update(common_kwargs)
point_kwargs.update(common_kwargs)
butterfly_kwargs.update(common_kwargs)
self.model.plot(energy_range=energy_range, ax=ax0, **fit_kwargs)
self.model.plot_error(energy_range=energy_range, ax=ax0, **butterfly_kwargs)
self.points.plot(ax=ax0, **point_kwargs)
point_kwargs.pop("flux_unit")
point_kwargs.pop("energy_power")
ax0.set_xlabel("")
self._plot_residuals(ax=ax1, **point_kwargs)
return ax0, ax1
def _plot_residuals(self, ax=None, energy_unit="TeV", **kwargs):
"""Plot residuals.
Parameters
----------
ax : `~matplotlib.axes.Axes`, optional
Axis
energy_unit : str, `~astropy.units.Unit`, optional
Unit of the energy axis
Returns
-------
ax : `~matplotlib.axes.Axes`, optional
Axis
"""
import matplotlib.pyplot as plt
ax = plt.gca() if ax is None else ax
kwargs.setdefault("fmt", ".")
y, y_err = self.flux_point_residuals
x = self.points.e_ref
x = x.to(energy_unit).value
ax.errorbar(x, y, yerr=y_err, **kwargs)
ax.axhline(0, color="black")
ax.set_xlabel("Energy [{}]".format(energy_unit))
ax.set_ylabel("Residuals")
return ax
def get_plot_axis(**kwargs):
"""Axis setup used for standard plots.
kwargs are forwarded to plt.figure()
Returns
-------
ax0 : `~matplotlib.axes.Axes`
Main plot
ax1 : `~matplotlib.axes.Axes`
Residuals
"""
from matplotlib import gridspec
import matplotlib.pyplot as plt
fig = plt.figure(**kwargs)
gs = gridspec.GridSpec(5, 1)
ax0 = plt.subplot(gs[:-2, :])
ax1 = plt.subplot(gs[3, :], sharex=ax0)
gs.update(hspace=0.1)
plt.setp(ax0.get_xticklabels(), visible=False)
ax0.set_xscale("log")
ax1.set_xscale("log")
return ax0, ax1