From 0b47905a1c2f4d81282aac23c475156ac7041723 Mon Sep 17 00:00:00 2001 From: cbb <2625274797@qq.com> Date: Mon, 19 Oct 2026 00:33:11 +0800 Subject: [PATCH] =?UTF-8?q?feat(prep/monitor):=20=E6=89=93=E5=8C=85?= =?UTF-8?q?=E6=AF=8F=2010%=20=E6=89=93=E4=B8=80=E8=A1=8C=E8=BF=9B=E5=BA=A6?= =?UTF-8?q?=20+=20=E7=9B=91=E6=8E=A7=E6=8C=89=E5=AE=9E=E6=B5=8B=E7=AE=97?= =?UTF-8?q?=E8=BF=9B=E5=BA=A6/ETA=EF=BC=9Bprep=20=E4=BC=B0=E6=97=B6=200.5?= =?UTF-8?q?=E2=86=922.0=20h?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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），与实际耗时一致。 --- ml/monitor.html | 42 ++++++++++++++++++++++++++++----------- ml/monitor.py | 19 +++++++++++++++--- ml/prep/build_dataset.cjs | 5 ++++- ml/pstate.py | 2 +- 4 files changed, 51 insertions(+), 17 deletions(-) diff --git a/ml/monitor.html b/ml/monitor.html index 1d3ce57..3d7153a 100644 --- a/ml/monitor.html +++ b/ml/monitor.html @@ -230,7 +230,7 @@
运行方式
- +
模型与数据
@@ -408,13 +408,26 @@ let logyOn = false, job = null, userPicked = false, conOff = 0; 下面这份只给示例数据用，数值与 pstate.py 一致。 */ let STAGE_META = [ ['setup', '安装依赖', 0.2], ['kin', '对拍', 0.2], ['r4', '推理延迟', 0.2], ['gate1', '一致性门禁', 0.5], ['gen', '生成老师轨迹', 5.0], - ['prep', '打包样本', 0.5], ['train', '预训练', 4.5], ['dagger', 'DAgger 纠偏', 3.5], ['m5', '头门禁', 0.5], ['apg', 'APG 微调', 4.0], ['mine', '挖错循环', 9.0], + ['prep', '打包样本', 2.0], ['train', '预训练', 4.5], ['dagger', 'DAgger 纠偏', 3.5], ['m5', '头门禁', 0.5], ['apg', 'APG 微调', 4.0], ['mine', '挖错循环', 9.0], ['export', '导出', 0.1], ['bench', '部署评测', 1.0], ['accept', '验收汇总', 0.5]]; let SM = Object.fromEntries(STAGE_META.map(([k, n, w]) => [k, { n, w, to: Math.max(1, 3 * w) }])); +/* 起始阶段下拉的编号按 README 步骤表（1 环境、2 对拍、2′ R4、3 门禁、…、12 验收）；setup 不进下拉（建议手动装依赖） */ +const FROM_NO = { setup: '1', kin: '2', r4: '2′', gate1: '3', gen: '4', prep: '5', train: '6', dagger: '6′', m5: '7', apg: '8', mine: '9', export: '10', bench: '11', accept: '12' }; +const FROM_NOTE = { gen: '（约 7 小时）' }; +function fillFrom() { + const sel = $('fFrom'); if (!sel) return; + const want = STAGE_META.map(x => x[0]).filter(k => k !== 'setup'); + if ([...sel.options].map(o => o.value).join() === want.join()) return; // 阶段表没变就别重建，免得每秒轮询把已选项重置 + const keep = sel.value; + sel.innerHTML = want.map(k => ``).join(''); + sel.value = want.includes(keep) ? keep : want.includes('train') ? 'train' : want[0]; +} function applyStageMeta(m) { if (!m || !m.length) return; STAGE_META = m.map(x => [x.id, x.name, x.est_h]); SM = Object.fromEntries(m.map(x => [x.id, { n: x.name, w: x.est_h, to: x.timeout_h }])); + fillFrom(); } +fillFrom(); /* 开始 / 停止请求带自定义头：跨站网页发不出这种请求（会触发 CORS 预检，控制台不应答），见 monitor.py 审查 N-8 */ const post = (path, body) => fetch(path, { method: 'POST', headers: { 'X-Console': '1', 'Content-Type': 'application/json' }, body: body || '{}' }).then(x => x.json()).catch(e => ({ error: String(e) })); const smoothv = () => +$('smooth').value; @@ -565,11 +578,15 @@ async function poll() { function setPill(cls, text) { const st = $('state'); st.className = 'pill ' + cls; st.lastElementChild.textContent = text; } /* ===================== 流水线 / GPU / 各阶段面板 ===================== */ +/* 「全流程」进度与上方步骤条同一口径：步骤 = 本次计划 ∪ 记录里已有的阶段。 + 从中间阶段重训时，前面已完成的阶段不在本次 plan 里，但记录还在（步骤条就靠这个显示打勾）， + 总进度/剩余时间/完成计数也必须算上它们，否则总进度会归零。 */ +const pipeKeys = (P, plan) => STAGE_META.map(x => x[0]).filter(k => plan.includes(k) || P.stages[k]); function stages(s) { const P = s.pipeline; if (P) { // 有流水线记录：以它为准 const plan = P.plan && P.plan.length ? P.plan : STAGE_META.map(x => x[0]).filter(k => k !== 'setup'); - const keys = STAGE_META.map(x => x[0]).filter(k => plan.includes(k) || P.stages[k]); + const keys = pipeKeys(P, plan); const el = P.elapsed || {}; // 累计用时（跨停止/续跑），没有就退回本段 return keys.map(k => { const r = P.stages[k], nm = SM[k].n; @@ -691,31 +708,32 @@ function renderPipe(s) { : `${job && job.rc != null ? (job.stopped ? '上次已手动停止，再点即从停下的那步继续' : job.rc === 0 ? '上次运行已全部完成' : `上次运行失败（返回码 ${job.rc}），看终端输出；修好后再点即续跑`) : '已完成的步骤会自动跳过'}`}
`; if (!P) { $('pipe').innerHTML = `

全流程尚未开始

${empty('还没运行过流水线。点「开始训练」从对拍一直跑到验收；中途每一步的进度都显示在这里。')}${btns}`; return; } const plan = P.plan && P.plan.length ? P.plan : Object.keys(P.stages); + const keys = pipeKeys(P, plan); // 含重训时保留的已完成阶段，与上方步骤条同口径 let tot = 0, got = 0; - for (const k of plan) { const w = SM[k]?.w || 0.1, r = P.stages[k]; tot += w; if (r && r.rc === 0 && P.current !== k) got += w; else if (P.current === k) got += w * ((P.progress && P.progress.frac) || 0); } + for (const k of keys) { const w = SM[k]?.w || 0.1, r = P.stages[k]; tot += w; if (r && r.rc === 0 && P.current !== k) got += w; else if (P.current === k) got += w * ((P.progress && P.progress.frac) || 0); } const frac = tot ? got / tot : 0, now = Date.now() / 1000; const cur = P.current, cr = cur && P.stages[cur], pr = P.progress || {}; - const doneN = plan.filter(k => P.stages[k] && P.stages[k].rc === 0 && P.current !== k).length; + const doneN = keys.filter(k => P.stages[k] && P.stages[k].rc === 0 && P.current !== k).length; const failed = plan.find(k => P.stages[k] && P.stages[k].rc != null && P.stages[k].rc !== 0); - const allDone = !cur && doneN === plan.length; - let remain = 0, calib = false; for (const k of plan) { const r = P.stages[k]; if (!r || r.rc == null) { if (k === cur && pr.eta_s != null) { remain += pr.eta_s / 3600; calib = true; } else remain += (SM[k]?.w || 0.1) * (k === cur ? 1 - (pr.frac || 0) : 1); } } + const allDone = !cur && doneN === keys.length; + let remain = 0, calib = false; for (const k of keys) { const r = P.stages[k]; if (!r || r.rc == null) { if (k === cur && pr.eta_s != null) { remain += pr.eta_s / 3600; calib = true; } else remain += (SM[k]?.w || 0.1) * (k === cur ? 1 - (pr.frac || 0) : 1); } } const head = cur ? `
正在运行${SM[cur]?.n || cur}${pr.label || ''}
` - : allDone ? `
状态全部完成共 ${plan.length} 步
` + : allDone ? `
状态全部完成共 ${keys.length} 步
` : failed ? `
状态停在「${SM[failed]?.n || failed}」${(P.stages[failed].info || {}).timed_out || P.stages[failed].rc === 124 ? '超过时间上限，判定卡死' : P.stages[failed].rc === -15 || P.stages[failed].rc === 143 ? '已停止' : '失败，看终端输出'}
` - : `
状态未在运行已完成 ${doneN} / ${plan.length} 步
`; + : `
状态未在运行已完成 ${doneN} / ${keys.length} 步
`; const esc2 = x => String(x).replace(/&/g, '&').replace(/ { const t = $('term'); if (!t) return ''; const ls = (t.innerText || '').split(/\r?\n/).filter(x => /Error|error:|Traceback|不通过|失败|超过时间上限/.test(x)); return ls.length ? ls[ls.length - 1].slice(0, 300) : ''; })(); $('alert').innerHTML = failed ? `
✕ 停在「${SM[failed]?.n || failed}」 · ${why}
${lastErr ? `
${esc2(lastErr)}
` : ''}
完整报错见下方「终端输出」；修好后点「继续训练」，已完成的步骤会自动跳过${stopped ? '、从停下的那一步接着跑' : ''}
` : ''; - $('pipe').innerHTML = `

全流程${doneN} / ${plan.length} 步完成

+ $('pipe').innerHTML = `

全流程${doneN} / ${keys.length} 步完成

${head}
${(100 * frac).toFixed(1)}%
${cur ? `
当前步骤 · ${SM[cur]?.n}${pr.frac != null ? `${pct(pr.frac)}` : '无法计量，按用时估'}
` : ''}
-
总用时
${hms(plan.reduce((a, k) => a + ((P.elapsed || {})[k] || 0), 0))}
+
总用时
${hms(keys.reduce((a, k) => a + ((P.elapsed || {})[k] || 0), 0))}
本步用时
${cur ? hms((P.elapsed || {})[cur] || 0) : '—'}
${cur && pr.eta_s != null ? '本步预计剩余（实测速率）' : '本步粗估 / 上限'}
${cur ? (pr.eta_s != null ? hmsShort(pr.eta_s) + ' / 上限 ' + hmsShort((cr && cr.timeout_h || SM[cur]?.to || 0) * 3600) : hmsShort((SM[cur]?.w || 0) * 3600) + ' / ' + hmsShort((cr && cr.timeout_h || SM[cur]?.to || 0) * 3600)) : '—'}
全流程剩余（${calib ? '实测外推' : '粗估'}）
${allDone ? '—' : '约 ' + hmsShort(remain * 3600)}
@@ -748,7 +766,7 @@ function openStart(preset) { const w = $('fWarn'); const upd = () => { const msgs = []; const done = P ? (P.plan || []).filter(k => P.stages[k] && P.stages[k].rc === 0) : []; if ($('fMode').value === 'resume' && done.length) msgs.push(`将跳过已完成的 ${done.length} 步：${done.map(k => SM[k]?.n || k).join('、')}。改了参数会自动改为从头重跑。`); - if ($('fMode').value === 'from' && P) { const nm = $('fFrom').selectedOptions[0].textContent; msgs.push(`从「${nm}」开始重训：该阶段及之后的旧产物移到 ml/archive/<时间>/（不删除），之前的阶段保持不动。${$('fFrom').value === 'gen' ? '注意：生成的数据会被归档，重跑约 7 小时。' : ''}`); } + if ($('fMode').value === 'from' && P) { const nm = SM[$('fFrom').value]?.n || $('fFrom').value; msgs.push(`从「${nm}」开始重训：该阶段及之后的旧产物移到 ml/archive/<时间>/（不删除），之前的阶段保持不动。${$('fFrom').value === 'gen' ? '注意：生成的数据会被归档，重跑约 7 小时。' : ''}`); } if ($('fMode').value === 'fresh' && P) msgs.push('全部重跑：上次的数据、模型、评测结果整体移到 ml/archive/<时间>/（不删除）。生成数据要重跑（约 7 小时）。'); $('fFrom').disabled = $('fMode').value !== 'from'; $('fGo').textContent = $('fMode').value === 'from' ? '▶ 开始重训' : ($('fMode').value === 'fresh' ? '▶ 全部重跑' : '▶ 开始'); diff --git a/ml/monitor.py b/ml/monitor.py index 152c8e3..297cce6 100644 --- a/ml/monitor.py +++ b/ml/monitor.py @@ -252,11 +252,24 @@ def stage_progress(st, s): 'eta_s': (N - n) / rate if rate > 0 else None} if st == 'prep': nsh = len(glob.glob(os.path.join(ROOT, DATA, 'shard_*.jsonl'))) - done = len(newer(f'{DATA}/samples/*.X.npy', t0)) + 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 = (nsh - done) / (done / el) if done > 0 and el > 1 and nsh > done else None - return {'frac': min(0.95, done / nsh) if nsh else None, 'label': f'已打包 {done} / {nsh} 个分片' + ('，R2 检查中' if nsh and done >= nsh else ''), 'eta_s': eta} + 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 diff --git a/ml/prep/build_dataset.cjs b/ml/prep/build_dataset.cjs index 85146e2..0a88787 100644 --- a/ml/prep/build_dataset.cjs +++ b/ml/prep/build_dataset.cjs @@ -31,7 +31,10 @@ let nAug = 0, nAugRej = 0, nGapDrop = 0; /* 2026-10-18：认证最小间隙 < 3 mm 的老师轨迹不进训练集；噪声增广接受地板同步为 3 mm（原 1 mm）。 */ const MIN_GAP_MM = +(process.env.MIN_GAP_MM || 3); const AUG_P = +(process.env.AUG_P || 0.4), AUG_SIGMA = +(process.env.AUG_SIGMA || 3), AUG_BACK = +(process.env.AUG_BACK || 12); -for (const line of lines) { +const NT = lines.length, PSTEP = Math.max(1, Math.floor(NT / 10)); +for (let li = 0; li < NT; li++) { + if (li % PSTEP === 0) console.log(`[prep] ${base} ${li}/${NT}`); // 每 10% 一行，28 个分片并行也看得见进度 + const line = lines[li]; const r = JSON.parse(line); if (!r.keep) continue; if (r.cert && r.cert.minGap != null && r.cert.minGap < MIN_GAP_MM) { nGapDrop++; continue; } // 与 stats.cjs 同一过滤 diff --git a/ml/pstate.py b/ml/pstate.py index c6b88bc..7b1eb3a 100644 --- a/ml/pstate.py +++ b/ml/pstate.py @@ -26,7 +26,7 @@ LOCK_STALE_S = 300 # 心跳每 60 s 一次；5 分钟没刷新 = 持锁 # 阶段顺序、名称、估时（小时，按 README §耗时表的共识分支上界；网页进度权重即此值） STAGES = [ ('setup', '安装依赖', 0.2), ('kin', '对拍', 0.2), ('r4', '推理延迟', 0.2), ('gate1', '一致性门禁', 0.5), - ('gen', '生成老师轨迹', 5.0), ('prep', '打包样本', 0.5), ('train', '预训练', 4.5), ('dagger', 'DAgger 纠偏', 3.5), ('m5', '头门禁', 0.5), + ('gen', '生成老师轨迹', 5.0), ('prep', '打包样本', 2.0), ('train', '预训练', 4.5), ('dagger', 'DAgger 纠偏', 3.5), ('m5', '头门禁', 0.5), ('apg', 'APG 微调', 4.0), ('mine', '挖错循环', 9.0), ('export', '导出', 0.1), ('bench', '部署评测', 1.0), ('accept', '验收汇总', 0.5)] ORDER = [s[0] for s in STAGES]