feat(all): NaN 伪收敛否决与 nl_tight 能量回退、emflux 检验 4π 修正、nst 行宽与种子边界修复、gfATO 谱线表接通与网格加密 9216 点、阶段分项统计与跳板机部署

物理修复(400 失败点归因,见 docs/failed400_nan_pseudo_convergence_2026_08_17.md):
- runner: 阶段 converged 后复查 fort.7,含 NaN/Inf 即否决(fort.9 全零伪收敛,
  曾致 261 点误跳过 nl_direct 回退);否决阶段不产出种子,阻断污染传播
- runner: nl_tight 回退——仅能量边际失败时以 CHMAX 收紧 10× 从自身模型续迭代,
  残差降幅 ~10×;execute_tlusty_stage 抽取供主链与回退共用
- conv_check: fort.14 为 Eddington 通量 Hλ,积分需乘 4π 再比 σTeff⁴
  (旧版 ratio 稳定 0.0796=1/4π,全点系统性假阳性)+ 回归测试
- nst_writer: 单行超 80 字符被 TLUSTY 静默截断,IFALI/JALI/TRAD 等从未生效;
  按 75 字符自动换行
- seed_finder: Teff 容忍度改含边界 <=,相邻 5000K 档恢复互为种子 + 回归测试

谱线表与网格:
- 默认线表 gfVIS99 → gfATO(全波段 18-23000Å),TaskSpec.linelist 支持工作流
  级覆盖,节点按需下载(进程互斥锁防并发重复下载 238MB)
- sdB_cno Teff 加密至 5000K 步长,432 → 9216 点;tlusty/synspec 静态二进制更新

统计与部署:
- grid 汇总改按 tlusty_status/synspec_status 分项计数,新增 tlusty_failed/
  synspec_failed/synspec_pending,前端详情页双视图适配
- deploy/fetch_results 支持跳板机 ProxyJump 与 SSH 主连接复用,fetch 新增 --force;
- Docker 构建支持 CARGO_MIRROR/USE_MIRRORS 国内镜像参数;移除 tools/ 拷贝
- 新增 tlusty-synspec-test skill 与 6 篇根因分析/验证文档
This commit is contained in:
fmq
2026-08-17 23:55:26 +08:00
parent 43b82b1ae2
commit b058e66722
54 changed files with 5624 additions and 347 deletions
+245
View File
@@ -0,0 +1,245 @@
#!/usr/bin/env python3
"""
check_fort9.py — 解析 {stage}.9 收敛轨迹,判数值收敛;检测 STOP / NaN / 雪崩。
复刻 DCTS check_fort9conv_check.rs:68)的核心逻辑。
退出码:0=收敛 /1=未收敛或异常 /2=文件缺失或无有效数据。
⚠️ **只对已完成的运行使用本脚本**:TLUSTY 是边算边写 fort.9 的(每次迭代的
深度行逐行写出),运行中实时读取,最新一拍的 max_relc 读到的是半写状态,数值
不可信(实测:运行中读到 iter6=0.477、iter9=0.00176 貌似收敛,进程退出后同一
迭代实为 1060 / 5.07e4)。等 run.sh 完全结束、进程退出后再跑本脚本。
判据:
- 每次 ITER 的 max_relc = 该迭代所有深度行 |MAXIMUM| 的最大值
- 末次迭代 max_relc < chmax(默认 0.001)且有限 = 收敛
- 单调下降=健康;骤涨几个数量级=雪崩
- fort.6stdout)里的 "STOP in SOLVE after ITER N" = 求解器发散中止
用法:
python3 check_fort9.py outputs/nl.9 --teff 60000
python3 check_fort9.py outputs/nc.9 --chmax 0.001 --stdout outputs/nc.6
"""
import argparse
import os
import re
import sys
# 复刻 conv_check.rs:44 parse_fortran_float
_NO_E_EXP_RE = re.compile(r"^([+-]?[\d.]+)([+-]\d+)$")
def parse_fortran_float(s):
s = s.strip()
try:
return float(s)
except ValueError:
pass
m = _NO_E_EXP_RE.match(s)
if m:
try:
return float(f"{m.group(1)}E{m.group(2)}")
except ValueError:
pass
return None
# 复刻 conv_check.rs:107 FORT9_REiter depth 5浮点 2整数
FORT9_RE = re.compile(
r"^\s*(\d+)\s+(\d+)\s+([-+\dE.]+)\s+([-+\dE.]+)\s+([-+\dE.]+)\s+([-+\dE.]+)\s+([-+\dE.]+)\s+(\d+)\s+(\d+)\s*$"
)
# NaN 模式(conv_check.rs:18
NAN_RE = re.compile(r"(?i)(\bnan\b|\binf(?:inity)?\b|\*{3,}|[eE]\+(?:3\d{2}|[4-9]\d{2,}))")
# STOP 模式(conv_check.rs SOLVER_STOP_RE
STOP_RE = re.compile(r"STOP\s+in\s+SOLVE\s+after\s+ITER\s+(\d+)", re.IGNORECASE)
def parse_fort9(path, chmax):
"""返回 (iters, last_iter, last_max_relc, converged, n_depths_last)。
iters = [(iter_no, max_relc, n_depths), ...]"""
iters = []
cur_iter = None
cur_max = 0.0
cur_n = 0
last_iter = None
with open(path, "r", errors="replace") as f:
for line in f:
m = FORT9_RE.match(line)
if not m:
continue
try:
it = int(m.group(1))
except ValueError:
continue
maximum = parse_fortran_float(m.group(7))
if maximum is None:
continue
if cur_iter != it:
if cur_iter is not None:
iters.append((cur_iter, cur_max, cur_n))
cur_iter, cur_max, cur_n = it, 0.0, 0
cur_n += 1
am = abs(maximum)
if am > cur_max:
cur_max = am
last_iter = it
if cur_iter is not None:
iters.append((cur_iter, cur_max, cur_n))
if not iters or last_iter is None:
return [], None, float("inf"), False, 0
last_max_relc = iters[-1][1]
converged = (last_max_relc < chmax) and (last_max_relc == last_max_relc) and (abs(last_max_relc) != float("inf"))
return iters, last_iter, last_max_relc, converged, iters[-1][2]
def detect_avalanche(iters):
"""检测雪崩:相邻迭代 max_relc 涨 >1e3 倍。返回告警列表。"""
warns = []
for i in range(1, len(iters)):
prev = iters[i - 1][1]
cur = iters[i][1]
if prev > 0 and cur / prev > 1e3:
warns.append(f" iter {iters[i-1][0]}{iters[i][0]}: max_relc {prev:.3g}{cur:.3g}(涨 {cur/prev:.0g}×,雪崩)")
return warns
def scan_stop(stdout_path):
"""扫 fort.6 找 STOP in SOLVE。返回 iter 号或 None。"""
if not stdout_path or not os.path.exists(stdout_path):
return None
last = None
with open(stdout_path, "r", errors="replace") as f:
for line in f:
m = STOP_RE.search(line)
if m:
last = int(m.group(1))
return last
def scan_nan(path):
if not path or not os.path.exists(path):
return 0
n = 0
with open(path, "r", errors="replace") as f:
for line in f:
if NAN_RE.search(line):
n += 1
return n
def main():
ap = argparse.ArgumentParser(description="fort.9 收敛轨迹分析")
ap.add_argument("fort9", help="{stage}.9 路径")
ap.add_argument("--teff", type=float, default=None, help="Teff(上下文,非必填)")
ap.add_argument("--chmax", type=float, default=0.001, help="收敛阈值(默认 0.001")
ap.add_argument("--stdout", default=None, help="对应 {stage}.6 路径,用于 STOP 检测(默认自动推断)")
ap.add_argument("--expected-niter", type=int, default=None,
help="配置的 NITER(来自 nst)。若末次 iter 远小于此且未收敛,提示疑似被 timeout 截断")
args = ap.parse_args()
if not os.path.exists(args.fort9):
print(f"⚠ 文件不存在: {args.fort9}")
return 2
# 自动推断 stdout(同目录同名 .6):把末尾 .9 换成 .6
stdout = args.stdout
if stdout is None:
if args.fort9.endswith(".9"):
cand = args.fort9[:-2] + ".6"
else:
cand = args.fort9 + "6"
stdout = cand if os.path.exists(cand) else None
if args.chmax <= 0:
print(f"❌ 非法 chmax={args.chmax}(须为正有限数)")
return 1
iters, last_iter, last_max_relc, converged, n_depths = parse_fort9(args.fort9, args.chmax)
if not iters:
print(f"{args.fort9} 无有效迭代数据")
return 2
stop_iter = scan_stop(stdout)
nan9 = scan_nan(args.fort9)
# 截断检测:① 附近有 .TIMEOUT 标记文件(run.sh 在 rc=124 时写到 outputs/{stage}.TIMEOUT);
# ② 末次 iter 远小于 --expected-niter 且未收敛(疑似 timeout)。
import glob
timeout_marker = None
d = os.path.dirname(os.path.abspath(args.fort9))
for pat in (os.path.join(d, "*.TIMEOUT"),
os.path.join(d, "outputs", "*.TIMEOUT"),
os.path.join(d, "..", "outputs", "*.TIMEOUT")):
hits = glob.glob(pat)
if hits:
timeout_marker = ",".join(os.path.basename(h) for h in hits)
break
likely_truncated = (timeout_marker is not None or
(args.expected_niter and last_iter < args.expected_niter - 1 and not converged and stop_iter is None))
print(f"=== 收敛轨迹: {args.fort9} (chmax={args.chmax}" +
(f", teff={args.teff}" if args.teff else "") + ") ===")
print(f" 迭代次数: {len(iters)}(末次 ITER {last_iter}")
print(f" 末 max_relc: {last_max_relc:.4g}{'✅ < chmax 收敛' if converged else '❌ ≥ chmax 未收敛'}")
print(f" 末次深度数: {n_depths}")
if stop_iter is not None:
print(f" STOP: ❌ STOP in SOLVE after ITER {stop_iter}(求解器发散中止)")
else:
print(f" STOP: 无")
print(f" fort.9 NaN行: {nan9}")
if likely_truncated:
detail = f"(标记文件: {timeout_marker}" if timeout_marker else (
f"(仅 iter {last_iter}/{args.expected_niter},疑似 timeout 截断)" if args.expected_niter else "")
print(f" 截断: ⚠ 疑似被 timeout 杀死 {detail}——轨迹不完整,结论需谨慎")
# 轨迹(最多显示首/末若干 + 全部若 ≤15)
print(" 轨迹 (iter: max_relc):")
show = iters if len(iters) <= 20 else iters[:8] + [None] + iters[-5:]
for it in show:
if it is None:
print(" ...")
else:
print(f" iter {it[0]:>3}: {it[1]:.4g} ({it[2]} 深度)")
warns = detect_avalanche(iters)
if warns:
print(" ⚠ 雪崩告警:")
for w in warns:
print(w)
# 伪收敛启发式:max_relc 恰为 0 或极小且迭代很少 → 极可能是"在 NaN 模型上迭代"。
# 真实收敛要多次迭代、max_relc 趋近但极少精确到 0。
pseudo_warn = False
if last_max_relc == 0.0 or (len(iters) <= 2 and last_max_relc < 1e-10):
pseudo_warn = True
# 结论
real_converged = converged and stop_iter is None and nan9 == 0
print()
if real_converged:
print(f"✅ 数值收敛(末 max_relc {last_max_relc:.4g} < {args.chmax},无 STOP,无 NaN")
if pseudo_warn:
print(" ⚠⚠ 伪收敛嫌疑:max_relc≈0 且迭代极少,极可能是『在 NaN 模型上迭代』(NaN 上相对变化=0)。")
print(" 务必运行 check_temperature.py 查温度结构——若 .7 全 NaN,则此『收敛』无效。")
else:
print(" 注意:数值收敛 ≠ 物理收敛。请再用 check_temperature.py 查温度结构(防伪收敛)。")
return 0
print("❌ 未数值收敛:" if not real_converged else "")
reasons = []
if not converged:
reasons.append(f"末 max_relc {last_max_relc:.4g} ≥ chmax {args.chmax}")
if stop_iter is not None:
reasons.append(f"STOP in SOLVE after ITER {stop_iter}")
if nan9 > 0:
reasons.append(f"fort.9 有 {nan9} 行 NaN/坏值")
if warns:
reasons.append("迭代轨迹有雪崩")
for r in reasons:
print(f" - {r}")
return 1
if __name__ == "__main__":
sys.exit(main())
+161
View File
@@ -0,0 +1,161 @@
#!/usr/bin/env python3
"""
check_inputs.py — TLUSTY 测试执行前输入校验(冷启动链 / 种子链)
复刻 docs/testing_workflow_2026_08_11.md §五 与 references/tlusty-stages.md 的"关键不变量"
退出码:0=全过,1=有违规。
用法:
python3 check_inputs.py <inputs_dir> --chain cold --teff 60000
python3 check_inputs.py <inputs_dir> --chain seed --teff 55000
"""
import argparse
import os
import re
import sys
# NaN 模式(与 conv_check.rs:18 NAN_RE_PATTERN 一致)
NAN_RE = re.compile(r"(?i)(\bnan\b|\binf(?:inity)?\b|\*{3,}|[eE]\+(?:3\d{2}|[4-9]\d{2,}))")
def read_lines(path):
with open(path, "r", errors="replace") as f:
return f.read().splitlines()
def check_dot5_l2(path, expect_lte, expect_ltgray):
"""校验 .5 第2行 LTE/LTGRAY 两个标志。行首允许空格。"""
lines = read_lines(path)
if len(lines) < 2:
return f"{path}: 文件不足2行"
l2 = lines[1].split("!")[0].split()
if len(l2) < 2:
return f"{path}: L2 字段不足2个: {lines[1]!r}"
lte, ltgray = l2[0].upper(), l2[1].upper()
if lte != expect_lte or ltgray != expect_ltgray:
return f"{path}: L2 期望 {expect_lte} {expect_ltgray}, 实际 {lte} {ltgray} ({lines[1].strip()!r})"
return None
def check_dot5_nst_name(path):
"""校验 .5 第3行 'nst' 文件名。
约定:生产/测试默认 nst 名都是字面 'nst'run.sh 会在执行时把 {stage}.nst 复制成 nst。
所以 'nst' 不算违规(运行时会有);只有引用了非标准名且该名在 inputs/ 找不到时才报错。"""
lines = read_lines(path)
if len(lines) < 3:
return None
l3 = lines[2]
m = re.search(r"'([^']+)'", l3)
if not m:
return None
nst_name = m.group(1)
if nst_name == "nst":
return None # 标准名,run.sh 运行时把 {stage}.nst → nst,无需 inputs/ 里有
base = os.path.dirname(path)
candidates = [os.path.join(base, nst_name), os.path.join(base, "..", nst_name)]
if not any(os.path.exists(c) for c in candidates):
return f"{path}: L3 指定 nst 文件 {nst_name!r} 未找到(非标准名,run.sh 不会自动改名)"
return None
def check_nst_niter(path):
"""校验 nst 第1行含 NITER=。"""
lines = read_lines(path)
if not lines:
return f"{path}: 空文件"
if "NITER" not in lines[0].upper():
return f"{path}: L1 缺 NITER=: {lines[0][:80]!r}"
return None
def check_file_exists(path, what):
return None if os.path.exists(path) else f"缺输入: {what} ({path})"
def check_no_nan(path, what):
if not os.path.exists(path):
return f"缺输入: {what} ({path})"
n = 0
with open(path, "r", errors="replace") as f:
for line in f:
if NAN_RE.search(line):
n += 1
return None if n == 0 else f"{what}{n} 行 NaN/Inf/坏值 ({path})"
def main():
ap = argparse.ArgumentParser(description="TLUSTY 测试执行前输入校验")
ap.add_argument("inputs_dir", help="inputs/ 目录路径")
ap.add_argument("--chain", choices=["cold", "seed"], required=True)
ap.add_argument("--teff", type=float, default=None, help="TeffK),用于上下文,非必填")
args = ap.parse_args()
d = args.inputs_dir
violations = []
def add(v):
if v:
violations.append(v)
if args.chain == "cold":
# 必须有 lte/nc/nl 的 .5 与 .nst
for f in ["lte.5", "nc.5", "nl.5", "lte.nst", "nc.nst", "nl.nst"]:
add(check_file_exists(os.path.join(d, f), f))
# L2: lte=T T, nc/nl=F F
if os.path.exists(os.path.join(d, "lte.5")):
add(check_dot5_l2(os.path.join(d, "lte.5"), "T", "T"))
for s in ("nc", "nl"):
p = os.path.join(d, f"{s}.5")
if os.path.exists(p):
add(check_dot5_l2(p, "F", "F"))
# nst 名一致
for s in ("lte", "nc", "nl"):
p = os.path.join(d, f"{s}.5")
if os.path.exists(p):
add(check_dot5_nst_name(p))
# nst L1 含 NITER
for s in ("lte", "nc", "nl"):
p = os.path.join(d, f"{s}.nst")
if os.path.exists(p):
add(check_nst_niter(p))
# 冷启动链不应有 fort.8
if os.path.exists(os.path.join(d, "fort.8")):
violations.append("冷启动链不应有 inputs/fort.8lte 阶段 ltgray=T 会删它;若有种子请用 --chain seed")
else: # seed
for f in ["seed_nc.5", "seed_nc.nst", "fort.8"]:
add(check_file_exists(os.path.join(d, f), f))
add(check_no_nan(os.path.join(d, "fort.8"), "fort.8 种子"))
if os.path.exists(os.path.join(d, "seed_nc.5")):
add(check_dot5_l2(os.path.join(d, "seed_nc.5"), "F", "F"))
add(check_dot5_nst_name(os.path.join(d, "seed_nc.5")))
if os.path.exists(os.path.join(d, "seed_nc.nst")):
add(check_nst_niter(os.path.join(d, "seed_nc.nst")))
# data 软链:run.sh / new_test.sh 会在执行时创建,inputs/ 阶段可能还没有。
# 这里只做提示(note),不算硬违规。
notes = []
data_link = None
for cand in [os.path.join(d, "..", "run", "data"), os.path.join(d, "data")]:
if os.path.islink(cand) or os.path.isdir(cand):
data_link = cand
break
if data_link is None:
notes.append("data 软链(run/data)暂不存在——run.sh 执行时会自动创建并指向 assets/data/,可忽略")
# 汇报
print(f"=== 输入校验: {d} (chain={args.chain}" +
(f", teff={args.teff}" if args.teff else "") + ") ===")
for n in notes:
print(f" {n}")
if not violations:
print("✅ 全部通过")
return 0
for v in violations:
print(f"{v}")
print(f"\n{len(violations)} 项违规。请修正后再执行(见 references/tlusty-stages.md)。")
return 1
if __name__ == "__main__":
sys.exit(main())
@@ -0,0 +1,152 @@
#!/usr/bin/env python3
"""
check_temperature.py — 对 fort.7/{stage}.7 做温度结构物理判定。
精确复刻 DCTS 权威判据 check_temperature_structureconv_check.rs:463)。
退出码:0=物理(三项全过)/1=不物理 /2=文件缺失或无法解析(跳过,不判失败)。
判据(valid = 三项全过):
1. 表层 T < max_factor × Teff (默认 max_factor=3.0
2. 每个深度 T ∈ [temp_floor, temp_ceiling] (默认 [10, 1e8] K
3. 无 NaN/Inf
用法:
python3 check_temperature.py outputs/nl.7 --teff 60000
python3 check_temperature.py outputs/nc.7 --teff 55000 --max-factor 3.0
"""
import argparse
import math
import re
import sys
# 复刻 conv_check.rs:44 parse_fortran_float
# Fortran 指数≥100 时挤掉 E,如 -1.35E+118 → -1.35+118
_NO_E_EXP_RE = re.compile(r"^([+-]?[\d.]+)([+-]\d+)$")
def parse_fortran_float(s):
"""返回 float(可能为 inf/-inf);无法解析返回 None。"""
s = s.strip()
try:
return float(s)
except ValueError:
pass
m = _NO_E_EXP_RE.match(s)
if m:
try:
return float(f"{m.group(1)}E{m.group(2)}") # 含 inf/-inf(溢出)
except ValueError:
pass
return None
def check_temperature_structure(path, teff, max_factor=3.0, temp_floor=10.0, temp_ceiling=1e8):
"""复刻 conv_check.rs:463。返回 (temps, violations) 或 (None, None)=无法解析。"""
try:
with open(path, "r", errors="replace") as f:
lines = f.readlines()
except OSError:
return None, None
if not lines:
return None, None
# 1. 第1行 → nd, numpar
tokens = lines[0].split()
if len(tokens) < 2:
return None, None
try:
nd = int(float(tokens[0]))
numpar = int(float(tokens[1]))
except ValueError:
return None, None
if nd <= 0 or numpar <= 0:
return None, None
# 2. 行数计算
dm_lines = (nd + 5) // 6 # FORMAT 502: 6/行
block_lines = (numpar + 4) // 5 # FORMAT 503: 5/行
# 3. 跳过 DM 列阵
idx = 1 # 第1行已读
for _ in range(dm_lines):
if idx >= len(lines):
return None, None # 截断 → 跳过
idx += 1
# 4. nd 个深度块,每块第一行第一token = TEMP
temps = []
for _ in range(nd):
if idx >= len(lines):
break # 截断:已解析的深度保留
first_line_tokens = lines[idx].split()
if not first_line_tokens:
idx += block_lines
continue
t = parse_fortran_float(first_line_tokens[0])
temps.append(t if t is not None else float("nan"))
idx += block_lines # 跳到下一块(含本块剩余 block_lines-1 行)
if not temps:
return None, None
# 5. 三项判据
violations = []
surface_ratio = temps[0] / teff if teff > 0 else float("inf")
if surface_ratio > max_factor:
violations.append(
f"表层 T={temps[0]:.4g} > {max_factor}×Teff({teff})ratio={surface_ratio:.4g}"
)
for i, t in enumerate(temps):
if not math.isfinite(t):
violations.append(f"深度 {i+1} T={t}(非有限,NaN/Inf")
break
if t < temp_floor or t > temp_ceiling:
violations.append(
f"深度 {i+1} T={t:.4g} 越界 [{temp_floor}, {temp_ceiling}]"
)
break
return temps, violations
def main():
ap = argparse.ArgumentParser(description="fort.7 温度结构物理判定(DCTS 权威判据)")
ap.add_argument("fort7", help="fort.7 或 {stage}.7 路径")
ap.add_argument("--teff", type=float, required=True, help="Teff (K)")
ap.add_argument("--max-factor", type=float, default=3.0, help="表层/Teff 上限(默认 3.0")
ap.add_argument("--floor", type=float, default=10.0, dest="temp_floor", help="温度下限 K(默认 10")
ap.add_argument("--ceiling", type=float, default=1e8, dest="temp_ceiling", help="温度上限 K(默认 1e8")
args = ap.parse_args()
import os
if not os.path.exists(args.fort7):
print(f"⚠ 文件不存在: {args.fort7}(跳过,不判失败)")
return 2
temps, violations = check_temperature_structure(
args.fort7, args.teff, args.max_factor, args.temp_floor, args.temp_ceiling
)
if temps is None:
print(f"⚠ 无法解析 {args.fort7}(缺失/截断/格式错)—— 跳过,不判失败")
return 2
surface_ratio = temps[0] / args.teff if args.teff > 0 else float("inf")
print(f"=== 温度结构判定: {args.fort7} (Teff={args.teff}) ===")
print(f" 深度数: {len(temps)}")
print(f" 表层 T: {temps[0]:.4g} K ({surface_ratio:.4g}×TeffEddington 参考 0.84)")
print(f" 底层 T: {temps[-1]:.4g} K")
nan_count = sum(1 for t in temps if not math.isfinite(t))
print(f" NaN/Inf 深度: {nan_count}")
if not violations:
print(f"✅ 物理(三项全过:表层<{args.max_factor}×Teff、各深度∈[{args.temp_floor},{args.temp_ceiling}]、无 NaN")
return 0
print("❌ 不物理 —— 违规:")
for v in violations:
print(f" - {v}")
print("\n参考 references/physics-checks.md 的判据说明与修正历史。")
return 1
if __name__ == "__main__":
sys.exit(main())
+132
View File
@@ -0,0 +1,132 @@
#!/usr/bin/env python3
"""
diff_configs.py — A/B 测试配置对照工具。
给定 2 个或多个 content 目录(或 inputs 目录),解析各自的 nstL1/L2 的 KEY=VAL
与 .5 关键字段(L1 TEFF/GRAV、L2 LTE/LTGRAY、L3 nst 名),输出并排对照表,高亮差异。
用于 A/B 测试时快速确认"只有想改的参数变了,其余一致"
用法:
python3 diff_configs.py dirA/inputs dirB/inputs dirC/inputs
python3 diff_configs.py dirA dirB --teff 50000 # 传 content 目录也行(自动找 inputs/
"""
import argparse
import os
import re
import sys
# nst 一行里逗号分隔的 KEY=VAL(L1 与 L2 都是这种格式)
KV_RE = re.compile(r"(\w+)=([^,]+)")
def parse_nst(path):
"""返回 nst 的 {KEY: VAL} 字典(合并 L1+L2+escape-hatch 行)。"""
kv = {}
if not os.path.exists(path):
return kv
for line in open(path, "r", errors="replace"):
for m in KV_RE.finditer(line):
kv[m.group(1).upper()] = m.group(2).strip()
return kv
def parse_dot5(path):
"""返回 .5 关键字段。"""
info = {}
if not os.path.exists(path):
return info
lines = open(path, "r", errors="replace").read().splitlines()
if len(lines) >= 1:
t = lines[0].split("!")[0].split()
info["TEFF"] = t[0] if len(t) > 0 else "?"
info["GRAV"] = t[1] if len(t) > 1 else "?"
if len(lines) >= 2:
t = lines[1].split("!")[0].split()
info["LTE/LTGRAY"] = f"{t[0]}/{t[1]}" if len(t) >= 2 else "?"
if len(lines) >= 3:
m = re.search(r"'([^']+)'", lines[2])
info["nst名"] = m.group(1) if m else "?"
return info
def resolve_inputs(d):
if os.path.basename(d) == "inputs" and os.path.isdir(d):
return d
cand = os.path.join(d, "inputs")
return cand if os.path.isdir(cand) else d
def main():
ap = argparse.ArgumentParser(description="A/B 配置对照(nst 关键字段 + .5 关键行)")
ap.add_argument("dirs", nargs="+", help="inputs/ 或 content 目录")
ap.add_argument("--teff", default=None, help="仅用于显示标题")
args = ap.parse_args()
dirs = [resolve_inputs(d) for d in args.dirs]
names = [os.path.basename(os.path.dirname(d)) if os.path.basename(d) == "inputs"
else os.path.basename(d) for d in dirs]
# 收集所有出现过的 key(nst + .5 字段),分两组
nst_keys_order = ["ND", "NLAMBD", "VTB", "ISPODF", "DDNU", "CNU1", "CHMAX", "ITEK", "NITER",
"ORELAX", "IDLTE", "IACC", "ICHANG", "IELCOR", "DPSILG"]
dot5_keys = ["TEFF", "GRAV", "LTE/LTGRAY", "nst名"]
nst_maps = []
dot5_maps = []
for d in dirs:
# nst: 各阶段都有,优先 nc.nstA/B 通常比 nc/nl);也读 lte/nl
nst_path = None
for stage in ("nc", "nl", "lte", "seed_nc"):
p = os.path.join(d, f"{stage}.nst")
if os.path.exists(p):
nst_path = p
break
nst_maps.append((parse_nst(nst_path) if nst_path else {}, nst_path))
# .5:优先 nc.5
dot5_path = None
for stage in ("nc", "nl", "lte", "seed_nc"):
p = os.path.join(d, f"{stage}.5")
if os.path.exists(p):
dot5_path = p
break
dot5_maps.append((parse_dot5(dot5_path) if dot5_path else {}, dot5_path))
# 收集所有 nst key(含非标准的)
all_nst_keys = list(nst_keys_order)
seen = set(k.upper() for k in all_nst_keys)
for nm, _ in nst_maps:
for k in nm:
if k not in seen:
all_nst_keys.append(k)
seen.add(k)
def print_table(title, keys, maps, getter):
print(f"\n=== {title} ===")
hdr = " 字段".ljust(16) + "".join(n.ljust(22) for n in names)
print(hdr)
print(" " + "-" * (len(hdr) - 2))
for k in keys:
vals = [getter(nm, k) for nm, _ in maps]
# 高亮:若不全相同,标记 *
differ = len(set(repr(v) for v in vals)) > 1
mark = " * " if differ else " "
row = mark + k.ljust(13) + "".join((v if v else "").ljust(22) for v in vals)
print(row)
print(f"\n配置对照({len(names)} 组)" + (f"Teff={args.teff}" if args.teff else ""))
print(" * = 该行各组不一致(A/B 测试应只有你想改的字段带 *)")
for d, n, (nm, np_), (dm, dp_) in zip(dirs, names, nst_maps, dot5_maps):
print(f" [{n}] nst={os.path.basename(np_) or '(无)'} .5={os.path.basename(dp_) or '(无)'}")
print_table(".5 关键字段", dot5_keys, dot5_maps, lambda m, k: m.get(k, ""))
print_table("nst 关键参数(nc/nl 阶段)", all_nst_keys, nst_maps, lambda m, k: m.get(k.upper(), ""))
# 源文件路径(便于核对)
print("\n源文件:")
for d, n, (_, np_), (_, dp_) in zip(dirs, names, nst_maps, dot5_maps):
print(f" [{n}] {np_}")
return 0
if __name__ == "__main__":
sys.exit(main())
+121
View File
@@ -0,0 +1,121 @@
#!/usr/bin/env bash
# ============================================================
# new_test.sh — 脚手架:在 dcts/test/ 下创建规范的四级测试目录
#
# 用法:
# bash new_test.sh <test_id> <type> <grid> <content> [--chain cold|seed|synspec]
#
# 示例:
# bash new_test.sh 20260813_my_div conv t60000_g5.0_he-2_cold baseline --chain cold
# bash new_test.sh 20260813_seed conv t55000_g5.0_he-2_seed doubleoff --chain seed
# bash new_test.sh 20260813_spec synspec t50000_g5.0_cno gfato_100_20000 --chain synspec
#
# 创建:
# dcts/test/<test_id>/<type>/<grid>/<content>/{inputs,scripts,run,outputs}
# scripts/run.sh <- 从 assets/run_{chain}_chain.sh 拷贝
# README.md <- 骨架
# ============================================================
set -euo pipefail
# --- 自定位 skill 目录(脚本在 scripts/ 下,skill 根是父目录)---
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
SKILL_DIR="$(cd "$SCRIPT_DIR/.." && pwd)"
ASSETS="$SKILL_DIR/assets"
# --- 找 dcts 根(含 assets/tlusty_static 的目录)---
# 优先用环境变量;否则从 skill 目录向上找第一个含 assets/tlusty_static 的祖先
DCTS_ROOT="${DCTS_ROOT:-}"
if [ -z "$DCTS_ROOT" ]; then
cur="$SCRIPT_DIR"
while [ "$cur" != "/" ]; do
if [ -f "$cur/assets/tlusty_static" ]; then DCTS_ROOT="$cur"; break; fi
cur="$(dirname "$cur")"
done
fi
TEST_ROOT="$DCTS_ROOT/test"
if [ -z "$DCTS_ROOT" ] || [ ! -f "$DCTS_ROOT/assets/tlusty_static" ]; then
echo "❌ 未找到 dcts 根(含 assets/tlusty_static 的目录)。用 DCTS_ROOT=... 环境变量指定。" >&2
exit 1
fi
# --- 解析参数 ---
if [ $# -lt 4 ]; then
sed -n '3,20p' "$0" | sed 's/^# \{0,1\}//' >&2
exit 1
fi
TEST_ID="$1"; TYPE="$2"; GRID="$3"; CONTENT="$4"; shift 4
CHAIN="cold"
while [ $# -gt 0 ]; do
case "$1" in
--chain) CHAIN="$2"; shift 2;;
*) echo "未知参数: $1" >&2; exit 1;;
esac
done
case "$CHAIN" in
cold) RUN_TEMPLATE="$ASSETS/run_cold_chain.sh"; STAGES="lte → nc → nl";;
seed) RUN_TEMPLATE="$ASSETS/run_seed_chain.sh"; STAGES="seed_nc → nl";;
synspec) RUN_TEMPLATE="$ASSETS/run_synspec.sh"; STAGES="SYNSPEC";;
*) echo "❌ --chain 必须是 cold|seed|synspec,收到: $CHAIN" >&2; exit 1;;
esac
# --- 创建四级目录 ---
TARGET="$TEST_ROOT/$TEST_ID/$TYPE/$GRID/$CONTENT"
if [ -e "$TARGET" ]; then
echo "❌ 目录已存在: $TARGET" >&2
exit 1
fi
mkdir -p "$TARGET/inputs" "$TARGET/scripts" "$TARGET/run" "$TARGET/outputs"
# --- 拷 run.sh 模板 ---
cp "$RUN_TEMPLATE" "$TARGET/scripts/run.sh"
chmod +x "$TARGET/scripts/run.sh"
# --- 建 data 软链到 run/TLUSTY/SYNSPEC 都要)---
ln -sfn "$DCTS_ROOT/assets/data" "$TARGET/run/data"
# --- README 骨架 ---
cat > "$TARGET/README.md" <<EOF
# $TEST_ID / $TYPE / $GRID / $CONTENT
## 目的
TODO: 一句话说清这次测试想验证什么。
## 配置
- 二进制: $DCTS_ROOT/assets/$([ "$CHAIN" = synspec ] && echo synspec || echo tlusty)_static
- 网格: $GRID
- 链: $STAGES
- 变体: TODO(与 baseline 不同的关键参数,如 ITEK/IACC/DPSILG/ORELAX/NITER
## 输入
TODO: inputs/ 生成方式(代码生成 / 模板修改,留 diff);关键参数。
- [ ] 见 references/directory-convention.md 的执行前检查清单
## 执行
- 命令: cd "$TARGET" && bash scripts/run.sh
- 各阶段: TODO
## 结果
TODO: 见 outputs/RESULT.md(用 assets/result_template.md
## 审查结论
- 数值: TODOmax_relc 是否达标;STOP/雪崩?)
- 物理: TODO(温度结构三项:表层<3×Teff、各深度∈[10,1e8]、无 NaN
- 结论: TODO
EOF
echo "✅ 已创建测试目录: $TARGET"
echo " 链类型: $CHAIN ($STAGES)"
echo ""
echo "下一步:"
echo " 1. 把输入文件放进 $TARGET/inputs/"
echo " 2. 执行前校验(TLUSTY 链):"
echo " python3 $SKILL_DIR/scripts/check_inputs.py $TARGET/inputs --chain $CHAIN --teff <Teff>"
echo " 3. 跑: cd $TARGET && bash scripts/run.sh"
echo " 4. 事后审查:"
if [ "$CHAIN" != "synspec" ]; then
echo " python3 $SKILL_DIR/scripts/check_fort9.py $TARGET/outputs/nl.9 --teff <Teff>"
echo " python3 $SKILL_DIR/scripts/check_temperature.py $TARGET/outputs/nl.7 --teff <Teff>"
else
echo " grep -ciE 'nan|inf|\\*{3,}' $TARGET/outputs/*.spec # 应为 0"
fi
@@ -0,0 +1,120 @@
#!/usr/bin/env python3
"""
stage_transform.py — 在冷启动链各阶段的 .5 输入文件之间做"阶段变换"
主要解决 nc.5 → nl.5 的 ilvlin 改写(这是手写最容易踩坑的一步:awk 会吞掉引号
里的空格、朴素正则会吞掉换行)。本脚本逐行**字节级**保留原格式,只改 ion 行的
ilvlin 列。
规则(见 references/tlusty-stages.md "ilvlin 规则":
- nc 阶段 (ilvlin=0):所有离子 ilvlin=0
- nl 阶段 (ilvlin=100):非裸核 (nlevs>1) → 100;裸核 (nlevs==1) 永远 0
- lte 阶段:同 ncilvlin=0),但 .5 第2行 LTE/LTGRAY = T T(本工具也支持改 L2)
ion 行识别:行首空白 + 6 个整数字段 (iat iz nlevs ilast ilvlin nonstd) + 字符串。
第 3 个字段 = nlevs,第 5 个 = ilvlin。
用法:
# nc.5 → nl.5(非裸核 ilvlin 0→100
python3 stage_transform.py nc.5 -o nl.5 --to nl
# nl.5 → nc.5(所有 ilvlin→0+ 第2行改 F F
python3 stage_transform.py nl.5 -o nc.5 --to nc
# 仅打印 diff,不写文件
python3 stage_transform.py nc.5 --to nl --diff
"""
import argparse
import re
import sys
# ion data 行:行首空白 + iat iz nlevs ilast ilvlin nonstd + 后续字符串
# 用分组保留各组间空白,只重写 ilvlin(第5个整数),字节级安全。
ION_RE = re.compile(
r"^(?P<lead>\s+)(?P<iat>\d+)(?P<s1>\s+)(?P<iz>\d+)(?P<s2>\s+)"
r"(?P<nlevs>\d+)(?P<s3>\s+)(?P<ilast>\d+)(?P<s4>\s+)"
r"(?P<ilvlin>\d+)(?P<rest>\s+\d+\s+.*)$"
)
# .5 第2行 LTE/LTGRAY(行首允许空格)
L2_RE = re.compile(r"^(\s*)(\S+)(\s+)(\S+)(\s*.*)$")
def transform_ions(text, target_ilvlin_for_nonbare):
"""改写 ion 行的 ilvlin。target_ilvlin_for_nonbare: nl=100, nc/lte=0。返回新文本。"""
out = []
changed = 0
bare_kept = 0
for line in text.splitlines(keepends=True):
m = ION_RE.match(line.rstrip("\n"))
if m:
nlevs = int(m.group("nlevs"))
if nlevs == 1:
new_ilvlin = "0" # 裸核永远 0
bare_kept += 1
else:
new_ilvlin = str(target_ilvlin_for_nonbare)
if m.group("ilvlin") != new_ilvlin:
changed += 1
rebuilt = (m.group("lead") + m.group("iat") + m.group("s1") + m.group("iz") +
m.group("s2") + m.group("nlevs") + m.group("s3") + m.group("ilast") +
m.group("s4") + new_ilvlin + m.group("rest") + "\n")
out.append(rebuilt)
else:
out.append(line)
return "".join(out), changed, bare_kept
def transform_l2(text, lte_flag, ltgray_flag):
"""改写 .5 第2行的 LTE/LTGRAY 两个标志。"""
lines = text.splitlines(keepends=True)
if len(lines) < 2:
return text, False
m = L2_RE.match(lines[1].rstrip("\n"))
if not m or not m.group(2).upper() in ("T", "F"):
return text, False
rebuilt = (m.group(1) + lte_flag + m.group(3) + ltgray_flag + m.group(5) + "\n")
if rebuilt != lines[1]:
lines[1] = rebuilt
return "".join(lines), True
return text, False
def main():
ap = argparse.ArgumentParser(description="冷启动链 .5 阶段变换(ilvlin / L2")
ap.add_argument("input", help="输入 .5 文件(如 nc.5")
ap.add_argument("-o", "--output", help="输出文件(省略则只打印摘要/diff")
ap.add_argument("--to", choices=["nl", "nc", "lte"], required=True,
help="目标阶段:nl=非裸核 ilvlin→100nc/lte=所有 ilvlin→0")
ap.add_argument("--diff", action="store_true", help="打印改动行的 diff(需系统带 diff)")
args = ap.parse_args()
with open(args.input, "r", errors="replace") as f:
text = f.read()
target = 100 if args.to == "nl" else 0
new_text, ion_changed, bare = transform_ions(text, target)
l2_changed = False
if args.to == "lte":
new_text, l2_changed = transform_l2(new_text, "T", "T")
elif args.to in ("nc", "nl"):
new_text, l2_changed = transform_l2(new_text, "F", "F")
print(f"=== stage_transform: {args.input} → 阶段 {args.to} ===")
print(f" ion 行 ilvlin 改写: {ion_changed} 行(裸核 nlevs==1 保持 0: {bare} 行)")
print(f" 第2行 LTE/LTGRAY 改写: {'' if l2_changed else '否(已是目标值)'}")
if args.diff:
import difflib
for line in difflib.unified_diff(text.splitlines(), new_text.splitlines(),
fromfile=args.input, tofile=f"({args.to})", lineterm=""):
print(" " + line)
if args.output:
with open(args.output, "w") as f:
f.write(new_text)
print(f"✅ 已写出 {args.output}")
else:
print("(未指定 -o,仅打印摘要。加 --diff 看改动详情。)")
return 0
if __name__ == "__main__":
sys.exit(main())