import abc
import numpy as np
import logging
from typing import List, Tuple, Any, Callable, Optional
import bokeh.models
from bokeh import __version__ as bokeh_version
if bokeh_version < '3.0.0':
from bokeh.models.markers import Scatter
else:
from bokeh.models.glyphs import Scatter
from shyft.time_series import TsVector, UtcPeriod, TimeAxis, statistics_property, utctime_now, Calendar
from shyft.dashboard.time_series.state import Quantity, UnitRegistry
from shyft.dashboard.time_series.bindable import Bindable
from shyft.dashboard.base.hashable import Hashable
from shyft.dashboard.time_series.data_utility import data_to_patch_values, calculate_dead_band_indices, convert_ts_to_plot_vectors
from shyft.dashboard.time_series.axes import YAxis
from shyft.dashboard.base.ports import States, StatePorts
VALID_LINE_STYLES = ['solid', 'dashed', 'dotted', 'dotdash', 'dashdot']
[docs]
class RendererError(RuntimeError):
pass
[docs]
def to_bokeh_datetime_rep(t: np.ndarray) -> np.ndarray:
""" for datetime axies bokeh uses ms, so we scale up time with 1000."""
return t*1000.0
[docs]
class SingleGlyphRenderer(BaseFigureRenderer):
"""
This object is the base class for single glyphs such as LineRendrer and FillInBetweenRenderer
"""
[docs]
def __init__(self, *, unit_registry: UnitRegistry,
notify_figure_y_range_update: Callable,
logger: logging.Logger = None) -> None:
"""
Parameters
----------
unit_registry: unit registry to use to verify data
notify_figure_y_range_update: function to trigger y_range_update of the figure renderer is connected to
"""
super().__init__(unit_registry=unit_registry,
notify_figure_y_range_update=notify_figure_y_range_update,
logger=logger)
self.bokeh_data_source = bokeh.models.ColumnDataSource({k: [] for k in self.bokeh_ds_keys})
self.reset_bokeh_data_source()
[docs]
def reset_bokeh_data_source(self):
"""
This function updates the properties of a glyph which can be dynamically set,
color, fill_alpha, etc.
"""
self.bokeh_data_source.data = dict({k: [] for k in self.bokeh_ds_keys})
@property
@abc.abstractmethod
def _glyph(self) -> bokeh.models.Model:
"""
This function returns the glyph of the current renderer
"""
# pass
@property
@abc.abstractmethod
def bokeh_ds_keys(self) -> List[str]:
"""
This function returns a list of strings of the data source keys which are in the renderer
"""
# pass
@property
def glyphs(self) -> List[Tuple[bokeh.models.ColumnDataSource, bokeh.models.Model]]:
"""
This function adds source data to bokeh glyphs
"""
return [(self.bokeh_data_source, self._glyph)]
[docs]
def color_callback(self, obj, attr, old_value, new_value) -> None:
if self._state == States.DEACTIVE:
return
if obj != self.view:
obj.remove_all_callbacks(self)
return
color_patches = [(i, self.view.color) for i, d in enumerate((self.bokeh_data_source.data["f"]))]
self.bokeh_data_source.patch({"color": color_patches})
[docs]
class LineRenderer(SingleGlyphRenderer):
"""
This object contains meta-data and initialisation/update functions for the renderers
of a single line
"""
[docs]
def __init__(self, *, unit_registry: UnitRegistry,
notify_figure_y_range_update: Callable,
logger: logging.Logger = None) -> None:
"""
Parameters
----------
unit_registry: unit registry to use to verify data
notify_figure_y_range_update: function to trigger y_range_update of the figure renderer is connected to
"""
super().__init__(unit_registry=unit_registry,
notify_figure_y_range_update=notify_figure_y_range_update,
logger=logger)
@property
def _glyph(self) -> bokeh.models.Model:
""" This function returns a glyph of a line
Remark: line_dash is not supposed to be changed with the data source so we need to change the bokeh.renderer
it self directly see self.line_style_callback
"""
return bokeh.models.MultiLine(xs="t", ys="f", line_color="color", line_dash="solid", line_width="line_width")
@property
def bokeh_ds_keys(self) -> List[str]:
"""
This function returns the keys of the dynamically changeable properties of the renderer
"""
tooltips = {"t", "f", "color", "label", "line_width"}
if self.view:
for tt in self.view.tooltips:
tooltips.add(tt[0])
return list(tooltips)
[docs]
def on_set_view(self) -> None:
self.view.on_change(obj=self, attr="index", callback=self.index_callback)
self.view.on_change(obj=self, attr="line_style", callback=self.line_style_callback)
self.view.on_change(obj=self, attr='line_width', callback=self.line_width_callback)
# init line sytle callback
self.line_style_callback(obj=self.view, attr='line_style', old_value='solid', new_value=self.view.line_style)
[docs]
def update_bokeh_data_source(self) -> None:
"""
This function updates the data source of the plot,
data points, color, etc.
"""
if self.do_log:
ts = utctime_now()
self.logger.debug(f"{self.__class__.__name__} {self.view.label} updating data source")
t, f = convert_ts_to_plot_vectors(ts=self.ts_vector.magnitude[self.view.index], cal=self.calendar, interpret_point_interpretation=True, crop_nan=True)
curve_data = {k: [] for k in self.bokeh_ds_keys}
curve_data["f"] = [f]
curve_data["t"] = [t]
curve_data["color"] = [self.view.color for _ in range(len(curve_data["f"]))]
curve_data["line_width"] = [self.view.line_width for _ in range(len(curve_data["f"]))]
curve_data["label"] = [self.view.label for _ in range(len(curve_data["f"]))]
self.bokeh_data_source.data = curve_data
if self.do_log:
self.logger.debug(f"{self.__class__.__name__} {self.view.label} update took {utctime_now() - ts}")
self.view.view_container.update_y_range()
[docs]
def index_callback(self, obj, attr, old_value, new_value) -> None:
if self._state == States.DEACTIVE:
return
if obj != self.view:
obj.remove_all_callbacks(self)
return
if not self.ts_vector:
self.reset_bokeh_data_source()
return
self.update_bokeh_data_source()
self.notify_figure_y_range_update()
[docs]
def line_style_callback(self, obj, attr, old_value, new_value) -> None:
if attr != 'line_style':
return
if self._state == States.DEACTIVE:
return
if obj != self.view:
obj.remove_all_callbacks(self)
return
if new_value not in VALID_LINE_STYLES:
obj.line_style = old_value
return
for renderer in self._bokeh_renderers:
renderer.glyph.line_dash = new_value
if not self.ts_vector:
self.reset_bokeh_data_source()
return
self.update_bokeh_data_source()
[docs]
def line_width_callback(self, obj, attr, old_value, new_value) -> None:
if attr != 'line_width':
return
if self._state == States.DEACTIVE:
return
if obj != self.view:
obj.remove_all_callbacks(self)
return
if not self.ts_vector:
self.reset_bokeh_data_source()
return
self.update_bokeh_data_source()
self.notify_figure_y_range_update()
[docs]
class ScatterRenderer(SingleGlyphRenderer):
"""
This object contains meta-data and initialisation/update functions for scatter renderers!
All scatter renderer are i principle the same just the _glyph function is different!
"""
[docs]
def __init__(self, *, unit_registry: UnitRegistry,
notify_figure_y_range_update: Callable,
logger: logging.Logger = None) -> None:
"""
Parameters
----------
unit_registry: unit registry to use to verify data
notify_figure_y_range_update: function to trigger y_range_update of the figure renderer is connected to
"""
super().__init__(unit_registry=unit_registry,
notify_figure_y_range_update=notify_figure_y_range_update,
logger=logger)
@property
@abc.abstractmethod
def _glyph(self) -> bokeh.models.Model:
"""
This function returns the glyph of the current renderer
"""
# pass
@property
def bokeh_ds_keys(self) -> List[str]:
"""
This function returns the keys of the dynamically changeable properties of the renderer
"""
tooltips = {"t", "f", "color", "label", "size", 'fill_alpha', 'line_alpha', 'fill_color'}
# if self.view:
# for tt in self.view.tooltips:
# tooltips.add(tt[0])
return list(tooltips)
[docs]
def on_set_view(self) -> None:
self.view.on_change(obj=self, attr="index", callback=self.index_callback)
self.view.on_change(obj=self, attr="size", callback=self.patch_attr_callback)
self.view.on_change(obj=self, attr="fill_alpha", callback=self.patch_attr_callback)
self.view.on_change(obj=self, attr="line_alpha", callback=self.patch_attr_callback)
self.view.on_change(obj=self, attr="fill_color", callback=self.patch_attr_callback)
[docs]
def update_bokeh_data_source(self) -> None:
"""
This function updates the data source of the plot,
data points, color, etc.
"""
if self.do_log:
ts = utctime_now()
self.logger.debug(f"{self.__class__.__name__} {self.view.label} updating data source")
t, f = convert_ts_to_plot_vectors(ts=self.ts_vector.magnitude[self.view.index], cal=self.calendar, interpret_point_interpretation=False, crop_nan=True)
curve_data = {k: [] for k in self.bokeh_ds_keys}
curve_data["f"] = f
curve_data["t"] = t
curve_data["color"] = [self.view.color for _ in range(len(curve_data["f"]))]
curve_data["label"] = [self.view.label for _ in range(len(curve_data["f"]))]
curve_data["size"] = [self.view.size for _ in range(len(curve_data["f"]))]
curve_data["fill_alpha"] = [self.view.fill_alpha for _ in range(len(curve_data["f"]))]
curve_data["line_alpha"] = [self.view.line_alpha for _ in range(len(curve_data["f"]))]
curve_data["fill_color"] = [self.view.fill_color for _ in range(len(curve_data["f"]))]
self.bokeh_data_source.data = curve_data
if self.do_log:
self.logger.debug(f"{self.__class__.__name__} {self.view.label} update took {utctime_now() - ts}")
[docs]
def index_callback(self, obj, attr, old_value, new_value) -> None:
if self._state == States.DEACTIVE:
return
if obj != self.view:
obj.remove_all_callbacks(self)
return
if not self.ts_vector:
self.reset_bokeh_data_source()
return
self.update_bokeh_data_source()
self.notify_figure_y_range_update()
[docs]
def patch_attr_callback(self, obj, attr, old_value, new_value) -> None:
if attr not in ['size', 'fill_alpha', 'line_alpha', 'fill_color']:
obj.__setattr__(attr, new_value)
return
if self._state == States.DEACTIVE:
return
if obj != self.view:
obj.remove_all_callbacks(self)
return
if not self.ts_vector:
self.reset_bokeh_data_source()
return
size_patches = [(i, new_value) for i, d in enumerate((self.bokeh_data_source.data["f"]))]
self.bokeh_data_source.patch({attr: size_patches})
[docs]
class DiamondScatterRenderer(ScatterRenderer):
"""This object contains meta-data and initialisation/update functions for diamond scatter renderer"""
[docs]
def __init__(self, *, unit_registry: UnitRegistry,
notify_figure_y_range_update: Callable,
logger: logging.Logger = None) -> None:
"""
Parameters
----------
unit_registry: unit registry to use to verify data
notify_figure_y_range_update: function to trigger y_range_update of the figure renderer is connected to
"""
super().__init__(unit_registry=unit_registry,
notify_figure_y_range_update=notify_figure_y_range_update,
logger=logger)
@property
def _glyph(self) -> bokeh.models.Model:
""" This function returns a glyph of a diamond """
return Scatter(marker="diamond", x="t", y="f", size="size", line_color="color", fill_color="fill_color",
fill_alpha="fill_alpha", line_alpha='line_alpha',
line_width=2)
[docs]
class CircleScatterRenderer(ScatterRenderer):
"""This object contains meta-data and initialisation/update functions for circle scatter renderer"""
[docs]
def __init__(self, *, unit_registry: UnitRegistry,
notify_figure_y_range_update: Callable,
logger: logging.Logger = None) -> None:
"""
Parameters
----------
unit_registry: unit registry to use to verify data
notify_figure_y_range_update: function to trigger y_range_update of the figure renderer is connected to
"""
super().__init__(unit_registry=unit_registry,
notify_figure_y_range_update=notify_figure_y_range_update,
logger=logger)
@property
def _glyph(self) -> bokeh.models.Model:
""" This function returns a glyph of a diamond """
return Scatter(marker="circle", x="t", y="f", size="size", line_color="color", fill_color="fill_color",
fill_alpha="fill_alpha", line_alpha='line_alpha',
line_width=2)
[docs]
class SquareScatterRenderer(ScatterRenderer):
"""This object contains meta-data and initialisation/update functions for square scatter renderer"""
[docs]
def __init__(self, *, unit_registry: UnitRegistry,
notify_figure_y_range_update: Callable,
logger: logging.Logger = None) -> None:
"""
Parameters
----------
unit_registry: unit registry to use to verify data
notify_figure_y_range_update: function to trigger y_range_update of the figure renderer is connected to
"""
super().__init__(unit_registry=unit_registry,
notify_figure_y_range_update=notify_figure_y_range_update,
logger=logger)
@property
def _glyph(self) -> bokeh.models.Model:
""" This function returns a glyph of a diamond """
return Scatter(marker="square", x="t", y="f", size="size", line_color="color", fill_color="fill_color",
fill_alpha="fill_alpha", line_alpha='line_alpha',
line_width=2)
[docs]
class TriangleScatterRenderer(ScatterRenderer):
"""This object contains meta-data and initialisation/update functions for triangle scatter renderer"""
[docs]
def __init__(self, *, unit_registry: UnitRegistry,
notify_figure_y_range_update: Callable,
logger: logging.Logger = None) -> None:
"""
Parameters
----------
unit_registry: unit registry to use to verify data
notify_figure_y_range_update: function to trigger y_range_update of the figure renderer is connected to
"""
super().__init__(unit_registry=unit_registry,
notify_figure_y_range_update=notify_figure_y_range_update,
logger=logger)
@property
def _glyph(self) -> bokeh.models.Model:
""" This function returns a glyph of a diamond """
return Scatter(marker="triangle", x="t", y="f", size="size", line_color="color", fill_color="fill_color",
fill_alpha="fill_alpha", line_alpha='line_alpha',
line_width=2)
[docs]
class FillInBetweenRenderer(SingleGlyphRenderer):
"""
This object contains the meta-data and initialisation/update functions of the renderers with
lines with filled color in between
"""
[docs]
def __init__(self, *, unit_registry: UnitRegistry,
notify_figure_y_range_update: Callable,
logger: logging.Logger = None) -> None:
"""
Parameters
----------
unit_registry: unit registry to use to verify data
notify_figure_y_range_update: function to trigger y_range_update of the figure renderer is connected to
"""
super().__init__(unit_registry=unit_registry,
notify_figure_y_range_update=notify_figure_y_range_update,
logger=logger)
@property
def _glyph(self) -> bokeh.models.Model:
""" This function returns a glyph of a line with spread """
return bokeh.models.Patches(xs="t", ys="f", fill_alpha="fill_alpha", line_alpha=0.0, line_width=0,
fill_color="color")
@property
def bokeh_ds_keys(self) -> List[str]:
"""
This function keeps the keys of the dynamically changeable properties of the renderer
"""
tooltips = {"t", "f", "color", "fill_alpha", "label"}
if self.view:
for tt in self.view.tooltips:
tooltips.add(tt[0])
return list(tooltips)
[docs]
def on_set_view(self) -> None:
self.view.on_change(obj=self, attr="indices", callback=self.indices_callback)
[docs]
def update_bokeh_data_source(self) -> None:
"""
This function updates the data source of the plot,
data points, color, etc.
"""
if self.do_log:
ts = utctime_now()
self.logger.debug(f"{self.__class__.__name__} {self.view.label} updating data source")
t1, f1 = convert_ts_to_plot_vectors(ts=self.ts_vector.magnitude[self.view.indices[0]], cal=self.calendar, interpret_point_interpretation=True, crop_nan=True)
t2, f2 = convert_ts_to_plot_vectors(ts=self.ts_vector.magnitude[self.view.indices[1]], cal=self.calendar, interpret_point_interpretation=True, crop_nan=True)
non_nan_slices = np.ma.masked_invalid(f1)
if len(non_nan_slices) != 0:
non_nan_slices = np.ma.clump_unmasked(non_nan_slices)
time_patches = data_to_patch_values(t1, t2, non_nan_slices)
data_patches = data_to_patch_values(f1, f2, non_nan_slices)
curve_data = {k: [] for k in self.bokeh_ds_keys}
curve_data["f"] = data_patches
curve_data["t"] = time_patches
curve_data["color"] = [self.view.color for _ in range(len(curve_data["f"]))]
curve_data["fill_alpha"] = [self.view.fill_alpha for _ in range(len(curve_data["f"]))]
curve_data["label"] = [self.view.label for _ in range(len(curve_data["f"]))]
self.bokeh_data_source.data = curve_data
if self.do_log:
self.logger.debug(f"{self.__class__.__name__} {self.view.label} update took {utctime_now() - ts}")
[docs]
def indices_callback(self, obj, attr, old_value, new_value) -> None:
if self._state == States.DEACTIVE:
return
if obj != self.view:
obj.remove_all_callbacks(self)
return
if not self.ts_vector:
self.reset_bokeh_data_source()
return
if len(new_value) != 2:
# print("ERROR ... wrong index assigned to {obj}.{label}")
return
self.update_bokeh_data_source()
self.notify_figure_y_range_update()
[docs]
class MultiLineRenderer(SingleGlyphRenderer):
"""
This object contains meta-data and initialisation/update functions for the renderers
of a MultiLine
"""
[docs]
def __init__(self, *, unit_registry: UnitRegistry,
notify_figure_y_range_update: Callable,
logger: logging.Logger = None) -> None:
"""
Parameters
----------
unit_registry: unit registry to use to verify data
notify_figure_y_range_update: function to trigger y_range_update of the figure renderer is connected to
"""
super().__init__(unit_registry=unit_registry,
notify_figure_y_range_update=notify_figure_y_range_update,
logger=logger)
@property
def _glyph(self) -> bokeh.models.Model:
""" This function returns a glyph of a multiline """
return bokeh.models.MultiLine(xs="t", ys="f", line_color="color", line_dash="line_style", line_width="line_width")
@property
def bokeh_ds_keys(self) -> List[str]:
"""
This function returns the keys of the dynamically changeable properties of the renderer
"""
tooltips = {"t", "f", "color", "line_width", "line_style"}
if self.view:
for tt in self.view.tooltips:
tooltips.add(tt[0])
return list(tooltips)
[docs]
def on_set_view(self) -> None:
self.view.on_change(obj=self, attr="indices", callback=self.index_callback)
self.view.on_change(obj=self, attr="line_styles", callback=self.line_style_callback)
self.view.on_change(obj=self, attr='line_widths', callback=self.line_width_callback)
# init line style callback
self.line_style_callback(obj=self.view, attr='line_styles', old_value='solid',
new_value=self.view.line_styles[0])
[docs]
def update_bokeh_data_source(self) -> None:
"""
This function updates the data source of the plot,
data points, color, etc.
"""
if self.do_log:
ts = utctime_now()
self.logger.debug(f"{self.__class__.__name__} {self.view.label} updating data source")
cal = self.calendar
tv, fv = [], []
for i, index in enumerate(self.view.indices):
t, f = convert_ts_to_plot_vectors(ts=self.ts_vector.magnitude[index], cal=cal, interpret_point_interpretation=True, crop_nan=True)
tv.append(t)
fv.append(f)
curve_data = {k: [] for k in self.bokeh_ds_keys}
curve_data["f"] = fv
curve_data["t"] = tv
curve_data["color"] = self.view.color
curve_data["line_width"] = self.view.line_widths
curve_data["line_style"] = self.view.line_styles
# curve_data["label"] = self.view.labels
for tt in self.view.tooltips:
curve_data[tt[0]] = getattr(self.view, tt[0])
self.bokeh_data_source.data = curve_data
if self.do_log:
self.logger.debug(f"{self.__class__.__name__} {self.view.label} update took {utctime_now() - ts}")
self.view.view_container.update_y_range()
[docs]
def index_callback(self, obj, attr, old_value, new_value) -> None:
if self._state == States.DEACTIVE:
return
if obj != self.view:
obj.remove_all_callbacks(self)
return
if not self.ts_vector:
self.reset_bokeh_data_source()
return
self.update_bokeh_data_source()
self.notify_figure_y_range_update()
[docs]
def line_style_callback(self, obj, attr, old_value, new_value) -> None:
if attr != 'line_styles':
return
if self._state == States.DEACTIVE:
return
if obj != self.view:
obj.remove_all_callbacks(self)
return
if new_value not in VALID_LINE_STYLES:
obj.line_styles = old_value
return
for renderer in self._bokeh_renderers:
renderer.glyph.line_dash = new_value
if not self.ts_vector:
self.reset_bokeh_data_source()
return
self.update_bokeh_data_source()
[docs]
def line_width_callback(self, obj, attr, old_value, new_value) -> None:
if attr != 'line_widths':
return
if self._state == States.DEACTIVE:
return
if obj != self.view:
obj.remove_all_callbacks(self)
return
if not self.ts_vector:
self.reset_bokeh_data_source()
return
self.update_bokeh_data_source()
self.notify_figure_y_range_update()
[docs]
class BackgroundDataRenderer(SingleGlyphRenderer):
"""
This object contains the meta-data and initialisation/update functions of the renderers with
lines with filled color in between
"""
[docs]
def __init__(self, *, unit_registry: UnitRegistry,
notify_figure_y_range_update: Callable,
logger: logging.Logger = None) -> None:
"""
Parameters
----------
unit_registry: unit registry to use to verify data
notify_figure_y_range_update: function to trigger y_range_update of the figure renderer is connected to
"""
super().__init__(unit_registry=unit_registry,
notify_figure_y_range_update=notify_figure_y_range_update,
logger=logger)
@property
def _glyph(self) -> bokeh.models.Model:
""" This function returns a glyph of a line with spread """
return bokeh.models.Quad(left="left", right="right",
top="top", bottom="bottom", fill_alpha="fill_alpha", line_alpha=0.0, line_width=0,
fill_color="colors")
@property
def bokeh_ds_keys(self) -> List[str]:
"""
This function keeps the keys of the dynamically changeable properties of the renderer
"""
return ["left", "right", "top", "bottom", "colors", "fill_alpha"]
[docs]
def on_set_view(self) -> None:
self.view.on_change(obj=self, attr="index", callback=self.redraw_callback)
self.view.on_change(obj=self, attr="fill_alpha", callback=self.redraw_callback)
self.view.on_change(obj=self, attr="values_color_map", callback=self.values_color_map_callback)
self.view.on_change(obj=self, attr="index", callback=self.redraw_callback)
self.view.on_change(obj=self, attr="y_max", callback=self.redraw_callback)
self.view.on_change(obj=self, attr="y_min", callback=self.redraw_callback)
self.view.on_change(obj=self, attr="show_not_defined", callback=self.redraw_callback)
[docs]
def update_bokeh_data_source(self) -> None:
"""
This function updates the data source of the plot,
data points, color, etc.
"""
if self.do_log:
ts = utctime_now()
self.logger.debug(f"{self.__class__.__name__} {self.view.label} updating data source")
t, f = convert_ts_to_plot_vectors(ts=self.ts_vector.magnitude[self.view.index], cal=self.calendar, interpret_point_interpretation=True, crop_nan=True)
line_indices, scatter_indices = calculate_dead_band_indices(ts_input=f)
left_indices = line_indices[:, 0]
right_indices = line_indices[:, 1]
values_color_map = {k: v['color'] for k, v in self.view.values_color_map.items()}
# remove values not in view.values_color_map if view.show_not_defined == True
if not self.view.show_not_defined:
to_keep = list(values_color_map.keys())
def map(entry):
return entry in to_keep
mask = np.nonzero(np.vectorize(map)(f))
left_indices = np.intersect1d(mask, left_indices)
right_indices = np.intersect1d(mask, right_indices)
n = len(left_indices)
curve_data = {k: [] for k in self.bokeh_ds_keys}
curve_data["left"] = t[left_indices]
curve_data["right"] = t[right_indices]
curve_data["top"] = [self.view.y_max]*n
curve_data["bottom"] = [self.view.y_min]*n
def map(entry):
return values_color_map.get(entry, self.view.color)
v = np.vectorize(map)
curve_data["colors"] = v(f[left_indices])
curve_data["fill_alpha"] = [self.view.fill_alpha]*n
self.bokeh_data_source.data = curve_data
if self.do_log:
self.logger.debug(f"{self.__class__.__name__} {self.view.label} update took {utctime_now() - ts}")
[docs]
def y_range(self, view_range: UtcPeriod) -> np.ndarray:
"""
This function returns an np.ndarray with min, and max y values for given view period
It is hardcoded set in the view that self.view.no_y_rescaling = False,
just in case some messes with that we return self.view.y_min, self.view.y_max if the check fails
"""
ts_vector = self.ts_vector
if not ts_vector or ts_vector is None or not self.view or \
self.view and self.view.no_y_rescaling:
return np.array([np.nan, np.nan])
return np.array([self.view.y_min, self.view.y_max])
[docs]
def redraw_callback(self, obj, attr, old_value, new_value) -> None:
"""
Callback to redraw the renderer user for all view. variables which require a redrawing of the renderer to
be effective
"""
if self._state == States.DEACTIVE:
return
if obj != self.view:
obj.remove_all_callbacks(self)
return
if not self.ts_vector:
self.reset_bokeh_data_source()
return
self.update_bokeh_data_source()
self.notify_figure_y_range_update()
[docs]
def values_color_map_callback(self, obj, attr, old_value, new_value) -> None:
"""
Callback to redraw the renderer user for all view. variables which require a redrawing of the renderer to
be effective
"""
if not isinstance(new_value, dict):
raise ValueError(
f"BackgroundData in values_color_map: expect dict got {type(new_value)}")
for k, v in new_value.items():
if not isinstance(v, dict):
raise ValueError(
f"BackgroundData in values_color_map: expect values in "
f"form of Dict[str, str] with keys `label` and `color` got {v}")
for label in ['color', 'label']:
if label not in v:
raise KeyError(f"BackgroundData in values_color_map: `{label}` not in dict of {k}: got {v}")
if not isinstance(v[label], str):
raise ValueError(f"BackgroundData in values_color_map: `{label}` not of type str: got {v[label]} {type(v[label])}")
if self._state == States.DEACTIVE:
return
if obj != self.view:
obj.remove_all_callbacks(self)
return
if not self.ts_vector:
self.reset_bokeh_data_source()
return
self.update_bokeh_data_source()
self.notify_figure_y_range_update()