🔍 Code Extractor

Validates that a callback function has an appropriate signature to work with DynamicMap by checking its arguments against key dimensions (kdims) and stream parameters.

Purpose

This function is used by DynamicMap to ensure callback functions have compatible signatures before execution. It validates that the callback can accept the required key dimensions and stream parameters, either as positional or keyword arguments. The function returns a mapping strategy (list of parameter names or None) that indicates how to pass kdim values to the callback, enabling proper argument binding between DynamicMap's data and the callback function.

Source Code

def validate_dynamic_argspec(callback, kdims, streams):
    """
    Utility used by DynamicMap to ensure the supplied callback has an
    appropriate signature.

    If validation succeeds, returns a list of strings to be zipped with
    the positional arguments, i.e. kdim values. The zipped values can then
    be merged with the stream values to pass everything to the Callable
    as keywords.

    If the callbacks use *args, None is returned to indicate that kdim
    values must be passed to the Callable by position. In this
    situation, Callable passes *args and **kwargs directly to the
    callback.

    If the callback doesn't use **kwargs, the accepted keywords are
    validated against the stream parameter names.
    """
    argspec = callback.argspec
    name = callback.name
    kdims = [kdim.name for kdim in kdims]
    stream_params = stream_parameters(streams)
    defaults = argspec.defaults if argspec.defaults else []
    all_posargs = argspec.args[:-len(defaults)] if defaults else argspec.args
    # Filter out any posargs for streams
    posargs = [arg for arg in all_posargs if arg not in stream_params]
    kwargs = argspec.args[-len(defaults):]

    if argspec.varkw is None:
        unassigned_streams = set(stream_params) - set(argspec.args)
        if unassigned_streams:
            unassigned = ','.join(unassigned_streams)
            raise KeyError('Callable {name!r} missing keywords to '
                           'accept stream parameters: {unassigned}'.format(name=name,
                                                                    unassigned=unassigned))


    if len(posargs) > len(kdims) + len(stream_params):
        raise KeyError('Callable {name!r} accepts more positional arguments than '
                       'there are kdims and stream parameters'.format(name=name))
    if kdims == []:                  # Can be no posargs, stream kwargs already validated
        return []
    if set(kdims) == set(posargs):   # Posargs match exactly, can all be passed as kwargs
        return kdims
    elif len(posargs) == len(kdims): # Posargs match kdims length, supplying names
        if argspec.args[:len(kdims)] != posargs:
            raise KeyError('Unmatched positional kdim arguments only allowed at '
                           'the start of the signature of {name!r}'.format(name=name))

        return posargs
    elif argspec.varargs:            # Posargs missing, passed to Callable directly
        return None
    elif set(posargs) - set(kdims):
        raise KeyError('Callable {name!r} accepts more positional arguments {posargs} '
                       'than there are key dimensions {kdims}'.format(name=name,
                                                                      posargs=posargs,
                                                                      kdims=kdims))
    elif set(kdims).issubset(set(kwargs)): # Key dims can be supplied by keyword
        return kdims
    elif set(kdims).issubset(set(posargs+kwargs)):
        return kdims
    elif argspec.varkw:
        return kdims
    else:
        raise KeyError('Callback {name!r} signature over {names} does not accommodate '
                       'required kdims {kdims}'.format(name=name,
                                                       names=list(set(posargs+kwargs)),
                                                       kdims=kdims))

Parameters

Parameter Details

callback: A Callable object that wraps the function to be validated. Must have 'argspec' attribute (function signature information) and 'name' attribute (function name). The argspec should contain 'args', 'defaults', 'varargs', and 'varkw' attributes similar to inspect.ArgSpec.

kdims: A list of key dimension objects, each having a 'name' attribute. These represent the dimensional parameters that the callback must accept. Can be an empty list if no key dimensions are required.

streams: A collection of stream objects that provide additional parameters to the callback. The function extracts parameter names from these streams using the stream_parameters() helper function to validate against the callback signature.

Return Value

Returns one of three possible values: (1) A list of strings representing kdim names that should be passed as keyword arguments to the callback, allowing the caller to zip these names with kdim values. (2) None if the callback uses *args (varargs), indicating kdim values should be passed positionally. (3) An empty list if there are no kdims. Raises KeyError if validation fails due to signature mismatches.

Dependencies

• param

Required Imports

from holoviews.util import stream_parameters

Usage Example

# Assuming you have a DynamicMap context with callback, kdims, and streams
import param
from holoviews.core.util import stream_parameters

# Example callback with proper signature
class MyCallback:
    def __init__(self, func):
        self.func = func
        self.argspec = inspect.getfullargspec(func)
        self.name = func.__name__
    
    def __call__(self, *args, **kwargs):
        return self.func(*args, **kwargs)

# Define a callback function
def my_func(x, y, stream_param=None):
    return x + y

# Create kdim-like objects
class KDim:
    def __init__(self, name):
        self.name = name

kdims = [KDim('x'), KDim('y')]
streams = []  # Empty streams for this example

# Wrap and validate
callback = MyCallback(my_func)
try:
    param_names = validate_dynamic_argspec(callback, kdims, streams)
    print(f"Validation successful. Parameter names: {param_names}")
    # param_names would be ['x', 'y'] in this case
except KeyError as e:
    print(f"Validation failed: {e}")

Best Practices

• Ensure callback objects have both 'argspec' and 'name' attributes before calling this function
• The callback's argspec should follow Python's inspect.ArgSpec or inspect.FullArgSpec structure
• When using *args in callbacks, be aware that None will be returned, requiring positional argument passing
• If callback doesn't use **kwargs, ensure all stream parameters are explicitly listed in the function signature
• Key dimensions should either match positional arguments exactly or be available as keyword arguments
• Stream parameters can appear as positional arguments and will be filtered out when counting kdim positions
• Handle KeyError exceptions that may be raised during validation to provide user-friendly error messages

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function dimensioned_streams 56.6% similar

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

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

• function dynamicmap_memoization 50.2% 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

• class DynamicMap 50.2% 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 rename_stream_kwargs 49.1% similar

  Maps parameter names in a kwargs dictionary to their renamed equivalents for a given stream object, with optional reverse mapping from renamed back to original names.

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

• class Callable 48.4% 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