views

View

Bases: Parameterized

Base view object.

Inspired by Holoviz Lumen's View objects

Source code in pyhdx/web/views.py

class View(param.Parameterized):
    """Base view object.

    Inspired by Holoviz Lumen's View objects"""

    _type = None

    opts = param.List(default=[], doc="list of opts dicts to apply on the plot", precedence=-1)

    dependencies = param.List(
        default=[],
        precedence=-1,
        doc="Additional dependencies which trigger update when their `updated` event fires",
    )

    def __init__(self, **params):
        super().__init__(**params)
        # todo allow for kwargs to be passed to DynamicMap's func

        for dep in self.dependencies:
            dep.param.watch(self._update, ["updated"])

        self.widgets = self.make_dict()

        self._panel = None
        self._updates = None  # what does this do?

    def _update(self, *events):
        self.update()  # todo or just catch events in the update function?  (probably this)

    def make_dict(self):
        # todo baseclass filter/controller/view with the same widget generation logic?
        """dict of widgets to be shown
        override this method to get custom mapping

        """
        return self.generate_widgets()

    def generate_widgets(self, **kwargs):
        """returns a dict with keys parameter names and values default mapped widgets"""

        names = [
            p
            for p in self.param
            if self.param[p].precedence is None or self.param[p].precedence > 1
        ]
        widgets = pn.Param(self.param, show_name=False, show_labels=True, widgets=kwargs)

        return {k: v for k, v in zip(names[1:], widgets)}

    def get_data(self):  # refactor get?
        """
        Queries the Source

        Returns
        -------
        DataFrame
            The queried table after filtering and transformations are
            applied.
        """
        # if self._cache is not None:
        #     return self._cache

        df = self.source.get()

        return df

    def _update_panel(self, *events):
        """
        Updates the cached Panel object and returns a boolean value
        indicating whether a rerender is required.
        """

        # todo clenaup
        if self._panel is not None:
            self._cleanup()
            self._updates = self._get_params()
            if self._updates is not None:
                return False
        self._panel = self.get_panel()
        return True

    @property  # todo move to init?
    def opts_dict(self):
        # Combine all opts and merge overlapping lists of opts
        # (currently to merge hooks, might be unwanted behaviour for others)
        opts_dict = {}
        for d in self.opts:  # Iterate over Opts on this view
            for k, v in d.opts.items():  # Iterate over all dict items in the Opt
                if k in opts_dict:
                    if isinstance(v, list) and isinstance(opts_dict[k], list):
                        combined_list = opts_dict[k] + v
                        opts_dict[k] = combined_list
                    else:
                        raise ValueError(
                            f"Overlapping key {k!r} in opt {d.name!r} on view {self.name!r}"
                        )
                else:
                    opts_dict[k] = v

        return opts_dict

generate_widgets(**kwargs)

returns a dict with keys parameter names and values default mapped widgets

Source code in pyhdx/web/views.py

def generate_widgets(self, **kwargs):
    """returns a dict with keys parameter names and values default mapped widgets"""

    names = [
        p
        for p in self.param
        if self.param[p].precedence is None or self.param[p].precedence > 1
    ]
    widgets = pn.Param(self.param, show_name=False, show_labels=True, widgets=kwargs)

    return {k: v for k, v in zip(names[1:], widgets)}

get_data()

Queries the Source

Returns

DataFrame The queried table after filtering and transformations are applied.

Source code in pyhdx/web/views.py

def get_data(self):  # refactor get?
    """
    Queries the Source

    Returns
    -------
    DataFrame
        The queried table after filtering and transformations are
        applied.
    """
    # if self._cache is not None:
    #     return self._cache

    df = self.source.get()

    return df

make_dict()

dict of widgets to be shown override this method to get custom mapping

Source code in pyhdx/web/views.py

def make_dict(self):
    # todo baseclass filter/controller/view with the same widget generation logic?
    """dict of widgets to be shown
    override this method to get custom mapping

    """
    return self.generate_widgets()

hvBarsAppView

Bases: hvXYView

Source code in pyhdx/web/views.py

class hvBarsAppView(hvXYView):
    _type = "bars"

    def get_plot(self) -> hv.DynamicMap:
        """Creates the scatter plot as DynamicMap

        Returns:
            Holoviews DynamicMap
        """

        func = partial(hv.Bars, kdims=self.x, vdims=self.y)
        param_stream = Params(
            parameterized=self,
            parameters=["x", "y"],
            rename={"x": "kdims", "y": "vdims"},
        )
        plot = hv.DynamicMap(func, streams=[self._stream, param_stream])
        plot = plot.apply.opts(**self.opts_dict)

        return plot

    @property
    def empty_df(self):
        dic = {self.x or "x": [], self.y or "y": []}
        if "color" in self.opts_dict:
            dic[self.opts_dict["color"]] = []
        return pd.DataFrame(dic)

get_plot()

Creates the scatter plot as DynamicMap

Returns:

Type	Description
DynamicMap	Holoviews DynamicMap

Source code in pyhdx/web/views.py

def get_plot(self) -> hv.DynamicMap:
    """Creates the scatter plot as DynamicMap

    Returns:
        Holoviews DynamicMap
    """

    func = partial(hv.Bars, kdims=self.x, vdims=self.y)
    param_stream = Params(
        parameterized=self,
        parameters=["x", "y"],
        rename={"x": "kdims", "y": "vdims"},
    )
    plot = hv.DynamicMap(func, streams=[self._stream, param_stream])
    plot = plot.apply.opts(**self.opts_dict)

    return plot

hvCurveView

Bases: hvXYView

Source code in pyhdx/web/views.py

class hvCurveView(hvXYView):
    _type = "curve"

    def get_plot(self) -> hv.DynamicMap:
        """Creates the curve plot as DynamicMap.

        Returns:
            Holoviews DynamicMap
        """

        func = partial(hv.Curve, kdims=self.y, vdims=self.x)
        param_stream = Params(
            parameterized=self,
            parameters=["x", "y"],
            rename={"x": "kdims", "y": "vdims"},
        )
        plot = hv.DynamicMap(func, streams=[self._stream, param_stream])
        plot = plot.apply.opts(**self.opts_dict)

        return plot

    @property
    def empty_df(self):
        dic = {self.x or "x": [], self.y or "y": []}
        return pd.DataFrame(dic)

get_plot()

Creates the curve plot as DynamicMap.

Returns:

Type	Description
DynamicMap	Holoviews DynamicMap

Source code in pyhdx/web/views.py

def get_plot(self) -> hv.DynamicMap:
    """Creates the curve plot as DynamicMap.

    Returns:
        Holoviews DynamicMap
    """

    func = partial(hv.Curve, kdims=self.y, vdims=self.x)
    param_stream = Params(
        parameterized=self,
        parameters=["x", "y"],
        rename={"x": "kdims", "y": "vdims"},
    )
    plot = hv.DynamicMap(func, streams=[self._stream, param_stream])
    plot = plot.apply.opts(**self.opts_dict)

    return plot

hvErrorBarsAppView

Bases: hvView

Source code in pyhdx/web/views.py

class hvErrorBarsAppView(hvView):
    _type = "errorbars"

    x = param.String("x", doc="Positions of the errorbars, x-values for vertical errorbars")

    y = param.String("y", doc="Values of the samples, y-values for vertical errorbars")

    err_type = param.Selector(
        default="size",
        objects=["size", "positions"],
        doc="The type of error specified, errors are either given by a size as distance from 'y'"
        "values, or as absolute positions",
    )

    err = param.String(None, doc="Error values in both directions")

    err_pos = param.String(None, doc="Error values in positive direction")

    err_neg = param.String(None, doc="Error values in negative direction")

    horizontal = param.Boolean(False, doc="error bar direction")

    def __init__(self, **params):
        # todo left and right cannot be none?
        super().__init__(**params)
        if self.horizontal:
            raise NotImplementedError("Horizontal error bars are not implemented")

    @property
    def value(self):
        return self.x if self.horizontal else self.y

    @property
    def pos(self):
        return self.y if self.horizontal else self.x

    def get_data(self):
        df = super().get_data()
        if df is None:
            return df
        if self.err_type == "size":
            return df
        else:
            new_df = df[[self.value]]
            if self.err is not None:
                raise ValueError("Must specify 'err_pos' and 'err_neg' with positional" "errors")
            new_df[self.err_pos] = df[self.err_pos] - df[self.value]
            new_df[self.err_neg] = df[self.value] - df[self.err_neg]

            return new_df

    def get_plot(self):
        """
        Dataframe df must have columns x0, y0, x1, y1 (in this order) for coordinates
        bottom-left (x0, y0) and top right (x1, y1). Optionally a fifth value-column can be provided for colors

        Parameters
        ----------
        df

        Returns
        -------

        """

        func = partial(hv.ErrorBars, kdims=self.kdims, vdims=self.vdims, horizontal=self.horizontal)
        plot = hv.DynamicMap(func, streams=[self._stream])

        if self.opts_dict:
            plot = plot.apply.opts(**self.opts_dict)

        return plot

    @property
    def vdims(self):
        if self.err is not None and self.err_pos is None and self.err_neg is None:
            return [self.value, self.err]
        elif self.err is None and self.err_pos is not None and self.err_neg is not None:
            return [self.value, self.err_pos, self.err_neg]
        else:
            raise ValueError("Must set either only 'err' or both 'err_pos' and 'err_neg'")

    @property
    def kdims(self):
        return [self.pos]

    @property
    def empty_df(self):
        columns = self.kdims + self.vdims
        return pd.DataFrame([[0] * len(columns)], columns=columns)

get_plot()

Dataframe df must have columns x0, y0, x1, y1 (in this order) for coordinates bottom-left (x0, y0) and top right (x1, y1). Optionally a fifth value-column can be provided for colors

Parameters

df

Returns

Source code in pyhdx/web/views.py

def get_plot(self):
    """
    Dataframe df must have columns x0, y0, x1, y1 (in this order) for coordinates
    bottom-left (x0, y0) and top right (x1, y1). Optionally a fifth value-column can be provided for colors

    Parameters
    ----------
    df

    Returns
    -------

    """

    func = partial(hv.ErrorBars, kdims=self.kdims, vdims=self.vdims, horizontal=self.horizontal)
    plot = hv.DynamicMap(func, streams=[self._stream])

    if self.opts_dict:
        plot = plot.apply.opts(**self.opts_dict)

    return plot

hvPlotView

Bases: hvView

Source code in pyhdx/web/views.py

class hvPlotView(hvView):
    _type = "hvplot"

    kind = param.String()

    def __init__(self, **params):
        self.kwargs = {k: v for k, v in params.items() if k not in self.param}
        super().__init__(**{k: v for k, v in params.items() if k in self.param})

    def get_plot(self):
        """

        Parameters
        ----------
        df

        Returns
        -------

        """

        def func(data, kind, **kwargs):
            return hvPlotTabular(data)(kind=kind, **kwargs)

        pfunc = partial(func, kind=self.kind, **self.kwargs)

        plot = hv.DynamicMap(pfunc, streams=[self._stream])
        plot = plot.apply.opts(**self.opts_dict)

        return plot

    @property
    def empty_df(self):
        df = pd.DataFrame({"null": [np.nan], "y2": [np.nan]})
        return df

get_plot()

Parameters

df

Returns

Source code in pyhdx/web/views.py

def get_plot(self):
    """

    Parameters
    ----------
    df

    Returns
    -------

    """

    def func(data, kind, **kwargs):
        return hvPlotTabular(data)(kind=kind, **kwargs)

    pfunc = partial(func, kind=self.kind, **self.kwargs)

    plot = hv.DynamicMap(pfunc, streams=[self._stream])
    plot = plot.apply.opts(**self.opts_dict)

    return plot

hvRectanglesAppView

Bases: hvView

Source code in pyhdx/web/views.py

class hvRectanglesAppView(hvView):
    _type = "rectangles"

    x0 = param.String("x0")

    x1 = param.String("x1")

    y0 = param.String("y0")

    y1 = param.String("y1")

    vdims = param.List(["value"])

    def __init__(self, **params):
        # todo left and right cannot be none?
        super().__init__(**params)

    def get_plot(self):
        """
        Dataframe df must have columns x0, y0, x1, y1 (in this order) for coordinates
        bottom-left (x0, y0) and top right (x1, y1). Optionally a fifth value-column can be provided for colors

        Parameters
        ----------
        df

        Returns
        -------

        """

        func = partial(hv.Rectangles, kdims=self.kdims, vdims=self.vdims)
        plot = hv.DynamicMap(func, streams=[self._stream])

        if self.opts_dict:
            plot = plot.apply.opts(**self.opts_dict)

        return plot

    @property
    def kdims(self):
        return [self.x0, self.y0, self.x1, self.y1]

    @property
    def empty_df(self):
        columns = self.kdims + self.vdims
        return pd.DataFrame([[0] * len(columns)], columns=columns)

get_plot()

Dataframe df must have columns x0, y0, x1, y1 (in this order) for coordinates bottom-left (x0, y0) and top right (x1, y1). Optionally a fifth value-column can be provided for colors

Parameters

df

Returns

Source code in pyhdx/web/views.py

def get_plot(self):
    """
    Dataframe df must have columns x0, y0, x1, y1 (in this order) for coordinates
    bottom-left (x0, y0) and top right (x1, y1). Optionally a fifth value-column can be provided for colors

    Parameters
    ----------
    df

    Returns
    -------

    """

    func = partial(hv.Rectangles, kdims=self.kdims, vdims=self.vdims)
    plot = hv.DynamicMap(func, streams=[self._stream])

    if self.opts_dict:
        plot = plot.apply.opts(**self.opts_dict)

    return plot

hvScatterAppView

Bases: hvXYView

Source code in pyhdx/web/views.py

class hvScatterAppView(hvXYView):
    _type = "scatter"

    def get_plot(self) -> hv.DynamicMap:
        """Creates the scatter plot as DynamicMap

        Returns:
            Holoviews DynamicMap
        """

        func = partial(hv.Scatter, kdims=self.x, vdims=self.y)
        param_stream = Params(
            parameterized=self,
            parameters=["x", "y"],
            rename={"x": "kdims", "y": "vdims"},
        )
        plot = hv.DynamicMap(func, streams=[self._stream, param_stream])
        plot = plot.apply.opts(**self.opts_dict)

        return plot

    @property
    def empty_df(self):
        dic = {self.x or "x": [], self.y or "y": []}
        if "color" in self.opts_dict:
            dic[self.opts_dict["color"]] = []
        return pd.DataFrame(dic)

get_plot()

Creates the scatter plot as DynamicMap

Returns:

Type	Description
DynamicMap	Holoviews DynamicMap

Source code in pyhdx/web/views.py

def get_plot(self) -> hv.DynamicMap:
    """Creates the scatter plot as DynamicMap

    Returns:
        Holoviews DynamicMap
    """

    func = partial(hv.Scatter, kdims=self.x, vdims=self.y)
    param_stream = Params(
        parameterized=self,
        parameters=["x", "y"],
        rename={"x": "kdims", "y": "vdims"},
    )
    plot = hv.DynamicMap(func, streams=[self._stream, param_stream])
    plot = plot.apply.opts(**self.opts_dict)

    return plot

hvView

Bases: View

Source code in pyhdx/web/views.py

class hvView(View):
    source = param.ClassSelector(
        class_=(Source, Transform),
        constant=True,
        precedence=-1,
        doc="""
        The Source to query for the data.""",
    )

    _type = None

    _stream = param.ClassSelector(class_=Pipe)

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

    @param.depends("source.updated", watch=True)
    def update(self):
        """
        Triggers an update in the View.

        Returns
        -------
        stale : bool
            Whether the panel on the View is stale and needs to be
            rerendered.
        """
        # Skip events triggered by a parameter change on this View
        # own_parameters = [self.param[p] for p in self.param]
        # own_events = events and all(
        #     isinstance(e.obj, ParamFilter) and
        #     (e.obj.parameter in own_parameters or
        #     e.new is self._ls.selection_expr)
        #     for e in events
        # )
        # if own_events:
        #     return False
        # if invalidate_cache:
        #     self._cache = None
        if self._stream is None:
            return self._update_panel()
        if self.get_data() is not None:
            self._stream.send(self.get_data())
        return False

    def get_panel(self):
        kwargs = self._get_params()
        # interactive? https://github.com/holoviz/panel/issues/1824
        return pn.pane.HoloViews(linked_axes=False, **kwargs)  # linked_axes=False??

    def _get_params(self):
        df = self.get_data()
        if df is None:
            df = self.empty_df

        self._stream = Pipe(data=df)
        return dict(object=self.get_plot(), sizing_mode="stretch_both")  # todo update sizing mode

    @property
    def panel(self):
        return self.get_panel()

update()

Triggers an update in the View.

Returns

stale : bool Whether the panel on the View is stale and needs to be rerendered.

Source code in pyhdx/web/views.py

@param.depends("source.updated", watch=True)
def update(self):
    """
    Triggers an update in the View.

    Returns
    -------
    stale : bool
        Whether the panel on the View is stale and needs to be
        rerendered.
    """
    # Skip events triggered by a parameter change on this View
    # own_parameters = [self.param[p] for p in self.param]
    # own_events = events and all(
    #     isinstance(e.obj, ParamFilter) and
    #     (e.obj.parameter in own_parameters or
    #     e.new is self._ls.selection_expr)
    #     for e in events
    # )
    # if own_events:
    #     return False
    # if invalidate_cache:
    #     self._cache = None
    if self._stream is None:
        return self._update_panel()
    if self.get_data() is not None:
        self._stream.send(self.get_data())
    return False

hvXYView

Bases: hvView

Base class for hv views with a single XY view

Source code in pyhdx/web/views.py

class hvXYView(hvView):
    """Base class for hv views with a single XY view"""

    x = param.Selector(default=None, doc="The column to render on the x-axis.")

    x_objects = param.ClassSelector(class_=(list, re.Pattern), default=None, precedence=-1)

    y = param.Selector(default=None, objects=[], doc="The column to render on the y-axis.")

    y_objects = param.ClassSelector(class_=(list, re.Pattern), default=None, precedence=-1)

    @param.depends("source.updated", watch=True)
    def update(self, *events) -> None:
        """Triggers an update of the view.

        The source is queried for new data and this is sent to the `_stream` object. The
        selector options for x and y are updated.

        """
        data = self.get_data()
        if data is not None:
            self.param["x"].objects = self.resolve_columns(data, self.x_objects)
            self.param["y"].objects = self.resolve_columns(data, self.y_objects)
            # todo check for case whe updated dataframe longer has current value of x in columns

            self._stream.send(data)

    @staticmethod
    def resolve_columns(data: pd.DataFrame, spec: Union[list, re.Pattern, None]) -> list[str]:
        """Resolve the columns of a dataframe to find the ones that match specification.

        Args:
            data: Dataframe to resolve
            spec: Column specification, either explicit list of allowed columns or a regex
                pattern.

        Returns:
            List of names of the columns (or index) which match specification.
        """
        all_objects = list(data.columns)
        if data.index.name is not None:
            all_objects += [data.index.name]

        if isinstance(spec, list):
            resolved_objects = [obj for obj in all_objects if obj in spec]
        elif isinstance(spec, re.Pattern):
            resolved_objects = [obj for obj in all_objects if spec.match(obj)]
        else:
            resolved_objects = all_objects

        return resolved_objects

resolve_columns(data, spec) staticmethod

Resolve the columns of a dataframe to find the ones that match specification.

Parameters:

Name	Type	Description	Default
data	DataFrame	Dataframe to resolve	required
spec	Union[list, Pattern, None]	Column specification, either explicit list of allowed columns or a regex pattern.	required

Returns:

Type	Description
list[str]	List of names of the columns (or index) which match specification.

Source code in pyhdx/web/views.py

@staticmethod
def resolve_columns(data: pd.DataFrame, spec: Union[list, re.Pattern, None]) -> list[str]:
    """Resolve the columns of a dataframe to find the ones that match specification.

    Args:
        data: Dataframe to resolve
        spec: Column specification, either explicit list of allowed columns or a regex
            pattern.

    Returns:
        List of names of the columns (or index) which match specification.
    """
    all_objects = list(data.columns)
    if data.index.name is not None:
        all_objects += [data.index.name]

    if isinstance(spec, list):
        resolved_objects = [obj for obj in all_objects if obj in spec]
    elif isinstance(spec, re.Pattern):
        resolved_objects = [obj for obj in all_objects if spec.match(obj)]
    else:
        resolved_objects = all_objects

    return resolved_objects

update(*events)

Triggers an update of the view.

The source is queried for new data and this is sent to the _stream object. The selector options for x and y are updated.

Source code in pyhdx/web/views.py

@param.depends("source.updated", watch=True)
def update(self, *events) -> None:
    """Triggers an update of the view.

    The source is queried for new data and this is sent to the `_stream` object. The
    selector options for x and y are updated.

    """
    data = self.get_data()
    if data is not None:
        self.param["x"].objects = self.resolve_columns(data, self.x_objects)
        self.param["y"].objects = self.resolve_columns(data, self.y_objects)
        # todo check for case whe updated dataframe longer has current value of x in columns

        self._stream.send(data)