Source code for panel.pane.vizzu

from __future__ import annotations

import datetime as dt
import sys

from typing import (
    TYPE_CHECKING, Any, Callable, ClassVar, Dict, List, Optional,

import numpy as np
import param

from bokeh.models import ColumnDataSource
from pyviz_comms import JupyterComm

from ..reactive import SyncableData
from ..util import isdatetime, lazy_load
from .base import ModelPane

    from bokeh.document import Document
    from bokeh.model import Model
    from pyviz_comms import Comm

[docs]class Vizzu(ModelPane, SyncableData): """ The `Vizzu` pane provides an interactive visualization component for large, real-time datasets built on the Vizzu project. Reference: :Example: >>> Vizzu(df) """ animation = param.Dict(default={}, nested_refs=True, doc=""" Animation settings (see""") config = param.Dict(default={}, nested_refs=True, doc=""" The config contains all of the parameters needed to render a particular static chart or a state of an animated chart (see""") click = param.Parameter(doc=""" Data associated with the latest click event.""") column_types = param.Dict(default={}, nested_refs=True, doc=""" Optional column definitions. If not defined will be inferred from the data.""") duration = param.Integer(default=500, doc=""" The config contains all of the parameters needed to render a particular static chart or a state of an animated chart.""") style = param.Dict(default={}, nested_refs=True, doc=""" Style configuration of the chart.""") tooltip = param.Boolean(default=False, doc=""" Whether to enable tooltips on the chart.""") _data_params: ClassVar[List[str]] = ['object'] _rename: ClassVar[Dict[str, str | None]] = { 'click': None, 'column_types': None, 'object': None } _rerender_params: ClassVar[List[str]] = [] _updates: ClassVar[bool] = True def __init__(self, object=None, **params): click_handler = params.pop('on_click', None) super().__init__(object, **params) self._event_handlers = [] if click_handler: self.on_click(click_handler)
[docs] @classmethod def applies(cls, object): if isinstance(object, dict) and all(isinstance(v, (list, np.ndarray)) for v in object.values()): return 0 if object else None elif 'pandas' in sys.modules: import pandas as pd if isinstance(object, pd.DataFrame): return 0 return False
def _get_data(self): if self.object is None: return {}, {} if isinstance(self.object, dict): cols = data = dict(self.object) else: data = self.object cols = ColumnDataSource.from_df(self.object) return data, {str(k): v for k, v in cols.items()} def _get_columns(self): import pandas as pd columns = [] for col, array in self._data.items(): if col in self.column_types: columns.append({'name': col, 'type': self.column_types[col]}) continue if not isinstance(array, np.ndarray): array = np.asarray(array) kind = array.dtype.kind if kind == 'M': columns.append({'name': col, 'type': 'datetime'}) elif kind in 'uif': columns.append({'name': col, 'type': 'measure'}) elif kind in 'bsU': columns.append({'name': col, 'type': 'dimension'}) else: if len(array): value = array[0] if isinstance(value, columns.append({'name': col, 'type': 'datetime'}) elif isdatetime(value) or isinstance(value, pd.Period): columns.append({'name': col, 'type': 'datetime'}) elif isinstance(value, str): columns.append({'name': col, 'type': 'dimension'}) elif isinstance(value, (float, np.float64, np.int_, int)): columns.append({'name': col, 'type': 'measure'}) else: columns.append({'name': col, 'type': 'dimension'}) else: columns.append({'name': col, 'type': 'dimension'}) return columns def _get_properties(self, doc, source=None): props = super()._get_properties(doc) props['duration'] = self.duration if source is None: props['source'] = ColumnDataSource(data=self._data) else: = self._data props['source'] = source return props def _process_param_change(self, params): if 'object' in params: self._processed, self._data = self._get_data() if 'object' in params or 'column_types' in params: params['columns'] = self._get_columns() return super()._process_param_change(params) def _get_model( self, doc: Document, root: Optional[Model] = None, parent: Optional[Model] = None, comm: Optional[Comm] = None ) -> Model: self._bokeh_model = lazy_load( 'panel.models.vizzu', 'VizzuChart', isinstance(comm, JupyterComm), root ) model = super()._get_model(doc, root, parent, comm) self._register_events('vizzu_event', model=model, doc=doc, comm=comm) return model def _process_event(self, event): = for handler in self._event_handlers: handler( def _update(self, ref: str, model: Model) -> None: pass
[docs] def animate( self, anim: Dict[str, Any], options: int | Dict[str, Any] | None = None ) -> None: """ Updates the chart with a new configuration. """ if not any(key in anim for key in ('config', 'data', 'style')): anim = {'config': anim} updates = {} for p, value in anim.items(): if p not in self.param: raise ValueError( f'Could not update {p!r}. You must pass either a dictionary ' 'containing config, data and/or style values OR a single ' 'config dictionary. ' ) updates[p] = dict(getattr(self, p), **value) if isinstance(options, int): self.duration = options elif isinstance(options, dict): self.animation = options self.param.update(updates)
# Public API
[docs] def on_click(self, callback: Callable[[Dict], None]): """ Register a callback to be executed when any element in the chart is clicked on. Arguments --------- callback: (callable) The callback to run on click events. """ self._event_handlers.append(callback)