Skip to content

Visualization Module

Interactive Panel/HoloViews visualization framework for OVRO-LWA datasets. All dependencies are optional — install with pip install 'ovro_lwa_portal[visualization]'.

For usage examples, see the Interactive Visualization guide.

Module Entry Points

Interactive visualization framework for OVRO-LWA datasets.

This module provides Panel/HoloViews-based interactive explorers for radio astronomy data. All dependencies are optional — install with:

pip install 'ovro_lwa_portal[visualization]'

Examples:

>>> ds = ovro_lwa_portal.open_dataset("path/to/data.zarr")
>>> ds.radport.explore_image()  # Launch interactive image explorer
>>> ds.radport.explore()        # Launch full exploration dashboard

ImageExplorer(ds, **kwargs)

Create an interactive image explorer.

Parameters:

Name Type Description Default
ds Dataset

OVRO-LWA dataset to explore.

 required
**kwargs Any

Passed to :class:~ovro_lwa_portal.viz.explorers.ImageExplorer.

 {}

Returns:

Type Description
ImageExplorer

Explorer instance. Call .panel() to get a panel.viewable.Viewable layout.
Source code in src/ovro_lwa_portal/viz/__init__.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def ImageExplorer(ds: xr.Dataset, **kwargs: Any) -> Any:  # noqa: N802
    """Create an interactive image explorer.

    Parameters
    ----------
    ds : xr.Dataset
        OVRO-LWA dataset to explore.
    **kwargs
        Passed to :class:`~ovro_lwa_portal.viz.explorers.ImageExplorer`.

    Returns
    -------
    ovro_lwa_portal.viz.explorers.ImageExplorer
        Explorer instance. Call ``.panel()`` to get a
        ``panel.viewable.Viewable`` layout.
    """
    check_viz_deps()
    from ovro_lwa_portal.viz.explorers import ImageExplorer as _IE

    return _IE(ds, **kwargs)

DynamicSpectrumExplorer(ds, **kwargs)

Create an interactive dynamic spectrum explorer.

Parameters:

Name Type Description Default
ds Dataset

OVRO-LWA dataset to explore.

 required
**kwargs Any

Passed to :class:~ovro_lwa_portal.viz.explorers.DynamicSpectrumExplorer.

 {}

Returns:

Type Description
DynamicSpectrumExplorer

Explorer instance. Call .panel() to get a panel.viewable.Viewable layout.
Source code in src/ovro_lwa_portal/viz/__init__.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def DynamicSpectrumExplorer(ds: xr.Dataset, **kwargs: Any) -> Any:  # noqa: N802
    """Create an interactive dynamic spectrum explorer.

    Parameters
    ----------
    ds : xr.Dataset
        OVRO-LWA dataset to explore.
    **kwargs
        Passed to :class:`~ovro_lwa_portal.viz.explorers.DynamicSpectrumExplorer`.

    Returns
    -------
    ovro_lwa_portal.viz.explorers.DynamicSpectrumExplorer
        Explorer instance. Call ``.panel()`` to get a
        ``panel.viewable.Viewable`` layout.
    """
    check_viz_deps()
    from ovro_lwa_portal.viz.explorers import DynamicSpectrumExplorer as _DSE

    return _DSE(ds, **kwargs)

CutoutExplorer(ds, **kwargs)

Create an interactive cutout explorer.

Parameters:

Name Type Description Default
ds Dataset

OVRO-LWA dataset to explore.

 required
**kwargs Any

Passed to :class:~ovro_lwa_portal.viz.explorers.CutoutExplorer.

 {}

Returns:

Type Description
CutoutExplorer

Explorer instance. Call .panel() to get a panel.viewable.Viewable layout.
Source code in src/ovro_lwa_portal/viz/__init__.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def CutoutExplorer(ds: xr.Dataset, **kwargs: Any) -> Any:  # noqa: N802
    """Create an interactive cutout explorer.

    Parameters
    ----------
    ds : xr.Dataset
        OVRO-LWA dataset to explore.
    **kwargs
        Passed to :class:`~ovro_lwa_portal.viz.explorers.CutoutExplorer`.

    Returns
    -------
    ovro_lwa_portal.viz.explorers.CutoutExplorer
        Explorer instance. Call ``.panel()`` to get a
        ``panel.viewable.Viewable`` layout.
    """
    check_viz_deps()
    from ovro_lwa_portal.viz.explorers import CutoutExplorer as _CE

    return _CE(ds, **kwargs)

SkyViewer(ds, **kwargs)

Create an interactive sky viewer with Aladin Lite.

Overlays OVRO-LWA data on astronomical survey backgrounds (DSS, WISE, Planck, etc.) with real-time panning, zooming, and coordinate exploration. Requires WCS header in the dataset.

Parameters:

Name Type Description Default
ds Dataset

OVRO-LWA dataset with WCS header to explore.

 required
**kwargs Any

Passed to :class:~ovro_lwa_portal.viz.sky_viewer.SkyViewer.

 {}

Returns:

Type Description
SkyViewer

Sky viewer instance. Call .panel() to get a panel.viewable.Viewable layout.
Source code in src/ovro_lwa_portal/viz/__init__.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def SkyViewer(ds: xr.Dataset, **kwargs: Any) -> Any:  # noqa: N802
    """Create an interactive sky viewer with Aladin Lite.

    Overlays OVRO-LWA data on astronomical survey backgrounds (DSS,
    WISE, Planck, etc.) with real-time panning, zooming, and
    coordinate exploration. Requires WCS header in the dataset.

    Parameters
    ----------
    ds : xr.Dataset
        OVRO-LWA dataset with WCS header to explore.
    **kwargs
        Passed to :class:`~ovro_lwa_portal.viz.sky_viewer.SkyViewer`.

    Returns
    -------
    ovro_lwa_portal.viz.sky_viewer.SkyViewer
        Sky viewer instance. Call ``.panel()`` to get a
        ``panel.viewable.Viewable`` layout.
    """
    check_viz_deps()
    from ovro_lwa_portal.viz.sky_viewer import SkyViewer as _SV

    return _SV(ds, **kwargs)

create_exploration_dashboard(ds, **kwargs)

Create a comprehensive exploration dashboard.

Combines image, dynamic spectrum, and cutout explorers into a tabbed dashboard layout.

Parameters:

Name Type Description Default
ds Dataset

OVRO-LWA dataset to explore.

 required
**kwargs Any

Passed to :func:~ovro_lwa_portal.viz.dashboards.create_exploration_dashboard.

 {}

Returns:

Type Description
Viewable

Panel tabbed layout with all explorers.
Source code in src/ovro_lwa_portal/viz/__init__.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def create_exploration_dashboard(ds: xr.Dataset, **kwargs: Any) -> Any:
    """Create a comprehensive exploration dashboard.

    Combines image, dynamic spectrum, and cutout explorers into a
    tabbed dashboard layout.

    Parameters
    ----------
    ds : xr.Dataset
        OVRO-LWA dataset to explore.
    **kwargs
        Passed to :func:`~ovro_lwa_portal.viz.dashboards.create_exploration_dashboard`.

    Returns
    -------
    panel.viewable.Viewable
        Panel tabbed layout with all explorers.
    """
    check_viz_deps()
    from ovro_lwa_portal.viz.dashboards import create_exploration_dashboard as _ced

    return _ced(ds, **kwargs)

Explorer Classes

ImageExplorer

ImageExplorer

Bases: Parameterized

Interactive image explorer with time/frequency/polarization selection.

Uses an LRU cache — first view of each (time, freq) slice takes one S3 read (~3s), subsequent views of the same slice are instant.

Parameters:

Name Type Description Default
ds Dataset

OVRO-LWA dataset to explore.

 required
max_size int

Maximum pixels per spatial side after downsampling (controls stride). Lower values are faster but coarser. Default 512.

 _MAX_DISPLAY_SIZE
Source code in src/ovro_lwa_portal/viz/explorers.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
class ImageExplorer(param.Parameterized):
    """Interactive image explorer with time/frequency/polarization selection.

    Uses an LRU cache — first view of each (time, freq) slice takes one
    S3 read (~3s), subsequent views of the same slice are instant.

    Parameters
    ----------
    ds : xr.Dataset
        OVRO-LWA dataset to explore.
    max_size : int, optional
        Maximum pixels per spatial side after downsampling (controls stride).
        Lower values are faster but coarser. Default 512.
    """

    time_idx = param.Integer(default=0, bounds=(0, 1), doc="Time step index")
    freq_idx = param.Integer(default=0, bounds=(0, 1), doc="Frequency channel index")
    pol = param.Integer(default=0, bounds=(0, 0), doc="Polarization index")
    var = param.Selector(
        default="SKY", objects=["SKY"], doc="Data variable to display"
    )
    cmap = param.Selector(
        default="inferno", objects=COLORMAPS, doc="Colormap"
    )
    robust = param.Boolean(default=True, doc="Use robust (percentile) color scaling")

    def __init__(self, ds: xr.Dataset, *, max_size: int = _MAX_DISPLAY_SIZE, **params: Any) -> None:
        # Set bounds before super().__init__ so incoming params validate
        # against dataset-derived ranges, not stale defaults.
        n_times = ds.sizes["time"]
        n_freqs = ds.sizes["frequency"]
        n_pols = ds.sizes["polarization"]

        self.param.time_idx.bounds = (0, max(0, n_times - 1))
        self.param.freq_idx.bounds = (0, max(0, n_freqs - 1))
        self.param.pol.bounds = (0, max(0, n_pols - 1))

        available_vars = [v for v in ["SKY", "BEAM"] if v in ds.data_vars]
        self.param.var.objects = available_vars
        if available_vars:
            params.setdefault("var", available_vars[0])

        super().__init__(**params)
        self._ds = ds
        self._max_size = max_size

        # LRU-cached cube — no upfront load, fetches one slice at a time.
        # Recreated when var or pol changes via _get_cube().
        self._cube: PreloadedCube | None = None
        self._cube_var: str | None = None
        self._cube_pol: int | None = None

        self._time_labels = {
            i: f"{float(ds.coords['time'].values[i]):.4f}"
            for i in range(n_times)
        }
        self._freq_labels = {
            i: f"{float(ds.coords['frequency'].values[i]) / 1e6:.1f} MHz"
            for i in range(n_freqs)
        }

    def _get_cube(self) -> PreloadedCube:
        """Get or create a PreloadedCube for the current var/pol."""
        if self._cube is None or self._cube_var != self.var or self._cube_pol != self.pol:
            self._cube = PreloadedCube(
                self._ds, var=self.var, pol=self.pol, max_size=self._max_size,
            )
            self._cube_var = self.var
            self._cube_pol = self.pol
        return self._cube

    @param.depends("time_idx", "freq_idx", "var", "pol", "cmap", "robust")
    def _image_view(self) -> hv.Image:
        img = sky_image_element(
            self._get_cube(),
            time_idx=self.time_idx,
            freq_idx=self.freq_idx,
            robust=self.robust,
        )
        return style_sky_image(img, cmap=self.cmap)

    @param.depends("time_idx")
    def _time_label(self) -> str:
        return f"**Time:** {self._time_labels.get(self.time_idx, '?')} MJD"

    @param.depends("freq_idx")
    def _freq_label(self) -> str:
        return f"**Frequency:** {self._freq_labels.get(self.freq_idx, '?')}"

    def panel(self) -> pn.viewable.Viewable:
        """Return the complete Panel layout."""
        time_label = pn.pane.Markdown(self._time_label, width=250)
        freq_label = pn.pane.Markdown(self._freq_label, width=250)

        controls = pn.Column(
            "## Image Explorer",
            pn.widgets.IntSlider.from_param(self.param.time_idx, name="Time Step"),
            time_label,
            pn.widgets.IntSlider.from_param(self.param.freq_idx, name="Frequency Channel"),
            freq_label,
            pn.widgets.Select.from_param(self.param.var, name="Variable"),
            pn.widgets.IntSlider.from_param(self.param.pol, name="Polarization"),
            pn.widgets.Select.from_param(self.param.cmap, name="Colormap"),
            pn.widgets.Checkbox.from_param(self.param.robust, name="Robust Scaling"),
            width=280,
        )

        image_pane = pn.pane.HoloViews(hv.DynamicMap(self._image_view))

        return pn.Row(controls, image_pane)

panel()

Return the complete Panel layout.

Source code in src/ovro_lwa_portal/viz/explorers.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def panel(self) -> pn.viewable.Viewable:
    """Return the complete Panel layout."""
    time_label = pn.pane.Markdown(self._time_label, width=250)
    freq_label = pn.pane.Markdown(self._freq_label, width=250)

    controls = pn.Column(
        "## Image Explorer",
        pn.widgets.IntSlider.from_param(self.param.time_idx, name="Time Step"),
        time_label,
        pn.widgets.IntSlider.from_param(self.param.freq_idx, name="Frequency Channel"),
        freq_label,
        pn.widgets.Select.from_param(self.param.var, name="Variable"),
        pn.widgets.IntSlider.from_param(self.param.pol, name="Polarization"),
        pn.widgets.Select.from_param(self.param.cmap, name="Colormap"),
        pn.widgets.Checkbox.from_param(self.param.robust, name="Robust Scaling"),
        width=280,
    )

    image_pane = pn.pane.HoloViews(hv.DynamicMap(self._image_view))

    return pn.Row(controls, image_pane)

DynamicSpectrumExplorer

DynamicSpectrumExplorer

Bases: Parameterized

Interactive dynamic spectrum explorer with linked views.

Uses the accessor's dynamic_spectrum() method which batches all pixel extractions via dask.compute() — much faster than 100 sequential reads. Linked spectrum/light curve use the accessor's single-frame methods (one chunk read each).

Parameters:

Name Type Description Default
ds Dataset

OVRO-LWA dataset to explore.

 required
Source code in src/ovro_lwa_portal/viz/explorers.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
class DynamicSpectrumExplorer(param.Parameterized):
    """Interactive dynamic spectrum explorer with linked views.

    Uses the accessor's ``dynamic_spectrum()`` method which batches all
    pixel extractions via ``dask.compute()`` — much faster than 100
    sequential reads. Linked spectrum/light curve use the accessor's
    single-frame methods (one chunk read each).

    Parameters
    ----------
    ds : xr.Dataset
        OVRO-LWA dataset to explore.
    """

    l_val = param.Number(default=0.0, bounds=(-1.2, 1.2), step=0.01, doc="l direction cosine")
    m_val = param.Number(default=0.0, bounds=(-1.2, 1.2), step=0.01, doc="m direction cosine")
    cmap = param.Selector(default="inferno", objects=COLORMAPS, doc="Colormap")
    robust = param.Boolean(default=True, doc="Use robust color scaling")

    def __init__(
        self,
        ds: xr.Dataset,
        *,
        l: float | None = None,
        m: float | None = None,
        **params: Any,
    ) -> None:
        # Set bounds before super().__init__ so incoming params validate
        # against dataset-derived ranges.
        l_vals = ds.coords["l"].values
        m_vals = ds.coords["m"].values
        self.param.l_val.bounds = (
            float(min(l_vals[0], l_vals[-1])),
            float(max(l_vals[0], l_vals[-1])),
        )
        self.param.m_val.bounds = (
            float(min(m_vals[0], m_vals[-1])),
            float(max(m_vals[0], m_vals[-1])),
        )

        if l is not None:
            params["l_val"] = l
        if m is not None:
            params["m_val"] = m

        super().__init__(**params)
        self._ds = ds

        self._tap = hv.streams.Tap(x=None, y=None)

        # Cache for dynamic spectra keyed by (l_val, m_val)
        self._dynspec_cache: dict[tuple[float, float], Any] = {}

    def _get_dynspec(self) -> Any:
        """Get dynamic spectrum DataArray, using cache if available."""
        key = (round(self.l_val, 4), round(self.m_val, 4))
        if key not in self._dynspec_cache:
            self._dynspec_cache[key] = self._ds.radport.dynamic_spectrum(
                l=self.l_val, m=self.m_val
            )
        return self._dynspec_cache[key]

    @param.depends("l_val", "m_val", "cmap", "robust")
    def _dynspec_view(self) -> hv.Image:
        from ovro_lwa_portal.viz._data import _ensure_extension

        _ensure_extension()

        dynspec = self._get_dynspec()
        time_mjd = dynspec.coords["time"].values
        freq_mhz = dynspec.coords["frequency"].values / 1e6
        data = dynspec.values

        pixel_l = dynspec.attrs.get("pixel_l", "?")
        pixel_m = dynspec.attrs.get("pixel_m", "?")
        if isinstance(pixel_l, (int, float)):
            title = f"Dynamic Spectrum at l={pixel_l:+.4f}, m={pixel_m:+.4f}"
        else:
            title = f"Dynamic Spectrum at l={self.l_val:+.4f}, m={self.m_val:+.4f}"

        # Add half-step padding so each pixel has visible extent.
        # Without this, a narrow time range (e.g., 0.001 MJD span) can
        # produce bounds where left ≈ right, rendering an empty image.
        if len(time_mjd) > 1:
            dt = float(time_mjd[1] - time_mjd[0]) / 2
        else:
            dt = 0.0001
        if len(freq_mhz) > 1:
            df = float(freq_mhz[1] - freq_mhz[0]) / 2
        else:
            df = 1.0

        bounds = (
            float(time_mjd[0]) - dt, float(freq_mhz[0]) - df,
            float(time_mjd[-1]) + dt, float(freq_mhz[-1]) + df,
        )

        img = hv.Image(
            data, kdims=["Time (MJD)", "Frequency (MHz)"], bounds=bounds,
        ).opts(
            xlabel="Time (MJD)", ylabel="Frequency (MHz)",
            title=title, colorbar=True, clabel="Jy/beam",
        )

        if self.robust:
            finite = data[np.isfinite(data)]
            if finite.size > 0:
                img = img.opts(
                    clim=(float(np.percentile(finite, 2)), float(np.percentile(finite, 98)))
                )

        img = style_spectrum_image(img, cmap=self.cmap)
        self._tap.source = img
        return img

    def _linked_spectrum(self, x: float | None, y: float | None) -> hv.Curve:
        """Spectrum at the clicked time step — one chunk read."""
        from ovro_lwa_portal.viz._data import _ensure_extension

        _ensure_extension()

        if x is None:
            return hv.Curve([], kdims=["Frequency (MHz)"], vdims=["Intensity (Jy/beam)"]).opts(
                title="Click on dynamic spectrum to show spectrum"
            )

        time_vals = self._ds.coords["time"].values
        time_idx = int(np.abs(time_vals - x).argmin())

        spec = self._ds.radport.spectrum(
            l=self.l_val, m=self.m_val, time_idx=time_idx
        )
        freq_mhz = spec.coords["frequency"].values / 1e6
        time_val = float(time_vals[time_idx])

        curve = hv.Curve(
            (freq_mhz, spec.values),
            kdims=["Frequency (MHz)"], vdims=["Intensity (Jy/beam)"],
        ).opts(title=f"Spectrum at t={time_val:.4f} MJD")
        return style_curve(curve)

    def _linked_light_curve(self, x: float | None, y: float | None) -> hv.Curve:
        """Light curve at the clicked frequency — one chunk read per time step."""
        from ovro_lwa_portal.viz._data import _ensure_extension

        _ensure_extension()

        if y is None:
            return hv.Curve([], kdims=["Time (MJD)"], vdims=["Intensity (Jy/beam)"]).opts(
                title="Click on dynamic spectrum to show light curve"
            )

        freq_hz = self._ds.coords["frequency"].values
        freq_mhz = freq_hz / 1e6
        freq_idx = int(np.abs(freq_mhz - y).argmin())

        lc = self._ds.radport.light_curve(
            l=self.l_val, m=self.m_val, freq_idx=freq_idx
        )

        curve = hv.Curve(
            (lc.coords["time"].values, lc.values),
            kdims=["Time (MJD)"], vdims=["Intensity (Jy/beam)"],
        ).opts(title=f"Light Curve at f={float(freq_mhz[freq_idx]):.1f} MHz")
        return style_curve(curve)

    def panel(self) -> pn.viewable.Viewable:
        """Return the complete Panel layout with linked views."""
        controls = pn.Column(
            "## Dynamic Spectrum Explorer",
            pn.widgets.FloatSlider.from_param(self.param.l_val, name="l", step=0.01),
            pn.widgets.FloatSlider.from_param(self.param.m_val, name="m", step=0.01),
            pn.widgets.Select.from_param(self.param.cmap, name="Colormap"),
            pn.widgets.Checkbox.from_param(self.param.robust, name="Robust Scaling"),
            "---",
            "*Initial load fetches pixel data across all time/freq slices.*",
            width=280,
        )

        dynspec_pane = pn.pane.HoloViews(hv.DynamicMap(self._dynspec_view))
        spectrum_pane = pn.pane.HoloViews(
            hv.DynamicMap(self._linked_spectrum, streams=[self._tap]),
        )
        lightcurve_pane = pn.pane.HoloViews(
            hv.DynamicMap(self._linked_light_curve, streams=[self._tap]),
        )

        return pn.Row(
            controls,
            dynspec_pane,
            pn.Column(spectrum_pane, lightcurve_pane),
        )

panel()

Return the complete Panel layout with linked views.

Source code in src/ovro_lwa_portal/viz/explorers.py 
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
def panel(self) -> pn.viewable.Viewable:
    """Return the complete Panel layout with linked views."""
    controls = pn.Column(
        "## Dynamic Spectrum Explorer",
        pn.widgets.FloatSlider.from_param(self.param.l_val, name="l", step=0.01),
        pn.widgets.FloatSlider.from_param(self.param.m_val, name="m", step=0.01),
        pn.widgets.Select.from_param(self.param.cmap, name="Colormap"),
        pn.widgets.Checkbox.from_param(self.param.robust, name="Robust Scaling"),
        "---",
        "*Initial load fetches pixel data across all time/freq slices.*",
        width=280,
    )

    dynspec_pane = pn.pane.HoloViews(hv.DynamicMap(self._dynspec_view))
    spectrum_pane = pn.pane.HoloViews(
        hv.DynamicMap(self._linked_spectrum, streams=[self._tap]),
    )
    lightcurve_pane = pn.pane.HoloViews(
        hv.DynamicMap(self._linked_light_curve, streams=[self._tap]),
    )

    return pn.Row(
        controls,
        dynspec_pane,
        pn.Column(spectrum_pane, lightcurve_pane),
    )

CutoutExplorer

CutoutExplorer

Bases: Parameterized

Interactive cutout explorer with linked light curve and spectrum.

Uses an LRU-cached cube for spatial subsetting — fast for single frames.

Parameters:

Name Type Description Default
ds Dataset

OVRO-LWA dataset to explore.

 required
max_size int

Maximum pixels per spatial side after downsampling (controls stride). Lower values are faster but coarser. Default 512.

 _MAX_DISPLAY_SIZE
Source code in src/ovro_lwa_portal/viz/explorers.py 
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
class CutoutExplorer(param.Parameterized):
    """Interactive cutout explorer with linked light curve and spectrum.

    Uses an LRU-cached cube for spatial subsetting — fast for single frames.

    Parameters
    ----------
    ds : xr.Dataset
        OVRO-LWA dataset to explore.
    max_size : int, optional
        Maximum pixels per spatial side after downsampling (controls stride).
        Lower values are faster but coarser. Default 512.
    """

    l_center = param.Number(default=0.0, step=0.01, doc="l center")
    m_center = param.Number(default=0.0, step=0.01, doc="m center")
    dl = param.Number(default=0.1, bounds=(0.01, 1.0), step=0.01, doc="l half-extent")
    dm = param.Number(default=0.1, bounds=(0.01, 1.0), step=0.01, doc="m half-extent")
    time_idx = param.Integer(default=0, bounds=(0, 1), doc="Time step")
    freq_idx = param.Integer(default=0, bounds=(0, 1), doc="Frequency channel")
    cmap = param.Selector(default="inferno", objects=COLORMAPS, doc="Colormap")
    robust = param.Boolean(default=True, doc="Robust scaling")

    def __init__(self, ds: xr.Dataset, *, max_size: int = _MAX_DISPLAY_SIZE, **params: Any) -> None:
        # Set bounds before super().__init__ so incoming params validate
        # against dataset-derived ranges.
        n_times = ds.sizes["time"]
        n_freqs = ds.sizes["frequency"]
        self.param.time_idx.bounds = (0, max(0, n_times - 1))
        self.param.freq_idx.bounds = (0, max(0, n_freqs - 1))

        super().__init__(**params)
        self._ds = ds

        var = "SKY" if "SKY" in ds.data_vars else list(ds.data_vars)[0]
        self._cube = PreloadedCube(ds, var=var, pol=0, max_size=max_size)

        self._tap = hv.streams.Tap(x=None, y=None)

    @param.depends("l_center", "m_center", "dl", "dm", "time_idx", "freq_idx", "cmap", "robust")
    def _cutout_view(self) -> hv.Image:
        img = cutout_image_element(
            self._cube,
            l_center=self.l_center, m_center=self.m_center,
            dl=self.dl, dm=self.dm,
            time_idx=self.time_idx, freq_idx=self.freq_idx,
            robust=self.robust,
        )
        img = style_sky_image(img, cmap=self.cmap)
        self._tap.source = img
        return img

    def _linked_spectrum(self, x: float | None, y: float | None) -> hv.Curve:
        from ovro_lwa_portal.viz._data import _ensure_extension, spectrum_element

        _ensure_extension()

        if x is None or y is None:
            return hv.Curve([], kdims=["Frequency (MHz)"], vdims=["Intensity (Jy/beam)"]).opts(
                title="Click on cutout to show spectrum"
            )
        curve = spectrum_element(self._cube, l=x, m=y, time_idx=self.time_idx)
        curve = curve.opts(title=f"Spectrum at l={x:.4f}, m={y:.4f}")
        return style_curve(curve)

    def _linked_light_curve(self, x: float | None, y: float | None) -> hv.Curve:
        from ovro_lwa_portal.viz._data import _ensure_extension, light_curve_element

        _ensure_extension()

        if x is None or y is None:
            return hv.Curve([], kdims=["Time (MJD)"], vdims=["Intensity (Jy/beam)"]).opts(
                title="Click on cutout to show light curve"
            )
        curve = light_curve_element(self._cube, l=x, m=y, freq_idx=self.freq_idx)
        curve = curve.opts(title=f"Light Curve at l={x:.4f}, m={y:.4f}")
        return style_curve(curve)

    def panel(self) -> pn.viewable.Viewable:
        """Return the complete Panel layout with linked views."""
        controls = pn.Column(
            "## Cutout Explorer",
            pn.widgets.FloatSlider.from_param(self.param.l_center, name="l center", step=0.01),
            pn.widgets.FloatSlider.from_param(self.param.m_center, name="m center", step=0.01),
            pn.widgets.FloatSlider.from_param(self.param.dl, name="dl (half-extent)", step=0.01),
            pn.widgets.FloatSlider.from_param(self.param.dm, name="dm (half-extent)", step=0.01),
            pn.widgets.IntSlider.from_param(self.param.time_idx, name="Time Step"),
            pn.widgets.IntSlider.from_param(self.param.freq_idx, name="Frequency"),
            pn.widgets.Select.from_param(self.param.cmap, name="Colormap"),
            pn.widgets.Checkbox.from_param(self.param.robust, name="Robust Scaling"),
            width=280,
        )

        cutout_pane = pn.pane.HoloViews(hv.DynamicMap(self._cutout_view))
        spectrum_pane = pn.pane.HoloViews(
            hv.DynamicMap(self._linked_spectrum, streams=[self._tap]),
        )
        lightcurve_pane = pn.pane.HoloViews(
            hv.DynamicMap(self._linked_light_curve, streams=[self._tap]),
        )

        return pn.Row(
            controls,
            cutout_pane,
            pn.Column(spectrum_pane, lightcurve_pane),
        )

panel()

Return the complete Panel layout with linked views.

Source code in src/ovro_lwa_portal/viz/explorers.py 
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
def panel(self) -> pn.viewable.Viewable:
    """Return the complete Panel layout with linked views."""
    controls = pn.Column(
        "## Cutout Explorer",
        pn.widgets.FloatSlider.from_param(self.param.l_center, name="l center", step=0.01),
        pn.widgets.FloatSlider.from_param(self.param.m_center, name="m center", step=0.01),
        pn.widgets.FloatSlider.from_param(self.param.dl, name="dl (half-extent)", step=0.01),
        pn.widgets.FloatSlider.from_param(self.param.dm, name="dm (half-extent)", step=0.01),
        pn.widgets.IntSlider.from_param(self.param.time_idx, name="Time Step"),
        pn.widgets.IntSlider.from_param(self.param.freq_idx, name="Frequency"),
        pn.widgets.Select.from_param(self.param.cmap, name="Colormap"),
        pn.widgets.Checkbox.from_param(self.param.robust, name="Robust Scaling"),
        width=280,
    )

    cutout_pane = pn.pane.HoloViews(hv.DynamicMap(self._cutout_view))
    spectrum_pane = pn.pane.HoloViews(
        hv.DynamicMap(self._linked_spectrum, streams=[self._tap]),
    )
    lightcurve_pane = pn.pane.HoloViews(
        hv.DynamicMap(self._linked_light_curve, streams=[self._tap]),
    )

    return pn.Row(
        controls,
        cutout_pane,
        pn.Column(spectrum_pane, lightcurve_pane),
    )

SkyViewer

SkyViewer

Bases: Parameterized

Interactive sky viewer with OVRO-LWA data overlaid on survey backgrounds.

Uses ipyaladin (Aladin Lite) to provide a real-time, pannable, zoomable celestial sky view. The OVRO-LWA image is projected onto the sky using the dataset's WCS header.

Parameters:

Name Type Description Default
ds Dataset

OVRO-LWA dataset with WCS header.

 required

Examples:

>>> viewer = SkyViewer(ds)
>>> viewer.panel()  # Launch in notebook or panel serve
Source code in src/ovro_lwa_portal/viz/sky_viewer.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
class SkyViewer(param.Parameterized):
    """Interactive sky viewer with OVRO-LWA data overlaid on survey backgrounds.

    Uses ipyaladin (Aladin Lite) to provide a real-time, pannable,
    zoomable celestial sky view. The OVRO-LWA image is projected onto
    the sky using the dataset's WCS header.

    Parameters
    ----------
    ds : xr.Dataset
        OVRO-LWA dataset with WCS header.

    Examples
    --------
    >>> viewer = SkyViewer(ds)
    >>> viewer.panel()  # Launch in notebook or panel serve
    """

    time_idx = param.Integer(default=0, bounds=(0, 1), doc="Time step")
    freq_idx = param.Integer(default=0, bounds=(0, 1), doc="Frequency channel")
    pol = param.Integer(default=0, bounds=(0, 0), doc="Polarization")
    var = param.Selector(default="SKY", objects=["SKY"], doc="Variable")

    survey = param.Selector(
        default="DSS Color",
        objects=list(SURVEY_PRESETS.keys()),
        doc="Background survey",
    )
    overlay_opacity = param.Number(
        default=0.7, bounds=(0.0, 1.0), step=0.05,
        doc="OVRO-LWA overlay opacity",
    )
    colormap = param.Selector(
        default="inferno", objects=COLORMAPS, doc="Overlay colormap"
    )
    stretch = param.Selector(
        default="linear",
        objects=["linear", "log", "sqrt", "pow2"],
        doc="Color stretch function",
    )
    robust = param.Boolean(default=True, doc="Robust percentile clipping")
    fov = param.Number(
        default=180.0, bounds=(0.1, 180.0), step=1.0,
        doc="Field of view (degrees)",
    )

    def __init__(self, ds: xr.Dataset, **params: Any) -> None:
        # Set bounds before super().__init__ so incoming params validate
        # against dataset-derived ranges.
        n_times = ds.sizes["time"]
        n_freqs = ds.sizes["frequency"]
        n_pols = ds.sizes["polarization"]

        self.param.time_idx.bounds = (0, max(0, n_times - 1))
        self.param.freq_idx.bounds = (0, max(0, n_freqs - 1))
        self.param.pol.bounds = (0, max(0, n_pols - 1))

        available_vars = [v for v in ["SKY", "BEAM"] if v in ds.data_vars]
        self.param.var.objects = available_vars
        if available_vars:
            params.setdefault("var", available_vars[0])

        super().__init__(**params)
        self._ds = ds

        # Precompute display labels
        self._time_labels = {
            i: f"{float(ds.coords['time'].values[i]):.4f}"
            for i in range(n_times)
        }
        self._freq_labels = {
            i: f"{float(ds.coords['frequency'].values[i]) / 1e6:.1f} MHz"
            for i in range(n_freqs)
        }

        # Get the phase center from WCS for initial Aladin target
        self._phase_center_ra = 0.0
        self._phase_center_dec = 0.0
        if ds.radport.has_wcs:
            try:
                wcs = ds.radport._get_wcs()
                self._phase_center_ra = float(wcs.wcs.crval[0])
                self._phase_center_dec = float(wcs.wcs.crval[1])
            except Exception:  # noqa: BLE001
                pass

        # Create the Aladin widget
        from astropy.coordinates import SkyCoord
        from ipyaladin import Aladin

        self._aladin = Aladin(
            target=SkyCoord(
                ra=self._phase_center_ra,
                dec=self._phase_center_dec,
                unit="deg",
                frame="fk5",
            ),
            fov=self.fov,
            survey=SURVEY_PRESETS[self.survey],
            projection="SIN",
            show_coo_grid=True,
            show_coo_grid_control=True,
            show_settings_control=True,
            height=600,
        )

        # Track the current overlay name for removal
        self._current_overlay_name: str | None = None

    def _update_overlay(self) -> None:
        """Update the OVRO-LWA image overlay on Aladin."""
        try:
            hdul = _build_fits_hdu(
                self._ds,
                time_idx=self.time_idx,
                freq_idx=self.freq_idx,
                pol=self.pol,
                var=self.var,
                robust=self.robust,
            )
        except ValueError:
            return  # No WCS header — skip overlay

        # Use a static name so add_fits replaces the existing layer
        # instead of accumulating new ones on each parameter change.
        overlay_name = "OVRO-LWA overlay"

        self._aladin.add_fits(
            hdul,
            name=overlay_name,
            opacity=self.overlay_opacity,
            colormap=self.colormap,
            stretch=self.stretch,
        )
        self._current_overlay_name = overlay_name

    def _on_param_change(self, event: Any) -> None:
        """React to parameter changes by updating the overlay."""
        if event.name == "survey":
            self._aladin.survey = SURVEY_PRESETS[self.survey]
        elif event.name == "fov":
            self._aladin.fov = self.fov
        elif event.name in (
            "time_idx", "freq_idx", "pol", "var",
            "overlay_opacity", "colormap", "stretch", "robust",
        ):
            self._update_overlay()

    @param.depends("time_idx")
    def _time_label(self) -> str:
        return f"**Time:** {self._time_labels.get(self.time_idx, '?')} MJD"

    @param.depends("freq_idx")
    def _freq_label(self) -> str:
        return f"**Frequency:** {self._freq_labels.get(self.freq_idx, '?')}"

    def panel(self) -> pn.viewable.Viewable:
        """Return the complete Panel layout with sky viewer.

        Returns
        -------
        pn.viewable.Viewable
            Panel layout with Aladin sky viewer and controls.
        """
        # Enable ipywidgets integration for Panel (required for ipyaladin)
        try:
            pn.extension("ipywidgets")
        except Exception:  # noqa: BLE001
            pass  # May fail outside notebook context; widget still works

        # Watch all relevant parameters
        self.param.watch(
            self._on_param_change,
            [
                "time_idx", "freq_idx", "pol", "var",
                "survey", "overlay_opacity", "colormap", "stretch",
                "robust", "fov",
            ],
        )

        time_label = pn.pane.Markdown(self._time_label, width=250)
        freq_label = pn.pane.Markdown(self._freq_label, width=250)

        controls = pn.Column(
            "## Sky Viewer",
            pn.widgets.IntSlider.from_param(self.param.time_idx, name="Time Step"),
            time_label,
            pn.widgets.IntSlider.from_param(self.param.freq_idx, name="Frequency"),
            freq_label,
            pn.widgets.IntSlider.from_param(self.param.pol, name="Polarization"),
            pn.widgets.Select.from_param(self.param.var, name="Variable"),
            "---",
            pn.widgets.Select.from_param(self.param.survey, name="Background Survey"),
            pn.widgets.FloatSlider.from_param(
                self.param.overlay_opacity, name="Overlay Opacity", step=0.05
            ),
            pn.widgets.Select.from_param(self.param.colormap, name="Colormap"),
            pn.widgets.Select.from_param(self.param.stretch, name="Stretch"),
            pn.widgets.Checkbox.from_param(self.param.robust, name="Robust Clipping"),
            pn.widgets.FloatSlider.from_param(
                self.param.fov, name="Field of View (\u00b0)", step=1.0
            ),
            width=280,
        )

        # Display the Aladin widget via ipywidgets Output to avoid
        # conflicts with Panel's comms="default" mode. IPyWidget pane
        # requires ipywidgets comms which may not be active.
        import ipywidgets as ipw

        output = ipw.Output()
        with output:
            from IPython.display import display
            display(self._aladin)

        aladin_pane = pn.pane.IPyWidget(output, width=700, height=600)

        # Load initial overlay after widget renders
        def _load_initial_overlay(event: Any = None) -> None:
            self._update_overlay()

        pn.state.onload(_load_initial_overlay)

        return pn.Row(controls, aladin_pane, sizing_mode="stretch_width", min_height=600)

panel()

Return the complete Panel layout with sky viewer.

Returns:

Type Description
Viewable

Panel layout with Aladin sky viewer and controls.
Source code in src/ovro_lwa_portal/viz/sky_viewer.py 
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
def panel(self) -> pn.viewable.Viewable:
    """Return the complete Panel layout with sky viewer.

    Returns
    -------
    pn.viewable.Viewable
        Panel layout with Aladin sky viewer and controls.
    """
    # Enable ipywidgets integration for Panel (required for ipyaladin)
    try:
        pn.extension("ipywidgets")
    except Exception:  # noqa: BLE001
        pass  # May fail outside notebook context; widget still works

    # Watch all relevant parameters
    self.param.watch(
        self._on_param_change,
        [
            "time_idx", "freq_idx", "pol", "var",
            "survey", "overlay_opacity", "colormap", "stretch",
            "robust", "fov",
        ],
    )

    time_label = pn.pane.Markdown(self._time_label, width=250)
    freq_label = pn.pane.Markdown(self._freq_label, width=250)

    controls = pn.Column(
        "## Sky Viewer",
        pn.widgets.IntSlider.from_param(self.param.time_idx, name="Time Step"),
        time_label,
        pn.widgets.IntSlider.from_param(self.param.freq_idx, name="Frequency"),
        freq_label,
        pn.widgets.IntSlider.from_param(self.param.pol, name="Polarization"),
        pn.widgets.Select.from_param(self.param.var, name="Variable"),
        "---",
        pn.widgets.Select.from_param(self.param.survey, name="Background Survey"),
        pn.widgets.FloatSlider.from_param(
            self.param.overlay_opacity, name="Overlay Opacity", step=0.05
        ),
        pn.widgets.Select.from_param(self.param.colormap, name="Colormap"),
        pn.widgets.Select.from_param(self.param.stretch, name="Stretch"),
        pn.widgets.Checkbox.from_param(self.param.robust, name="Robust Clipping"),
        pn.widgets.FloatSlider.from_param(
            self.param.fov, name="Field of View (\u00b0)", step=1.0
        ),
        width=280,
    )

    # Display the Aladin widget via ipywidgets Output to avoid
    # conflicts with Panel's comms="default" mode. IPyWidget pane
    # requires ipywidgets comms which may not be active.
    import ipywidgets as ipw

    output = ipw.Output()
    with output:
        from IPython.display import display
        display(self._aladin)

    aladin_pane = pn.pane.IPyWidget(output, width=700, height=600)

    # Load initial overlay after widget renders
    def _load_initial_overlay(event: Any = None) -> None:
        self._update_overlay()

    pn.state.onload(_load_initial_overlay)

    return pn.Row(controls, aladin_pane, sizing_mode="stretch_width", min_height=600)

Data Bridge

Internal utilities for converting accessor outputs to HoloViews elements.

PreloadedCube

PreloadedCube

Cached, spatially downsampled accessor for OVRO-LWA datasets.

Instead of loading the full data cube into memory, this class loads individual 2D slices on demand with strided downsampling and caches them. For the dynamic spectrum (all times x freqs at one pixel), it precomputes a small downsampled cube lazily.

Parameters:

Name Type Description Default
ds Dataset

OVRO-LWA dataset (may be dask-backed / remote).

 required
var str

Data variable to use (default "SKY").

 'SKY'
pol int

Polarization index.

 0
max_size int

Maximum spatial dimension size after downsampling.

 _MAX_DISPLAY_SIZE
Source code in src/ovro_lwa_portal/viz/_data.py 
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
class PreloadedCube:
    """Cached, spatially downsampled accessor for OVRO-LWA datasets.

    Instead of loading the full data cube into memory, this class loads
    individual 2D slices on demand with strided downsampling and caches
    them. For the dynamic spectrum (all times x freqs at one pixel),
    it precomputes a small downsampled cube lazily.

    Parameters
    ----------
    ds : xr.Dataset
        OVRO-LWA dataset (may be dask-backed / remote).
    var : str
        Data variable to use (default ``"SKY"``).
    pol : int
        Polarization index.
    max_size : int
        Maximum spatial dimension size after downsampling.
    """

    def __init__(
        self,
        ds: xr.Dataset,
        var: str = "SKY",
        pol: int = 0,
        max_size: int = _MAX_DISPLAY_SIZE,
    ) -> None:
        self._ds = ds
        self.var = var
        self.pol = pol
        self._max_size = max_size

        # Compute stride factors
        n_l = ds.sizes["l"]
        n_m = ds.sizes["m"]
        self.stride_l = max(1, -(-n_l // max_size))
        self.stride_m = max(1, -(-n_m // max_size))

        # Cache coordinate arrays (strided to match display resolution)
        self.l_vals = ds.coords["l"].values[::self.stride_l]
        self.m_vals = ds.coords["m"].values[::self.stride_m]
        self.time_vals = ds.coords["time"].values
        self.freq_vals = ds.coords["frequency"].values
        self.freq_mhz = self.freq_vals / 1e6
        self.n_times = len(self.time_vals)
        self.n_freqs = len(self.freq_vals)

        # Dynamic spectrum cache (lazily computed)
        self._dynspec_cache: dict[tuple[int, int], np.ndarray] = {}

        # Per-instance LRU cache for _load_slice (avoids caching self
        # in a module-level lru_cache, which would pin the instance and
        # prevent garbage collection).
        self._load_slice = lru_cache(maxsize=_CACHE_SIZE)(self._load_slice_impl)

        print(  # noqa: T201
            f"PreloadedCube ready: {var} "
            f"({self.n_times}t x {self.n_freqs}f, "
            f"display {len(self.l_vals)}x{len(self.m_vals)} "
            f"from {n_l}x{n_m}, stride {self.stride_l}x{self.stride_m})"
        )

    def _load_slice_impl(self, time_idx: int, freq_idx: int) -> np.ndarray:
        """Load and downsample a single (l, m) slice.

        Returns
        -------
        np.ndarray
            2D array of shape (n_l_display, n_m_display), float32.
        """
        da = self._ds[self.var].isel(
            time=time_idx, frequency=freq_idx, polarization=self.pol
        )
        # Stride-based downsampling — only reads the strided elements,
        # which for contiguous chunks still reads the full chunk but
        # the resulting array is small and fast to work with.
        data = da.values[::self.stride_l, ::self.stride_m]
        return data.astype(np.float32, copy=False)

    def image(self, time_idx: int = 0, freq_idx: int = 0) -> np.ndarray:
        """Get a 2D image slice, transposed for display (m, l) → (y, x)."""
        return self._load_slice(time_idx, freq_idx).T

    def dynamic_spectrum(self, l_idx: int, m_idx: int) -> np.ndarray:
        """Get a 2D (time, freq) dynamic spectrum at a display pixel.

        Loads all slices not yet in cache, then extracts the pixel.
        """
        key = (l_idx, m_idx)
        if key not in self._dynspec_cache:
            out = np.empty((self.n_times, self.n_freqs), dtype=np.float32)
            for ti in range(self.n_times):
                for fi in range(self.n_freqs):
                    slc = self._load_slice(ti, fi)
                    out[ti, fi] = slc[l_idx, m_idx]
            self._dynspec_cache[key] = out
        return self._dynspec_cache[key]

    def light_curve(self, l_idx: int, m_idx: int, freq_idx: int) -> np.ndarray:
        """Get a 1D time series at a display pixel and frequency."""
        out = np.empty(self.n_times, dtype=np.float32)
        for ti in range(self.n_times):
            slc = self._load_slice(ti, freq_idx)
            out[ti] = slc[l_idx, m_idx]
        return out

    def spectrum(self, l_idx: int, m_idx: int, time_idx: int) -> np.ndarray:
        """Get a 1D frequency spectrum at a display pixel and time."""
        out = np.empty(self.n_freqs, dtype=np.float32)
        for fi in range(self.n_freqs):
            slc = self._load_slice(time_idx, fi)
            out[fi] = slc[l_idx, m_idx]
        return out

    def nearest_lm_idx(self, l: float, m: float) -> tuple[int, int]:
        """Find nearest display pixel indices for given l, m values."""
        l_idx = int(np.argmin(np.abs(self.l_vals - l)))
        m_idx = int(np.argmin(np.abs(self.m_vals - m)))
        return l_idx, m_idx

    @property
    def bounds(self) -> tuple[float, float, float, float]:
        """Image bounds as (l_left, m_bottom, l_right, m_top)."""
        return (
            float(self.l_vals[0]),
            float(self.m_vals[0]),
            float(self.l_vals[-1]),
            float(self.m_vals[-1]),
        )

bounds property

Image bounds as (l_left, m_bottom, l_right, m_top).

image(time_idx=0, freq_idx=0)

Get a 2D image slice, transposed for display (m, l) → (y, x).

Source code in src/ovro_lwa_portal/viz/_data.py 
123
124
125
def image(self, time_idx: int = 0, freq_idx: int = 0) -> np.ndarray:
    """Get a 2D image slice, transposed for display (m, l) → (y, x)."""
    return self._load_slice(time_idx, freq_idx).T

dynamic_spectrum(l_idx, m_idx)

Get a 2D (time, freq) dynamic spectrum at a display pixel.

Loads all slices not yet in cache, then extracts the pixel.

Source code in src/ovro_lwa_portal/viz/_data.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def dynamic_spectrum(self, l_idx: int, m_idx: int) -> np.ndarray:
    """Get a 2D (time, freq) dynamic spectrum at a display pixel.

    Loads all slices not yet in cache, then extracts the pixel.
    """
    key = (l_idx, m_idx)
    if key not in self._dynspec_cache:
        out = np.empty((self.n_times, self.n_freqs), dtype=np.float32)
        for ti in range(self.n_times):
            for fi in range(self.n_freqs):
                slc = self._load_slice(ti, fi)
                out[ti, fi] = slc[l_idx, m_idx]
        self._dynspec_cache[key] = out
    return self._dynspec_cache[key]

light_curve(l_idx, m_idx, freq_idx)

Get a 1D time series at a display pixel and frequency.

Source code in src/ovro_lwa_portal/viz/_data.py 
142
143
144
145
146
147
148
def light_curve(self, l_idx: int, m_idx: int, freq_idx: int) -> np.ndarray:
    """Get a 1D time series at a display pixel and frequency."""
    out = np.empty(self.n_times, dtype=np.float32)
    for ti in range(self.n_times):
        slc = self._load_slice(ti, freq_idx)
        out[ti] = slc[l_idx, m_idx]
    return out

spectrum(l_idx, m_idx, time_idx)

Get a 1D frequency spectrum at a display pixel and time.

Source code in src/ovro_lwa_portal/viz/_data.py 
150
151
152
153
154
155
156
def spectrum(self, l_idx: int, m_idx: int, time_idx: int) -> np.ndarray:
    """Get a 1D frequency spectrum at a display pixel and time."""
    out = np.empty(self.n_freqs, dtype=np.float32)
    for fi in range(self.n_freqs):
        slc = self._load_slice(time_idx, fi)
        out[fi] = slc[l_idx, m_idx]
    return out

nearest_lm_idx(l, m)

Find nearest display pixel indices for given l, m values.

Source code in src/ovro_lwa_portal/viz/_data.py 
158
159
160
161
162
def nearest_lm_idx(self, l: float, m: float) -> tuple[int, int]:
    """Find nearest display pixel indices for given l, m values."""
    l_idx = int(np.argmin(np.abs(self.l_vals - l)))
    m_idx = int(np.argmin(np.abs(self.m_vals - m)))
    return l_idx, m_idx

Element Factories

sky_image_element(cube, *, time_idx=0, freq_idx=0, robust=True, mask_radius=None)

Create an hv.Image from the cached cube.

Parameters:

Name Type Description Default
cube PreloadedCube

Data cube.

 required
time_idx int

Slice indices.

 0
freq_idx int

Slice indices.

 0
robust bool

Use 2nd/98th percentile for color limits.

 True
mask_radius int

Circular mask radius in pixels.

 None

Returns:

Type Description
Image
Source code in src/ovro_lwa_portal/viz/_data.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def sky_image_element(
    cube: PreloadedCube,
    *,
    time_idx: int = 0,
    freq_idx: int = 0,
    robust: bool = True,
    mask_radius: int | None = None,
) -> hv.Image:
    """Create an hv.Image from the cached cube.

    Parameters
    ----------
    cube : PreloadedCube
        Data cube.
    time_idx, freq_idx : int
        Slice indices.
    robust : bool
        Use 2nd/98th percentile for color limits.
    mask_radius : int, optional
        Circular mask radius in pixels.

    Returns
    -------
    hv.Image
    """
    import holoviews as hv

    _ensure_extension()

    data = cube.image(time_idx, freq_idx)

    if mask_radius is not None:
        data = data.copy()
        ny, nx = data.shape
        cy, cx = ny // 2, nx // 2
        yy, xx = np.ogrid[:ny, :nx]
        distance = np.sqrt((yy - cy) ** 2 + (xx - cx) ** 2)
        data[distance > mask_radius] = np.nan

    time_val = float(cube.time_vals[time_idx])
    freq_mhz = float(cube.freq_mhz[freq_idx])

    img = hv.Image(data, kdims=["l", "m"], bounds=cube.bounds).opts(
        xlabel="l (direction cosine)",
        ylabel="m (direction cosine)",
        title=f"{cube.var} t={time_val:.4f} MJD, f={freq_mhz:.1f} MHz, pol={cube.pol}",
        aspect="equal",
        colorbar=True,
        clabel="Jy/beam",
    )

    if robust:
        finite = data[np.isfinite(data)]
        if finite.size > 0:
            img = img.opts(
                clim=(float(np.percentile(finite, 2)), float(np.percentile(finite, 98)))
            )

    return img

cutout_image_element(cube, *, l_center=0.0, m_center=0.0, dl=0.1, dm=0.1, time_idx=0, freq_idx=0, robust=True)

Create an hv.Image of a spatial cutout from the cached cube.

Parameters:

Name Type Description Default
cube PreloadedCube

Data cube.

 required
l_center float

Center coordinates.

 0.0
m_center float

Center coordinates.

 0.0
dl float

Half-extent.

 0.1
dm float

Half-extent.

 0.1
time_idx int

Slice indices.

 0
freq_idx int

Slice indices.

 0
robust bool

Use robust color scaling.

 True

Returns:

Type Description
Image
Source code in src/ovro_lwa_portal/viz/_data.py 
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
def cutout_image_element(
    cube: PreloadedCube,
    *,
    l_center: float = 0.0,
    m_center: float = 0.0,
    dl: float = 0.1,
    dm: float = 0.1,
    time_idx: int = 0,
    freq_idx: int = 0,
    robust: bool = True,
) -> hv.Image:
    """Create an hv.Image of a spatial cutout from the cached cube.

    Parameters
    ----------
    cube : PreloadedCube
        Data cube.
    l_center, m_center : float
        Center coordinates.
    dl, dm : float
        Half-extent.
    time_idx, freq_idx : int
        Slice indices.
    robust : bool
        Use robust color scaling.

    Returns
    -------
    hv.Image
    """
    import holoviews as hv

    _ensure_extension()

    l_min, l_max = l_center - dl, l_center + dl
    m_min, m_max = m_center - dm, m_center + dm

    # Get the full display-resolution slice
    full = cube._load_slice(time_idx, freq_idx)
    l_vals = cube.l_vals
    m_vals = cube.m_vals

    # Mask for the cutout region (handles ascending or descending coords)
    lo_l, hi_l = min(l_min, l_max), max(l_min, l_max)
    lo_m, hi_m = min(m_min, m_max), max(m_min, m_max)
    l_mask = (l_vals >= lo_l) & (l_vals <= hi_l)
    m_mask = (m_vals >= lo_m) & (m_vals <= hi_m)

    if not np.any(l_mask) or not np.any(m_mask):
        return hv.Image(
            np.zeros((2, 2)), kdims=["l", "m"],
            bounds=(l_min, m_min, l_max, m_max),
        ).opts(title="Empty cutout — adjust center/extent")

    sub_data = full[np.ix_(l_mask, m_mask)].T
    sub_l = l_vals[l_mask]
    sub_m = m_vals[m_mask]
    bounds = (float(sub_l[0]), float(sub_m[0]), float(sub_l[-1]), float(sub_m[-1]))

    img = hv.Image(sub_data, kdims=["l", "m"], bounds=bounds).opts(
        xlabel="l (direction cosine)",
        ylabel="m (direction cosine)",
        title=f"{cube.var} Cutout at l={l_center:.4f}, m={m_center:.4f}",
        aspect="equal",
        colorbar=True,
        clabel="Jy/beam",
    )

    if robust:
        finite = sub_data[np.isfinite(sub_data)]
        if finite.size > 0:
            img = img.opts(
                clim=(float(np.percentile(finite, 2)), float(np.percentile(finite, 98)))
            )

    return img

dynamic_spectrum_element(cube, *, l=0.0, m=0.0, robust=True)

Create an hv.Image of a dynamic spectrum from the cached cube.

Parameters:

Name Type Description Default
cube PreloadedCube

Data cube.

 required
l float

Direction cosine coordinates.

 0.0
m float

Direction cosine coordinates.

 0.0
robust bool

Use 2nd/98th percentile for color limits.

 True

Returns:

Type Description
Image
Source code in src/ovro_lwa_portal/viz/_data.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
def dynamic_spectrum_element(
    cube: PreloadedCube,
    *,
    l: float = 0.0,
    m: float = 0.0,
    robust: bool = True,
) -> hv.Image:
    """Create an hv.Image of a dynamic spectrum from the cached cube.

    Parameters
    ----------
    cube : PreloadedCube
        Data cube.
    l, m : float
        Direction cosine coordinates.
    robust : bool
        Use 2nd/98th percentile for color limits.

    Returns
    -------
    hv.Image
    """
    import holoviews as hv

    _ensure_extension()

    l_idx, m_idx = cube.nearest_lm_idx(l, m)
    data = cube.dynamic_spectrum(l_idx, m_idx)

    pixel_l = float(cube.l_vals[l_idx])
    pixel_m = float(cube.m_vals[m_idx])

    bounds = (
        float(cube.time_vals[0]),
        float(cube.freq_mhz[0]),
        float(cube.time_vals[-1]),
        float(cube.freq_mhz[-1]),
    )

    img = hv.Image(
        data,
        kdims=["Time (MJD)", "Frequency (MHz)"],
        bounds=bounds,
    ).opts(
        xlabel="Time (MJD)",
        ylabel="Frequency (MHz)",
        title=f"{cube.var} Dynamic Spectrum at l={pixel_l:+.4f}, m={pixel_m:+.4f}",
        colorbar=True,
        clabel="Jy/beam",
    )

    if robust:
        finite = data[np.isfinite(data)]
        if finite.size > 0:
            img = img.opts(
                clim=(float(np.percentile(finite, 2)), float(np.percentile(finite, 98)))
            )

    return img

light_curve_element(cube, *, l=0.0, m=0.0, freq_idx=0)

Create an hv.Curve of a light curve from the cached cube.

Parameters:

Name Type Description Default
cube PreloadedCube

Data cube.

 required
l float

Direction cosine coordinates.

 0.0
m float

Direction cosine coordinates.

 0.0
freq_idx int

Frequency index.

 0

Returns:

Type Description
Curve
Source code in src/ovro_lwa_portal/viz/_data.py 
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
def light_curve_element(
    cube: PreloadedCube,
    *,
    l: float = 0.0,
    m: float = 0.0,
    freq_idx: int = 0,
) -> hv.Curve:
    """Create an hv.Curve of a light curve from the cached cube.

    Parameters
    ----------
    cube : PreloadedCube
        Data cube.
    l, m : float
        Direction cosine coordinates.
    freq_idx : int
        Frequency index.

    Returns
    -------
    hv.Curve
    """
    import holoviews as hv

    _ensure_extension()

    l_idx, m_idx = cube.nearest_lm_idx(l, m)
    values = cube.light_curve(l_idx, m_idx, freq_idx)
    freq_mhz = float(cube.freq_mhz[freq_idx])

    return hv.Curve(
        (cube.time_vals, values),
        kdims=["Time (MJD)"],
        vdims=["Intensity (Jy/beam)"],
    ).opts(
        xlabel="Time (MJD)",
        ylabel="Intensity (Jy/beam)",
        title=f"Light Curve at f={freq_mhz:.1f} MHz",
    )

spectrum_element(cube, *, l=0.0, m=0.0, time_idx=0)

Create an hv.Curve of a frequency spectrum from the cached cube.

Parameters:

Name Type Description Default
cube PreloadedCube

Data cube.

 required
l float

Direction cosine coordinates.

 0.0
m float

Direction cosine coordinates.

 0.0
time_idx int

Time index.

 0

Returns:

Type Description
Curve
Source code in src/ovro_lwa_portal/viz/_data.py 
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
def spectrum_element(
    cube: PreloadedCube,
    *,
    l: float = 0.0,
    m: float = 0.0,
    time_idx: int = 0,
) -> hv.Curve:
    """Create an hv.Curve of a frequency spectrum from the cached cube.

    Parameters
    ----------
    cube : PreloadedCube
        Data cube.
    l, m : float
        Direction cosine coordinates.
    time_idx : int
        Time index.

    Returns
    -------
    hv.Curve
    """
    import holoviews as hv

    _ensure_extension()

    l_idx, m_idx = cube.nearest_lm_idx(l, m)
    values = cube.spectrum(l_idx, m_idx, time_idx)
    time_val = float(cube.time_vals[time_idx])

    return hv.Curve(
        (cube.freq_mhz, values),
        kdims=["Frequency (MHz)"],
        vdims=["Intensity (Jy/beam)"],
    ).opts(
        xlabel="Frequency (MHz)",
        ylabel="Intensity (Jy/beam)",
        title=f"Spectrum at t={time_val:.4f} MJD",
    )

Components

Reusable styled HoloViews components for OVRO-LWA visualization.

Provides consistent styling defaults and common plot configurations used across the explorer classes.

COLORMAPS = ['inferno', 'viridis', 'plasma', 'magma', 'cividis', 'gray'] module-attribute

style_sky_image(img, *, cmap='inferno')

Apply standard styling to a sky image element.

Parameters:

Name Type Description Default
img Image

HoloViews image element.

 required
cmap str

Colormap name.

 'inferno'

Returns:

Type Description
Image

Styled image.
Source code in src/ovro_lwa_portal/viz/components.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def style_sky_image(img: hv.Image, *, cmap: str = "inferno") -> hv.Image:
    """Apply standard styling to a sky image element.

    Parameters
    ----------
    img : hv.Image
        HoloViews image element.
    cmap : str
        Colormap name.

    Returns
    -------
    hv.Image
        Styled image.
    """
    return img.opts(
        cmap=cmap,
        width=IMAGE_WIDTH,
        height=IMAGE_HEIGHT,
        tools=["hover", "tap", "crosshair"],
        active_tools=["tap"],
    )

style_spectrum_image(img, *, cmap='inferno')

Apply standard styling to a dynamic spectrum image.

Parameters:

Name Type Description Default
img Image

HoloViews image element.

 required
cmap str

Colormap name.

 'inferno'

Returns:

Type Description
Image

Styled image.
Source code in src/ovro_lwa_portal/viz/components.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def style_spectrum_image(img: hv.Image, *, cmap: str = "inferno") -> hv.Image:
    """Apply standard styling to a dynamic spectrum image.

    Parameters
    ----------
    img : hv.Image
        HoloViews image element.
    cmap : str
        Colormap name.

    Returns
    -------
    hv.Image
        Styled image.
    """
    return img.opts(
        cmap=cmap,
        width=IMAGE_WIDTH,
        height=IMAGE_HEIGHT,
        tools=["hover", "tap", "crosshair"],
        active_tools=["tap"],
    )

style_curve(curve)

Apply standard styling to a curve element.

Parameters:

Name Type Description Default
curve Curve

HoloViews curve element.

 required

Returns:

Type Description
Curve

Styled curve.
Source code in src/ovro_lwa_portal/viz/components.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def style_curve(curve: hv.Curve) -> hv.Curve:
    """Apply standard styling to a curve element.

    Parameters
    ----------
    curve : hv.Curve
        HoloViews curve element.

    Returns
    -------
    hv.Curve
        Styled curve.
    """
    return curve.opts(
        width=CURVE_WIDTH,
        height=CURVE_HEIGHT,
        tools=["hover"],
        line_width=1.5,
    )

Dashboards

create_exploration_dashboard(ds, *, max_size=512)

Create a comprehensive exploration dashboard.

Parameters:

Name Type Description Default
ds Dataset

OVRO-LWA dataset to explore.

 required
max_size int

Maximum pixels per spatial side after downsampling. Lower values are faster but coarser. Default 512.

 512

Returns:

Type Description
Tabs

Tabbed layout with all explorers.
Source code in src/ovro_lwa_portal/viz/dashboards.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def create_exploration_dashboard(
    ds: xr.Dataset, *, max_size: int = 512,
) -> pn.Tabs:
    """Create a comprehensive exploration dashboard.

    Parameters
    ----------
    ds : xr.Dataset
        OVRO-LWA dataset to explore.
    max_size : int, optional
        Maximum pixels per spatial side after downsampling. Lower
        values are faster but coarser. Default 512.

    Returns
    -------
    panel.Tabs
        Tabbed layout with all explorers.
    """
    image_explorer = ImageExplorer(ds, max_size=max_size)
    dynspec_explorer = DynamicSpectrumExplorer(ds)
    cutout_explorer = CutoutExplorer(ds, max_size=max_size)

    tabs: list[tuple[str, Any]] = [
        ("Image", image_explorer.panel()),
        ("Dynamic Spectrum", dynspec_explorer.panel()),
        ("Cutout", cutout_explorer.panel()),
    ]

    if ds.radport.has_wcs:
        from ovro_lwa_portal.viz.sky_viewer import SkyViewer

        sky_viewer = SkyViewer(ds)
        tabs.append(("Sky Viewer", sky_viewer.panel()))

    return pn.Tabs(*tabs)

Accessor Integration

The radport xarray accessor exposes the visualization framework through convenience methods:

Accessor Method Explorer
ds.radport.explore() Full tabbed dashboard
ds.radport.explore_image() ImageExplorer
ds.radport.explore_dynamic_spectrum() DynamicSpectrumExplorer
ds.radport.explore_sky() SkyViewer