adam-community 1.0.29

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")

功能特性

• Python 文件解析: 自动解析 Python 类和函数的文档字符串
• JSON Schema 验证: 将 Python 类型转换为 JSON Schema 并验证
• 项目构建: 检查配置文件、文档文件并创建 zip 包
• 类型检查: 支持多种 Python 类型注解格式
• 自动更新: 智能检查和更新到最新版本，支持用户配置
• SIF 镜像构建: 将 SIF 文件切片并构建 Docker 镜像推送到仓库

开发

安装依赖：

make install

运行测试：

make test

构建包：

make build