import warnings
from astropy.nddata import CCDData
from astropy.wcs import WCS
from glue.core import Data
from gwcs import WCS as GWCS
from specutils.manipulation import spectral_slab
from traitlets import List, Unicode, observe
from jdaviz.core.events import SnackbarMessage
from jdaviz.core.registries import tray_registry
from jdaviz.core.template_mixin import (PluginTemplateMixin,
DatasetSelectMixin,
SelectPluginComponent,
SpectralSubsetSelectMixin,
AddResultsMixin,
with_spinner)
from jdaviz.core.user_api import PluginUserApi
__all__ = ['Collapse']
[docs]
@tray_registry('g-collapse', label="Collapse", category="data:reduction")
class Collapse(PluginTemplateMixin, DatasetSelectMixin, SpectralSubsetSelectMixin, AddResultsMixin):
"""
See the :ref:`Collapse Plugin Documentation <collapse>` for more details.
Only the following attributes and methods are available through the
:ref:`public plugin API <plugin-apis>`:
* :meth:`~jdaviz.core.template_mixin.PluginTemplateMixin.show`
* :meth:`~jdaviz.core.template_mixin.PluginTemplateMixin.open_in_tray`
* :meth:`~jdaviz.core.template_mixin.PluginTemplateMixin.close_in_tray`
* ``dataset`` (:class:`~jdaviz.core.template_mixin.DatasetSelect`):
Dataset to use for computing line statistics.
* ``function`` (:class:`~jdaviz.core.template_mixin.SelectPluginComponent`):
Function to use for the collapse operation (Mean, Median, Min, Max, Sum).
* ``spectral_subset`` (:class:`~jdaviz.core.template_mixin.SubsetSelect`):
Subset to use for the line, or ``Entire Spectrum``.
* ``add_results`` (:class:`~jdaviz.core.template_mixin.AddResults`)
* :meth:`collapse`
"""
template_file = __file__, "collapse.vue"
function_items = List().tag(sync=True)
function_selected = Unicode('Sum').tag(sync=True)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._label_counter = 0
self.collapsed_flux = None
self.function = SelectPluginComponent(self,
items='function_items',
selected='function_selected',
manual_options=['Mean', 'Median', 'Min', 'Max', 'Sum']) # noqa
self.dataset.add_filter('is_flux_cube')
self.add_results.viewer.filters = ['is_image_viewer']
# description displayed under plugin title in tray
self._plugin_description = 'Collapse a spectral cube along one axis.'
if self.config == "deconfigged":
self.observe_traitlets_for_relevancy(traitlets_to_observe=['dataset_items'])
@property
def _default_spectrum_viewer_reference_name(self):
return self.jdaviz_helper._default_spectrum_viewer_reference_name
@property
def user_api(self):
return PluginUserApi(self, expose=('dataset', 'function', 'spectral_subset',
'add_results', 'collapse'))
@observe("dataset_selected", "dataset_items")
def _set_default_results_label(self, event={}):
label_comps = []
if hasattr(self, 'dataset'):
label_comps += [self.dataset_selected]
label_comps += ["collapsed"]
self.results_label_default = " ".join(label_comps)
[docs]
@with_spinner()
def collapse(self, add_data=True):
"""
Collapse over the spectral axis.
Parameters
----------
add_data : bool
Whether to load the resulting data back into the application according to
``add_results``.
"""
# Collapsing over the spectral axis. Cut out the desired spectral
# region. Defaults to the entire spectrum.
cube = self.dataset.selected_obj
spec_min, spec_max = self.spectral_subset.selected_min_max(cube)
# Extract 2D WCS from input cube.
data = self.dataset.selected_dc_item
# Similar to coords_info logic.
if '_orig_spec' in getattr(data, 'meta', {}):
w = data.meta['_orig_spec'].wcs
else:
w = data.coords
if isinstance(w, GWCS):
data_wcs = WCS(w.to_fits_sip())
else:
data_wcs = getattr(w, 'celestial', None)
with warnings.catch_warnings():
warnings.filterwarnings('ignore', message='No observer defined on WCS')
spec = spectral_slab(cube, spec_min, spec_max)
# Spatial-spatial image only.
collapsed_flux = spec.collapse(self.function_selected.lower(),
axis=cube.spectral_axis_index) # Quantity
# stuff for exporting to file
self.collapsed_flux = CCDData(collapsed_flux, wcs=data_wcs)
if add_data:
data = Data(coords=data_wcs)
data['flux'] = collapsed_flux.value
data.get_component('flux').units = collapsed_flux.unit.to_string()
self.add_results.add_results_from_plugin(data,
format='Image')
snackbar_message = SnackbarMessage(
f"Data set '{self.dataset_selected}' collapsed successfully.",
color="success",
sender=self)
self.hub.broadcast(snackbar_message)
return collapsed_flux
[docs]
def vue_collapse(self, *args, **kwargs):
try:
self.collapse(add_data=True)
except Exception as e:
snackbar_fail_message = SnackbarMessage(
f"Data set '{self.dataset_selected}' could not collapse successfully: {repr(e)}",
color="error",
traceback=e,
sender=self)
self.hub.broadcast(snackbar_fail_message)
