工作流构建 (Pipeline Build)

本章汇总编写 Pipeline 创建脚本时 所有 GDIM 应用类型都要走的步骤：确认需求、选择包、核对表名字段、查找模块、实例化与连线、暴露属性、设置返回结果、保存和本地测试。

不同的 GDIM 应用类型（自定义应用、自定义数据更新配置、地图兴趣点弹窗、数据对接、自动报告）在这些步骤之上还有各自的规则，例如必须返回什么端口、哪些模块不能出现在 .pipe 里。先按 GDIM应用类型 (App Types) 判断应用类型，再读对应的专题页；本章只写共用部分。

开始前确认需求

写代码前先确认下列信息。缺哪一项就先问清楚，不要猜测或编造。

需要的信息	用途	缺失时
应用类型	决定返回端口、必需 / 禁用模块，见 GDIM应用类型 (App Types)	问：「这个 Pipeline 要上传成 GDIM 的哪类应用？」
数据来源（`proj_id`、`proj_name` 或 `tpl_id`）	所有 GDIM 读写模块都需要	问：「请提供项目 ID，或项目名称及所在空间」
要读取的表 / 字段	`GdimTableReader.table_fields` 等参数	用下文「核对 GDIM 表名与字段」查询
输出形式	决定用哪些模块（图表、表格、文件、报告、写库……）	问：「希望得到什么结果？」
关键列名（图表轴、过滤条件、关联键）	如 `x_column`、`query_template`、`join_key`	用下文「核对 GDIM 表名与字段」查询

选择 `gdi` 或 `gdisdk`

示例统一写 gdisdk。确定导入包的顺序：

用户明确指定 gdi 或 gdisdk 时，照做。
未指定时，看项目根目录：有 gdi/ 目录用 gdi，否则用 gdisdk。
仍不确定时，可以用 try: from gdi... except ImportError: from gdisdk... 兼容写法。两种包保存出的 .pipe 完全一致。

查看 gdisdk 的类签名时，先读 .py 源码；若只有编译产物，读同目录的 .pyi 存根。不要凭记忆猜参数名。

核对 GDIM 表名与字段

table_fields、列名、关联键等必须来自真实的数字模型结构，或由用户明确给出。

在浏览器里查看：打开 GDIM 开发模式，见 GDIM开发模式 (Dev Mode)。

在命令行查询（proj_id 已知时，从项目根目录运行）。python 用当前 skill 的 Running Python 选定的解释器。

python <pkg>/agents/tools/gdimTplQueryTools.py --proj-id <id> --list-tables
python <pkg>/agents/tools/gdimTplQueryTools.py --proj-id <id> --fields <table_name>
python <pkg>/agents/tools/gdimTplQueryTools.py --proj-id <id> --search <关键词>
python <pkg>/agents/tools/gdimTplQueryTools.py --proj-id <id> --validate <table_name> --check <field1> <field2>

只有模板 ID 时把 --proj-id 换成 --tpl-id；--force-refresh 跳过 10 分钟缓存。

读结果时注意：

--list-tables 输出 name | title | sub_tables，代码里一律用 name。
--validate 对每个字段给出 [OK] / [FAIL] 和候选项：得分 ≥ 0.9 可直接改正；0.6–0.9 先向用户确认；< 0.6 让用户选择。

查找模块

通读模块总览（AI 助手读 <pkg>/agents/knowledge/modulesOverview.md；人工查阅见模块帮助 (Module Help)），挑出需要的模块。
对每个要用的模块查看完整帮助：端口、参数、示例。AI 助手可用 search_module_help("ModuleName")，或直接读 <pkg>/agents/knowledge/modulesHelp/ 下对应分类的文件（分类以该目录中实际存在的文件为准）。
需要对一组条目重复同一段子流程（多个项目、多个文件、上游给出的 list[dict]）时，用 ForEachController，不要在 Pipeline 外写 Python 循环。设计前先读流程控制 (Flow Control)。

没有合适模块时

先停下，不写 Pipeline 代码。
说明缺的是哪项能力、为什么现有模块覆盖不了。
建议用 PythonCoder 承载这段自定义逻辑，等用户同意 后再写。
同意后在脚本里配置 PythonCoder``（``local_function_name、动态端口），本地函数按 modules.widgets 模块帮助的 PythonCoder 约定编写。

不要用不对口的模块凑合，也不要编造不存在的模块或端口。

步骤 1：创建 PipeLine 与 workspace

workspace 由脚本位置推导，指向该 Pipeline 自己文件夹下的 workspace/，不要写本机绝对路径。

from pathlib import Path

from gdisdk.connectors import log_in
from gdisdk.pipeline import PipeLine

_SCRIPT_DIR = Path(__file__).resolve().parent
WORKSPACE_DIR = _SCRIPT_DIR / "workspace"

pipeline = PipeLine(
    app_name="settlementLine",      # 内部标识，也用作 .pipe 文件名
    app_title="沉降曲线",            # 界面显示标题
    description="基于沉降数据绘制沉降曲线",
    workspace=WORKSPACE_DIR,
)

步骤 2：配置 GDIM 状态

只要有模块读写 GDIM 就需要；纯本地文件处理可以跳过。

pipeline.update_gdim_state(token=log_in(), proj_id="你的GDIM项目ID")

只有项目名称时可以不写 proj_id。run() 会在模块执行前把名称解析成 id：

pipeline.update_gdim_state(token=log_in(), proj_name="项目名称")

proj_id 优先于 proj_name。空间怎么选、模块名称如何覆盖 Pipeline 项目，见用项目名称定位。

update_gdim_state() 可重复调用。不要把用户名、密码写进脚本；log_in() 会读取本机已保存的凭证。上传后 token、proj_id、workspace、local_functions_path 都由平台接管。模块自己的 proj_name 仍会覆盖平台写入的项目 id。

步骤 3：实例化与配置模块

构造时只传模块名：GdimTableReader("ReadTable")``（``mname 关键字可省略）。
参数在构造之后赋值：read_table.table_fields = [...]。
输入端口用 mod.InputX = data 赋值，不要用构造参数代替端口，也不要给模块单独传 token / proj_id。需要读另一个项目时，再给该模块设置 proj_name 或 proj_id。
Pipeline 脚本里不要调用 mod.execute()，由 pipeline.run() 调度。
模块名用 PascalCase，同一类用两次时起不同的名字（ReadTable1、ReadTable2）。

单独测试某个模块时：先赋值全部参数和端口，再调用一次 execute()，从 mod.OutputX.data 取结果。模块默认 auto_run=True，构造时参数齐全就会自动执行一次，再调 execute() 会重复执行。

步骤 4：连接端口

links = (
    read_table.OutputTable >> query_table.InputTables
    | query_table.OutputTables >> plot_line.InputData
    | query_table.OutputTables >> table_plot.InputData   # 一个输出可以接多个输入
)
pipeline.add_links(links)

每个输入端口 只能接一条 连线。给同一个输入端口再接一条会静默覆盖前一条，不报错。需要汇总多路数据时，用支持动态输入端口的模块（如 MergeTables、MergeSingleResult、MergeDocData）。
端口名单复数要和模块帮助完全一致：OutputTable 和 OutputTables 是两个端口，写错会静默生成一个新的动态端口。
没有接入连线图的模块不会执行。

步骤 5：暴露 Pipeline 属性

GDIM 只能通过 Pipeline 属性给模块参数赋值。只在用户明确要求、或设计上确实需要用户调整时才暴露，不要把每个参数都暴露出去。

pipeline.add_attribute(
    attr_name="tpl_number",     # Pipeline 层唯一名称
    module_name="QueryTable",   # 模块 mname
    param_name="tpl_number",    # 模块参数名
    attr_title="沉降点编号",     # 界面标签
)

调整默认值、范围、下拉选项或隐藏控件：ui_schema_overrides，见 UI控件 (UI Schema)。
PythonCoder 的自定义属性要暴露给前端时，必须在模块上设置 ui_schema_function_name，否则 add_attribute() 会失败，见 modules.widgets 模块帮助。
auto_bind 只用于两种特定应用：地图兴趣点弹窗（poi_field，见 GDIM地图弹窗 (Map POI Popup)）和子表数据更新（pk_field，见 GDIM数据更新 (Data Update)）。其它应用不要用。

步骤 6：设置返回结果

return_results_config 决定 GDIM 前端拿到哪些端口的数据。按应用类型选择，见 GDIM应用类型 (App Types)。

pipeline.return_results_config = {"PlotLine": "OutputPlotData"}          # 指定端口（推荐）
pipeline.return_results_config = {"Mod": ["OutputA", "OutputB"]}         # 同一模块多个端口
pipeline.return_results_config = True                                    # 所有末端输出（仅在不确定时）
pipeline.return_results_config = False                                   # 不返回

正式上传时优先写明确的 dict，避免把大量中间表发给前端。

initial_run_to 只在用户要求、或存在 auto_select 下拉需要预先取数时才设置。打开应用时运行到这些模块之前，列出的模块及其下游不执行。用户之后点击运行时由 rerun_initial_upstream 决定上游是否重新执行：

``True``（默认）：从起点重新执行全部模块，上游会读取最新数据。
False：initial_run_to 的上游不再执行，沿用打开时已经得到的输出；其余模块照常执行，包括不在该上游路径上的模块。

pipeline.initial_run_to = ["QueryTable"]   # 打开应用时先运行到 QueryTable
pipeline.rerun_initial_upstream = False    # 仅当用户运行时不需要重读上游数据

步骤 7：保存 `.pipe`

把构建过程放进 build_pipeline()，在函数末尾保存到 workspace；本地测试放在 if __name__ == "__main__": 里。这样上传的 .pipe 不会混进只用于本地测试的改动。

def build_pipeline() -> PipeLine:
    pipeline = PipeLine(app_name="settlementLine", app_title="沉降曲线", workspace=WORKSPACE_DIR)
    # ... 步骤 2–6 ...
    pipeline.save_pipeline(f"{pipeline.workspace}/{pipeline.app_name}.pipe")
    return pipeline

步骤 8：本地测试（用户要求时）

先按当前 skill 的 Running Python 选定解释器，再用它运行脚本。用户没有指定 conda 环境、且项目根目录有 gdisdk/ 时，只先试 gdisdk-env``（``gdisdk-env-setup 创建的环境），并且只有该环境能导入项目里的 gdisdk 时才使用。否则向用户询问要用哪个 conda 环境。

if __name__ == "__main__":
    pipeline = build_pipeline()
    pipeline.set_attributes(tpl_number=["点1", "点2"])   # 与 add_attribute 的 attr_name 一致
    result = pipeline.run(
        verbose=True,
        run_trace_file="run_trace.json",
        return_results=pipeline.return_results_config,
    )
    print(result.get_result("PlotLine", "OutputPlotData"))

run() 默认只收集末端模块端口，而不是 return_results_config。要让 get_result(...) 和 GDIM 看到的一致，传 return_results=pipeline.return_results_config。
module_result.port_data 就是原始数据，没有 .data 属性。
run_trace_file="run_trace.json" 会同时生成 run_trace.html，里面有工作流拓扑图和各模块的执行状态，查看拓扑或排查连线时直接打开它即可。详见运行调试 (Debug)。

加载已保存的 `.pipe`

要检查或复跑一个已保存的 Pipeline，用 restore_pipeline() 读回，参数可以是本地路径或 URL：

pipeline = PipeLine()
pipeline.restore_pipeline("workspace/SettlementCurve.pipe")
# pipeline.restore_pipeline("https://host/path/SettlementCurve.pipe")
pipeline.update_gdim_state(token=log_in(), proj_id="你的GDIM项目ID")

读回后先调用 update_gdim_state() 设置当前的 token 和项目，再运行需要访问 GDIM 的模块。
reset_to_auto_run 默认为 True，所有模块都会恢复成 auto_run=True，和 GDIM 加载 .pipe 时的行为一致。要保留文件中保存的 auto_run 状态，传 reset_to_auto_run=False。
本地函数默认从 .pipe 里记录的 local_functions_path 加载；路径变了就传 local_functions_path=...。

完整示例

下例是一个自定义应用：读取沉降数据，按用户选择的沉降点绘制曲线。TemplateVariableConfig 让 TablesQuery 生成 auto_select 下拉框。

from pathlib import Path

from gdisdk.connectors import log_in
from gdisdk.modules import GdimTableReader, LineChartPlotter, TablesQuery
from gdisdk.pipeline import PipeLine
from gdisdk.pipeline.pipeData import TemplateVariableConfig

_SCRIPT_DIR = Path(__file__).resolve().parent
WORKSPACE_DIR = _SCRIPT_DIR / "workspace"


def build_pipeline() -> PipeLine:
    pipeline = PipeLine(
        app_name="settlementLine",
        app_title="沉降曲线",
        description="基于沉降数据绘制沉降曲线",
        workspace=WORKSPACE_DIR,
    )
    pipeline.update_gdim_state(token=log_in(), proj_id="你的GDIM项目ID")

    read_table = GdimTableReader("ReadTable")
    read_table.table_fields = ["累计沉降数据"]

    query_table = TablesQuery("QueryTable")
    query_table.query_template = "`沉降点编号` in {tpl_number}"
    query_table.template_variables = {
        "tpl_number": TemplateVariableConfig(
            title="沉降点编号",
            value_type="list[str]",
            schema_type="auto_select",
        )
    }

    plot_line = LineChartPlotter("PlotLine")
    plot_line.x_column = "时间"
    plot_line.y_columns = "累计沉降"
    plot_line.color_column = "沉降点编号"
    plot_line.output_name = "沉降曲线"
    plot_line.generate_files = False

    pipeline.add_links(
        read_table.OutputTable >> query_table.InputTables
        | query_table.OutputTables >> plot_line.InputData
    )

    pipeline.add_attribute(
        attr_name="tpl_number",
        module_name="QueryTable",
        param_name="tpl_number",
        attr_title="沉降点编号",
    )

    pipeline.return_results_config = {"PlotLine": "OutputPlotData"}
    pipeline.initial_run_to = ["QueryTable"]   # 预先取数，填充 auto_select 下拉框
    # rerun_initial_upstream 默认为 True：用户运行时上游会重新读取。
    # 不需要重读上游时设为 False。
    pipeline.save_pipeline(f"{pipeline.workspace}/{pipeline.app_name}.pipe")
    return pipeline


if __name__ == "__main__":
    pipeline = build_pipeline()
    pipeline.set_attributes(tpl_number=["点1", "点2"])
    result = pipeline.run(return_results=pipeline.return_results_config)
    print(result.get_result("PlotLine", "OutputPlotData"))

常见错误

端口名单复数写错。 OutputTable ≠ OutputTables，写错会静默新建动态端口，模块间没有真正连上。
忘了 add_links() 。模块没接进连线图就不会执行。
同一输入端口接了两条线。 后一条静默覆盖前一条；汇总多路数据用合并类模块。
正式上传时 return_results_config = True 。会把所有末端输出（包括大表）发给前端。
在 Pipeline 脚本里调用 execute() 或在构造函数里传业务参数。 可能导致模块重复执行。
把用户名、密码或本机绝对路径写进脚本。 用 log_in() 和由脚本位置推导的 workspace。
没按应用类型检查。 比如自定义数据更新配置应用把 GdimTableWriter 存进了 .pipe，自动报告应用没设 save_to_gdim=True。见 GDIM应用类型 (App Types)。

与其它文档的关系

应用类型与各类型规则：GDIM应用类型 (App Types)
属性、auto_bind、界面控件：用户指南 (User Guide)、UI控件 (UI Schema)
批量处理：流程控制 (Flow Control)
上传前检查：上传GDIM检查 (GDIM Upload Check)
在页面上核对表名 / 字段名：GDIM开发模式 (Dev Mode)