adam-community

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"},  # 合并到当前环境
)

参数说明：

返回值 CmdResult：

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 别名

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

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

按领域分组：

@tool_io 装饰器

在 Tool 子类上声明输出类型，无需实例化即可提取合约。强烈推荐为每个端口用元组 (Type, "描述") 写明语义——描述会被 DAG Planner 用来判断工具用途、做连线决策，缺失描述会显著降低自动规划质量：

from adam_community.tool_types import tool_io, SDFFile, CSVFile

@tool_io(
    outputs={
        "docking_result": (SDFFile, "对接产生的小分子构象"),
        "scores":         (CSVFile, "每个构象的打分表"),
        "best_affinity":  (float,   "最优构象的结合亲和力 (kcal/mol)"),
        "num_poses":      (int,     "生成的构象总数"),
    },
)
class DockingTool(Tool):
    def run(self, tool_kwargs):
        # tool_kwargs["protein"] 和 tool_kwargs["ligand"] 是文件路径字符串
        ...

参数说明：

FileType 别名（SDFFile、PDBFile 等）声明文件输出，int/float/str/bool 声明标量参数输出。

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

• 自动连线时做类型兼容检查（PDB → PDB 兼容，PDB → SDF 不兼容）
• 缺少类型转换时自动插入转换节点
• 读取 outputs 中的标量类型声明，获知工具产出的标量参数及其类型
• 读取端口描述理解工具语义——这是 Planner 判断"这个输出能不能喂给下一个节点"的关键信息

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

Kit 类型的 Tool（calltype="python"）在 call() 方法中直接返回 flat dict，框架根据 @tool_io 声明自动区分文件和标量参数，写入 _outputs.json。每个端口都应该带描述——Kit 通常是工作流的关键节点，Planner 依赖描述才能正确连线：

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

@tool_io(
    outputs={
        "predictions":     (CSVFile, "每条输入的预测结果表（含 id、score、label）"),
        "report":          (MDFile,  "人类可读的分析报告，含可视化摘要"),
        "num_predictions": (int,     "本次产生的预测条目总数"),
        "best_score":      (float,   "最高置信度分值，范围 0–1"),
    },
)
class MyKit(Tool, calltype="python"):
    def call(self, kwargs):
        # ... 计算逻辑 ...
        return {
            "predictions": "./results.csv",
            "report": "./report.md",
            "num_predictions": 42,
            "best_score": 0.95,
        }

规则：

• 返回 flat dict，key 必须与 @tool_io(outputs={...}) 声明的 key 完全一致
• 框架根据 @tool_io 中 FileType（如 CSVFile）和标量类型（如 int）自动分类
• 文件路径为相对于工作目录的相对路径
• 描述写清楚产出是什么、单位/范围/格式——这比写了多长重要，一句话能讲清就够

Toolbox declare_outputs（Bash 类型 Tool）

Toolbox 类型的 Tool（calltype="bash"）在 call() 末尾调用 self.declare_outputs()。同样推荐在 @tool_io 里给每个端口写描述：

@tool_io(
    outputs={
        "docked":        (SDFFile, "AutoDock Vina 输出的对接构象（多模型 SDF）"),
        "log":           (TXTFile, "Vina 运行日志，含收敛信息与警告"),
        "best_affinity": (float,   "最优构象的结合自由能 (kcal/mol)，越负越好"),
    },
)
class DockingTool(Tool):
    def call(self, kwargs):
        self.declare_outputs({"docked": "result.sdf", "log": "docking.log", "best_affinity": -8.5})
        return "run_docking ..."  # 返回 bash 命令

提取工具合约

无需实例化 Tool 即可读取输出声明：

from adam_community.tool_types import get_tool_contract

contract = get_tool_contract(DockingTool)
# {
#   "outputs": {
#     "docking_result": {"ext": ".sdf", "description": "对接产生的小分子构象"},
#     "scores":         {"ext": ".csv", "description": "每个构象的打分表"},
#   },
#   "output_params": {
#     "best_affinity": {"type": "float", "description": "最优构象的结合亲和力 (kcal/mol)"},
#     "num_poses":     {"type": "int",   "description": "生成的构象总数"},
#   },
# }

功能特性

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

开发

安装依赖：

make install

运行测试：

make test

构建包：

make build