Spaces:
Running
on
Zero
Running
on
Zero
| #----------------------------------------------------------------------------- | |
| # Copyright (c) 2012 - 2022, 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 a | |
| a function parameter in a specified way each time the function is | |
| called. | |
| These decorators can be especially useful in conjunction with periodic | |
| callbacks 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__ import annotations | |
| import logging # isort:skip | |
| log = logging.getLogger(__name__) | |
| #----------------------------------------------------------------------------- | |
| # Imports | |
| #----------------------------------------------------------------------------- | |
| # Standard library imports | |
| from functools import partial | |
| from typing import ( | |
| Any, | |
| Callable, | |
| Iterable, | |
| Iterator, | |
| Sequence, | |
| TypeVar, | |
| ) | |
| #----------------------------------------------------------------------------- | |
| # Globals and constants | |
| #----------------------------------------------------------------------------- | |
| __all__ = ( | |
| 'bounce', | |
| 'cosine', | |
| 'count', | |
| 'force', | |
| 'linear', | |
| 'repeat', | |
| 'sine', | |
| ) | |
| #----------------------------------------------------------------------------- | |
| # General API | |
| #----------------------------------------------------------------------------- | |
| def bounce(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) | |
| def f(i: int) -> int: | |
| div, mod = divmod(i, N) | |
| if div % 2 == 0: | |
| return sequence[mod] | |
| else: | |
| return sequence[N-mod-1] | |
| return partial(force, sequence=_advance(f)) | |
| def cosine(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 | |
| ''' | |
| from math import cos | |
| def f(i: float) -> float: | |
| return A * cos(w*i + phi) + offset | |
| return partial(force, sequence=_advance(f)) | |
| def count() -> partial[Callable[[], None]]: | |
| ''' Return a driver function that can advance a simple count. | |
| ''' | |
| return partial(force, sequence=_advance(lambda x: x)) | |
| def force(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 | |
| ''' | |
| def wrapper() -> None: | |
| f(next(sequence)) | |
| return wrapper | |
| def linear(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 | |
| x (float) : an offset for the linear driver | |
| ''' | |
| def f(i: float) -> float: | |
| return m * i + b | |
| return partial(force, sequence=_advance(f)) | |
| def repeat(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) | |
| def f(i: int) -> int: | |
| return sequence[i%N] | |
| return partial(force, sequence=_advance(f)) | |
| def sine(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 | |
| ''' | |
| from math import sin | |
| def f(i: float) -> float: | |
| return A * sin(w*i + phi) + offset | |
| return partial(force, sequence=_advance(f)) | |
| #----------------------------------------------------------------------------- | |
| # Dev API | |
| #----------------------------------------------------------------------------- | |
| #----------------------------------------------------------------------------- | |
| # Private API | |
| #----------------------------------------------------------------------------- | |
| T = TypeVar("T") | |
| def _advance(f: Callable[[int], T]) -> Iterable[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 = 0 | |
| while True: | |
| yield f(i) | |
| i += 1 | |
| #----------------------------------------------------------------------------- | |
| # Code | |
| #----------------------------------------------------------------------------- | |