InfoRSI/tasks/circle_packing/harness.py

"""填圆任务的运行 harness（宿主的固定工具，候选不得修改）。

它按文件路径导入候选程序、调用候选定义的 ``run_packing()``、把原始解原子落盘。
**它不判定解是否合法**：成绩只由独立校验器 ``validate.py`` 事后从实际产物重算。

指标的分工（docs/方法.md 第五部分"指标"）：
  - 最终指标和合法性判定归 ``validate.py``，跟候选无关；
  - 假设专用指标由研究者在协议里登记、实验员按定义在候选代码里实现并写出
    ``metrics.json``，宿主直接采信。harness 不内置任何指标算法。

命令行：python -B harness.py --program <program.py> --output <solution.json>
种子从环境变量 INFORSI_SEED 读，必须给出。
成功退出 0；任何失败都带 traceback 写 stderr 并非零退出。

工作目录里的三个文件（给实验员看的完整契约在同目录 ``TASK.md`` 的"指标怎么交给宿主"）：
  - ``metrics-spec.json``：本次运行登记的假设专用指标定义，由宿主在准备运行目录时写好，
    harness 不碰它。
  - ``solution.json``（``--output`` 指定）：候选返回的原始解。
  - ``metrics.json``：实验员的代码写出的假设专用指标取值，**键必须是 metrics-spec.json 里
    那条指标的 ``id`` 原文**，值是有限数或 null，路径相对工作目录，
    且必须在 ``run_packing()`` 返回之前写完。没有这个文件就表示这一次的假设专用指标一个
    都没测到。另有两个保留键（docs/方法.md 第五部分"答案路径"）：``answer-path``（字符串，
    最终交的答案出自哪条路径）和 ``safeguard-triggers``（对象，每条保障动手几次）。
    它们不是指标、不是数，harness 原样保留，宿主读成辅助记录。
"""

import argparse
import hashlib
import importlib.util
import json
import math
import os
import random
import sys
from typing import Any, cast

METRICS_FILE = "metrics.json"
# metrics.json 里不是数的两个保留键（docs/方法.md 第五部分"答案路径"）。finalize_metrics 只改写指标值，这两个原样留着：
# 改写成 null 的话，候选写对了的答案路径到宿主那里就成了"不认识的路径：null"，出错后无法察觉。
RESERVED_KEYS = ("answer-path", "safeguard-triggers")


def load_program(program_path: str):
    """按显式文件路径导入候选模块（不假定固定的 main.py 模式）。"""
    spec = importlib.util.spec_from_file_location("candidate_program", program_path)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"无法从 {program_path} 加载候选模块")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def write_json_atomically(path: str, payload) -> None:
    """先写临时文件再 os.replace，让崩溃的运行永远不会留下半截产物。"""
    temporary = path + ".tmp"
    with open(temporary, "w", encoding="utf-8") as handle:
        json.dump(payload, handle, allow_nan=False)
        handle.flush()
        os.fsync(handle.fileno())
    os.replace(temporary, path)
    directory_fd = os.open(
        os.path.dirname(os.path.abspath(path)), os.O_RDONLY | os.O_DIRECTORY
    )
    try:
        os.fsync(directory_fd)
    finally:
        os.close(directory_fd)


def is_finite_number(value) -> bool:
    """有限数的定义：不是布尔值的 int/float，且 math.isfinite。

    布尔值排除在外，否则 ``true`` 会被当成 1 记进证据。
    超出浮点范围的大整数 math.isfinite 会抛 OverflowError，一并算作非有限。
    """
    if not isinstance(value, (int, float)) or isinstance(value, bool):
        return False
    try:
        return math.isfinite(value)
    except OverflowError:
        return False


def finalize_metrics() -> None:
    """把实验员写出的 metrics.json 整理成宿主读得进去的形状，然后原子落盘。

    只做一件事：指标的非有限值（NaN、±inf、布尔、非数值）**原样改写成 null**，其余照抄；
    两个保留键（RESERVED_KEYS）不是指标，原样保留。
    这正是"测不到就说测不到"。被改写的编号在 stderr 记一行，**退出码保持 0**。

    为什么不抛错——这里抛错会让情况变坏，不是变好：
      1. 它防不住自述要防的那件事。实验员把测不到的量写成 0.0 照样放行。
      2. 它实际拦的几种情况宿主已经逐条拦过，而且处理温和得多：那一条记 null
         加一句原因，成绩照常。
      3. 抛错时 solution.json 已经完整落盘、解合法、成绩算得出来，但非零退出码
         会让宿主判定这次运行没干净结束，最终指标也记成 null——一条坏指标白白
         作废掉整次运行，而运行名额照扣。更糟的是那份裸 NaN 的 metrics.json 会
         留在盘上，JS 的 JSON.parse 不接受裸 NaN，于是**这一轮所有候选指标一起
         丢**，不只是坏的那一条。

    它防的是"模型正常工作时就会做"的那类失手：实验员算某条比值时碰上退化布局
    （分母是零半径）拿到 nan 没接住。改写成 null 之后，坏的那条变成 null 加原因，
    好的那些照常，成绩和运行名额都不受影响。

    **不**校验键必须是 metrics-spec.json 里声明过的那些：宿主只按登记的编号取值，
    多出来的键不读。

    文件不存在表示候选指标一个都没测到，不补零也不补空文件。
    """
    if not os.path.exists(METRICS_FILE):
        return
    with open(METRICS_FILE, encoding="utf-8") as handle:
        try:
            measured = json.load(handle)
        except ValueError as error:
            # 同样不抛错，理由同上：整份读不出来是宿主那边看得见的一个状态
            # （它会记一句"metrics.json 读不出来"），不该连累已经算出来的成绩。
            print(f"{METRICS_FILE} 不是合法 JSON，原样留在盘上: {error}", file=sys.stderr)
            return
    if not isinstance(measured, dict):
        print(
            f"{METRICS_FILE} 必须是 JSON 对象（指标编号 -> 值），实际是 "
            f"{type(measured).__name__}，原样留在盘上",
            file=sys.stderr,
        )
        return

    rewritten: dict = {}
    replaced: list[str] = []
    for metric_id, value in measured.items():
        if metric_id in RESERVED_KEYS or value is None or is_finite_number(value):
            rewritten[metric_id] = value
        else:
            rewritten[metric_id] = None
            replaced.append(f"{metric_id}={value!r}")
    if replaced:
        print(
            f"{METRICS_FILE} 里这些指标的值不是有限数，已按'测不到就写 null'改写成 "
            f"null: {', '.join(replaced)}",
            file=sys.stderr,
        )
    write_json_atomically(METRICS_FILE, rewritten)


def main() -> int:
    parser = argparse.ArgumentParser(
        description="运行填圆候选程序并落盘它返回的原始解。"
    )
    parser.add_argument(
        "--program",
        required=True,
        help="定义了 run_packing() 的候选程序文件路径",
    )
    parser.add_argument(
        "--output",
        required=True,
        help="原始解 JSON 的写出路径（centers、radii、reported_sum）",
    )
    args = parser.parse_args()

    # 运行目录是候选目录整个复制来的，里面可能留着以前某次运行写的这两个输出文件。
    # 候选这次没写时，宿主会把旧值当成这次的产物读走，所以先删掉，让读到的只可能是这次写的。
    for leftover in (args.output, METRICS_FILE):
        if os.path.exists(leftover):
            os.remove(leftover)

    # 在导入候选模块之前设种子。宿主给的是一个种子字符串（同一份研究的全部运行共用，
    # docs/方法.md 第十部分要求评测确定），取它的 sha256 前 4 字节大端整数。
    # 没给就报 KeyError 停下，不替它补一个。
    np = importlib.import_module("numpy")

    seed = int.from_bytes(
        hashlib.sha256(os.environ["INFORSI_SEED"].encode()).digest()[:4],
        "big",
    )
    random.seed(seed)
    np.random.seed(seed)

    module = load_program(args.program)
    run_packing = getattr(module, "run_packing", None)
    if not callable(run_packing):
        raise RuntimeError(f"{args.program} 没有定义可调用的 run_packing()")

    # 候选模块是动态加载的，下面的转换就是协议本身的强制点。
    centers, radii, reported_sum = cast(Any, run_packing())
    solution = {
        "centers": [[float(x), float(y)] for x, y in centers],
        "radii": [float(r) for r in radii],
        "reported_sum": float(reported_sum),
    }
    write_json_atomically(args.output, solution)

    finalize_metrics()
    return 0


if __name__ == "__main__":
    sys.exit(main())