app.py

import dask.dataframe as dd
import holoviews as hv
import numpy as np
import panel as pn
import param
from holoviews.operation.datashader import dynspread, rasterize

from utils import (
    DATASETS,
    DATASHADER_LOGO,
    DATASHADER_URL,
    DESCRIPTION,
    MAJOR_TOM_LOGO,
    MAJOR_TOM_LYRICS,
    MAJOR_TOM_PICTURE,
    MAJOR_TOM_REF_URL,
    PANEL_LOGO,
    PANEL_URL,
    get_closest_rows,
    get_image,
    get_meta_data,
)


class DatasetInput(pn.viewable.Viewer):
    value = param.Selector(objects=DATASETS, allow_None=False, label="Dataset")

    data = param.DataFrame(allow_None=False)

    def __panel__(self):
        return pn.widgets.RadioButtonGroup.from_param(
            self.param.value, button_style="outline"
        )

    @pn.depends("value", watch=True, on_init=True)
    def _update_data(self):
        self.data = pn.cache(get_meta_data)(dataset=self.value)


class MapInput(pn.viewable.Viewer):
    data = param.DataFrame(allow_refs=True, allow_None=False)

    data_in_view = param.DataFrame(allow_None=False)
    data_selected = param.DataFrame(allow_None=False)

    _plot = param.Parameter(allow_None=False)
    _pointer_x = param.Parameter(allow_None=False)
    _pointer_y = param.Parameter(allow_None=False)
    _range_xy = param.Parameter(allow_None=False)
    _tap = param.Parameter(allow_None=False)

    updating = param.Boolean()

    def __panel__(self):
        return pn.pane.HoloViews(
            self._plot, height=550, width=800, loading=self.param.updating
        )

    @param.depends("data", watch=True, on_init=True)
    def _handle_data_dask_change(self):
        with self.param.update(updating=True):
            data_dask = dd.from_pandas(self.data).persist()
            points = hv.Points(
                data_dask, kdims=["centre_easting", "centre_northing"], vdims=[]
            )

            mean_easting = np.mean(points.range("centre_easting"))
            mean_northing = np.mean(points.range("centre_northing"))
            rangexy = hv.streams.RangeXY(source=points)
            tap = hv.streams.Tap(source=points, x=mean_easting, y=mean_northing)

            agg = rasterize(
                points, link_inputs=True, x_sampling=0.0001, y_sampling=0.0001
            )
            dyn = dynspread(agg)
            dyn.opts(cmap="kr_r", colorbar=True)

            pointerx = hv.streams.PointerX(x=mean_easting, source=points)
            pointery = hv.streams.PointerY(y=mean_northing, source=points)
            vline = hv.DynamicMap(lambda x: hv.VLine(x), streams=[pointerx])
            hline = hv.DynamicMap(lambda y: hv.HLine(y), streams=[pointery])
            tiles = hv.Tiles(
                "https://tile.openstreetmap.org/{Z}/{X}/{Y}.png", name="OSM"
            ).opts(width=600, height=550, xlabel="Longitude", ylabel="Latitude")

            self.param.update(
                _plot=tiles * agg * dyn * hline * vline,
                _pointer_x=pointerx,
                _pointer_y=pointery,
                _range_xy=rangexy,
                _tap=tap,
            )

            update_viewed = pn.bind(
                self._update_data_in_view,
                rangexy.param.x_range,
                rangexy.param.y_range,
                watch=True,
            )
            update_viewed()

            update_selected = pn.bind(
                self._update_data_selected, tap.param.x, tap.param.y, watch=True
            )
            update_selected()

    def _update_data_in_view(self, x_range, y_range):
        if not x_range or not y_range:
            self.data_in_view = self.data.head(0)
            return

        data = self.data
        data = data[
            (data.centre_easting.between(*x_range))
            & (data.centre_northing.between(*y_range))
        ]
        self.data_in_view = data.head(10).reset_index(drop=True)

    def _update_data_selected(self, tap_x, tap_y):
        self.data_selected = get_closest_rows(self.data, tap_x, tap_y)


class ImageInput(pn.viewable.Viewer):
    data = param.DataFrame(allow_refs=True, allow_None=False)

    updating = param.Boolean()
    image = param.Parameter()
    plot = param.Parameter()

    _timestamp = param.Selector(label="Timestamp", objects=[None])

    def __panel__(self):
        return pn.Column(
            pn.widgets.RadioButtonGroup.from_param(
                self.param._timestamp, button_style="outline"
            ),
            pn.Tabs(
                pn.pane.HoloViews(
                    hv.DynamicMap(pn.bind(lambda plot: plot, self.param.plot)),
                    loading=self.param.updating,
                    height=800,
                    width=800,
                    name="Interactive Image",
                ),
                pn.pane.Image(
                    self.param.image,
                    name="Static Image",
                    loading=self.param.updating,
                    width=800,
                ),
                dynamic=True,
            ),
        )

    @pn.depends("data", watch=True, on_init=True)
    def _update_timestamp(self):
        if self.data.empty:
            default_value = None
            options = [None]
            print("empty options")
        else:
            options = sorted(self.data["timestamp"].unique())
            default_value = options[0]
            print("options", options)

        self.param._timestamp.objects = options
        if not self._timestamp in options:
            self._timestamp = default_value

    @pn.depends("_timestamp", watch=True, on_init=True)
    def tap_image(self):
        if self.data.empty or not self._timestamp:
            self.image = hv.RGB(np.array([]))
        else:
            with self.param.update(updating=True):
                row = self.data[self.data.timestamp == self._timestamp].iloc[0]
                self.image = image = pn.cache(get_image)(row)
                image_array = np.array(image)
                self.plot = hv.RGB(image_array).opts(xaxis=None, yaxis=None)


class App(param.Parameterized):
    sidebar = param.Parameter()
    main = param.Parameter()

    def __init__(self, **params):
        super().__init__(**params)

        self.sidebar = self._create_sidebar()
        self.main = pn.FlexBox(
            pn.Column(
                pn.Row(
                    pn.indicators.LoadingSpinner(value=True, size=50),
                    "**Loading data...**",
                ),
                MAJOR_TOM_LYRICS,
            )
        )

        pn.state.onload(self._update_main)

    def _create_sidebar(self):
        return pn.Column(
            pn.pane.Image(
                MAJOR_TOM_LOGO, link_url=MAJOR_TOM_REF_URL, sizing_mode="stretch_width"
            ),
            pn.pane.Image(
                MAJOR_TOM_PICTURE,
                link_url=MAJOR_TOM_REF_URL,
                sizing_mode="stretch_width",
            ),
            DESCRIPTION,
            pn.pane.Image(PANEL_LOGO, link_url=PANEL_URL, width=200, margin=(10, 20)),
            pn.pane.Image(
                DATASHADER_LOGO, link_url=DATASHADER_URL, width=200, margin=(10, 20)
            ),
        )

    def _create_main_content(self):
        dataset = DatasetInput()
        map_input = MapInput(data=dataset.param.data)
        image_input = ImageInput(data=map_input.param.data_selected)

        return pn.Column(dataset, map_input), image_input

    def _update_main(self):
        self.main[:] = list(self._create_main_content())


pn.extension("tabulator", design="fast")

app = App()

pn.template.FastListTemplate(
    title="Major TOM Explorer",
    main=[app.main],
    sidebar=[app.sidebar],
    main_layout=None,
    accent="#003247",  # "#A01346"
).servable()