如何在Taipy中设置并自动触发场景触发器？

一、设置触发操作 #

Taipy 中触发操作可通过 UI 交互绑定、后台事件监听两种方式实现：

UI 交互触发 #

直接在 UI 组件中通过 on_action 参数绑定回调函数，响应用户操作（如按钮点击）：

from taipy.gui import Gui, notify

def execute_trigger(state):
    notify(state, "success", "触发操作执行完毕")
    # 在此添加业务逻辑：更新数据、调用计算任务等

page = """
<|执行触发|button|on_action=execute_trigger|>
"""

Gui(page).run()

也可通过 taipy.gui.trigger 手动触发回调，比如在输入框内容满足条件时触发：

def on_input_update(state, var_name, var_value):
    if var_name == "trigger_keyword" and var_value == "run":
        trigger(state, execute_trigger)

后台事件触发 #

通过监听数据节点变化触发任务/场景，适用于非 UI 驱动的业务逻辑：

from taipy import Core, Config, Scenario

# 定义数据节点与任务
data_node = Config.configure_data_node(id="input_data")
task = Config.configure_task(id="process_task", function=lambda x: x*2, input=data_node, output=data_node)
scenario_config = Config.configure_scenario(id="auto_scenario", task_configs=[task])

Core().run()
scenario = Scenario(scenario_config)

# 绑定数据节点变化回调：数据更新时自动提交场景
def on_data_update(scenario, data_node):
    scenario.submit()

scenario.data_nodes["input_data"].on_change(on_data_update)

# 更新数据节点，触发回调
scenario.data_nodes["input_data"].write(10)

二、自动触发场景 #

自动触发场景支持三种常见模式：

定时触发 #

社区版可结合 schedule 库实现定时任务，企业版支持内置调度器：

import schedule
import time
from taipy import Scenario

def auto_run_scenario():
    scenario = Scenario(scenario_config)
    scenario.submit()
    print("场景已自动执行")

# 配置每日10点触发
schedule.every().day.at("10:00").do(auto_run_scenario)

# 维持调度运行
while True:
    schedule.run_pending()
    time.sleep(60)

Taipy Enterprise 可直接通过配置实现调度：

from taipy import Config

Config.configure_scheduler(
    scenario_config_id="auto_scenario",
    cron_expression="0 10 * * *"  # Cron表达式：每日10点执行
)

数据变化触发 #

利用数据节点的 on_change 回调，实现数据更新即触发场景：

scenario.data_nodes["input_data"].on_change(lambda s, dn: s.submit())

外部事件触发 #

监听外部事件（如文件变化、消息队列）触发场景，以文件监听为例：

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import pandas as pd
import time

class FileUpdateHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if not event.is_directory and event.src_path.endswith("data_source.csv"):
            # 更新数据节点
            new_data = pd.read_csv("data_source.csv")
            scenario.data_nodes["input_data"].write(new_data)
            # 触发场景
            scenario.submit()

observer = Observer()
observer.schedule(FileUpdateHandler(), path="./", recursive=False)
observer.start()

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    observer.stop()
observer.join()

三、场景触发器自动化完全可行 #

Taipy 支持完整的场景触发器自动化：

• 社区版结合第三方库（schedule、watchdog 等）可覆盖定时、外部事件、数据变化等触发需求；
• 企业版提供内置调度器、事件总线，支持复杂触发规则（如依赖场景完成后触发、多节点联动），还支持可视化配置触发器。

例如企业版中配置场景依赖触发：当场景 A 执行完成后自动触发场景 B：

from taipy import Config

scenario_a_config = Config.configure_scenario(id="scenario_a")
scenario_b_config = Config.configure_scenario(id="scenario_b")

# 配置触发规则：场景A完成后启动场景B
Config.configure_trigger(
    id="a_trigger_b",
    source_scenario_id="scenario_a",
    target_scenario_id="scenario_b",
    condition=lambda s: s.status == "COMPLETED"
)

内容的提问来源于stack exchange，提问作者Rym Guerbi Michaut