cbb
0b47905a1c
- build_dataset.cjs：每 10% 打 `[prep] shard_xxxx a/b`（只加日志，不改计算）。新增广 （40%/3°/12，每个增广样本 6 次碰撞检查）让这批打包从上次的 25 分钟涨到约 1 小时， 而原来整片完成前一行输出都没有，看起来像卡死。 - monitor.py prep 分支：解析这些行，按 (已完成分片 + 进行中分片的任务比例)/总分片 算 frac， 按实测速率外推 eta；首批完成前显示"首波处理中，第一批完成后显示预估"。 - pstate/monitor.html：prep 估时 0.5 h → 2.0 h（超时 1.5 h → 6 h），与实际耗时一致。
657 lines
39 KiB
Python
657 lines
39 KiB
Python
"""训练控制台（本地网页，零第三方依赖，离线可用）：一键开始全流程、看进度、可停止。
|
||
双击 ml/console.bat（Windows）或运行 bash ml/console.sh（Linux/WSL）→ 自动打开 http://localhost:8100
|
||
也可：python -m ml.monitor [--port 8100] [--host 0.0.0.0] [--open]
|
||
网页上点「开始训练」→ 选参数 → 后台执行 bash ml/pipeline.sh all（已完成的阶段自动跳过 = 续跑；勾「从头重跑」= FRESH=1），
|
||
输出写 ml/runs/console.log，网页实时显示；「停止」结束整个进程树。网页和控制台窗口关掉都不影响训练，重新打开即可继续看。
|
||
只读数据来源：
|
||
ml/runs/pipeline.json 各阶段开始/结束/返回码（pipeline.sh 自动写）→ 全流程进度
|
||
ml/runs/*/log.jsonl 预训练（base / wta / bench_*）与 APG（apg / apg_r*）逐步日志与闭环评估
|
||
ml/runs/m5_gate_*.json、mine/r*/report.json + gate.json、policy_bench.json、acceptance.json
|
||
ml/data/stats.json / consistency.json / r2.json / samples/index.json、nvidia-smi
|
||
安全：默认只监听 127.0.0.1；--host 0.0.0.0 时其他机器只能看，开始/停止只接受本机请求（--remote-control 放开）。
|
||
2026-10-18 新增（用户要求：训练全程可视化，并在网页上点按钮一键开始）。
|
||
2026-10-18 第五轮（审查第四轮）：
|
||
N-7 是否在跑看 ml/runs/pipeline.lock（与终端同一把锁），不再靠"日志 120 s 内有更新"去猜；
|
||
N-8 开始/停止要求自定义头 X-Console + 同源 Origin/Sec-Fetch-Site，外站网页发不进来；
|
||
N-10 启动自检（训练用 Python 能否 import torch、GPU、node、bash），不过就拒绝「开始训练」并在顶栏标红；
|
||
N-11 Windows 子进程用 CREATE_NO_WINDOW 自带隐藏控制台：关掉控制台黑窗口训练照跑，重开控制台可接着看、可停止；
|
||
N-9 生成阶段的行数按文件增量统计（只读新增字节），其他阶段不扫 shard；
|
||
N-12 停止后的状态改写走 ml.pstate（单一写者、原子替换）；N-14 阶段名/估时/权重由 ml.pstate.STAGES 下发。"""
|
||
import argparse, glob, json, os, platform, re, shutil, signal, subprocess, sys, threading, time, webbrowser
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
from ml import pstate
|
||
from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
|
||
from urllib.parse import urlparse, parse_qs
|
||
|
||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
HTML = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'monitor.html')
|
||
RUNS, DATA = 'ml/runs', 'ml/data'
|
||
|
||
|
||
def jload(p):
|
||
try:
|
||
with open(p, encoding='utf-8') as f: return json.load(f)
|
||
except Exception: return None
|
||
|
||
|
||
def kind_of(name):
|
||
if name.startswith('apg'): return 'apg'
|
||
if name.startswith('bench'): return 'bench'
|
||
return 'train'
|
||
|
||
|
||
def list_runs():
|
||
out = []
|
||
for f in glob.glob(os.path.join(ROOT, RUNS, '*', 'log.jsonl')):
|
||
d = os.path.basename(os.path.dirname(f)); st = os.stat(f)
|
||
out.append({'name': d, 'kind': kind_of(d), 'size': st.st_size, 'mtime': st.st_mtime,
|
||
'has_best': os.path.exists(os.path.join(os.path.dirname(f), 'best.pt')),
|
||
'diverged': os.path.exists(os.path.join(os.path.dirname(f), 'diverged.pt'))})
|
||
out.sort(key=lambda r: -r['mtime'])
|
||
return out
|
||
|
||
|
||
def read_log(name, offset):
|
||
if not re.fullmatch(r'[\w.\-]+', name or ''): return {'error': 'bad run'}
|
||
f = os.path.join(ROOT, RUNS, name, 'log.jsonl')
|
||
if not os.path.exists(f): return {'error': 'no log'}
|
||
size = os.path.getsize(f)
|
||
if offset > size: offset = 0 # 日志被新训练覆盖 → 从头读
|
||
recs = []
|
||
with open(f, 'rb') as fh:
|
||
fh.seek(offset); buf = fh.read(min(size - offset, 64 << 20))
|
||
end = buf.rfind(b'\n') + 1 # 只取完整行，半行留到下次
|
||
for line in buf[:end].decode('utf-8', 'replace').splitlines():
|
||
if line.strip():
|
||
try: recs.append(json.loads(line))
|
||
except Exception: pass
|
||
return {'records': recs, 'offset': offset + end, 'size': size, 'reset': offset == 0, 'mtime': os.path.getmtime(f)}
|
||
|
||
|
||
_CPU = {'v': {}}
|
||
|
||
|
||
def cpu_name():
|
||
"""CPU 型号：psutil 只有核数，型号得自己读——Windows 在注册表，Linux 在 /proc/cpuinfo。"""
|
||
try:
|
||
if os.name == 'nt':
|
||
import winreg
|
||
with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, r'HARDWARE\DESCRIPTION\System\CentralProcessor\0') as k:
|
||
return re.sub(r'\s+', ' ', winreg.QueryValueEx(k, 'ProcessorNameString')[0]).strip()
|
||
with open('/proc/cpuinfo') as f:
|
||
for ln in f:
|
||
if ln.startswith('model name'):
|
||
return ln.split(':', 1)[1].strip()
|
||
except Exception:
|
||
pass
|
||
return platform.processor().strip()
|
||
|
||
|
||
class CpuSampler(threading.Thread):
|
||
"""后台每秒阻塞采样一次 CPU。cpu_percent 是"距上次调用"的差值：放在请求里算，窗口取决于两次
|
||
HTTP 请求的间隔（网页 1 秒轮询 + 手动 curl 会把它压成几毫秒，读数恒为 0）——所以单独一个线程按固定
|
||
1 秒窗口取，status 只读最近一次。"""
|
||
def __init__(self): super().__init__(daemon=True); self.start()
|
||
|
||
def run(self):
|
||
try:
|
||
import psutil
|
||
name = cpu_name() # 型号只读一次，别每秒翻注册表
|
||
psutil.cpu_percent(interval=None, percpu=True) # 打桩：第一次恒为 0，丢掉
|
||
while True:
|
||
per = psutil.cpu_percent(interval=1.0, percpu=True)
|
||
_CPU['v'] = {'total': round(sum(per) / max(1, len(per)), 1), 'per': [round(x) for x in per], 'n': len(per), 'model': name}
|
||
except Exception:
|
||
_CPU['v'] = {}
|
||
|
||
|
||
def cpu():
|
||
return {**_CPU['v'], **mahm()}
|
||
|
||
|
||
_MAHM = {'t': 0.0, 'v': {}}
|
||
|
||
|
||
def mahm():
|
||
"""借 MSI Afterburner 的共享内存（MAHMSharedMemory）读 CPU 温度 / 封装功耗——Windows 上没有别的免驱动口子：
|
||
psutil 在 Windows 没有 sensors_temperatures；ACPI 热区只有个 27.9 °C 的板温；Power Meter 计数器没有实例。
|
||
条目布局实测：头 32 字节（签名 MAHM / 版本 / 头大小 / 条目数 / 条目大小 / 时间戳），每条 1324 字节，
|
||
名字在前 260 字节、单位在其后 260、数值是偏移 1300 的 float（用 GPU1 power 对 nvidia-smi 校准过）。
|
||
读数比 30 秒旧才当没读到：Afterburner 的硬件轮询周期实测 10~13 秒（不是 1 秒），阈值卡 10 秒会
|
||
正好落在两次刷新之间、把还在用的读数丢掉（表现为温度和功耗一闪一闪地消失）。"""
|
||
now = time.time()
|
||
if now - _MAHM['t'] < 1.0: return _MAHM['v']
|
||
_MAHM['t'] = now
|
||
out = {}
|
||
try:
|
||
import ctypes, struct
|
||
k = ctypes.windll.kernel32
|
||
k.OpenFileMappingW.restype = ctypes.c_void_p
|
||
k.OpenFileMappingW.argtypes = [ctypes.c_uint32, ctypes.c_int, ctypes.c_wchar_p]
|
||
k.MapViewOfFile.restype = ctypes.c_void_p
|
||
k.MapViewOfFile.argtypes = [ctypes.c_void_p, ctypes.c_uint32, ctypes.c_uint32, ctypes.c_uint32, ctypes.c_size_t]
|
||
h = k.OpenFileMappingW(0x0004, 0, 'MAHMSharedMemory')
|
||
p = k.MapViewOfFile(h, 0x0004, 0, 0, 0) if h else 0
|
||
if p:
|
||
hd = ctypes.string_at(p, 32)
|
||
num, esize, hsize = (int.from_bytes(hd[i * 4:i * 4 + 4], 'little') for i in (3, 4, 2))
|
||
upd = int.from_bytes(hd[20:24], 'little')
|
||
if 0 < num < 5000 and 32 < esize < 65536 and now - upd <= 30:
|
||
buf = ctypes.string_at(p, hsize + num * esize)
|
||
core = {}
|
||
for i in range(num):
|
||
e = buf[hsize + i * esize: hsize + (i + 1) * esize]
|
||
nm = e[:260].split(b'\x00')[0].decode('latin1')
|
||
if not nm: continue
|
||
val = struct.unpack('<f', e[1300:1304])[0]
|
||
if nm == 'CPU temperature' and 0 < val < 130: out['temp'] = round(val)
|
||
elif nm == 'CPU power' and 0 < val < 1000: out['power'] = round(val, 1)
|
||
elif nm.startswith('CPU') and nm.endswith(' temperature'):
|
||
try:
|
||
j = int(nm[3:-12])
|
||
if 0 < val < 130: core[j] = round(val)
|
||
except ValueError: pass
|
||
if core: out['coreT'] = [core[k] for k in sorted(core)]
|
||
except Exception:
|
||
out = {}
|
||
_MAHM['v'] = out
|
||
return out
|
||
|
||
|
||
_GPU_CACHE = {'t': 0.0, 'v': []}
|
||
|
||
|
||
def gpu():
|
||
# 网页 1 秒轮询：nvidia-smi 每次要 150–600 ms，缓存 2 秒（GPU 数字不需要更快），避免每轮都起子进程
|
||
if time.time() - _GPU_CACHE['t'] < 2.0: return _GPU_CACHE['v']
|
||
try:
|
||
q = 'name,utilization.gpu,memory.used,memory.total,temperature.gpu,power.draw'
|
||
r = subprocess.run(['nvidia-smi', f'--query-gpu={q}', '--format=csv,noheader,nounits'], capture_output=True, text=True, timeout=3)
|
||
out = []
|
||
for line in r.stdout.strip().splitlines():
|
||
n, u, mu, mt, t, p = [x.strip() for x in line.split(',')]
|
||
num = lambda x: float(x) if re.fullmatch(r'[\d.]+', x) else None
|
||
out.append({'name': n, 'util': num(u), 'mem_used': num(mu), 'mem_total': num(mt), 'temp': num(t), 'power': num(p)})
|
||
except Exception:
|
||
out = []
|
||
_GPU_CACHE.update(t=time.time(), v=out)
|
||
return out
|
||
|
||
|
||
def last_rec(name):
|
||
"""读日志最后一条非评估记录（只读文件尾 16 KB）。"""
|
||
f = os.path.join(ROOT, RUNS, name, 'log.jsonl')
|
||
try:
|
||
with open(f, 'rb') as fh:
|
||
fh.seek(0, 2); n = fh.tell(); fh.seek(max(0, n - 16384)); tail = fh.read().decode('utf-8', 'replace').splitlines()
|
||
for line in reversed(tail):
|
||
try:
|
||
r = json.loads(line)
|
||
if 'eval' not in r: return r
|
||
except Exception: pass
|
||
except Exception: pass
|
||
return None
|
||
|
||
|
||
class LineCount:
|
||
"""按文件增量数行：记住每个文件已读到的偏移与行数，只读新增字节（审查 N-9：此前每 5 s 全量重读全部 shard）。"""
|
||
def __init__(self): self.c = {}; self.lock = threading.Lock()
|
||
|
||
def count(self, f):
|
||
with self.lock:
|
||
try: size = os.path.getsize(f)
|
||
except OSError: return 0
|
||
off, n = self.c.get(f, (0, 0))
|
||
if size < off: off, n = 0, 0
|
||
if size > off:
|
||
with open(f, 'rb') as fh:
|
||
fh.seek(off)
|
||
while True:
|
||
b = fh.read(1 << 22)
|
||
if not b: break
|
||
n += b.count(b'\n'); off += len(b)
|
||
self.c[f] = (off, n)
|
||
return n
|
||
|
||
|
||
LC = LineCount()
|
||
|
||
|
||
def newer(pattern, t0):
|
||
return [f for f in glob.glob(os.path.join(ROOT, pattern)) if os.path.getmtime(f) >= t0 - 1]
|
||
|
||
|
||
def stage_progress(st, s):
|
||
"""当前阶段的进度：返回 {'frac': 0..1 或 None（无法计量）, 'label': 文字, 'eta_s': 按实测速率外推的剩余秒数或 None}。"""
|
||
t0, info = s['start'], s.get('info') or {}
|
||
if st == 'gen':
|
||
# 分子算全部（含之前轮次已完成的分片），分母是累计目标：补量时显示 50,008/79,688 而不是 0/79,688。
|
||
# 2026-10-18 实测：只统计本轮开始后改过的文件会让"补量"看起来像从零开始。
|
||
allf = glob.glob(os.path.join(ROOT, DATA, 'shard_*.jsonl')) + glob.glob(os.path.join(ROOT, DATA, 'shard_*.jsonl.part'))
|
||
n = sum(LC.count(f) for f in allf)
|
||
tgt = info.get('target_tasks') or 0
|
||
new = sum(LC.count(f) for f in newer(f'{DATA}/shard_*.jsonl', t0) + newer(f'{DATA}/shard_*.jsonl.part', t0))
|
||
# 剩余时间按"本轮已写条数 ÷ 本轮已用时"外推（条数是实时数出来的，速率就是实测的），不再用静态估时
|
||
el = max(1.0, time.time() - t0)
|
||
rate = new / el if new > 0 else 0
|
||
eta = ((tgt - n) / rate if rate > 0 else None) if tgt > n else (0.0 if rate > 0 else None)
|
||
lab = f'已生成 {n:,} / {tgt:,} 条任务' + (f'（本轮补 {new:,} 条 · {rate * 60:.0f} 条/分）' if 0 < new < n else '')
|
||
return {'frac': min(1, n / tgt) if tgt else None, 'label': lab, 'eta_s': eta}
|
||
if st == 'gate1':
|
||
# 一致性门禁每 10 个任务往 stderr 打一行 "k/300"（consistency.cjs:41）。进度只读控制台日志最后 64 KB，
|
||
# 不跑任何计算、不占 CPU —— 正在生成的节点不受影响。日志里可能还留着上一次门禁的进度行，
|
||
# 所以只数最后一个 "consistency.cjs" 命令行之后的部分。
|
||
txt = console(0).get('text') or ''
|
||
cut = txt.rfind('consistency.cjs')
|
||
m = re.findall(r'^(\d+)/(\d+)\s*$', txt[cut:] if cut >= 0 else '', re.M)
|
||
n, N = (int(m[-1][0]), int(m[-1][1])) if m else (0, 300)
|
||
el = time.time() - t0
|
||
rate = n / el if n > 0 and el > 2 else 0
|
||
return {'frac': n / N if N else None,
|
||
'label': f'门禁 {n} / {N} 个任务' + (f' · {rate * 60:.0f} 个/分' if rate else '') if n else '准备中（日志里还没有进度行）',
|
||
'eta_s': (N - n) / rate if rate > 0 else None}
|
||
if st == 'prep':
|
||
nsh = len(glob.glob(os.path.join(ROOT, DATA, 'shard_*.jsonl')))
|
||
fin = newer(f'{DATA}/samples/*.X.npy', t0)
|
||
done = len(fin)
|
||
if os.path.exists(os.path.join(ROOT, DATA, 'r2.json')) and newer(f'{DATA}/r2.json', t0): return {'frac': 1, 'label': '打包完成，R2 检查完成', 'eta_s': 0}
|
||
# 2026-10-19：build_dataset 每 10% 打一行 [prep] shard_xxxx a/b；用它把"整片完成"之间的进度和 ETA 算出来
|
||
fin_n = {os.path.basename(p).split('.')[0] for p in fin}
|
||
part = {}
|
||
for ln in console(0)['text'].splitlines()[-800:]:
|
||
m = re.search(r'\[prep\] (shard_\d+) (\d+)/(\d+)', ln)
|
||
if m: part[m.group(1)] = int(m.group(2)) / max(1, int(m.group(3)))
|
||
run = {k: v for k, v in part.items() if k not in fin_n}
|
||
frac = (done + sum(run.values())) / nsh if nsh else None
|
||
el = time.time() - t0
|
||
eta = el * (1 - frac) / frac if frac and 0.02 < frac < 1 and el > 5 else None
|
||
lab = f'已打包 {done} / {nsh} 个分片'
|
||
if run: lab += f' · 进行中 {len(run)} 个（平均 {100 * sum(run.values()) / len(run):.0f}%）'
|
||
elif done == 0: lab += ' · 首波处理中，第一批完成后显示预估'
|
||
elif nsh and done >= nsh: lab += '，R2 检查中'
|
||
return {'frac': min(0.95, frac) if frac is not None else None, 'label': lab, 'eta_s': eta}
|
||
if st == 'train':
|
||
tag, bs = info.get('tag', 'base'), info.get('bench_steps', 2000)
|
||
b, m = last_rec(f'bench_{tag}') if newer(f'{RUNS}/bench_{tag}/log.jsonl', t0) else None, last_rec(tag) if newer(f'{RUNS}/{tag}/log.jsonl', t0) else None
|
||
if m and m.get('total'):
|
||
return {'frac': (bs + m['step']) / (bs + m['total']), 'label': f'正式训练 {m["step"]:,} / {m["total"]:,} 步', 'run': tag,
|
||
'eta_s': m['eta_h'] * 3600 if m.get('eta_h') else None} # 训练脚本自己按实测样本/s 算的
|
||
if b: return {'frac': b['step'] / (bs + 60000), 'label': f'测速 {b["step"]:,} / {bs:,} 步', 'run': f'bench_{tag}'}
|
||
return {'frac': 0, 'label': '加载数据中'}
|
||
if st == 'dagger': # 2026-10-18 新增：每轮 = 采起点 → 老师重规划 → 打包 → 微调（微调占大头，进度按 0.1/0.25/0.05/0.6 分）
|
||
tag, R = info.get('tag', 'base'), int(info.get('rounds', 2))
|
||
k = len(newer(f'{RUNS}/{tag}_dag*/done', t0)); k = min(k, R)
|
||
base = f'第 {min(k + 1, R)} / {R} 轮'
|
||
if k >= R: return {'frac': 1, 'label': 'DAgger 完成'}
|
||
D, run = f'{DATA}/dag_{tag}_r{k + 1}', f'{tag}_dag{k + 1}'
|
||
m = last_rec(run) if newer(f'{RUNS}/{run}/log.jsonl', t0) else None
|
||
if m and m.get('total'):
|
||
return {'frac': (k + 0.4 + 0.6 * m['step'] / m['total']) / R, 'label': f'{base}：微调 {m["step"]:,} / {m["total"]:,} 步', 'run': run,
|
||
'eta_s': m['eta_h'] * 3600 if m.get('eta_h') else None}
|
||
if os.path.exists(os.path.join(ROOT, D, 'collect.json')):
|
||
nt = sum(LC.count(f) for f in glob.glob(os.path.join(ROOT, D, 'tasks_*.jsonl')))
|
||
nd = sum(LC.count(f) for f in glob.glob(os.path.join(ROOT, D, 'shard_dag_*.jsonl')) + glob.glob(os.path.join(ROOT, D, 'shard_dag_*.jsonl.part')))
|
||
return {'frac': (k + 0.1 + 0.25 * (nd / nt if nt else 0)) / R, 'label': f'{base}：老师重新规划 {nd:,} / {nt:,} 个起点'}
|
||
return {'frac': k / R, 'label': f'{base}：闭环采起点'}
|
||
if st == 'apg':
|
||
r = last_rec('apg') if newer(f'{RUNS}/apg/log.jsonl', t0) else None
|
||
eta = None
|
||
if r and r.get('total'):
|
||
it, mn = r.get('it') or 0, r.get('min') or 0
|
||
eta = (r['total'] - it) / (it / mn) * 60 if it > 0 and mn > 0.1 else None # 步/分实测外推
|
||
return {'frac': r['it'] / r['total'] if r and r.get('total') else 0, 'label': f'APG {r["it"]:,} / {r["total"]:,} 步' if r else '准备中', 'run': 'apg', 'eta_s': eta}
|
||
if st == 'mine':
|
||
R = info.get('rounds', 6)
|
||
k = len(newer(f'{RUNS}/mine/r*/gate.json', t0))
|
||
cur = sorted(newer(f'{RUNS}/apg_r*/log.jsonl', t0), key=os.path.getmtime)
|
||
lab = f'第 {k + 1} 轮：挖错出题'
|
||
if cur:
|
||
name = os.path.basename(os.path.dirname(cur[-1])) # ml/runs/apg_rN
|
||
rn = re.sub(r'\D', '', name) or '1'
|
||
gate = os.path.join(ROOT, RUNS, 'mine', f'r{rn}', 'gate.json')
|
||
r = last_rec(name)
|
||
if r and r.get('total') and not os.path.exists(gate): # 这一轮还在跑（还没出 gate）
|
||
frac_in = 0.3 + 0.6 * r['it'] / r['total']
|
||
lab = f'第 {k + 1} 轮：APG {r["it"]:,} / {r["total"]:,} 步'
|
||
it, mn = r.get('it') or 0, r.get('min') or 0
|
||
eta = (r['total'] - it) / (it / mn) * 60 if it > 0 and mn > 0.1 else None
|
||
if eta is not None and k >= 1: # 剩余轮次按已完成轮次的平均时长外推
|
||
eta += max(0, R - k - 1) * (time.time() - t0) / k
|
||
return {'frac': min(1, (k + frac_in) / R), 'label': f'{lab}（共最多 {R} 轮，提升不足会提前停）', 'run': name, 'eta_s': eta}
|
||
return {'frac': min(1, k / R), 'label': f'{lab}（共最多 {R} 轮，提升不足会提前停）'}
|
||
if st == 'm5':
|
||
tag = info.get('tag', 'base')
|
||
n = len(newer(f'{RUNS}/m5_{tag}_policy.json', t0)) + len(newer(f'{RUNS}/m5_{tag}_safety.json', t0))
|
||
el = time.time() - t0
|
||
eta = (2 - n) * (el / n) if n == 1 else (0.0 if n >= 2 else None) # 两次评估（纯策略 / 加安全层）按已完成的平均耗时外推
|
||
return {'frac': n / 2, 'label': ['纯策略评估中', '加安全层评估中', '汇总'][min(n, 2)], 'eta_s': eta}
|
||
return {'frac': None, 'label': '进行中'}
|
||
|
||
|
||
def stage_elapsed(pl, now=None):
|
||
"""每个阶段的累计用时（秒）：pipeline.json 的 time{阶段: 秒} 是不含当前段的累计，运行中的阶段
|
||
再把"这一段"实时加上。老记录（没有 time 字段）用 start/end 的差值兜底。
|
||
网页的「总用时」＝各阶段之和，「本步用时」＝当前阶段这条 —— 确定/续跑都不归零。"""
|
||
now = now or time.time()
|
||
acc = pl.get('time') or {}
|
||
cur = pl.get('current')
|
||
out = {}
|
||
for k, r in (pl.get('stages') or {}).items():
|
||
t = acc.get(k)
|
||
if t is None and r.get('start') and r.get('end'): t = r['end'] - r['start']
|
||
t = t or 0
|
||
if k == cur and r.get('start') and r.get('end') is None: t += max(0.0, now - r['start'])
|
||
out[k] = round(t, 1)
|
||
return out
|
||
|
||
|
||
def pipeline():
|
||
d = jload(os.path.join(ROOT, RUNS, 'pipeline.json'))
|
||
if not d: return None
|
||
d['elapsed'] = stage_elapsed(d)
|
||
cur = d.get('current')
|
||
if cur and cur in d['stages']:
|
||
try: d['progress'] = stage_progress(cur, d['stages'][cur])
|
||
except Exception as e: d['progress'] = {'frac': None, 'label': f'进度读取失败：{e}'}
|
||
return d
|
||
|
||
|
||
def status():
|
||
P = lambda *a: os.path.join(ROOT, *a)
|
||
mine = []
|
||
for d in sorted(glob.glob(P(RUNS, 'mine', 'r*')), key=lambda x: int(re.sub(r'\D', '', os.path.basename(x)) or 0)):
|
||
rep, gate = jload(os.path.join(d, 'report.json')), jload(os.path.join(d, 'gate.json'))
|
||
if rep or gate:
|
||
mine.append({'round': os.path.basename(d), 'report': rep and {k: rep.get(k) for k in ('n', 'E1', 'E2', 'E3', 'E4', 'E5', 'ok', 'pool')},
|
||
'gate': gate and {k: gate.get(k) for k in ('gain_pp', 'A_drop_pp', 'fixed_kept', 'R7_left_share_shift_pp', 'R7_travel_median_ratio', 'accept')}})
|
||
stats = jload(P(DATA, 'stats.json'))
|
||
if stats: stats = {k: stats.get(k) for k in ('tasks', 'keep', 'gapDrop', 'minGapFilterMm', 'keepRate', 'samplesEvery3', 'samplesPerTask', 'projectedSamplesFor50k', 'consensusDrop', 'vertexMinGapMm', 'execS', 'travelDeg')}
|
||
shards = glob.glob(P(DATA, 'shard_*.jsonl'))
|
||
pl = pipeline()
|
||
if pl and pl.get('current') == 'gen': # 只在生成阶段数行（增量）；其他阶段用 stats.json 的 tasks
|
||
gen_lines = sum(LC.count(f) for f in shards + glob.glob(P(DATA, 'shard_*.jsonl.part')))
|
||
else: gen_lines = (stats or {}).get('tasks') or 0
|
||
pb = jload(P(RUNS, 'policy_bench.json'))
|
||
if pb: pb = {k: {kk: v.get(kk) for kk in ('n', 'ok', 'reached', 'collisionFrames', 'brakeRate', 'recoveryRate', 'livelock', 'inferMsP99', 'safetyMsP99')}
|
||
for k, v in pb.items() if isinstance(v, dict) and 'n' in v}
|
||
return {'now': time.time(), 'gpu': gpu(), 'cpu': cpu(),
|
||
'vram_reserve_gb': float(os.environ.get('VRAM_RESERVE_GIB', 1)), # 给系统留的余量（GB）：训练进程上限 = 整卡 − 已用 − 这个余量（浮动，见 ml/common/vram.py）
|
||
'vram_budget_gb': float(os.environ['VRAM_BUDGET_GB']) if os.environ.get('VRAM_BUDGET_GB') else None, # 只有显式指定才覆盖上面的口径
|
||
'pipeline': pl, 'check': CHECK.get(), 'lock': lock_info(), # 2026-10-18 第八轮修：上一版这三项被行尾注释吞掉，网页步骤条/锁状态拿不到
|
||
'stage_meta': [{'id': k, 'name': n, 'est_h': h, 'timeout_h': pstate.TIMEOUT_H[k]} for k, n, h in pstate.STAGES],
|
||
'r4': jload(P(RUNS, 'r4.json')),
|
||
'data': {'consistency': (jload(P(DATA, 'consistency.json')) or {}).get('consistentByGateRule'),
|
||
'gen_tasks_written': gen_lines, 'shards': len(shards), 'stats': stats,
|
||
'index': jload(P(DATA, 'samples', 'index.json')), 'r2': jload(P(DATA, 'r2.json'))},
|
||
'm5': [x for x in (jload(f) for f in sorted(glob.glob(P(RUNS, 'm5_gate_*.json')))) if x],
|
||
'selected': open(P(RUNS, 'selected_ckpt')).read().strip() if os.path.exists(P(RUNS, 'selected_ckpt')) else None,
|
||
'mine': mine, 'bench': pb, 'acceptance': jload(P(RUNS, 'acceptance.json'))}
|
||
|
||
|
||
STAGES = pstate.ORDER
|
||
ENV_KEYS = {'N_TASKS': r'\d+', 'NPROC': r'\d+', 'TARGET_SAMPLES': r'\d+', 'CFG': r'ml/configs/(base|wta|small)\.json',
|
||
'CONSENSUS': r'[01]?', 'ROUNDS': r'\d+', 'APG_STEPS': r'\d+', 'MINE_N': r'\d+', 'MIN_GAP_MM': r'\d+(\.\d+)?',
|
||
'FRESH': r'[01]', 'FRESH_FROM': r'(setup|kin|r4|gate1|gen|prep|train|dagger|m5|apg|mine|export|bench|accept)', 'R2_STRICT': r'[01]', 'AUTO_WTA': r'[01]'}
|
||
CONSOLE = os.path.join(ROOT, RUNS, 'console.log')
|
||
|
||
|
||
def find_bash():
|
||
"""Windows 优先 Git Bash（System32\\bash.exe 是 WSL，不能用）；其他系统用 PATH 里的 bash。"""
|
||
if os.name == 'nt':
|
||
for p in [os.environ.get('BASH', ''), r'C:\Program Files\Git\bin\bash.exe', r'C:\Program Files (x86)\Git\bin\bash.exe',
|
||
os.path.expandvars(r'%LOCALAPPDATA%\Programs\Git\bin\bash.exe')]:
|
||
if p and os.path.exists(p): return p
|
||
g = shutil.which('git')
|
||
if g:
|
||
root = os.path.dirname(os.path.dirname(g)) # 便携版 git 常放在 <根>\mingw64\bin\git.exe，要再上一层才找到 bash
|
||
for base in dict.fromkeys([root, os.path.dirname(root)]):
|
||
for sub in ('bin', os.path.join('usr', 'bin')):
|
||
p = os.path.join(base, sub, 'bash.exe')
|
||
if os.path.exists(p): return p
|
||
b = shutil.which('bash')
|
||
if b and 'windows\\system32' not in b.lower(): return b
|
||
return None
|
||
return shutil.which('bash')
|
||
|
||
|
||
def train_python():
|
||
"""流水线用的 Python：尊重 PYTHON 环境变量（与 console.sh / console.bat 一致），否则用控制台自己的解释器。"""
|
||
return os.environ.get('PYTHON') or sys.executable
|
||
|
||
|
||
class SelfCheck:
|
||
"""启动自检（审查 N-10）：在后台线程跑一次，结果缓存；「开始训练」前不过就拒绝，网页顶栏显示。"""
|
||
def __init__(self): self.r = {'pending': True}; threading.Thread(target=self.run, daemon=True).start()
|
||
|
||
def run(self):
|
||
py = train_python(); r = {'python': py, 'bash': find_bash(), 'node': shutil.which('node')}
|
||
try:
|
||
o = subprocess.run([py, '-c', 'import json,torch;c=torch.cuda.is_available();print(json.dumps({"torch":torch.__version__,"cuda":c,"gpu":torch.cuda.get_device_name(0) if c else None}))'],
|
||
capture_output=True, text=True, timeout=120)
|
||
r.update(json.loads(o.stdout.strip().splitlines()[-1]) if o.returncode == 0 else {'torch': None, 'torch_error': (o.stderr or o.stdout).strip()[-300:]})
|
||
except Exception as e: r.update(torch=None, torch_error=str(e))
|
||
probs = []
|
||
if not r.get('torch'): probs.append(f'这个 Python 没有 torch：{py}（先激活训练环境再启动控制台，或设 PYTHON=...）')
|
||
if not r['bash']: probs.append('找不到 bash：Windows 请安装 Git for Windows（自带 Git Bash）')
|
||
if not r['node']: probs.append('找不到 node')
|
||
if r.get('torch') and not r.get('cuda'): r['warn'] = 'torch 看不到 GPU，训练会极慢'
|
||
r['problems'] = probs; r['ok'] = not probs
|
||
self.r = r
|
||
|
||
def get(self): return self.r
|
||
|
||
|
||
def lock_info():
|
||
lk = pstate.lock_read()
|
||
if not lk: return None
|
||
return dict(lk, alive=pstate.lock_alive(lk), age_s=time.time() - lk.get('beat', 0))
|
||
|
||
|
||
NT_NAMES = "'node.exe','xargs.exe','bash.exe','sh.exe','timeout.exe'"
|
||
NT_SWEEP = (r"pipeline\.sh|run_prep\.sh|run_parallel\.sh|build_dataset\.cjs|"
|
||
r"gen_dataset\.cjs|teacher\.cjs|policy_bench\.cjs")
|
||
|
||
|
||
def sweep_nt(keep_pid):
|
||
"""taskkill /T 在 Git Bash 下只杀到 bash，xargs/node 孙进程会逃（10-18 实测：停止后还有 28 个
|
||
gen_dataset 在跑；当晚二次踩到：两批 build_dataset 抢同一批 shard）。按命令行补扫整条流水线，
|
||
keep_pid 的祖先链（= 控制台自己拉起的 monitor bash）不动。先杀包装再杀 node，否则 node 一死
|
||
xargs 会立刻补下一个 shard。返回仍存活个数，-1 = pwsh 没跑成。"""
|
||
q = ("$anc=@(); $c=%d; while ($c -gt 0 -and ($p=Get-CimInstance Win32_Process -Filter \"ProcessId=$c\" "
|
||
"-ErrorAction SilentlyContinue)) { $anc+=$p.ProcessId; $c=$p.ParentProcessId }; "
|
||
"$m={ $_.Name -in %s -and $_.CommandLine -match '%s' -and $anc -notcontains $_.ProcessId }; "
|
||
"foreach ($x in @(Get-CimInstance Win32_Process | Where-Object $m | "
|
||
"Sort-Object { if ($_.Name -eq 'node.exe') { 1 } else { 0 } })) "
|
||
"{ Stop-Process -Id $x.ProcessId -Force -ErrorAction SilentlyContinue }; "
|
||
"Start-Sleep -Seconds 2; "
|
||
"@(Get-CimInstance Win32_Process | Where-Object $m).Count" % (int(keep_pid), NT_NAMES, NT_SWEEP))
|
||
r = subprocess.run(['pwsh', '-NoProfile', '-Command', q], capture_output=True, text=True, timeout=90)
|
||
lines = [s for s in (r.stdout or '').split() if s.isdigit()]
|
||
return int(lines[-1]) if lines else -1
|
||
|
||
|
||
def kill_session(sid):
|
||
"""POSIX：结束会话 sid 里的所有进程（含 timeout 另起的进程组）。"""
|
||
pids = []
|
||
for d in os.listdir('/proc') if os.path.isdir('/proc') else []:
|
||
if d.isdigit():
|
||
try:
|
||
if os.getsid(int(d)) == sid: pids.append(int(d))
|
||
except Exception: pass
|
||
if not pids:
|
||
try: os.killpg(sid, signal.SIGTERM)
|
||
except Exception: pass
|
||
return
|
||
for p in pids:
|
||
try: os.kill(p, signal.SIGTERM)
|
||
except Exception: pass
|
||
|
||
|
||
class Job:
|
||
def __init__(self):
|
||
self.p = None; self.info = {}; self.lock = threading.Lock()
|
||
|
||
def state(self):
|
||
with self.lock:
|
||
run = self.p is not None and self.p.poll() is None
|
||
if self.p is not None and not run and self.info.get('rc') is None:
|
||
self.info['rc'] = self.p.returncode; self.info['ended'] = time.time()
|
||
lk = lock_info()
|
||
ext = bool(lk and lk['alive']) and not run # 终端或上一个控制台窗口起的流水线
|
||
return {'running': run or ext, 'external': ext, 'lock': lk, **self.info, 'bash': find_bash() is not None}
|
||
|
||
def start(self, frm, to, env):
|
||
with self.lock:
|
||
if self.p is not None and self.p.poll() is None: return {'error': '已经在运行'}
|
||
lk = lock_info()
|
||
if lk and lk['alive']:
|
||
return {'error': f"已有一条流水线在运行（{lk.get('owner')}，pid {lk.get('pid')}，{lk['age_s']:.0f} s 前心跳）；先停止它"}
|
||
ck = CHECK.get()
|
||
if ck.get('pending'): return {'error': '环境自检还没完成，稍等几秒再点'}
|
||
if not ck.get('ok'): return {'error': '环境自检未通过：' + '；'.join(ck['problems'])}
|
||
if frm not in STAGES or to not in STAGES or STAGES.index(frm) > STAGES.index(to): return {'error': '起止阶段无效'}
|
||
bash = find_bash()
|
||
e = dict(os.environ, FROM=frm, TO=to, MONITOR='0', PYTHONUNBUFFERED='1', PYTHONIOENCODING='utf-8', PYTHONUTF8='1', PIPE_OWNER='网页')
|
||
py = train_python()
|
||
if ' ' not in py: e['PYTHON'] = py.replace('\\', '/')
|
||
for k, v in (env or {}).items():
|
||
if k in ENV_KEYS and v not in (None, '') and re.fullmatch(ENV_KEYS[k], str(v)): e[k] = str(v)
|
||
os.makedirs(os.path.dirname(CONSOLE), exist_ok=True)
|
||
shown = ' '.join(f'{k}={e[k]}' for k in ('FRESH', 'FROM', 'TO') if k in e)
|
||
with open(CONSOLE, 'w', encoding='utf-8') as f: f.write(f'$ {shown} bash ml/pipeline.sh all [{time.strftime("%Y-%m-%d %H:%M:%S")}]\n')
|
||
log = open(CONSOLE, 'a', encoding='utf-8') # 追加模式：网页写入的"已停止"提示不会被子进程覆盖
|
||
# Windows：CREATE_NO_WINDOW 给子进程树一个隐藏的独立控制台 → 关掉控制台黑窗口不会连带杀训练（审查 N-11）
|
||
kw = {'creationflags': subprocess.CREATE_NEW_PROCESS_GROUP | 0x08000000} if os.name == 'nt' else {'start_new_session': True}
|
||
self.p = subprocess.Popen([bash, 'ml/pipeline.sh', 'all'], cwd=ROOT, env=e, stdout=log, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL, **kw)
|
||
self.info = {'pid': self.p.pid, 'from': frm, 'to': to, 'started': time.time(), 'rc': None, 'ended': None,
|
||
'env': {k: e[k] for k in ENV_KEYS if k in e}}
|
||
return {'ok': True, 'pid': self.p.pid}
|
||
|
||
def stop(self):
|
||
# 2026-10-18 体面停止：训练类阶段先写 ml/runs/STOP，等训练进程自己存 last.pt 并退出（最多 90 s），再硬杀剩下的。
|
||
# Windows 的 taskkill /F 收不到信号，不这样做会丢掉上次存档之后的进度。生成/打包本身按 shard/条断点续，直接杀。
|
||
with self.lock:
|
||
ours = self.p is not None and self.p.poll() is None
|
||
lk = lock_info()
|
||
if not ours and not (lk and lk['alive']): return {'error': '没有在运行'}
|
||
pid = self.p.pid if ours else int(lk.get('pid') or 0)
|
||
if not pid: return {'error': '锁里没有 pid，无法停止'}
|
||
cur = (pstate.load().get('current') or '')
|
||
if cur in ('train', 'dagger', 'm5', 'apg', 'mine'):
|
||
open(STOPF, 'w').write(time.strftime('%H:%M:%S'))
|
||
with open(CONSOLE, 'a', encoding='utf-8') as f: f.write(f'\n[停止请求已发出，等待 {cur} 存档 last.pt…]\n')
|
||
for _ in range(180):
|
||
time.sleep(0.5)
|
||
l2 = pstate.lock_read()
|
||
if (self.p is not None and self.p.poll() is not None) if ours else not (l2 and pstate.lock_alive(l2)): break
|
||
with self.lock:
|
||
if os.name == 'nt':
|
||
subprocess.run(['taskkill', '/T', '/F', '/PID', str(pid)], capture_output=True)
|
||
# 补扫漏网的子孙；有残留就写进网页日志，别让用户以为停干净了
|
||
try:
|
||
left = sweep_nt(os.getpid())
|
||
except Exception:
|
||
left = -1
|
||
if left != 0:
|
||
with open(CONSOLE, 'a', encoding='utf-8') as f:
|
||
f.write(f'\n[停止补扫：仍有 {left} 个流水线子进程存活（-1=补扫没跑成），把这条发我]\n')
|
||
else:
|
||
if ours:
|
||
kill_session(pid)
|
||
try: self.p.wait(15)
|
||
except subprocess.TimeoutExpired: kill_session(pid); self.p.kill()
|
||
else: # 只动它自己的会话 / 进程组，绝不按会话杀到用户的终端
|
||
try:
|
||
if os.getsid(pid) == pid: kill_session(pid)
|
||
elif os.getpgid(pid) == pid: os.killpg(pid, signal.SIGTERM)
|
||
else: os.kill(pid, signal.SIGTERM)
|
||
except ProcessLookupError: pass
|
||
self.info['stopped'] = True
|
||
with open(CONSOLE, 'a', encoding='utf-8') as f: f.write(f'\n[已在网页上停止 {time.strftime("%H:%M:%S")}]\n')
|
||
time.sleep(0.5)
|
||
pstate.mark_stopped(-15) # 当前阶段记为"已停止"、释放锁（单一写者）
|
||
try: os.remove(STOPF)
|
||
except FileNotFoundError: pass
|
||
return {'ok': True}
|
||
|
||
|
||
STOPF = os.path.join(ROOT, 'ml', 'runs', 'STOP') # 与 ml/common/ckpt.py 的 STOP 同一文件
|
||
JOB = Job()
|
||
CHECK = None # main() 里创建（自检要等 find_bash 定义完）
|
||
|
||
|
||
def console(offset):
|
||
if not os.path.exists(CONSOLE): return {'text': '', 'offset': 0, 'reset': True}
|
||
size = os.path.getsize(CONSOLE); reset = offset > size or offset == 0
|
||
if reset: offset = max(0, size - 65536) # 首次只给最后 64 KB
|
||
with open(CONSOLE, 'rb') as f: f.seek(offset); b = f.read(min(size - offset, 1 << 20))
|
||
return {'text': b.decode('utf-8', 'replace').replace('\r', ''), 'offset': offset + len(b), 'reset': reset}
|
||
|
||
|
||
class H(BaseHTTPRequestHandler):
|
||
def log_message(self, *a): pass
|
||
|
||
def send(self, code, body, ctype):
|
||
b = body if isinstance(body, bytes) else body.encode('utf-8')
|
||
self.send_response(code); self.send_header('Content-Type', ctype); self.send_header('Content-Length', str(len(b)))
|
||
self.send_header('Cache-Control', 'no-store'); self.end_headers(); self.wfile.write(b)
|
||
|
||
def do_GET(self):
|
||
u = urlparse(self.path); q = parse_qs(u.query)
|
||
try:
|
||
if u.path in ('/', '/index.html'):
|
||
with open(HTML, 'rb') as f: return self.send(200, f.read(), 'text/html; charset=utf-8')
|
||
if u.path == '/api/runs': return self.send(200, json.dumps(list_runs()), 'application/json')
|
||
if u.path == '/api/log':
|
||
return self.send(200, json.dumps(read_log(q.get('run', [''])[0], int(q.get('offset', ['0'])[0]))), 'application/json')
|
||
if u.path == '/api/status': return self.send(200, json.dumps(status()), 'application/json')
|
||
if u.path == '/api/job': return self.send(200, json.dumps(JOB.state()), 'application/json')
|
||
if u.path == '/api/console': return self.send(200, json.dumps(console(int(q.get('offset', ['0'])[0]))), 'application/json')
|
||
self.send(404, 'not found', 'text/plain')
|
||
except Exception as e:
|
||
self.send(500, json.dumps({'error': str(e)}), 'application/json')
|
||
|
||
|
||
def do_POST(self):
|
||
u = urlparse(self.path)
|
||
if not (self.server.remote_control or self.client_address[0] in ('127.0.0.1', '::1')):
|
||
return self.send(403, json.dumps({'error': '开始/停止只接受本机操作'}), 'application/json')
|
||
# 审查 N-8：本机任意网页都能向 localhost 发 POST（源地址同为 127.0.0.1）。要求自定义头（跨站请求会触发 CORS 预检，
|
||
# 本服务不应答 OPTIONS → 浏览器不发），并且 Origin / Sec-Fetch-Site 若存在必须同源。
|
||
org, sfs, host = self.headers.get('Origin'), self.headers.get('Sec-Fetch-Site'), self.headers.get('Host', '')
|
||
if self.headers.get('X-Console') != '1' or (sfs and sfs not in ('same-origin', 'none')) or \
|
||
(org and urlparse(org).netloc not in (host, f'localhost:{self.server.server_port}', f'127.0.0.1:{self.server.server_port}')):
|
||
return self.send(403, json.dumps({'error': '拒绝：非本控制台页面发起的请求'}), 'application/json')
|
||
try:
|
||
n = int(self.headers.get('Content-Length') or 0); body = json.loads(self.rfile.read(n) or b'{}')
|
||
if u.path == '/api/start':
|
||
env = dict(body.get('env') or {})
|
||
if body.get('fresh'): env['FRESH'] = '1'
|
||
r = JOB.start(body.get('from', 'kin'), body.get('to', 'accept'), env)
|
||
elif u.path == '/api/stop': r = JOB.stop()
|
||
else: return self.send(404, 'not found', 'text/plain')
|
||
self.send(200 if 'ok' in r else 409, json.dumps(r, ensure_ascii=False), 'application/json')
|
||
except Exception as e:
|
||
self.send(500, json.dumps({'error': str(e)}), 'application/json')
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument('--port', type=int, default=8100); ap.add_argument('--host', default='127.0.0.1')
|
||
ap.add_argument('--open', action='store_true', help='启动后自动打开浏览器')
|
||
ap.add_argument('--remote-control', action='store_true', help='允许其他机器点开始/停止（默认只能看）')
|
||
a = ap.parse_args()
|
||
global CHECK
|
||
CHECK = SelfCheck()
|
||
CPUS = CpuSampler() # 每秒采一次 CPU（线程常驻，status 只读最近一次）
|
||
srv = ThreadingHTTPServer((a.host, a.port), H); srv.remote_control = a.remote_control
|
||
url = f'http://localhost:{a.port}/'
|
||
print(f'训练控制台：{url} （关闭本窗口只关控制台，已开始的训练继续跑；重新打开控制台即可接着看、可停止）', flush=True)
|
||
if a.open: threading.Timer(0.8, lambda: webbrowser.open(url)).start()
|
||
try: srv.serve_forever()
|
||
except KeyboardInterrupt: pass
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|