TAIPY长时回调：不传递state返回多DataFrame并更新图表方案

背景 #

在实现长时回调时，我编写了以下代码：

def start_analytics(project_id,instance_id):
    missing_signals = []
    signal_dict = get_all_timeseriesdata(project_id,instance_id)
    print("got dict")
    print(signal_dict)
    
    # 检查字典中是否存在'X'、'Y'和'Z'键
    if 'X' not in signal_dict:
        missing_signals.append('X')

    if 'Y' not in signal_dict:
        missing_signals.append('Y')

    if 'Z' not in signal_dict:
        missing_signals.append('Z')

    if missing_signals:
        print("signals missing")
        #notify(state, "info", f"The following signals are missing: {', '.join(missing_signals)}", True)
    else:
        print("checking inxed of x,y,z")
        x_id = signal_dict.get('X')
        y_id = signal_dict.get('Y')
        z_id = signal_dict.get('Z')

        print(project_id,instance_id,x_id)
        x_df = read_timeseriesdata(project_id,instance_id,x_id)
        y_df = read_timeseriesdata(project_id,instance_id,y_id)
        z_df = read_timeseriesdata(project_id,instance_id,z_id)
        print("got all df")

        # 沿列轴合并DataFrame
        merged_df = pd.concat([x_df, y_df, z_df], axis=1)
        selected_df = merged_df[['X', 'Y', 'Z']]
        print(selected_df.shape[0])
        print(selected_df.head())
        total_points_rsi = selected_df.shape[0]
        print("total rsi points",total_points_rsi)
        
        #state.total_points_rsi = total_points_rsi

        # 使用Plotly生成交互式3D图表
        fig = go.Figure(data=[go.Scatter3d(
            x=selected_df['X'],  # X轴数值
            y=selected_df['Y'],  # Y轴数值
            z=selected_df['Z'],  # Z轴数值
            mode='lines',
            line=dict(width=1, color='red'),
            name='RSI Data',
            hovertext=[f"Index of RSI point: {index}" for index in merged_df.index]
        )])

        files = get_all_files(project_id,instance_id)
        #print(files)
        src_df = read_src_data(project_id,instance_id,files)
        #print(src_df)

        print("got src file")

        # 对SRC路径应用特定线性插值
        df_src_interpolated_specific = linear_interpolate_path_specific_points(src_df, total_points_rsi)

        # 添加插值后的SRC数据作为标记到图表中
        fig.add_trace(go.Scatter3d(
            x=df_src_interpolated_specific['X'],
            y=df_src_interpolated_specific['Y'],
            z=df_src_interpolated_specific['Z'],
            mode='markers',  # 以单个标记形式绘制
            marker=dict(size=2, color='green'),  # 调整大小和颜色
            name='Interpolated SRC Path',
            hovertext=[f'Index: {i}' for i in df_src_interpolated_specific.index]  # 自定义悬浮文本包含索引
        ))
        return True

def heavy_function_status(state, status):
    if status:
        notify(state, "success", "The heavy function has finished!")
    else:
        notify(state, "error", "The heavy function has failed")

def plot_path(state):
        notify(state, "info", "Gathering resourses started",duration=3000)
        project_id = state.form_visualize.v_selected_projects.id
        instance_id = state.form_visualize.v_selected_instance.id
        invoke_long_callback(state, start_analytics,[project_id,instance_id],heavy_function_status,None,period=0)
        notify(state, "info", "REST STARTED",duration=3000)

问题 #

我了解到不可将state作为参数传入耗时函数，但我需要在流程中更新图表。请问：

1. 如何在不传递state的前提下实现图表更新？
2. 若需从耗时函数返回DataFrame等资源，再调用其他函数处理，该如何编排执行顺序？
3. 我需要从start_analytics返回3个DataFrame，用于在heavy_function_status中更新图表，但尝试返回时出现“bool object”错误，该如何解决？

解答 #

1. 不传递state实现图表更新 #

可以通过回调函数的返回值传递数据，结合后续状态处理函数操作state来更新图表，也可借助全局存储（内存缓存、临时文件或数据库）：

• 把start_analytics生成的图表数据（比如序列化后的Plotly fig对象）存入全局缓存，用project_id+instance_id作为唯一标识；
• 在heavy_function_status中通过缓存键取出数据，再用state更新对应的图表组件（例如state.components.your_plot.figure = fig）。

你当前使用的invoke_long_callback已经支持传入后续处理函数heavy_function_status，只需让start_analytics返回图表所需数据，即可在heavy_function_status里结合state完成更新。

2. 编排耗时函数与后续处理的执行顺序 #

利用长时回调的回调链机制来串联流程：

1. 让start_analytics返回需要处理的DataFrame或缓存键；
2. 在heavy_function_status内部调用后续处理函数，把返回值作为输入传递；
3. 若需多步处理，拆分每一步为独立函数，前一步的返回值作为后一步的参数，通过回调机制依次执行。

示例改造思路：

def process_dataframes(state, dfs):
    # 这里用state处理DataFrame并更新图表
    selected_df, src_df, interpolated_df = dfs
    # 图表更新逻辑...
    notify(state, "success", "数据处理完成，图表已更新")

def heavy_function_status(state, result):
    if isinstance(result, tuple):
        # 拿到DataFrame元组，调用处理函数
        process_dataframes(state, result)
    elif result is False:
        notify(state, "error", "耗时函数执行失败")
    else:
        notify(state, "success", "耗时函数执行完成")

3. 解决返回DataFrame时的"bool object"错误 #

当前start_analytics默认返回布尔值True，当你改为返回DataFrame时，后续heavy_function_status仍把它当作布尔值处理，就会触发类型错误。解决步骤如下：

• 修改start_analytics的返回值：返回包含3个DataFrame的元组，异常场景返回False

  # 替换原来的return True
  return selected_df, src_df, df_src_interpolated_specific
  # 异常分支返回False
  if missing_signals:
      print("signals missing")
      return False
  

• 调整heavy_function_status的逻辑：根据返回值类型做不同处理

  def heavy_function_status(state, result):
      if isinstance(result, tuple) and len(result) == 3:
          # 成功获取三个DataFrame，生成并更新图表
          selected_df, src_df, interpolated_df = result
          fig = go.Figure(data=[go.Scatter3d(
              x=selected_df['X'],
              y=selected_df['Y'],
              z=selected_df['Z'],
              mode='lines',
              line=dict(width=1, color='red'),
              name='RSI Data'
          )])
          fig.add_trace(go.Scatter3d(
              x=interpolated_df['X'],
              y=interpolated_df['Y'],
              z=interpolated_df['Z'],
              mode='markers',
              marker=dict(size=2, color='green'),
              name='Interpolated SRC Path'
          ))
          # 更新到对应的图表组件
          state.form_visualize.v_3d_plot.figure = fig
          notify(state, "success", "图表更新完成！")
      elif result is False:
          notify(state, "error", "缺少必要的信号数据")
      else:
          notify(state, "error", "耗时函数执行异常")
  

内容的提问来源于stack exchange，提问作者Aravind R K