adam-community 1.0.30rc3

Adam Community Tools and Utilities

Adam Community

Adam Community 是一个 Python 工具包，提供了 CLI 命令行工具和 Python 模块，用于解析和构建 Python 项目包。

安装

pip install -e .

使用方式

CLI 命令行

查看帮助：

adam-cli --help

初始化新项目：

adam-cli init

解析 Python 文件生成 functions.json：

adam-cli parse .

构建项目包：

adam-cli build .

更新 CLI 到最新版本：

adam-cli update

SIF 文件管理

管理 SIF 文件，包括上传到 Docker 镜像仓库等操作：

# 查看帮助
adam-cli sif --help

# 上传 SIF 文件到镜像仓库（自适应切片）
adam-cli sif upload ./xxx.sif registry.example.com/image:1.0.0

# 带认证上传
adam-cli sif upload ./app.sif registry.cn-hangzhou.aliyuncs.com/ns/app:latest \
  --username user --password pass

Python 模块导入

from adam_community.cli.parser import parse_directory, parse_python_file
from adam_community.cli.build import build_package

# 解析目录下的 Python 文件
classes = parse_directory(Path("./"))

# 构建项目包
success, errors, zip_name = build_package(Path("./"))

命令执行

execCmd - 复杂执行

execCmd 是新的命令执行函数，支持异常处理、超时控制、实时输出等特性。

from adam_community import execCmd, CmdResult
from subprocess import CalledProcessError, TimeoutExpired

# 基本用法
result = execCmd("python train.py")
print(result.stdout)      # 标准输出
print(result.stderr)      # 错误输出
print(result.returncode)  # 退出码
print(result.duration)    # 执行耗时（秒）

# 异常处理
try:
    result = execCmd("python train.py", timeout=3600)
except TimeoutExpired as e:
    print(f"超时: {e.output}")
except CalledProcessError as e:
    print(f"失败 (exit {e.returncode}): {e.stderr}")

# 实时输出到控制台
result = execCmd("python train.py", echo=True)

# 自定义回调（如写日志、发送到前端）
result = execCmd(
    "python train.py",
    on_stdout=lambda line: logger.info(f"[OUT] {line}"),
    on_stderr=lambda line: logger.error(f"[ERR] {line}"),
)

# 指定工作目录和环境变量
result = execCmd(
    "python train.py",
    cwd="/workspace/project",
    env={"CUDA_VISIBLE_DEVICES": "0,1"},  # 合并到当前环境
)

参数说明：

参数	类型	默认值	说明
cmd	str	-	要执行的命令
timeout	float	None	超时秒数，None 表示无限制
cwd	str	None	工作目录
env	dict	None	环境变量（合并到当前环境）
shell	str	/bin/bash	shell 路径
echo	bool	False	是否实时打印到控制台
on_stdout	Callable	None	stdout 回调 (line: str) -> None
on_stderr	Callable	None	stderr 回调 (line: str) -> None

返回值 CmdResult：

字段	类型	说明
stdout	str	完整标准输出
stderr	str	完整错误输出
returncode	int	退出码
duration	float	执行耗时（秒）
command	str	原始命令
timed_out	bool	是否超时

runCmd - 最简执行

runCmd 是最简化版命令执行函数，实时输出到控制台，失败时直接退出进程。

from adam_community import runCmd

runCmd("echo 'Hello, World!'")

States Management（任务状态管理）

用于在任务执行过程中记录和读取状态，与服务端共享 states.json 文件。

from adam_community.util import setState, getState, trackPath

# 记录文件列表（自动与服务端和其他 Tool 的文件合并）
setState("files", [
    {"path": "output/result.json", "is_dir": False, "mtime": 1704100800},
    {"path": "cache", "is_dir": True, "mtime": 1704100000}
])

# 获取合并后的文件列表（来自 server + 所有 tools）
files = getState("files")

# 记录自定义状态（支持嵌套 key）
setState("stage", "data_cleaning")
setState("config.threshold", 0.5)

# 获取状态
stage = getState("stage")              # -> "data_cleaning"
threshold = getState("config.threshold")  # -> 0.5

# 追踪文件/目录（自动检测 is_dir 和 mtime，先进先出，最多 30 条）
trackPath("/path/to/output/result.json")
trackPath("/path/to/cache")

Tool I/O 类型系统

用于 DAG 工作流的数据传递——为每个 Tool 声明输入/输出文件类型，供 Planner 规划时做类型检查，运行时做结构化输出注册。

FileType 别名

预定义了 39 种科学计算常用文件类型，运行时值就是 str（文件路径）：

from adam_community.tool_types import PDBFile, SDFFile, CSVFile, FASTAFile

按领域分组：

分类	类型别名
结构文件	PDBFile, CIFFile, MMCIFFile, MMTFFile, PDBQTFile, GROFile, SDFFile, MOLFile, MOL2File, XYZFile, PQRFile, STRUFile, VASPFile
序列/比对	FASTAFile, A3MFile
拓扑文件	PRMTOPFile, PSFFile, TOPFile
轨迹文件	DCDFile, XTCFile, TRRFile, NetCDFFile, NCTRAJFile, LAMMPSFile
电子密度/体积	MRCFile, MAPFile, CCP4File, DSN6File, DXFile, CubeFile
表格/数据	CSVFile, JSONFile, TXTFile, MDFile, NPYFile
图片/网格	PNGFile, OBJFile, PLYFile

@tool_io 装饰器

在 Tool 子类上声明输入输出类型，无需实例化即可提取合约：

from adam_community.tool_types import tool_io, PDBFile, SDFFile, CSVFile

@tool_io(
    inputs={"protein": PDBFile, "ligand": SDFFile},
    outputs={
        "docking_result": SDFFile,
        "scores": CSVFile,
        "best_affinity": float,
        "num_poses": int,
    },
)
class DockingTool(Tool):
    def run(self, tool_kwargs):
        # tool_kwargs["protein"] 和 tool_kwargs["ligand"] 是文件路径字符串
        ...

参数说明：

参数	说明
inputs	输入文件端口声明，{port_name: FileType}
outputs	输出声明，{port_name: FileType | type}。FileType 别名声明文件输出，int/float/str/bool 声明标量参数输出

DAG Planner 会在规划时调用 get_tool_contract(DockingTool) 读取合约，用于：

• 自动连线时做类型兼容检查（PDB → PDB 兼容，PDB → SDF 不兼容）
• 缺少类型转换时自动插入转换节点
• 读取 outputs 中的标量类型声明，获知工具产出的标量参数及其类型

Kit 结构化返回（推荐方式）

Kit 类型的 Tool（calltype="python"）在 call() 方法中直接返回 {"files": {...}, "params": {...}} 字典，框架自动写入 _outputs.json：

from adam_community.tool_types import tool_io, PDBFile, CSVFile, MDFile

@tool_io(
    inputs={"input_structure": PDBFile},
    outputs={
        "predictions": CSVFile,
        "report": MDFile,
        "num_predictions": int,
        "best_score": float,
    },
)
class MyKit(Tool, calltype="python"):
    def call(self, kwargs):
        # ... 计算逻辑 ...
        return {
            "files": {
                "predictions": "./results.csv",    # key 必须与 @tool_io outputs 中的 FileType key 一致
                "report": "./report.md",
            },
            "params": {                            # key 必须与 @tool_io outputs 中的标量类型 key 一致
                "num_predictions": 42,
                "best_score": 0.95,
            },
        }

规则：

• files 的 key 必须与 @tool_io(outputs={...}) 声明的 key 完全一致
• params 的 key 必须与 outputs 中声明的标量类型 key 完全一致
• 文件路径为相对于工作目录的相对路径

Toolbox declare_outputs（Bash 类型 Tool）

Toolbox 类型的 Tool（calltype="bash"）在 call() 末尾调用 self.declare_outputs()：

@tool_io(
    inputs={"protein": PDBFile, "ligand": SDFFile},
    outputs={
        "docked": SDFFile,
        "log": TXTFile,
        "best_affinity": float,
    },
)
class DockingTool(Tool):
    def call(self, kwargs):
        self.declare_outputs(
            files={"docked": "result.sdf", "log": "docking.log"},
            params={"best_affinity": -8.5},
        )
        return "run_docking ..."  # 返回 bash 命令

ToolOutput 统一出口

Tool 基类提供 output() 方法，返回 ToolOutput 对象。只要加了 @tool_io 装饰器，框架会自动扫描工作目录、按扩展名匹配到声明的输出端口，开发者无需实现 output() 方法。

output() 的优先级：

1. 子类 override → 直接使用子类实现
2. @tool_io 自动匹配 → 扫描 directory_path 下所有文件，按扩展名匹配到声明的 outputs
3. 兜底 → 包装旧的 outputShow() + summary() 返回值

零开发负担用法（推荐）：只需声明 @tool_io，框架自动完成一切：

@tool_io(
    inputs={"protein": PDBFile, "ligand": SDFFile},
    outputs={"docking_result": SDFFile, "scores": CSVFile},
)
class DockingTool(Tool):
    def call(self, kwargs):
        # 返回 bash 命令，Slurm 执行完毕后：
        # - 目录下的 .sdf 文件自动匹配到 "docking_result"
        # - 目录下的 .csv 文件自动匹配到 "scores"
        return f"run_docking --protein {kwargs['protein']} --ligand {kwargs['ligand']}"

自动匹配规则：

• 声明 {"structure": PDBFile} + 目录下 1 个 .pdb → ToolFile(path="xxx.pdb")
• 声明 {"designs": PDBFile} + 目录下 3 个 .pdb → ToolFile(paths=["d0.pdb", "d1.pdb", "d2.pdb"])
• 声明 {"scores": CSVFile} + 目录下 0 个 .csv → 跳过该 key
• 没有 @tool_io → 走旧的 outputShow 兜底

手动 override（精确控制场景）：

from adam_community.tool_types import ToolOutput, ToolFile

class MyTool(Tool):
    def output(self, tool_kwargs) -> ToolOutput:
        return ToolOutput(
            files={
                "structure": ToolFile(
                    type="pdb",
                    description="Predicted protein structure",
                    path="/path/to/result.pdb",
                ),
            },
            text="Prediction completed successfully.",
            summary="Generated PDB structure with confidence 0.95.",
        )

多文件输出：当一个输出端口产出 N 个同类型文件时（如 RFdiffusion 生成多个 design），使用 paths 替代 path：

class RFdiffusionTool(Tool):
    def output(self, tool_kwargs) -> ToolOutput:
        return ToolOutput(
            files={
                "designs": ToolFile(
                    type="pdb",
                    description="Design variants",
                    paths=["/out/design_0.pdb", "/out/design_1.pdb", "/out/design_2.pdb"],
                ),
            },
            ...
        )

也可以用 fill_paths 快捷方法，支持传入 list[str]：

files = self.fill_paths({
    "designs": ["/out/design_0.pdb", "/out/design_1.pdb"],  # 多文件
    "scores": "/out/scores.csv",                              # 单文件
})

提取工具合约

无需实例化 Tool 即可读取 I/O 声明：

from adam_community.tool_types import get_tool_contract

contract = get_tool_contract(DockingTool)
# {
#   "inputs":  {"protein": {"ext": ".pdb", "description": "..."}, "ligand": {"ext": ".sdf", ...}},
#   "outputs": {"docking_result": {"ext": ".sdf", ...}, "scores": {"ext": ".csv", ...}},
#   "output_params": {"best_affinity": "float", "num_poses": "int"},
# }

功能特性

• Python 文件解析: 自动解析 Python 类和函数的文档字符串
• JSON Schema 验证: 将 Python 类型转换为 JSON Schema 并验证
• 项目构建: 检查配置文件、文档文件并创建 zip 包
• 类型检查: 支持多种 Python 类型注解格式
• 自动更新: 智能检查和更新到最新版本，支持用户配置
• SIF 镜像构建: 将 SIF 文件切片并构建 Docker 镜像推送到仓库
• Tool I/O 类型系统: 为 DAG 工作流提供文件类型声明、结构化输出、合约提取

开发

安装依赖：

make install

运行测试：

make test

构建包：

make build