import warnings
from astropy.io.fits import getheader
from astropy.utils import deprecated
import astropy.units as u
import ipyvue
import os
from lightkurve import LightCurve
from glue.config import settings as glue_settings
from glue.core.link_helpers import LinkSame
from glue.core.units import unit_converter
from jdaviz.configs.default.plugins.viewers import JdavizViewerWindow
from jdaviz.core.helpers import ConfigHelper
from lcviz import __version__
from lcviz.viewers import TimeScatterView
__all__ = ['LCviz']
@unit_converter('custom-lcviz')
class UnitConverter:
def equivalent_units(self, data, cid, units):
return set(list(map(str, u.Unit(units).find_equivalent_units(include_prefix_units=True))))
def to_unit(self, data, cid, values, original_units, target_units):
# for some reason, glue is trying to request a change for cid='flux' from d to electron / s
if target_units not in self.equivalent_units(data, cid, original_units):
return values
return (values * u.Unit(original_units)).to_value(u.Unit(target_units))
glue_settings.UNIT_CONVERTER = 'custom-lcviz'
custom_components = {'plugin-ephemeris-select': 'components/plugin_ephemeris_select.vue'}
# Register pure vue component. This allows us to do recursive component instantiation only in the
# vue component file
for name, path in custom_components.items():
ipyvue.register_component_from_file(None, name,
os.path.join(os.path.dirname(__file__), path))
def _get_range_subset_bounds(self, subset_state, *args, **kwargs):
time_viewers = [v for v in self._viewer_store.values()
if isinstance(v, TimeScatterView)]
if not time_viewers: # pragma: no cover
raise ValueError("Unable to find time viewer")
viewer = time_viewers[0]
light_curve = viewer.data()[0]
reference_time = light_curve.meta['reference_time']
if viewer:
# TODO: use display units once implemented in Glue for ScatterViewer
# units = u.Unit(viewer.state.x_display_unit)
units = u.Unit(viewer.time_unit)
else: # pragma: no cover
raise ValueError("Unable to find time axis units")
region = reference_time + u.Quantity([subset_state.lo * units, subset_state.hi * units])
return [{"name": subset_state.__class__.__name__,
"glue_state": subset_state.__class__.__name__,
"region": region,
"subset_state": subset_state}]
def _link_new_data(app, reference_data=None, data_to_be_linked=None):
dc = app.data_collection
ref_data = dc[reference_data] if reference_data else dc[0]
linked_data = dc[data_to_be_linked] if data_to_be_linked else dc[-1]
new_link = LinkSame(
cid1=ref_data.world_component_ids[0],
cid2=linked_data.world_component_ids[0],
data1=ref_data,
data2=linked_data,
labels1=ref_data.label,
labels2=linked_data.label
)
dc.add_link(new_link)
return
def _get_display_unit(app, axis):
if app._jdaviz_helper is None or app._jdaviz_helper.plugins.get('Unit Conversion') is None: # noqa
# fallback on native units (unit conversion is not enabled)
if axis == 'time':
return u.d
elif axis == 'flux':
try:
time_viewers = [v for v in app._viewer_store.values()
if isinstance(v, TimeScatterView)]
if time_viewers and len(time_viewers[0].data()) > 0:
return time_viewers[0].data()[0].flux.unit
return u.electron / u.s
except (ValueError, IndexError):
return u.electron / u.s
else:
raise ValueError(f"could not find units for axis='{axis}'")
try:
# TODO: need to implement and add unit conversion plugin for this to be able to work
return getattr(app._jdaviz_helper.plugins.get('Unit Conversion')._obj,
f'{axis}_unit_selected')
except AttributeError:
raise ValueError(f"could not find display unit for axis='{axis}'")
def _apply_lcviz_patches(jdaviz_application):
"""
Apply lcviz-specific instance patches to a JdavizApplication for temporal data support.
Each patch falls back to the original jdaviz implementation when no temporal data is
present, so the patched app remains safe to use with spectral data as well.
Idempotent: calling multiple times on the same instance is safe.
"""
if getattr(jdaviz_application, '_lcviz_patched', False):
return
from lcviz.utils import TimeCoordinates
_original_get_range_subset_bounds = type(jdaviz_application)._get_range_subset_bounds
_original_get_display_unit = type(jdaviz_application)._get_display_unit
def _patched_get_range_subset_bounds(*args, **kwargs):
if any(isinstance(v, TimeScatterView)
for v in jdaviz_application._viewer_store.values()):
return _get_range_subset_bounds(jdaviz_application, *args, **kwargs)
return _original_get_range_subset_bounds(jdaviz_application, *args, **kwargs)
def _patched_link_new_data(reference_data=None, data_to_be_linked=None):
dc = jdaviz_application.data_collection
linked = dc[data_to_be_linked] if data_to_be_linked else (dc[-1] if len(dc) else None)
if linked is not None and isinstance(linked.coords, TimeCoordinates):
return _link_new_data(jdaviz_application, reference_data, data_to_be_linked)
# for non-time data in deconfigged config the base implementation is a no-op
def _patched_get_display_unit(*args, **kwargs):
axis = args[0] if args else kwargs.get('axis', '')
if axis in ('time', 'flux'):
return _get_display_unit(jdaviz_application, *args, **kwargs)
return _original_get_display_unit(jdaviz_application, *args, **kwargs)
jdaviz_application._get_range_subset_bounds = _patched_get_range_subset_bounds
jdaviz_application._link_new_data = _patched_link_new_data
jdaviz_application._get_display_unit = _patched_get_display_unit
jdaviz_application._lcviz_patched = True
if jdaviz_application.config == 'deconfigged':
jdaviz_application.update_tray_items_from_registry()
jdaviz_application.update_loaders_from_registry()
jdaviz_application.update_new_viewers_from_registry()
elif not jdaviz_application.state.new_viewer_items:
# jdaviz 5.0 only calls update_new_viewers_from_registry for deconfigged;
# populate state.new_viewer_items for lcviz config so new_viewers works.
import jdaviz.core.viewer_creators # noqa - ensure built-in creators are registered
from jdaviz.core.registries import viewer_creator_registry
new_viewer_items = []
for vc_registry_member in viewer_creator_registry.members.values():
try:
item = jdaviz_application._create_new_viewer_item(vc_registry_member)
except Exception: # nosec
continue
new_viewer_items.append(item)
jdaviz_application.state.new_viewer_items = new_viewer_items
try:
about = jdaviz_application.get_tray_item_from_name('about')
about.register_downstream_package('lcviz', __version__, abbreviation='lc')
except (KeyError, AttributeError): # pragma: no cover
pass
[docs]
class LCviz(ConfigHelper):
_default_configuration = {
'settings': {'configuration': 'lcviz',
'visible': {'menu_bar': False,
'toolbar': True,
'tray': True,
'tab_headers': True},
'dense_toolbar': False,
'context': {'notebook': {'max_height': '600px'}}},
'toolbar': ['g-data-tools', 'g-subset-tools', 'g-viewer-creator', 'g-coords-info'],
'tray': ['g-metadata-viewer', 'flux-column',
'g-plot-options', 'g-subset-tools',
'g-markers', 'time-selector', 'photometric-extraction',
'stitch', 'flatten', 'frequency-analysis', 'ephemeris',
'binning', 'export', 'logger'],
'viewer_area': [{'container': 'col'}]}
_component_ids = {}
def __init__(self, *args, **kwargs):
warnings.warn(
"The LCviz configuration class is deprecated and will be removed in a future release. "
"Please use jdaviz.show() instead.",
DeprecationWarning,
stacklevel=2
)
super().__init__(*args, **kwargs)
# override jdaviz behavior to support temporal subsets
_apply_lcviz_patches(self._app)
# inject custom css from lcviz_style.vue (on top of jdaviz styles)
self._app._add_style((__file__, 'lcviz_style.vue'))
# enable loaders (currently requires dev-flag in jdaviz)
self._app.state.dev_loaders = True
self.load = self._load
# set the link to read the docs
self._app.vdocs = 'latest' if 'dev' in __version__ else 'v'+__version__
self._app.docs_link = f"https://lcviz.readthedocs.io/en/{self._app.vdocs}"
for plugin in self.plugins.values():
# NOTE that plugins that need to override upstream docs_link should do so in
# an @observe('vdocs') rather than the init, since plugin instances have
# already been initialized
plugin._obj.vdocs = self._app.vdocs
[docs]
@deprecated(since="1.2", alternative="load")
def load_data(self, data, data_label=None, extname=None):
"""
Load data into LCviz.
Parameters
----------
data : obj or str
File name or object to be loaded. Supported formats include:
* ``'filename.fits'`` (or any extension that ``astropy.io.fits``
supports)
* `~lightkurve.LightCurve` (extracts the default flux column)
data_label : str or `None`
Data label to go with the given data. If not given, this is
automatically determined from filename or randomly generated.
The final label shown in LCviz may have additional information
appended for clarity.
extname : str or `None`
Used for DVT parsing if only a single TCE from a multi-TCE file should be
loaded. Formatted as 'TCE_1', 'TCE_2', etc.
"""
kwargs = {}
# Determine if we're loading a DVT file, which has a separate parser
if isinstance(data, str):
header = getheader(data)
if (header.get('TELESCOP', '') == 'TESS' and 'CREATOR' in header and
'DvTimeSeriesExporter' in header['CREATOR']):
kwargs['extension'] = extname
self.load(data, data_label=data_label, format=['Light Curve', 'TPF'], **kwargs)
[docs]
def get_data(self, data_label=None, cls=LightCurve, subset=None):
"""
Returns data with name equal to data_label of type cls with subsets applied from
subset_to_apply.
Parameters
----------
data_label : str, optional
Provide a label to retrieve a specific data set from data_collection.
cls : light curve class, optional
The type that data will be returned as.
subset : str, optional
Subset that is to be applied (as a mask) to the data before it is returned.
Returns
-------
data : cls
Data is returned as type cls with subsets applied.
"""
return super()._get_data(data_label=data_label, mask_subset=subset, cls=cls)
@property
def default_time_viewer(self):
tvs = [viewer for vid, viewer in self._app._viewer_store.items()
if isinstance(viewer, TimeScatterView)]
if not len(tvs):
raise ValueError("no time viewers exist")
return JdavizViewerWindow(tvs[0], app=self.app).user_api
@property
def _has_cube_data(self):
for data in self._app.data_collection:
if data.ndim == 3:
return True
return False
@property
def new_viewers(self):
# jdaviz 5.0 only exposes new_viewers for deconfigged; override to support lcviz config.
from ipywidgets.widgets import widget_serialization
return {item['label']: widget_serialization['from_json'](item['widget'], None).user_api
for item in self._app.state.new_viewer_items if item['is_relevant']}
@property
def _tray_tools(self):
"""
Access API objects for plugins in the app toolbar.
Returns
-------
plugins : dict
dict of plugin objects
"""
# TODO: provide user-friendly labels, user API, and move upstream to be public
# for now this is just useful for dev-debugging access to toolbar entries
from ipywidgets.widgets import widget_serialization
return {item['name']: widget_serialization['from_json'](item['widget'], None)
for item in self._app.state.tool_items}
