#-----------------------------------------------------------------------------# Copyright (c) Anaconda, Inc., and Bokeh Contributors.# All rights reserved.## The full license is in the file LICENSE.txt, distributed with this software.#-----------------------------------------------------------------------------''' Provide a set of decorators useful for repeatedly updating aa function parameter in a specified way each time the function iscalled.These decorators can be especially useful in conjunction with periodiccallbacks in a Bokeh server application.Example: As an example, consider the ``bounce`` forcing function, which advances a sequence forwards and backwards: .. code-block:: python from bokeh.driving import bounce @bounce([0, 1, 2]) def update(i): print(i) If this function is repeatedly called, it will print the following sequence on standard out: .. code-block:: none 0 1 2 2 1 0 0 1 2 2 1 ...'''#-----------------------------------------------------------------------------# Boilerplate#-----------------------------------------------------------------------------from__future__importannotationsimportlogging# isort:skiplog=logging.getLogger(__name__)#-----------------------------------------------------------------------------# Imports#-----------------------------------------------------------------------------# Standard library importsfromfunctoolsimportpartialfromtypingimport(Any,Callable,Iterator,Sequence,TypeVar,)#-----------------------------------------------------------------------------# Globals and constants#-----------------------------------------------------------------------------__all__=('bounce','cosine','count','force','linear','repeat','sine',)#-----------------------------------------------------------------------------# General API#-----------------------------------------------------------------------------
[docs]defbounce(sequence:Sequence[int])->partial[Callable[[],None]]:''' Return a driver function that can advance a "bounced" sequence of values. .. code-block:: none seq = [0, 1, 2, 3] # bounce(seq) => [0, 1, 2, 3, 3, 2, 1, 0, 0, 1, 2, ...] Args: sequence (seq) : a sequence of values for the driver to bounce '''N=len(sequence)deff(i:int)->int:div,mod=divmod(i,N)ifdiv%2==0:returnsequence[mod]else:returnsequence[N-mod-1]returnpartial(force,sequence=_advance(f))
[docs]defcosine(w:float,A:float=1,phi:float=0,offset:float=0)->partial[Callable[[],None]]:''' Return a driver function that can advance a sequence of cosine values. .. code-block:: none value = A * cos(w*i + phi) + offset Args: w (float) : a frequency for the cosine driver A (float) : an amplitude for the cosine driver phi (float) : a phase offset to start the cosine driver with offset (float) : a global offset to add to the driver values '''frommathimportcosdeff(i:float)->float:returnA*cos(w*i+phi)+offsetreturnpartial(force,sequence=_advance(f))
[docs]defcount()->partial[Callable[[],None]]:''' Return a driver function that can advance a simple count. '''returnpartial(force,sequence=_advance(lambdax:x))
[docs]defforce(f:Callable[[Any],None],sequence:Iterator[Any])->Callable[[],None]:''' Return a decorator that can "force" a function with an arbitrary supplied generator Args: sequence (iterable) : generator to drive f with Returns: decorator '''defwrapper()->None:f(next(sequence))returnwrapper
[docs]deflinear(m:float=1,b:float=0)->partial[Callable[[],None]]:''' Return a driver function that can advance a sequence of linear values. .. code-block:: none value = m * i + b Args: m (float) : a slope for the linear driver b (float) : an offset for the linear driver '''deff(i:float)->float:returnm*i+breturnpartial(force,sequence=_advance(f))
[docs]defrepeat(sequence:Sequence[int])->partial[Callable[[],None]]:''' Return a driver function that can advance a repeated of values. .. code-block:: none seq = [0, 1, 2, 3] # repeat(seq) => [0, 1, 2, 3, 0, 1, 2, 3, 0, 1, ...] Args: sequence (seq) : a sequence of values for the driver to bounce '''N=len(sequence)deff(i:int)->int:returnsequence[i%N]returnpartial(force,sequence=_advance(f))
[docs]defsine(w:float,A:float=1,phi:float=0,offset:float=0)->partial[Callable[[],None]]:''' Return a driver function that can advance a sequence of sine values. .. code-block:: none value = A * sin(w*i + phi) + offset Args: w (float) : a frequency for the sine driver A (float) : an amplitude for the sine driver phi (float) : a phase offset to start the sine driver with offset (float) : a global offset to add to the driver values '''frommathimportsindeff(i:float)->float:returnA*sin(w*i+phi)+offsetreturnpartial(force,sequence=_advance(f))
#-----------------------------------------------------------------------------# Dev API#-----------------------------------------------------------------------------#-----------------------------------------------------------------------------# Private API#-----------------------------------------------------------------------------T=TypeVar("T")def_advance(f:Callable[[int],T])->Iterator[T]:''' Yield a sequence generated by calling a given function with successively incremented integer values. Args: f (callable) : The function to advance Yields: f(i) where i increases each call '''i=0whileTrue:yieldf(i)i+=1#-----------------------------------------------------------------------------# Code#-----------------------------------------------------------------------------