Source code for holoviews.core.decollate

from collections import namedtuple

import param

from .. import (
from ..plotting.util import initialize_dynamic
from ..streams import Derived, Stream
from . import AdjointLayout, ViewableTree
from .operation import OperationCallable

Expr = namedtuple("HoloviewsExpr", ["fn", "args", "kwargs"])
StreamIndex = namedtuple("StreamIndex", ["index"])
KDimIndex = namedtuple("KDim", ["index"])

[docs]def to_expr_extract_streams( hvobj, kdims, streams, original_streams, stream_mapping, container_key=None ): """ Build a HoloViewsExpr expression tree from a potentially nested dynamic HoloViews object, extracting the streams and replacing them with StreamIndex objects. This function is recursive an assumes that initialize_dynamic has already been called on the input object. Args: hvobj: Element or DynamicMap or Layout Potentially dynamic HoloViews object to represent as a HoloviewsExpr kdims: list of Dimensions List that DynamicMap key-dimension objects should be added to streams: list of Stream List that cloned extracted streams should be added to original_streams: list of Stream List that original extracted streams should be added to stream_mapping: dict dict to be populated with mappings from container keys to extracted Stream objects, as described by the Callable parameter of the same name. container_key: int or tuple key into parent container that is associated to hvobj, or None if hvobj is not in a container Returns: HoloviewsExpr expression representing hvobj if hvobj is dynamic. Otherwise, return hvobj itself """ if isinstance(hvobj, DynamicMap): args = [] kwargs = [] dm_streams = hvobj.streams # Process callback inputs recursively input_exprs = [ to_expr_extract_streams( v, kdims, streams, original_streams, stream_mapping, container_key=container_key, ) for i, v in enumerate(hvobj.callback.inputs) ] # Record all key dimensions kdim_args = [] for kdim in hvobj.kdims: current_kdim_names = [ for k in kdims] if in current_kdim_names: # Find index to existing kdim idx = current_kdim_names.index( kdim_index = KDimIndex(index=idx) # Overwrite so that we end up with dimension object highest in the # object tree kdims[idx] = kdim else: # Add new kdim index kdim_index = KDimIndex(index=len(kdims)) kdims.append(kdim) kdim_args.append(kdim_index) # Determine function expand_kwargs = True if len(input_exprs) > 1: fn = Overlay args.extend([input_exprs]) elif isinstance(hvobj.callback, OperationCallable): fn = hvobj.callback.operation.instance(streams=[]) fn.dynamic = False args.extend(input_exprs) if "kwargs" in fn.param: expand_kwargs = False if "kwargs" in hvobj.callback.operation_kwargs: kwargs.append(hvobj.callback.operation_kwargs["kwargs"]) elif hvobj.callback.operation_kwargs: # Preserve custom operation kwargs kwargs.append(hvobj.callback.operation_kwargs) else: fn = hvobj.callback.callable args.extend(kdim_args) for dm_stream in dm_streams: stream_arg = to_expr_extract_streams( dm_stream, kdims, streams, original_streams, stream_mapping, container_key, ) if hvobj.positional_stream_args: args.append(stream_arg) else: kwargs.append(stream_arg) if expand_kwargs: expr = Expr(fn, args, kwargs) else: expr = Expr(fn, args, [{"kwargs": Expr(dict, [], kwargs)}]) return expr elif isinstance(hvobj, Stream): if isinstance(hvobj, Derived): stream_arg_fn = hvobj.transform_function stream_indexes = [] for input_stream in hvobj.input_streams: stream_indexes.append( to_expr_extract_streams( input_stream, kdims, streams, original_streams, stream_mapping, container_key, ) ) constants = hvobj.constants return Expr( stream_arg_fn, [stream_indexes, constants], [] ) else: # Get index for stream # Compute stream index if hvobj in original_streams: # Reuse index to existing stream stream_index = StreamIndex(index=original_streams.index(hvobj)) else: # Add new stream stream_index = StreamIndex(index=len(streams)) cloned_stream = hvobj.clone() original_streams.append(hvobj) streams.append(cloned_stream) if container_key is not None: stream_mapping.setdefault(container_key, []).append(cloned_stream) return stream_index elif isinstance(hvobj, (Layout, GridSpace, NdOverlay, HoloMap, Overlay, AdjointLayout)): fn = hvobj.clone(data={}).clone args = [] data_expr = [] for i, (key, v) in enumerate( el = to_expr_extract_streams( v, kdims, streams, original_streams, stream_mapping, i ) # Replace "DynamicMap" with type of the non-dynamic return element if isinstance(v, DynamicMap): initialize_dynamic(v) if (v.type is not None and isinstance(key, tuple) and isinstance(key[0], str)): type_str = v.type.__name__ key = (key[0].replace("DynamicMap", type_str), "I") data_expr.append((key, el)) if isinstance(hvobj, ViewableTree): # Use _process_items to ensure that keys are unique data_expr = ViewableTree._process_items(data_expr) kwargs = [{"data": data_expr}] return Expr(fn, args, kwargs) elif isinstance(hvobj, Element): return hvobj.clone(link=False) else: raise NotImplementedError(f"Type {type(hvobj)} not implemented")
def expr_to_fn_of_stream_contents(expr, nkdims): def eval_expr(expr, kdim_values, stream_values): if isinstance(expr, Expr): fn = expr.fn args = [eval_expr(arg, kdim_values, stream_values) for arg in expr.args] kwargs_list = [eval_expr(kwarg, kdim_values, stream_values) for kwarg in expr.kwargs] kwargs = {} for kwargs_el in kwargs_list: kwargs.update(**eval_expr(kwargs_el, kdim_values, stream_values)) # For a ParameterizedFunction (e.g. an Operation), drop keys that are not # accepted as params to avoid warnings if isinstance(fn, param.ParameterizedFunction): kwargs = {k: v for k, v in kwargs.items() if k in fn.param} return fn(*args, **kwargs) elif isinstance(expr, StreamIndex): return stream_values[expr.index] elif isinstance(expr, KDimIndex): return kdim_values[expr.index] elif isinstance(expr, dict): return {k: eval_expr(v, kdim_values, stream_values) for k, v in expr.items()} elif isinstance(expr, (list, tuple)): return type(expr)([eval_expr(v, kdim_values, stream_values) for v in expr]) else: return expr def expr_fn(*args): kdim_values = args[:nkdims] stream_values = args[nkdims:] return eval_expr(expr, kdim_values, stream_values) return expr_fn
[docs]def decollate(hvobj): """ Decollate transforms a potentially nested dynamic HoloViews object into single DynamicMap that returns a non-dynamic HoloViews object. All nested streams in the input object are copied and attached to the resulting DynamicMap. Args: hvobj: Holoviews object Returns: DynamicMap """ kdims = [] original_streams = [] streams = [] stream_mapping = {} initialize_dynamic(hvobj) expr = to_expr_extract_streams(hvobj, kdims, streams, original_streams, stream_mapping) expr_fn = expr_to_fn_of_stream_contents(expr, nkdims=len(kdims)) callback = Callable(expr_fn, stream_mapping=stream_mapping) return DynamicMap( callback, kdims=kdims, streams=streams, positional_stream_args=True )