modules.widgets 模块帮助

本章节包含 modules.widgets 包中常用「小部件 / 辅助模块」的使用说明和示例。

PythonCoder

模块简介与适用场景

PythonCoder 是一个通用「脚本执行」模块，用于在 Pipeline 中调用本地 Python 脚本中的 本地函数 (local function)。
它通过 动态端口 接收任意类型的数据（如 TableData、 TableCollection、普通 dict / list 等），执行时会把 模块实例 module 作为本地函数的第一个参数传入；本地函数从 module.dynamic_ports_in[...] 读取输入，并以 dict 形式返回输出，由 PythonCoder 写回到动态输出端口。
典型适用场景：
- 已有一段较为独立的业务逻辑（例如复杂统计、报表拼装、特殊格式转换），不希望为其单独封装一个 PipeModule 子类；
- 需要在不改动主代码库的情况下，为某些项目/模板增加「可插拔」的计算逻辑；
- 希望让熟悉 Python 的工程师以「写脚本 + 定义函数」的方式扩展 Pipeline 能力。

端口说明

输入端口 - （无固定端口）：需要通过 add_dynamic_ports_in(...) 动态声明输入端口 - 动态输入端口命名要求：端口名必须以 "Input" 开头（例如 InputTable、InputTables） - 在本地函数中读取方式：module.dynamic_ports_in["InputX"].data
输出端口 - （无固定端口）：需要通过 add_dynamic_ports_out(...) 动态声明输出端口 - 动态输出端口命名要求：端口名必须以 "Output" 开头（例如 OutputTable、OutputSingleResult） - 写出机制：本地函数必须返回一个 dict，其中 key 为输出端口名、value 为要写入端口的数据；PythonCoder 会将其写回到对应的动态输出端口

Note

推荐使用 gdisdk.pipeline.nameSpace.local_function 装饰器来定义本地函数（文件方式 / 字符串函数名）。
Inline 方式**（直接传入可调用对象）**不需要 @local_function 装饰器。
旧方案 LocalFunctionProvider``（类 + ``run_function）仅用于兼容历史代码，未来版本将逐步停止使用；本文档示例统一采用 @local_function 新方案。

快速上手示例：计算工程地质纵剖面总长度

下面以「计算工程地质纵剖面总长度」为例，展示一个完整的 PythonCoder 使用流程。该示例基于真实案例的实际代码进行了简化。

1）在 Pipeline 中创建并配置 PythonCoder

from gdisdk.modules.readers import GdimTableReader
from gdisdk.modules.filters import TableSelector
from gdisdk.modules.widgets import PythonCoder
from gdisdk.pipeline.pipeline import PipeLine

# 1. 创建 Pipeline，并指定本地函数脚本路径
pipeline = PipeLine(
    app_name="SectionLengthDemo",
    app_title="剖面总长度计算示例",
    version="1.0.0",
)
pipeline.update_gdim_state(token="你的GDIM Token", proj_id="你的GDIM项目ID")
pipeline.workspace = "test"

# 关键：指定本地函数脚本路径
# （相对于运行该脚本的工作目录；也可以使用绝对路径）
# 通过pipeline属性指定可以避免有多个PythonCoder时每个PythonCoder都要设置脚本路径
pipeline.local_functions_path = (
    "test/Doc1LocationOverview.py"
)

# 2. 前置模块：读取「剖面数据表」
read_tables = GdimTableReader("ReadTables")
read_tables.table_fields = ["剖面数据表"]

section_lines_data_table = TableSelector("SectionLinesDataTable")
section_lines_data_table.table_name = "剖面数据表"

# 3. 创建 PythonCoder 模块，并指定要调用的本地函数名（脚本中的函数名字符串）
section_lines_length = PythonCoder("SectionLinesLength")
section_lines_length.local_function_name = "get_sections_length"

# 4. 声明动态输入 / 输出端口
#    端口名称需要与脚本中本地函数读取/返回的端口名保持一致
section_lines_length.add_dynamic_ports_in("InputTable")
section_lines_length.add_dynamic_ports_out("OutputSingleResult")

# 5. 像普通模块一样连线并运行
links = (
    read_tables.OutputTables >> section_lines_data_table.InputTables
    | section_lines_data_table.OutputTable >> section_lines_length.InputTable
)
pipeline.add_links(links)

result = pipeline.run()
print(section_lines_length.OutputSingleResult.data)

要点：

pipeline.local_functions_path 指向包含本地函数定义的 .py 脚本；
section_lines_length.local_function_name 指定要调用的函数名（例如 get_sections_length）；
add_dynamic_ports_in("InputTable") / add_dynamic_ports_out("OutputSingleResult") 负责在模块上创建动态端口，这些名称需要和本地函数读取 module.dynamic_ports_in[...] / 返回字典的键保持一致。

2）在脚本中定义本地函数（@local_function）

与上文 pipeline.local_functions_path 对应的脚本中，需要定义一个使用 @local_function 装饰的函数，并确保 第一个参数为 module。例如：

import numpy as np
import pandas as pd

from gdisdk.dataclass.results import SingleResult, UnitResult
from gdisdk.dataclass.tables import TableData
from gdisdk.dataclass.terminologies import Units
from gdisdk.pipeline.nameSpace import local_function


@local_function
def get_sections_length(module, **kwargs) -> dict[str, SingleResult | None]:
    """计算工程地质纵剖面总长度（供 PythonCoder 调用）"""

    # 重要：PythonCoder 访问动态输入端口数据应使用 dynamic_ports_in
    table: TableData | None = module.dynamic_ports_in["InputTable"].data
    if table is None:
        return {"OutputSingleResult": None}

    sections_length = 0.0
    for _, group_df in table.groupby("剖面编号"):
        dx = group_df["x_coordinate"].diff()
        dy = group_df["y_coordinate"].diff()
        distances = np.sqrt(dx**2 + dy**2)
        sections_length += float(distances.sum())

    unit_result = UnitResult(
        name="sections_length",
        title="工程地质纵剖面总长度",
        unit=Units.m,
        value=sections_length,
    )

    # 返回值必须是一个 dict：
    #   key 为 PythonCoder 动态输出端口名称，
    #   value 为要写入该端口的数据
    return {"OutputSingleResult": SingleResult([unit_result])}

在运行时，PythonCoder 内部会通过 gdisdk.pipeline.nameSpace.load_local_function()：

动态加载上述 get_sections_length 函数（要求使用 @local_function 装饰；支持调试模式 debug_mode）；
调用 get_sections_length(module) 获取返回字典；
将返回字典中的各个键写入对应的动态输出端口。

Note

本地函数第一个参数必须是 ``module``（即 PythonCoder 模块实例）。
你可以通过 module 访问/修改的不仅仅是端口数据，还包括：
- 模块属性：例如 module.threshold、module.some_flag （也可以在函数内 setattr(module, ...) 动态修改）；
- Pipeline 上下文：module.pipeline （可进一步 module.pipeline.get_module("OtherModuleName") 访问其它模块实例，或遍历 module.pipeline.modules）；
- 端口数据：推荐从动态端口读取输入：module.dynamic_ports_in["InputX"].data。

参数说明

PythonCoder 参数一览
参数名	类型	默认值	说明
`local_functions_path`	`str \| Path \| None`	`None`	本地函数脚本路径。若 pipeline 已设置 `local_functions_path`，则优先使用 `pipeline.local_functions_path`。
`local_function_name`	`str \| Callable \| None`	`None`	要调用的本地函数：文件方式（可上传 GDIM，推荐）：填写脚本中函数名（字符串），该函数需用 `@local_function` 装饰； Inline 方式（仅适合脚本内运行）：直接传入可调用对象 ``callable``（如函数对象），此时不依赖脚本文件加载。
`ui_schema_function_name`	`str \| Callable \| None`	`None`	（可选）用于生成 UI Schema 的本地函数（字符串或可调用对象）。若设置，则会在 `update_ui_schema(reset=...)` 中调用，并返回 `dict[str, UIAttributeSchema]`。建议签名：def my_ui_schema(module, reset: bool = False) -> dict[str, UIAttributeSchema]``（PythonCoder 会传入 ``reset 参数；如不想显式声明，可用 `**kwargs` 接收）。
`debug_mode`	`bool`	`False`	调试模式。为 `True` 时，加载本地函数失败会给出包含完整堆栈的详细错误信息，便于排查脚本语法错误 / 导入错误等问题。

此外，还需要通过以下方法声明端口：

动态端口相关方法
方法	说明
`add_dynamic_ports_in(name: str, pdoc: str \| None = None)`	添加一个动态输入端口。 `name` 必须以 `"Input"` 开头，并与本地函数中读取 `module.dynamic_ports_in["..."]` 的键对应。
`add_dynamic_ports_out(name: str, pdoc: str \| None = None)`	添加一个动态输出端口。 `name` 必须以 `"Output"` 开头，并与本地函数返回字典的键对应。

除了端口相关方法外，PythonCoder 还提供了一个便捷的 属性配置方法：

add_attributes(**kwargs) - 为模块实例动态增加属性，相当于对当前模块执行一系列 setattr(self, key, value) 操作。 - 这些属性可以在两个地方被使用：
- 在脚本侧：本地函数可以直接读取 module.<attr>``（例如 ``module.threshold）；
- 在 UI 侧：如果你通过 ui_schema_function_name 定义了 UI Schema，UI schema 本地函数可以直接读取这些属性，或者配合 Pipeline 的 add_attribute 将其暴露给前端。
- 一个简单示例：
```
python_coder = PythonCoder("MyCoder")
python_coder.local_function_name = "MyFunc"
python_coder.add_dynamic_ports_in("InputTable")
python_coder.add_dynamic_ports_out("OutputText")

# 通过属性传递额外参数给脚本中的本地函数
python_coder.add_attributes(
    threshold=0.5,
    description="自定义分析说明",
)
```

上传 GDIM 前检查

是否需要特别处理：是。
必须检查：
- local_function_name 上传前应使用 字符串函数名，不要直接传入 inline callable；
- 执行函数应定义在独立 .py 脚本中，并使用 @local_function 装饰；
- 本地函数脚本路径应优先在 pipeline 层 提供；上传平台后通常由平台注入或替换，而不是依赖写死在 .pipe 中的本机路径。
建议检查：
- ui_schema_function_name 也优先使用字符串函数名，而不是 inline callable；
- 动态端口名与本地函数中读取 / 返回的键完全一致，避免上传后因端口名不匹配而失败。
若遗漏，常见现象：
- 本地脚本运行正常，但保存为 .pipe 后，GDIM 无法还原 inline 函数对象，导致模块加载或执行失败。

几种常见用法补充

1）Inline 本地函数 vs 文件本地函数（GDIM 上传要求）

PythonCoder 支持 Inline 方式：直接把 local_function_name 设为一个可调用对象（函数）。
但如果你要把 pipeline 保存为 .pipe 并上传 GDIM 平台运行，必须使用文件方式 （即把函数定义在独立 .py 脚本中，并用字符串指定函数名）。

原因是：Inline 的函数对象无法被序列化到 .pipe 文件中，GDIM 侧也无法反序列化还原该可调用对象。

一个 Inline 的简化示例：

from typing import Any
from gdisdk.modules.widgets import PythonCoder

def inline_func(module, **kwargs) -> dict[str, Any]:
    table = module.dynamic_ports_in["InputTable"].data
    return {"OutputText": f"rows={0 if table is None else len(table.data)}"}

coder = PythonCoder("MyCoder", local_function_name=inline_func)
coder.add_dynamic_ports_in("InputTable")
coder.add_dynamic_ports_out("OutputText")

2）@local_function 的 module 参数可以做什么

使用 @local_function 装饰的函数时，第一个参数必须是 module。你可以通过 module：

读取输入端口数据：module.dynamic_ports_in["InputX"].data （推荐写法）
写入输出：通过返回字典 {"OutputY": data} （由 PythonCoder 写回端口）
读取/修改模块属性：module.some_attr / setattr(module, "some_attr", value)
访问 pipeline 与其它模块：
- module.pipeline 获取当前 Pipeline
- module.pipeline.get_module("OtherModuleName") 获取其它模块实例并读取/修改其属性

Warning

在本地函数里修改其它模块属性会影响后续执行行为，属于 “高级用法” 。建议只在你明确需要动态联动配置时使用，并保持修改可追踪、可回滚。

Note

为了让本地函数在未来版本「可能增加额外参数」时仍保持兼容，建议你采用更稳健的函数签名：

执行函数（execute）：def my_func(module, **kwargs): ...
UI schema 函数：def my_ui_schema(module, reset: bool = False, **kwargs): ...

3）UI Schema 的本地函数与 reset 参数

PythonCoder 的 update_ui_schema 会把 reset 参数传给 UI schema 本地函数：

Inline 方式：调用 ui_schema_function_name(module, reset)
文件方式：调用 ui_schema_function_name(module, reset=reset)

因此 建议 UI schema 本地函数签名包含 ``reset: bool = False``，并额外接收 ``**kwargs``，例如：

from gdisdk.pipeline.nameSpace import local_function
from gdisdk.pipeline.pipeData import StringAttributeSchema

@local_function
def my_ui_schema(module, reset: bool = False, **kwargs):
    return {
        "prompt": StringAttributeSchema(
            title="提示词",
            default=getattr(module, "prompt", ""),
        )
    }

4）兼容 gdi / gdisdk 的包导入写法

若本地函数脚本需要在 gdi``（源码环境）和 ``gdisdk``（编译发布包）两种环境下都能正常运行。推荐在脚本顶部使用 ``try/except 兼容导入：

try:
    from gdi.pipeline.nameSpace import local_function
    from gdi.dataclass.tables import TableData
except ImportError:
    from gdisdk.pipeline.nameSpace import local_function
    from gdisdk.dataclass.tables import TableData

按需导入实际使用到的类即可。

5）本地函数可用的第三方包

gdi / gdisdk 环境已内置以下第三方包，可在本地函数脚本中直接 import：

数据处理与科学计算

pandas — DataFrame 操作（import pandas as pd）
numpy — 数值计算（由 pandas 依赖引入，可直接使用）
scipy — 统计检验、插值、信号处理
shapely — 几何运算（坐标点、线段、多边形）
pyproj — 坐标投影与转换

可视化

plotly — 交互式图表
seaborn — 统计图表（基于 matplotlib）
pillow — 图像处理（from PIL import Image）

文件读写

openpyxl — Excel 文件读写
docxtpl — Word 文档模板渲染
pymupdf — PDF 读写（import fitz）
pypandoc_binary — 文档格式转换（import pypandoc）
rasterio — 栅格地理数据读写
graphviz — 流程图 / 有向图渲染

网络与配置

requests — HTTP 请求
pyyaml — YAML 解析（import yaml）
python-dotenv — 读取 .env 文件环境变量
pyodbc — ODBC 数据库连接（仅 Windows）

数据验证与 AI

pydantic — 数据验证与结构化对象（BaseModel）
pandasai — AI 驱动的 DataFrame 查询
pydantic_ai — pydantic-ai agent 框架
chromadb — 向量数据库

Note

numpy 和 matplotlib 未在 requirements.txt 中显式声明，但它们是 pandas / seaborn 等包的依赖项，在任何 gdi/gdisdk 环境中均可直接使用。

进阶用法与配置示例（可选）

使用 ui_schema_function_name 为 PythonCoder 的自定义属性生成 UI Schema，从而在 GDIM / 自定义前端中暴露参数；
在脚本的本地函数中读取多个输入端口（例如 InputTable1、 InputTable2），并返回多个输出端口（如 OutputTable、 OutputSingleResult）；
配合 gdisdk.pipeline.pipeline.PipeLine 的 add_attribute，可将 PythonCoder 的自定义属性暴露为 Pipeline 属性，实现高度可配置的「脚本模块」。

在 pipeline 中的使用方式小结

与其他模块的最大区别在于： 计算逻辑在外部脚本中，通过函数实现；
在 Pipeline 中，你只需要：
1. 设置 local_functions_path 和 local_function_name；
2. 通过 add_dynamic_ports_in / add_dynamic_ports_out 声明好端口；
3. 像普通模块一样通过 >> / | 将其接入数据流；
4. （可选）通过 Pipeline 属性将 PythonCoder 的参数暴露给前端。

更多信息