4. Python 脚本驱动外部计算程序的子进程封装

任务拆解：把外部可执行文件当作被监控的“黑盒”

本节要封装的对象是一个典型的科学计算可执行文件，例如 solver。它通过命令行参数接收输入文件与输出目录，正常结束时返回码为 0，并把关键指标打印到标准输出；参数错误、输入缺失或数值不收敛时返回非零码，并把错误信息写入标准错误。Python 脚本的职责不是重写计算，而是可靠地启动它、收集它的输出、把状态与指标结构化落盘，并在失败时给出可判断的结论。

封装的关键在于把“进程级事实”与“业务级判断”分开：subprocess.run 负责拿到返回码、stdout、stderr 与耗时；脚本再根据返回码和输出内容决定状态是 success、failed 还是 timeout，并把关键指标从 stdout 中解析出来。下图展示了这一条从参数到结构化日志的完整信号链。

封装脚本的关键实现

下面的脚本是一个可直接运行的完整封装。它用 subprocess.run 启动外部程序，设置超时，捕获 stdout 与 stderr，解析指标，并把一条 JSON 记录追加到日志文件。为了让示例可独立验证，脚本内置了一个“假 solver”生成逻辑：当传入 --demo 时，它会先写一个小的 Python 脚本充当外部程序，再对它进行封装调用，从而完整演示成功、失败与超时三种路径。

import json
import subprocess
import sys
import time
from datetime import datetime, timezone
from pathlib import Path

LOG_PATH = Path("run_log.jsonl")

def parse_metrics(stdout: str) -> dict:
    """从 solver 的标准输出中提取关键指标。
    约定输出形如：
      iterations: 42
      residual: 1.2e-08
      energy: -3.14159
    """
    metrics = {}
    for line in stdout.splitlines():
        if ":" not in line:
            continue
        key, _, value = line.partition(":")
        key = key.strip()
        value = value.strip()
        if key in {"iterations", "residual", "energy"}:
            try:
                metrics[key] = int(value) if key == "iterations" else float(value)
            except ValueError:
                metrics[key] = value
    return metrics

def run_solver(cmd: list[str], timeout: float = 30.0) -> dict:
    """执行外部程序并返回结构化结果。"""
    started = time.perf_counter()
    record = {
        "run_id": datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ"),
        "cmd": cmd,
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }
    try:
        proc = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=timeout,
            check=False,
        )
        elapsed = time.perf_counter() - started
        record.update({
            "returncode": proc.returncode,
            "elapsed_sec": round(elapsed, 3),
            "stdout_tail": proc.stdout[-500:],
            "stderr_tail": proc.stderr[-500:],
        })
        if proc.returncode == 0:
            record["status"] = "success"
            record["metrics"] = parse_metrics(proc.stdout)
        else:
            record["status"] = "failed"
            record["metrics"] = {}
    except subprocess.TimeoutExpired as exc:
        elapsed = time.perf_counter() - started
        record.update({
            "returncode": None,
            "status": "timeout",
            "elapsed_sec": round(elapsed, 3),
            "stdout_tail": (exc.stdout or "")[-500:] if isinstance(exc.stdout, str) else "",
            "stderr_tail": (exc.stderr or "")[-500:] if isinstance(exc.stderr, str) else "",
            "metrics": {},
        })
    return record

def append_log(record: dict, path: Path = LOG_PATH) -> None:
    with path.open("a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")

def make_demo_solver(path: Path) -> None:
    """生成一个可切换成功/失败/超时行为的演示用外部程序。"""
    path.write_text(
        "import sys, time\n"
        "mode = sys.argv[1] if len(sys.argv) > 1 else 'ok'\n"
        "if mode == 'ok':\n"
        "    print('iterations: 42')\n"
        "    print('residual: 1.2e-08')\n"
        "    print('energy: -3.14159')\n"
        "    sys.exit(0)\n"
        "elif mode == 'fail':\n"
        "    print('input file not found', file=sys.stderr)\n"
        "    sys.exit(2)\n"
        "elif mode == 'slow':\n"
        "    time.sleep(5)\n"
        "    sys.exit(0)\n",
        encoding="utf-8",
    )

if __name__ == "__main__":
    demo = Path("demo_solver.py")
    make_demo_solver(demo)
    cases = [
        ("ok", 30.0),
        ("fail", 30.0),
        ("slow", 1.0),
    ]
    for mode, timeout in cases:
        rec = run_solver([sys.executable, str(demo), mode], timeout=timeout)
        rec["case_name"] = mode
        append_log(rec)
        print(f"{mode:5s} -> status={rec['status']:8s} rc={rec['returncode']} "
              f"elapsed={rec['elapsed_sec']}s metrics={rec['metrics']}")

运行后终端会输出三行摘要，同时 run_log.jsonl 中追加三条记录。预期现象是：ok 对应 status=success、返回码 0、指标被解析为数值；fail 对应 status=failed、返回码 2、stderr_tail 含 “input file not found”；slow 在 1 秒超时下对应 status=timeout、返回码为 None。这些是脚本逻辑直接推出的结果，不是实测数据。

失败场景的处理要点

• 区分“程序失败”与“封装失败”：返回码非零属于程序失败，应记录 failed 并保留 stderr 尾部；而可执行文件不存在会抛 FileNotFoundError，属于封装层问题，应单独捕获并记录为 launch_error，避免与计算失败混淆。
• 超时要有独立状态：TimeoutExpired 时子进程已被终止，返回码不可用，日志中写 null 比写 0 更诚实，便于后续统计区分。
• 截断而非全量落盘：stdout/stderr 可能很长，日志只保留尾部若干字符用于定位，完整输出仍留在输出目录中，避免日志文件膨胀。
• 日志追加而非覆盖：使用 JSON Lines 追加写入，每行一条独立记录，便于后续用脚本按 status 或 case_name 过滤统计。

验证清单