企业官网建设流程全解析

Bokehbokeh.driving模块全解析：用装饰器驱动周期回调的动画数据源

【免费下载链接】bokehInteractive Data Visualization in the browser, from Python项目地址: https://gitcode.com/GitHub_Trending/bo/bokeh

导读

bokeh.driving是 Bokeh 官方提供的装饰器工具集，专门用于在函数每次被调用时，按照预定规则自动推进一个参数值，从而免去手写计数器与状态变量的繁琐。该模块的核心应用场景是配合 Bokeh 服务器应用（bokeh serve）中的curdoc().add_periodic_callback()周期回调，实现数据流动画、仿真推演与动态仪表盘。读完本文，你将掌握count、repeat、bounce、linear、sine、cosine六种内置驱动器的数学含义与用法，理解底层force装饰器与_advance生成器的工作机制，并能独立写出可运行的实时动画 Bokeh 应用。

本文以 API 参考文档 为骨架，其内容由 Sphinx 的automodule:: bokeh.driving指令从模块源码自动生成，因此模块 docstring 即文档正文；全文结合 driving.py 源码、单元测试 与 examples/server/app 下的真实示例进行纵深扩充。

一、bokeh.driving的设计动机与适用场景

bokeh.driving模块 docstring 明确写道：它提供"一组装饰器（decorators），用于在函数每次被调用时以指定方式反复更新该函数的一个参数"，并强调"这些装饰器在与 Bokeh 服务器应用的周期回调（periodic callbacks）结合时尤其有用"（见 driving.py 模块 docstring）。

在实际动画开发中，最常见的模式是：每过一段固定时间（如 50ms、100ms）调用一次更新函数，把时间步长t传入并刷新图形数据。若手动管理，通常需要闭包计数器或全局变量：

counter = 0 def update(): global counter t = counter counter += 1 # ... 用 t 更新图形

bokeh.driving将这一模式封装为装饰器——被装饰的函数签名自动从update(t)变为无参的update()，每次调用时t由内部生成器自动推进。这使得回调函数职责单一、代码整洁，且取值策略（递增、循环、往返、正弦等）可以随时通过更换装饰器来切换，无需改动函数体。

从源码结构看，该模块暴露的公开 API 稳定为 7 个函数（见 driving.py 的__all__）：

除force外的六个工厂函数返回的都是一个经由functools.partial部分应用后的装饰器；force本身则是实现其余全部驱动器的最底层装饰器。单元测试 tests/unit/bokeh/test_driving.py 中的ALL元组与verify_all(bd, ALL)调用也印证了这 7 个名称构成模块的完整公开接口。

二、底层机制：force装饰器与_advance生成器

理解bokeh.driving只需抓住两个核心构件：无状态的生成器_advance，以及把生成器"喂"给目标函数的装饰器force。

2.1_advance：无界的整数推进器

_advance是一个私有生成器函数（见 driving.py 中的_advance），其实现只有几行：

def _advanceT -> Iterator[T]: i = 0 while True: yield f(i) i += 1

它接收一个"值函数"f（把整数步进i映射为目标值），然后从i = 0开始无限地yield f(i)并递增。所有驱动器工厂的取值公式都集中在传给_advance的这个小函数里，例如：

• count()直接传入恒等函数lambda x: x，故产生0, 1, 2, ...（driving.py）；
• linear(m, b)传入lambda i: m*i + b（driving.py）；
• sine/cosine分别传入A*sin(w*i + phi) + offset与A*cos(w*i + phi) + offset（[driving.py](https://link.gitcode.com/i/7b439739f712a1b24f764fb7bfcc773d#L97-L114、L172-L189）。

单元测试 test__advance 验证了其行为：连续四次next(s)依次产出0, 1, 2, 3，确认这是一个无界、无状态的递增流。

2.2force：把生成器推进与函数调用绑定

force是模块内唯一的"真正"装饰器（driving.py 中的force）：

def force(f: Callable[[Any], None], sequence: Iterator[Any]) -> Callable[[], None]: def wrapper() -> None: f(next(sequence)) return wrapper

它接受一个目标函数f和一个可迭代对象sequence，返回一个无参的wrapper。每次调用wrapper()时，内部执行next(sequence)取出序列的下一个值，并将其作为唯一参数传给f。这正是"每次调用推进一个参数"这一核心语义的落点。

force的测试用例直接展示了它可接受任意可迭代对象——包括字符串生成器（test_force）：

seq = (x for x in ["foo", "bar", "baz"]) w = bd.force(_collector(results), seq) w() # results == ["foo"] w() # results == ["foo", "bar"] w() # results == ["foo", "bar", "baz"]

2.3 工厂函数如何组装两者

以linear为例（driving.py）：

def linear(m: float = 1, b: float = 0) -> partial[Callable[[], None]]: def f(i: float) -> float: return m * i + b return partial(force, sequence=_advance(f))

它先用闭包捕获斜率m与截距b构造值函数f，再由_advance(f)生成无限序列，最后通过functools.partial(force, sequence=...)把sequence预绑定到force上。于是linear(m=2.5, b=3.7)本身就是一个装饰器，直接装饰更新函数即可。全部六个工厂（bounce、cosine、count、linear、repeat、sine）都遵循这一"闭包值函数 +_advance+partial(force, ...)"的统一组装模式。

三、内置驱动器逐一详解

以下每个驱动器的数学定义与参数语义均直接继承自 driving.py 中各函数的 docstring，并由单元测试给出数值验证。

3.1count()：最简单的整数推进

from bokeh.driving import count @count() def update(t): print(t)

不接收任何参数，每次调用产生一个递增整数：0, 1, 2, 3, ...。由于它只做恒等映射，适合直接用作"时间步"或"帧序号"。单元测试 test_count 断言连续 8 次调用的结果为[0, 1, 2, 3, 4, 5, 6, 7]。

3.2repeat(sequence)：循环重复序列

from bokeh.driving import repeat seq = [0, 1, 2, 3] # repeat(seq) => [0, 1, 2, 3, 0, 1, 2, 3, 0, 1, ...]

docstring 中明确给出了上述输出序列（driving.py）。实现上通过sequence[i % N]取模索引实现循环，其中N = len(sequence)。单元测试 test_repeat 用序列[0, 1, 5, -1]验证了 8 次调用严格周期复现[0, 1, 5, -1, 0, 1, 5, -1]。

典型场景：周期性参数化动画。官方示例 fourier_animated.py 用@repeat(range(N))装饰更新函数，令傅里叶级数谐波相位在 0~N 之间循环，每 100ms 滚动一次数据，实现流式傅里叶动画：

@repeat(range(N)) def update(ind): ... items_source.data.update(update_term_data(ind)) curdoc().add_periodic_callback(update, 100)

3.3bounce(sequence)：往返反弹序列

from bokeh.driving import bounce @bounce([0, 1, 2]) def update(i): print(i)

模块 docstring 中的完整示例（driving.py）指出：反复调用该函数会在标准输出打印0 1 2 2 1 0 0 1 2 2 1 ...。docstring 还给出了更长的示意：

seq = [0, 1, 2, 3] # bounce(seq) => [0, 1, 2, 3, 3, 2, 1, 0, 0, 1, 2, ...]

实现要点（driving.py 中的bounce）：设N = len(sequence)，对步进i计算div, mod = divmod(i, N)；当div为偶数时正向取sequence[mod]，为奇数时反向取sequence[N-mod-1]——即在序列两端各驻留一拍后折返。单元测试 test_bounce 用[0, 1, 5, -1]验证 8 次调用输出[0, 1, 5, -1, -1, 5, 1, 0]。

典型场景：需要"来回扫动"的动画，如探针扫描、往返运动的指示器。

3.4linear(m, b)：线性推进

from bokeh.driving import linear @linear(m=1, b=0) def update(x): print(x)

docstring 给出公式value = m * i + b（driving.py），参数含义为：

单元测试 test_linear 以m=2.5, b=3.7验证输出为[3.7, 6.2, 8.7, 11.2]，与公式逐项吻合。适合匀速递增（如按固定步长增长的数值字段）。

3.5sine(w, A, phi, offset)与cosine(w, A, phi, offset)：三角函数驱动

两个驱动器公式分别为value = A * sin(w*i + phi) + offset与value = A * cos(w*i + phi) + offset（[driving.py](https://link.gitcode.com/i/7b439739f712a1b24f764fb7bfcc773d#L97-L114、L172-L189），参数语义完全一致：

单元测试 test_sine 与 test_cosine 使用同一组参数w=0.3, A=3, phi=0.1, offset=2并借助numpy.testing.assert_allclose验证了数值序列——例如cosine前四值为[4.985012495834077, 4.763182982008655, 4.294526561853465, 3.6209069176044197]。这类驱动器非常适合平滑波动、呼吸灯效果、按三角函数变化的物理模拟量。

四、实战：把驱动器接入 Bokeh 服务器周期回调

bokeh.driving的"标准用法"是让驱动器装饰的更新函数成为curdoc().add_periodic_callback(callback, period_ms)的回调，从而驱动ColumnDataSource的数据刷新。仓库 examples/server/app 下提供了多个可直接bokeh serve运行的真实案例。

4.1 官方示例一：count+ 等值线动画

contour_animated.py 用@count()提供时间步，每 40ms 重算一次等值线数据并调用set_data刷新渲染器：

from bokeh.driving import count from bokeh.plotting import curdoc, figure @count() def callback(timestep): z = get_z(timestep) # 时间步决定波动相位 new_contour_data = contour_data(x, y, z, levels) contour_renderer.set_data(new_contour_data) curdoc().add_periodic_callback(callback, 40) curdoc().add_root(fig)

启动方式（见文件头部注释）：在examples/server/app目录下执行bokeh serve contour_animated.py，然后访问http://localhost:5006/contour_animated。

4.2 官方示例二：count+ 流式金融图

ohlc/main.py 中，@count()装饰的update(t)每次用步进t生成一组新的开盘/最高/最低/收盘价与 MACD 指标，并通过source.stream(new_data, 300)以滑动窗口方式追加到ColumnDataSource；同时t还被用作序列化窗口长度计算的基准。回调周期为 50ms：

@count() def update(t): open, high, low, close, average = _create_prices(t) ... source.stream(new_data, 300) curdoc().add_periodic_callback(update, 50)

这展示了"驱动器推进的参数直接参与每帧数据计算"的典型模式——count()不仅提供帧序号，还参与均线、MACD 等指标的时间对齐。

4.3 官方示例三：repeat+ 流式傅里叶动画

fourier_animated.py 中，@repeat(range(N))让相位索引在0..N-1内循环，配合numpy.roll滚动各谐波数据，实现方波分解的连续动画，回调周期 100ms。

4.4 官方示例四：count+ 自定义扩展 3D 曲面

surface3d/main.py 中，@count()的update(t)把时间步代入compute(t)重算整个 3D 网格的z值并整体替换source.data，每 100ms 刷新一次——是"整体重算型"动画的简洁示范。

五、从源码看设计细节与使用约束

1. partial而非直接闭包：六个工厂函数均返回partial(force, sequence=_advance(f))（[driving.py](https://link.gitcode.com/i/7b439739f712a1b24f764fb7bfcc773d#L95、L114、L120、L152、L170、L189）。这样的好处是force的逻辑只实现一份，参数预绑定由标准库functools.partial完成，装饰器本身可被反复复用。

2. 序列长度必须合法：bounce与repeat都依赖len(sequence)与取模运算（[driving.py](https://link.gitcode.com/i/7b439739f712a1b24f764fb7bfcc773d#L88、L167），传入空序列会产生除零/索引错误；传入的应是支持len()与下标访问的Sequence。docstring 的类型注解也标明二者参数为Sequence[int]。

3. force接受任意迭代器：与其他工厂不同，force对第二个参数没有任何len要求（见 test_force 使用生成器表达式），因此它是最灵活的低层接口，可用于驱动"自定义随机序列、读取文件行、事件流"等任意取值逻辑；其局限是序列耗尽后wrapper将抛出StopIteration，需要使用者自行保证无限或足够长。

4. 模块导出稳定：模块通过__all__显式声明 7 个公开名称（driving.py），测试中verify_all(bd, ALL)（test_driving.py）用于防止 API 意外增删，保证文档与实现的同步。

六、快速自测：用测试文件验证你的理解

仓库自带的单元测试 tests/unit/bokeh/test_driving.py 覆盖了模块全部公开 API 与私有_advance。其核心断言可当作"行为契约"来校验上文的数值推导：

• count()8 次调用 →[0..7]；
• repeat([0, 1, 5, -1])8 次调用 →[0, 1, 5, -1, 0, 1, 5, -1]；
• bounce([0, 1, 5, -1])8 次调用 →[0, 1, 5, -1, -1, 5, 1, 0]；
• linear(m=2.5, b=3.7)4 次调用 →[3.7, 6.2, 8.7, 11.2]；
• cosine(w=0.3, A=3, phi=0.1, offset=2)前 4 值 →[4.985012495834077, 4.763182982008655, 4.294526561853465, 3.6209069176044197]；
• sine(w=0.3, A=3, phi=0.1, offset=2)前 4 值 →[2.2995002499404844, 3.1682550269259515, 3.932653061713073, 4.524412954423689]。

你可以直接运行测试复现这些结果，或在交互环境中用同样方法收集调用输出，验证任一驱动器的取值规律。

结语：何时选用哪种驱动器

bokeh.driving以极小的 API 表面积（7 个函数、约 60 行核心实现）覆盖了绝大多数动画驱动的取值模式：把"取值策略"从"业务逻辑"中剥离，交由装饰器表达。配合curdoc().add_periodic_callback，即可在 Bokeh 服务器应用中轻松实现等值线动画、流式金融图、傅里叶级数演示与 3D 曲面刷新——这些能力均可在本仓库 examples/server/app 目录中实际运行验证。

【免费下载链接】bokehInteractive Data Visualization in the browser, from Python项目地址: https://gitcode.com/GitHub_Trending/bo/bokeh