gammapy.estimators.points.core.FluxPoints.to_table()

Complexity

Conditions

22

Size

Total Lines	153
Code Lines	75

Duplication

Lines	0
Ratio	0 %

Importance

Changes

0

# Licensed under a 3-clause BSD style license - see LICENSE.rst
    def to_table(self, sed_type=None, format="gadf-sed", formatted=False):
        """Create table for a given SED type.

        Parameters
        ----------
        sed_type : {"likelihood", "dnde", "e2dnde", "flux", "eflux"}
            Sed type to convert to. Default is `likelihood`
        format : {"gadf-sed", "lightcurve", "binned-time-series", "profile"}
            Format specification. The following formats are supported:

                * "gadf-sed": format for sed flux points see :ref:`gadf:flux-points`
                  for details
                * "lightcurve": Gammapy internal format to store energy dependent
                  lightcurves. Basically a generalisation of the "gadf" format, but
                  currently there is no detailed documentation available.
                * "binned-time-series": table format support by Astropy's
                  `~astropy.timeseries.BinnedTimeSeries`.
                * "profile": Gammapy internal format to store energy dependent
                  flux profiles. Basically a generalisation of the "gadf" format, but
                  currently there is no detailed documentation available.

        formatted : bool
            Formatted version with column formats applied. Numerical columns are
            formatted to .3f and .3e respectively.

        Returns
        -------
        table : `~astropy.table.Table`
            Flux points table

        Examples
        --------

        This is how to read and plot example flux points:

        >>> from gammapy.estimators import FluxPoints
        >>> fp = FluxPoints.read("$GAMMAPY_DATA/hawc_crab/HAWC19_flux_points.fits")
        >>> table = fp.to_table(sed_type="flux", format="gadf-sed", formatted=True)
        >>> print(table[:2])
        e_ref e_min e_max     flux      flux_err    flux_ul      ts    sqrt_ts is_ul
         TeV   TeV   TeV  1 / (cm2 s) 1 / (cm2 s) 1 / (cm2 s)
        ----- ----- ----- ----------- ----------- ----------- -------- ------- -----
        1.334 1.000 1.780   1.423e-11   3.135e-13         nan 2734.000  52.288 False
        2.372 1.780 3.160   5.780e-12   1.082e-13         nan 4112.000  64.125 False
        """
        if sed_type is None:
            sed_type = self.sed_type_init

        if format == "gadf-sed":
            # TODO: what to do with GTI info?
            if not self.geom.axes.names == ["energy"]:
                raise ValueError(
                    "Only flux points with a single energy axis "
                    "can be converted to 'gadf-sed'"
                )

            idx = (Ellipsis, 0, 0)
            table = self.energy_axis.to_table(format="gadf-sed")
            table.meta["SED_TYPE"] = sed_type

            if not self.is_convertible_to_flux_sed_type:
                table.remove_columns(["e_min", "e_max"])

            if self.n_sigma_ul:
                table.meta["UL_CONF"] = np.round(
                    1 - 2 * stats.norm.sf(self.n_sigma_ul), 7
                )

            if sed_type == "likelihood":
                table["ref_dnde"] = self.dnde_ref[idx]
                table["ref_flux"] = self.flux_ref[idx]
                table["ref_eflux"] = self.eflux_ref[idx]

            for quantity in self.all_quantities(sed_type=sed_type):
                data = getattr(self, quantity, None)
                if data:
                    table[quantity] = data.quantity[idx]

            if self.has_stat_profiles:
                norm_axis = self.stat_scan.geom.axes["norm"]
                table["norm_scan"] = norm_axis.center.reshape((1, -1))
                table["stat"] = self.stat.data[idx]
                table["stat_scan"] = self.stat_scan.data[idx]

            table["is_ul"] = self.is_ul.data[idx]

        elif format == "lightcurve":
            time_axis = self.geom.axes["time"]

            tables = []
            for idx, (time_min, time_max) in enumerate(time_axis.iter_by_edges):
                table_flat = Table()
                table_flat["time_min"] = [time_min.mjd]
                table_flat["time_max"] = [time_max.mjd]

                fp = self.slice_by_idx(slices={"time": idx})
                table = fp.to_table(sed_type=sed_type, format="gadf-sed")

                for column in table.columns:
                    table_flat[column] = table[column][np.newaxis]

                tables.append(table_flat)

            table = vstack(tables)
        elif format == "binned-time-series":
            message = (
                "Format 'binned-time-series' support a single time axis "
                f"only. Got {self.geom.axes.names}"
            )

            if not self.geom.axes.is_unidimensional:
                raise ValueError(message)

            axis = self.geom.axes.primary_axis

            if not isinstance(axis, TimeMapAxis):
                raise ValueError(message)

            table = Table()
            table["time_bin_start"] = axis.time_min
            table["time_bin_size"] = axis.time_delta

            for quantity in self.all_quantities(sed_type=sed_type):
                data = getattr(self, quantity, None)
                if data:
                    table[quantity] = data.quantity.squeeze()
        elif format == "profile":
            x_axis = self.geom.axes["projected-distance"]

            tables = []
            for idx, (x_min, x_max) in enumerate(x_axis.iter_by_edges):
                table_flat = Table()
                table_flat["x_min"] = [x_min]
                table_flat["x_max"] = [x_max]
                table_flat["x_ref"] = [(x_max + x_min) / 2]

                fp = self.slice_by_idx(slices={"projected-distance": idx})
                table = fp.to_table(sed_type=sed_type, format="gadf-sed")

                for column in table.columns:
                    table_flat[column] = table[column][np.newaxis]

                tables.append(table_flat)

            table = vstack(tables)

        else:
            raise ValueError(f"Not a supported format {format}")

        if formatted:
            table = self._format_table(table=table)

        return table