🔍 Code Extractor

A utility class that manages periodic event updates for DynamicMap objects, allowing scheduled triggering of stream updates that can be started and stopped.

Purpose

The periodic class provides a mechanism to periodically trigger stream updates on a DynamicMap object at specified intervals. It wraps a periodic utility function to create non-blocking loops that can update stream parameters repeatedly, either for a fixed count or indefinitely. This is useful for creating animations, real-time data updates, or any scenario requiring scheduled event-driven updates to visualizations.

Source Code

class periodic(object):
    """
    Implements the utility of the same name on DynamicMap.

    Used to defined periodic event updates that can be started and
    stopped.
    """
    _periodic_util = util.periodic

    def __init__(self, dmap):
        self.dmap = dmap
        self.instance = None

    def __call__(self, period, count=None, param_fn=None, timeout=None, block=True):
        """Periodically trigger the streams on the DynamicMap.

        Run a non-blocking loop that updates the stream parameters using
        the event method. Runs count times with the specified period. If
        count is None, runs indefinitely.

        Args:
            period: Timeout between events in seconds
            count: Number of events to trigger
            param_fn: Function returning stream updates given count
               Stream parameter values should be returned as dictionary
            timeout: Overall timeout in seconds
            block: Whether the periodic callbacks should be blocking
        """

        if self.instance is not None and not self.instance.completed:
            raise RuntimeError('Periodic process already running. '
                               'Wait until it completes or call '
                               'stop() before running a new periodic process')
        def inner(i):
            kwargs = {} if param_fn is None else param_fn(i)
            if kwargs:
                self.dmap.event(**kwargs)
            else:
                Stream.trigger(self.dmap.streams)

        instance = self._periodic_util(period, count, inner,
                                       timeout=timeout, block=block)
        instance.start()
        self.instance = instance

    def stop(self):
        "Stop the periodic process."
        self.instance.stop()

    def __str__(self):
        return "<holoviews.core.spaces.periodic method>"

Parameters

Parameter Details

dmap: A DynamicMap object that this periodic instance will manage. The DynamicMap's streams will be triggered periodically according to the configured schedule.

Return Value

Instantiation returns a periodic object that manages the lifecycle of periodic updates for the associated DynamicMap. The __call__ method returns None but starts the periodic process. The stop method returns None. The __str__ method returns a string representation '<holoviews.core.spaces.periodic method>'.

Class Interface

Methods

Attributes

Dependencies

• holoviews.util
• holoviews.streams

Required Imports

from holoviews.core.spaces import periodic
from holoviews.streams import Stream

Usage Example

# Assuming you have a DynamicMap object
import holoviews as hv
from holoviews.streams import Stream

# Create a DynamicMap with a stream
stream = Stream.define('MyStream', x=0)()
def plot_fn(x):
    return hv.Curve([(0, x), (1, x+1)])

dmap = hv.DynamicMap(plot_fn, streams=[stream])

# Create periodic instance
per = periodic(dmap)

# Start periodic updates every 0.5 seconds, 10 times
def update_params(i):
    return {'x': i}

per(period=0.5, count=10, param_fn=update_params)

# Stop the periodic process before completion if needed
# per.stop()

Best Practices

• Always check if a periodic process is already running before starting a new one, or call stop() first to avoid RuntimeError
• Use the param_fn parameter to dynamically generate stream parameters based on the iteration count
• Set count=None for indefinite periodic updates, but ensure you have a way to stop it later
• Use timeout parameter to automatically stop the periodic process after a specified duration
• The block parameter controls whether callbacks are blocking; set to False for non-blocking behavior in interactive environments
• Store the periodic instance to call stop() when needed to clean up resources
• If param_fn returns an empty dictionary or None, the method will trigger all streams without parameters

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class DynamicMap 59.3% similar

  A DynamicMap is a type of HoloMap where the elements are dynamically generated by a callable. The callable is invoked with values associated with the key dimensions or with values supplied by stream parameters.

  From: /tf/active/vicechatdev/patches/spaces.py

• function dynamicmap_memoization 55.5% similar

  A context manager that temporarily controls memoization behavior of a callable object based on the state of associated streams, disabling memoization when transient streams are triggering.

  From: /tf/active/vicechatdev/patches/spaces.py

• function dimensioned_streams 50.1% similar

  Filters and returns streams from a DynamicMap that have parameters matching the DynamicMap's key dimensions.

  From: /tf/active/vicechatdev/patches/util.py

• class periodic 49.1% similar

  A threading class that executes a callback function periodically at specified intervals, with optional count limits and timeout controls.

  From: /tf/active/vicechatdev/patches/util.py

• class Callable 47.7% similar

  Callable is a wrapper class for callback functions used with DynamicMaps, providing memoization, stream management, and input/output tracking capabilities.

  From: /tf/active/vicechatdev/patches/spaces.py