Source code for panel.pane.streamz

Renders Streamz Stream objects.
from __future__ import annotations

import sys

from typing import (
    TYPE_CHECKING, Any, ClassVar, Mapping, Optional,

import param

from .base import ReplacementPane

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

[docs]class Streamz(ReplacementPane): """ The `Streamz` pane renders streamz `Stream` objects emitting arbitrary objects, unlike the DataFrame pane which specifically handles streamz DataFrame and Series objects and exposes various formatting objects. Reference: :Example: >>> Streamz(some_streamz_stream_object, always_watch=True) """ always_watch = param.Boolean(default=False, doc=""" Whether to watch even when not displayed.""") rate_limit = param.Number(default=0.1, bounds=(0, None), doc=""" The minimum interval between events.""") _rename: ClassVar[Mapping[str, str | None]] = {'rate_limit': None, 'always_watch': None} def __init__(self, object=None, **params): super().__init__(object, **params) self._stream = None if self.always_watch: self._setup_stream() @param.depends('always_watch', 'object', 'rate_limit', watch=True) def _setup_stream(self): if self.object is None or (self.always_watch and self._stream): return elif self._stream: self._stream.destroy() self._stream = None if self._pane._models or self.always_watch: self._stream = self.object.latest().rate_limit(self.rate_limit).gather() self._stream.sink(self._update_inner) def _get_model( self, doc: Document, root: Optional[Model] = None, parent: Optional[Model] = None, comm: Optional[Comm] = None ) -> Model: model = super()._get_model(doc, root, parent, comm) self._setup_stream() return model def _cleanup(self, root: Model | None = None): super()._cleanup(root) if not self._pane._models and self._stream: self._stream.destroy() self._stream = None #---------------------------------------------------------------- # Public API #----------------------------------------------------------------
[docs] @classmethod def applies(cls, obj: Any) -> float | bool | None: if 'streamz' in sys.modules: from streamz import Stream return isinstance(obj, Stream) return False