Orange3插件开发：如何处理部件输出的连接/断开事件？

在Orange3开发附加部件时，要实现“仅当输出与其他部件连接时才执行大量计算”的需求，核心是监听输出端口的连接订阅/取消订阅事件，然后在合适的时机触发计算。下面是具体的实现方案和代码示例：

核心思路 #

Orange3的OWWidget基类提供了专门的钩子方法，用于捕获输出信号的连接状态变化：

• on_signal_subscribed(signal_name)：当某个输出端口被其他部件连接时触发
• on_signal_unsubscribed(signal_name)：当某个输出端口与其他部件断开连接时触发

我们可以利用这两个方法，结合结果缓存，实现延迟计算的逻辑：

1. 初始状态下不执行计算，仅缓存输入数据（如果有）
2. 当输出被连接时，检查是否已有缓存结果，无则执行计算并缓存，然后发送结果
3. 当输入数据更新时，清空缓存；若此时输出已连接，则重新计算并发送
4. 输出断开时可选择清空缓存以释放内存

代码示例 #

from Orange.widgets.widget import OWWidget, Input, Output
from Orange.data import Table
from Orange.widgets.utils.widgetpreview import WidgetPreview

class OWHeavyComputation(OWWidget):
    name = "Heavy Computation"
    description = "延迟大量计算直至输出被连接"
    icon = "icons/my_widget.svg"  # 替换成你的部件图标
    priority = 10  # 部件在工具栏中的优先级

    # 定义输入输出端口
    inputs = [("Input Data", Table, "on_input_data")]
    outputs = [("Computed Result", Table)]

    def __init__(self):
        super().__init__()
        self._input_data = None
        self._cached_result = None
        self._is_output_connected = False

        # 可选：添加进度条（计算时显示）
        self.progressBar = self.progressBar()

    # 捕获输入数据变化
    def on_input_data(self, data):
        self._input_data = data
        # 输入更新，清空缓存结果
        self._cached_result = None
        # 如果输出已连接，立即重新计算并发送
        if self._is_output_connected and data is not None:
            self._run_heavy_computation()
            self.send("Computed Result", self._cached_result)

    # 输出被连接时触发
    def on_signal_subscribed(self, signal_name):
        if signal_name == "Computed Result" and self._input_data is not None:
            self._is_output_connected = True
            # 如果还没计算过，执行计算
            if self._cached_result is None:
                self._run_heavy_computation()
            # 发送缓存的结果
            self.send("Computed Result", self._cached_result)

    # 输出被断开时触发
    def on_signal_unsubscribed(self, signal_name):
        if signal_name == "Computed Result":
            self._is_output_connected = False
            # 可选：清空缓存以释放内存（根据你的计算量决定）
            # self._cached_result = None

    # 封装大量计算逻辑
    def _run_heavy_computation(self):
        self.progressBar.set(0)
        # --------------------------
        # 这里替换成你的实际计算逻辑
        # 比如复杂的数据预处理、模型训练等
        self.progressBar.set(50)
        # 示例：基于输入数据生成结果（这里用iris数据集模拟）
        self._cached_result = self._input_data.copy()
        # --------------------------
        self.progressBar.set(100)
        self.progressBarFinished()

# 用于预览部件（开发时用）
if __name__ == "__main__":
    WidgetPreview(OWHeavyComputation).run()

关键细节说明 #

• 信号名称匹配：在on_signal_subscribed和on_signal_unsubscribed中，一定要通过signal_name判断目标输出端口，避免处理其他输出的事件。
• 缓存策略：缓存计算结果可以避免重复计算，尤其是当输出断开后重新连接时，无需再次执行耗时操作；但如果计算结果占用内存过大，可在输出断开时清空缓存。
• 输入联动：当输入数据更新时，必须清空缓存并在输出已连接的情况下重新计算，保证结果的时效性。
• 用户体验：添加进度条可以让用户感知计算过程，避免误以为部件无响应。

内容的提问来源于stack exchange，提问作者Dmitriy Kalugin-Balashov