cbb
6de085fd37
- Kin.cert_lever 用三角不等式给全姿态位移界（替代 JS 随机采样×1.05），界量级 5.93–48.86 mm/整步 - Kin.certify_segments 批量前沿二分+每根查询预算：48 固定段中预算32判安全37/碰撞1/未知拒10， 第180题1.08–1.10s漏检段以5次查询判碰撞；未知只拒不放 - 修 seg_seg_dist 第二条退化成点不投影（模型内退化胶囊686受影响）；但7586个加密点的组最小间隙 逐位不变、2000样本JS对拍仍 PASS → 不解释轮次30漏检、不改既有成绩 - 负结论：B=1 单段认证 P50 22.93/P90 58.43/P99 101.57 ms，432次中225次>20ms； 自适应宽相位 cap 交错复测耗时无收益，先前1.6×为GPU顺序漂移，已在日志撤回 - 成本模型 L=ceil(log2(bound/(ga+gb)))、查询=2^L+1 与实测对齐（安全段2–17次，未知段33–65次） - 新增8项段认证测试+1项回放测试（14→23项通过）；变异实测证明张量 max_depth 回绕会误放行
401 lines
22 KiB
Python
401 lines
22 KiB
Python
#!/usr/bin/env python
|
||
"""闭环碰撞定位：第一次碰撞发生在第几秒、碰撞时最近的是哪两条连杆（J几-J几）。
|
||
|
||
用法（仓库根目录）：
|
||
PYTHONUTF8=1 python -m ml.eval.diag_collision --ckpt ml/runs/base/last.pt --n 512 --out ml/runs/diag_collision.json
|
||
|
||
口径与 closed_loop.rollout 一致：每步同时检查当前状态 q 与相邻状态中点 mid 的间隙（cap=0.5 mm）；
|
||
间隙组对索引 = gA*7+gB（gA<gB，g 为连杆组 0..5 → J1..J6）。
|
||
"""
|
||
import argparse, json, math, os, sys, time
|
||
from collections import Counter
|
||
|
||
import torch
|
||
|
||
from ml.common.kin_torch import Kin
|
||
from ml.data import load_tasks
|
||
from ml.model.policy import build
|
||
from ml.sim.closed_loop import DT, rollout
|
||
|
||
|
||
def pair_names():
|
||
# kin.json 有 7 个连杆组：g=0 是固定基座/立柱，g=1..6 是 J1..J6
|
||
def nm(g):
|
||
return '基座' if g == 0 else f'J{g}'
|
||
return {a * 7 + b: f'{nm(a)}-{nm(b)}' for a in range(7) for b in range(a + 1, 7)}
|
||
|
||
|
||
@torch.no_grad()
|
||
def replay_first_hits(path, out=None):
|
||
saved = torch.load(path, map_location='cpu', weights_only=True)
|
||
k32, k64 = Kin(), Kin(dtype=torch.float64)
|
||
rows, offset = [], 0
|
||
for _, per, count in saved['batches']:
|
||
for b in (per['col'] > 0).nonzero().flatten().tolist():
|
||
it = int(round(float(per['first_hit'][b]) / DT))
|
||
prev, q = per['traj'][b, it - 1:it + 1]
|
||
mid = prev + 0.5 * (q - prev)
|
||
points = torch.stack([per['traj'][b, 0], prev, mid, q])
|
||
gm32 = k32.gaps(points, 0.5)['gm']
|
||
gm64 = k64.gaps(points.double(), 0.5)['gm']
|
||
gap32, gp32 = gm32.min(-1)
|
||
gap64, gp64 = gm64.min(-1)
|
||
nseg = max(1, math.ceil(float((q - prev).abs().max()) / 0.01))
|
||
dense = prev.double() + torch.linspace(0, 1, nseg + 1, dtype=torch.float64)[:, None] * (q.double() - prev.double())
|
||
dense_gap = torch.cat([k64.gaps(chunk, 0.5)['gm'].min(-1).values for chunk in dense.split(32)])
|
||
boundary = (~k32.cont) & ((q - k32.lo).abs().lt(1e-5) | (q - k32.hi).abs().lt(1e-5))
|
||
row = {
|
||
'i': offset + b, 'task': saved['tasks'][offset + b],
|
||
'first_hit_s': float(per['first_hit'][b]), 'dls_it': int(per['dls_it'][b]),
|
||
'point_order': ['q0', 'previous', 'midpoint', 'endpoint'],
|
||
'gap32_mm': gap32.tolist(), 'gap64_mm': gap64.tolist(),
|
||
'pair32': gp32.tolist(), 'pair64': gp64.tolist(),
|
||
'boundary_joints': (boundary.nonzero().flatten() + 1).tolist(),
|
||
'previous': prev.tolist(), 'endpoint': q.tolist(),
|
||
'dense_nsegments': nseg, 'dense_gap64_min_mm': float(dense_gap.min()),
|
||
}
|
||
if 'plans' in per:
|
||
plans = per['plans']
|
||
pi = int(torch.searchsorted(plans['it'], it - 1, right=True)) - 1
|
||
step = it - 1 - int(plans['it'][pi])
|
||
ncommit = plans['commit'].shape[2]
|
||
command = plans['commit'][b, pi, step]
|
||
cp, cm = plans['points'][b, pi, step], plans['midpoints'][b, pi, step]
|
||
checked = bool(plans['certified'][b, pi, step])
|
||
predicted = k32.clamp_box((prev + command)[None])[0]
|
||
row['certificate'] = {
|
||
'plan_start_it': int(plans['it'][pi]), 'step': step, 'checked': checked,
|
||
'endpoint': cp.tolist() if checked else None,
|
||
'midpoint': cm.tolist() if checked else None,
|
||
'recorded_gap32_mm': plans['gaps'][b, pi, [step, ncommit + step]].tolist() if checked else None,
|
||
'replayed_gap64_mm': k64.gaps(torch.stack([cp, cm]).double(), 0.5)['gm'].min(-1).values.tolist() if checked else None,
|
||
'command_delta': command.tolist(),
|
||
'clamped_reconstruction_max_abs_deg': float((predicted - q).abs().max()),
|
||
'certified_vs_executed_max_abs_deg': float((cp - q).abs().max()) if checked else None,
|
||
}
|
||
rows.append(row)
|
||
print('task %d first_hit %.2fs gap32 %s gap64 %s boundary %s' %
|
||
(row['i'], row['first_hit_s'], row['gap32_mm'], row['gap64_mm'], row['boundary_joints']), flush=True)
|
||
offset += count
|
||
result = {
|
||
'trace': path, 'n': offset, 'reported_collision_tasks': len(rows),
|
||
'replayed_first_hits32': sum(min(r['gap32_mm'][2:]) <= 0 for r in rows),
|
||
'replayed_first_hits64': sum(min(r['gap64_mm'][2:]) <= 0 for r in rows),
|
||
'scope': 'Replay reported first hits only; same geometry kernel at float32/float64, not a full safety audit.',
|
||
'rows': rows,
|
||
}
|
||
if out:
|
||
with open(out, 'w', encoding='utf-8') as f:
|
||
json.dump(result, f, indent=2, allow_nan=False)
|
||
print('replayed first hits: %d/%d float32, %d/%d float64' %
|
||
(result['replayed_first_hits32'], len(rows), result['replayed_first_hits64'], len(rows)), flush=True)
|
||
return result
|
||
|
||
|
||
@torch.no_grad()
|
||
def replay_dense(path, indices, max_step_deg=0.05, device='cpu', out=None):
|
||
saved = torch.load(path, map_location='cpu', weights_only=True)
|
||
kin = Kin(device=device, dtype=torch.float64)
|
||
if saved.get('kin_fp') != kin.fp:
|
||
raise ValueError('trace/kin fingerprint mismatch')
|
||
if not indices or min(indices) < 0 or max(indices) >= len(saved['tasks']) or max_step_deg <= 0:
|
||
raise ValueError('valid task indices and positive max_step_deg required')
|
||
started = time.perf_counter()
|
||
rows, offset = [], 0
|
||
for _, per, count in saved['batches']:
|
||
for index in indices:
|
||
if not offset <= index < offset + count:
|
||
continue
|
||
b = index - offset
|
||
nstep = min(per['traj'].shape[1] - 1, math.floor((float(per['limit'][b]) + 1e-6) / DT))
|
||
if torch.isfinite(per['t_done'][b]):
|
||
nstep = min(nstep, round(float(per['t_done'][b]) / DT))
|
||
trajectory = per['traj'][b, :nstep + 1].double()
|
||
points, segment_ids, fractions = [trajectory[:1]], [torch.tensor([-1])], [torch.zeros(1)]
|
||
for it in range(nstep):
|
||
prev, end = trajectory[it:it + 2]
|
||
nseg = max(2, 2 * math.ceil(float((end - prev).abs().max()) / (2 * max_step_deg)))
|
||
alpha = torch.arange(1, nseg + 1, dtype=torch.float64) / nseg
|
||
points.append(prev + alpha[:, None] * (end - prev))
|
||
segment_ids.append(torch.full((nseg,), it))
|
||
fractions.append(alpha)
|
||
points = torch.cat(points)
|
||
segments, alpha = torch.cat(segment_ids), torch.cat(fractions)
|
||
print('dense task %d steps %d points %d start' % (index, nstep, len(points)), flush=True)
|
||
gaps = []
|
||
for start in range(0, len(points), 64):
|
||
gaps.append(kin.gaps(points[start:start + 64].to(device), 0.5)['gm'].min(-1).values.cpu())
|
||
if (start // 64 + 1) % 64 == 0:
|
||
print('dense task %d points %d/%d wall %.1fs' %
|
||
(index, min(start + 64, len(points)), len(points), time.perf_counter() - started), flush=True)
|
||
gaps = torch.cat(gaps)
|
||
hit = gaps <= 0
|
||
first = int(hit.nonzero()[0]) if hit.any() else None
|
||
rows.append({
|
||
'i': index, 'task': saved['tasks'][index], 'steps': nstep, 'sampled_points': len(points),
|
||
'initial_gap64_mm': float(gaps[0]), 'min_gap64_mm': float(gaps.min()),
|
||
'collision_segments': int(segments[hit & (segments >= 0)].unique().numel()),
|
||
'first_collision_s': max(0.0, float(segments[first] + alpha[first]) * DT) if first is not None else None,
|
||
})
|
||
print('dense task %d min_gap %.9f mm collision_segments %d' %
|
||
(index, rows[-1]['min_gap64_mm'], rows[-1]['collision_segments']), flush=True)
|
||
offset += count
|
||
result = {
|
||
'trace': str(path), 'tasks_sha256': saved.get('tasks_sha256'), 'kin_fp': kin.fp,
|
||
'indices': indices, 'max_step_deg': max_step_deg, 'dtype': 'float64', 'device': device,
|
||
'scope': 'Executed trajectory through each task deadline or arrival, including initial state; finite sampling, not continuous certification.',
|
||
'collision_tasks': sum(r['min_gap64_mm'] <= 0 for r in rows), 'rows': rows,
|
||
'audit_wall_s': time.perf_counter() - started,
|
||
}
|
||
if out:
|
||
with open(out, 'w', encoding='utf-8') as f:
|
||
json.dump(result, f, indent=2, allow_nan=False)
|
||
return result
|
||
|
||
|
||
@torch.no_grad()
|
||
def replay_certificates(path, indices, device='cpu', out=None, repeats=5, extra_segments=()):
|
||
saved = torch.load(path, map_location='cpu', weights_only=True)
|
||
kin = Kin(device=device, dtype=torch.float64)
|
||
if saved.get('kin_fp') != kin.fp:
|
||
raise ValueError('trace/kin fingerprint mismatch')
|
||
if not indices or min(indices) < 0 or max(indices) >= len(saved['tasks']) or repeats < 1:
|
||
raise ValueError('valid task indices and positive repeats required')
|
||
rows, offset = [], 0
|
||
for _, per, count in saved['batches']:
|
||
for index in indices:
|
||
if not offset <= index < offset + count:
|
||
continue
|
||
b = index - offset
|
||
nstep = min(per['traj'].shape[1] - 1, math.floor((float(per['limit'][b]) + 1e-6) / DT))
|
||
if torch.isfinite(per['t_done'][b]):
|
||
nstep = min(nstep, round(float(per['t_done'][b]) / DT))
|
||
traj = per['traj'][b, :nstep + 1]
|
||
moving = ((traj[1:] - traj[:-1]).abs().max(-1).values > 1e-6).nonzero().flatten().tolist()
|
||
if not moving:
|
||
continue
|
||
plans = per['plans']
|
||
nearest, nearest_gap = None, float('inf')
|
||
for it in moving:
|
||
pi = int(torch.searchsorted(plans['it'], it, right=True)) - 1
|
||
step = it - int(plans['it'][pi])
|
||
if 0 <= step < plans['commit'].shape[2] and plans['certified'][b, pi, step]:
|
||
gap = float(plans['gaps'][b, pi, [step, plans['commit'].shape[2] + step]].min())
|
||
if gap < nearest_gap:
|
||
nearest, nearest_gap = it, gap
|
||
selected = {moving[0], moving[len(moving) // 2], moving[-1]}
|
||
if nearest is not None:
|
||
selected.add(nearest)
|
||
for task_index, it in extra_segments:
|
||
if task_index == index:
|
||
if it not in moving:
|
||
raise ValueError('extra segment must be a moving segment before deadline or arrival')
|
||
selected.add(it)
|
||
for it in sorted(selected):
|
||
rows.append({'i': index, 'start_it': it, 'a': traj[it].tolist(), 'b': traj[it + 1].tolist()})
|
||
offset += count
|
||
if not rows:
|
||
raise ValueError('no moving segments in selected tasks')
|
||
a = torch.tensor([r['a'] for r in rows], dtype=torch.float64, device=device)
|
||
b = torch.tensor([r['b'] for r in rows], dtype=torch.float64, device=device)
|
||
started = time.perf_counter()
|
||
print('certificate replay: %d tasks, %d fixed moving segments' % (len(indices), len(rows)), flush=True)
|
||
summaries = {}
|
||
for budget in (2, 8, 32):
|
||
result = kin.certify_segments(a, b, max_queries=budget)
|
||
summaries[str(budget)] = {key: int(result[key].sum()) for key in ('safe', 'collision', 'unresolved', 'outside_box')}
|
||
summaries[str(budget)]['queries_p50'] = float(result['queries'].double().quantile(0.5))
|
||
summaries[str(budget)]['queries_max'] = int(result['queries'].max())
|
||
for j, row in enumerate(rows):
|
||
row.setdefault('budgets', {})[str(budget)] = {key: value[j].item() for key, value in result.items()}
|
||
print('budget %d: %s' % (budget, summaries[str(budget)]), flush=True)
|
||
latencies = []
|
||
for j, row in enumerate(rows):
|
||
aa, bb = a[j:j + 1], b[j:j + 1]
|
||
discrete = kin.gaps(torch.cat([aa, aa + 0.5 * (bb - aa), bb]), 0.5)['gm'].min(-1).values
|
||
row['previous_midpoint_endpoint_gap_mm'] = discrete.tolist()
|
||
nseg = max(2, 2 * math.ceil(float((bb - aa).abs().max()) / 0.02))
|
||
alpha = torch.linspace(0, 1, nseg + 1, dtype=torch.float64, device=device)
|
||
points = aa + alpha[:, None] * (bb - aa)
|
||
dense = torch.cat([kin.gaps(chunk, 0.5)['gm'].min(-1).values for chunk in points.split(64)])
|
||
row['dense_points'] = len(points)
|
||
row['dense_min_gap_mm'] = float(dense.min())
|
||
row['dense_min_fraction'] = float(alpha[dense.argmin()])
|
||
for _ in range(3):
|
||
kin.certify_segments(aa, bb)
|
||
elapsed = []
|
||
for _ in range(repeats):
|
||
if a.is_cuda:
|
||
torch.cuda.synchronize(a.device)
|
||
t0 = time.perf_counter()
|
||
result = kin.certify_segments(aa, bb)
|
||
if a.is_cuda:
|
||
torch.cuda.synchronize(a.device)
|
||
elapsed.append((time.perf_counter() - t0) * 1000)
|
||
row['batch_single_match'] = all(value.item() == row['budgets']['32'][key] for key, value in result.items())
|
||
row['latency_ms'] = elapsed
|
||
latencies.extend(elapsed)
|
||
print('segment %d/%d task %d it %d dense %.6f mm queries %d latency %.2f ms' %
|
||
(j + 1, len(rows), row['i'], row['start_it'], row['dense_min_gap_mm'],
|
||
row['budgets']['32']['queries'], sum(elapsed) / len(elapsed)), flush=True)
|
||
elapsed = torch.tensor(latencies, dtype=torch.float64)
|
||
result = {
|
||
'trace': str(path), 'tasks_sha256': saved.get('tasks_sha256'), 'kin_fp': kin.fp,
|
||
'device': str(device), 'gpu': torch.cuda.get_device_name(a.device) if a.is_cuda else None,
|
||
'torch_version': str(torch.__version__), 'cpu_threads': torch.get_num_threads(), 'dtype': 'float64',
|
||
'indices': indices, 'extra_segments': list(extra_segments),
|
||
'selection': 'First/middle/last moving segment and minimum original certified gap per task; plus explicit extra_segments.',
|
||
'cap_mm': 20.0, 'eps_mm': 1e-4, 'max_depth': 12, 'dense_max_step_deg': 0.01,
|
||
'timing': {'scope': 'B=1 certificate only, resident inputs, 3 warmups per segment; synchronized wall time, not policy or scheduling latency.',
|
||
'repeats_per_segment': repeats, 'calls': len(latencies),
|
||
'p50_ms': float(elapsed.quantile(0.5)), 'p99_ms': float(elapsed.quantile(0.99)),
|
||
'max_ms': float(elapsed.max()), 'over_20ms': int((elapsed > 20).sum())},
|
||
'budgets': summaries,
|
||
'dense_collision_segments': sum(r['dense_min_gap_mm'] <= 0 for r in rows),
|
||
'accepted_dense_collision_segments': sum(r['dense_min_gap_mm'] <= 0 and r['budgets']['32']['safe'] for r in rows),
|
||
'batch_single_match': all(r['batch_single_match'] for r in rows),
|
||
'scope': 'Fixed recorded segments, not a new rollout. Dense replay is finite sampling; floating-point certification error is not formally bounded.',
|
||
'wall_s': time.perf_counter() - started, 'rows': rows,
|
||
}
|
||
if out:
|
||
with open(out, 'w', encoding='utf-8') as f:
|
||
json.dump(result, f, indent=2, allow_nan=False)
|
||
return result
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument('--ckpt')
|
||
ap.add_argument('--trace')
|
||
ap.add_argument('--dense_indices', help='逗号分隔的任务下标；回放实际轨迹，不依赖原 col 标记')
|
||
ap.add_argument('--cert_indices', help='固定实际段的连续认证、加密对照和单段计时')
|
||
ap.add_argument('--cert_segments', default='', help='额外指定任务号:零基段号，逗号分隔')
|
||
ap.add_argument('--repeats', type=int, default=5)
|
||
ap.add_argument('--max_step_deg', type=float, default=0.05)
|
||
ap.add_argument('--samples', default='ml/data/samples')
|
||
ap.add_argument('--split', default='val')
|
||
ap.add_argument('--n', type=int, default=512)
|
||
ap.add_argument('--bs', type=int, default=64)
|
||
ap.add_argument('--chunk', type=int, default=64, help='回放轨迹分块步数（显存小就调小）')
|
||
ap.add_argument('--K', type=int, default=8)
|
||
ap.add_argument('--cap', type=float, default=0.5)
|
||
ap.add_argument('--device', default='cuda' if torch.cuda.is_available() else 'cpu')
|
||
ap.add_argument('--out', default=None)
|
||
a = ap.parse_args()
|
||
if a.dense_indices and a.cert_indices:
|
||
ap.error('--dense_indices and --cert_indices are mutually exclusive')
|
||
if a.cert_segments and not a.cert_indices:
|
||
ap.error('--cert_segments requires --cert_indices')
|
||
if a.trace:
|
||
if a.cert_indices:
|
||
extra = [tuple(map(int, segment.split(':'))) for segment in a.cert_segments.split(',') if segment]
|
||
indices = [int(i) for i in a.cert_indices.split(',')]
|
||
if any(len(segment) != 2 or segment[0] not in indices for segment in extra):
|
||
ap.error('--cert_segments requires task:segment pairs from --cert_indices')
|
||
replay_certificates(a.trace, indices, a.device, a.out, a.repeats, extra)
|
||
elif a.dense_indices:
|
||
replay_dense(a.trace, [int(i) for i in a.dense_indices.split(',')], a.max_step_deg, a.device, a.out)
|
||
else:
|
||
replay_first_hits(a.trace, a.out)
|
||
return
|
||
if a.dense_indices or a.cert_indices:
|
||
ap.error('--dense_indices and --cert_indices require --trace')
|
||
if not a.ckpt:
|
||
ap.error('--ckpt or --trace is required')
|
||
|
||
ck = torch.load(a.ckpt, map_location=a.device)
|
||
model = build(ck['cfg']).to(a.device)
|
||
model.load_state_dict(ck['model'])
|
||
model.eval()
|
||
kin = Kin(device=a.device)
|
||
if ck.get('fp') and ck['fp'] != kin.fp:
|
||
sys.exit(f'checkpoint 指纹 {ck["fp"]} 与 kin.json {kin.fp} 不符')
|
||
tasks = [t for t in load_tasks([a.samples]) if t['split'] == a.split][:a.n]
|
||
PN = pair_names()
|
||
rows = []
|
||
tau = [0, 0.5, 1.0, 2.0, 3.0, 5.0, 1e9]
|
||
hist = Counter()
|
||
pair_first = Counter(); pair_steps = Counter()
|
||
with torch.no_grad():
|
||
for i in range(0, len(tasks), a.bs):
|
||
chunk = tasks[i:i + a.bs]
|
||
res, per = rollout(model, kin, chunk, a.device, K=a.K, safety=False, seed=i, return_traj=True)
|
||
traj = per['traj'].to(a.device) # [B,T,6]
|
||
B, Tn, _ = traj.shape
|
||
first_t = torch.full((B,), float('nan'), device=a.device)
|
||
first_pair = torch.full((B,), -1, dtype=torch.long, device=a.device)
|
||
first_gap = torch.full((B,), float('nan'), device=a.device)
|
||
min_gap = torch.full((B,), float('inf'), device=a.device)
|
||
col_steps = torch.zeros(B, device=a.device)
|
||
for k in range(Tn):
|
||
qk = traj[:, k]
|
||
mid = 0.5 * (traj[:, max(0, k - 1)] + qk)
|
||
pts = torch.stack([qk, mid], 1).reshape(2 * B, 6) # 每帧只喂 2B 个点（与 rollout 同量级；一次 2048 点会爆显存）
|
||
gm = kin.gaps(pts, a.cap)['gm'].view(B, 2, 49)
|
||
gmin, gpair = gm.min(-1) # [B,2]
|
||
g1 = gmin.min(1).values
|
||
min_gap = torch.minimum(min_gap, g1)
|
||
newly = (g1 <= 0) & torch.isnan(first_t)
|
||
if bool(newly.any()):
|
||
for b in newly.nonzero().flatten().tolist():
|
||
pos = 0 if float(gmin[b, 0]) <= 0 else 1
|
||
first_t[b] = k * DT
|
||
first_pair[b] = gpair[b, pos]
|
||
first_gap[b] = gmin[b, pos]
|
||
if not bool(torch.isnan(first_t).any()):
|
||
break
|
||
for b, t in enumerate(chunk):
|
||
ft = float(first_t[b])
|
||
fp = int(first_pair[b])
|
||
name = PN.get(fp, '—')
|
||
if not math.isnan(ft):
|
||
pair_first[name] += 1
|
||
for j in range(len(tau) - 1):
|
||
if tau[j] <= ft < tau[j + 1]:
|
||
hist[f'{tau[j]}–{tau[j + 1]} s' if tau[j + 1] < 1e8 else '> 5 s'] += 1
|
||
break
|
||
rows.append({
|
||
'tid': t.get('tid'), 'cat': t.get('cat'), 'mode': t.get('mode'),
|
||
'collided': bool(per['col'][b] > 0),
|
||
'first_collision_s': None if math.isnan(ft) else round(ft, 2),
|
||
'first_pair': None if name == '—' else name, 'first_pair_idx': None if fp < 0 else fp,
|
||
'first_gap_mm': None if math.isnan(float(first_gap[b])) else round(float(first_gap[b]), 3),
|
||
'min_gap_mm': round(float(min_gap[b]), 3),
|
||
'collision_steps': int(per['col'][b]),
|
||
})
|
||
print(f'[diag] {min(i + a.bs, len(tasks))}/{len(tasks)} 已回放', flush=True)
|
||
|
||
collided = [r for r in rows if r['collided']]
|
||
def pct(v, p):
|
||
v = sorted(v); return v[min(len(v) - 1, int(len(v) * p))] if v else None
|
||
times = [r['first_collision_s'] for r in collided]
|
||
out = {
|
||
'ckpt': a.ckpt, 'step': ck.get('step'), 'n': len(rows), 'n_collided': len(collided),
|
||
'collision_rate': len(collided) / max(1, len(rows)),
|
||
'first_time_s': {'p10': pct(times, .1), 'p50': pct(times, .5), 'p90': pct(times, .9)},
|
||
'first_time_hist': dict(hist),
|
||
'first_pair_top': pair_first.most_common(),
|
||
'min_gap_mm_p10/p50': [pct([r['min_gap_mm'] for r in rows], .1), pct([r['min_gap_mm'] for r in rows], .5)],
|
||
'by_cat': {},
|
||
'tasks': rows,
|
||
}
|
||
for k in sorted({f"{r['cat']}/{r['mode']}" for r in rows}):
|
||
g = [r for r in rows if f"{r['cat']}/{r['mode']}" == k]
|
||
out['by_cat'][k] = {'n': len(g), 'first_t_p50': pct([r['first_collision_s'] for r in g if r['collided']], .5),
|
||
'first_pairs': Counter([r['first_pair'] for r in g if r['first_pair']]).most_common(3)}
|
||
s = json.dumps(out, indent=1, ensure_ascii=False)
|
||
if a.out:
|
||
open(a.out + '.tmp', 'w', encoding='utf-8').write(s); os.replace(a.out + '.tmp', a.out)
|
||
print('== 汇总 ==')
|
||
print('题数 %d，撞过 %d（%.1f%%）' % (len(rows), len(collided), 100 * out['collision_rate']))
|
||
print('首次碰撞时间 p10/p50/p90 = %s / %s / %s s' % (
|
||
out['first_time_s']['p10'], out['first_time_s']['p50'], out['first_time_s']['p90']))
|
||
print('首次碰撞时间分布：', out['first_time_hist'])
|
||
print('首次碰撞的连杆对 Top：', pair_first.most_common(8))
|
||
print('全程最小间隙 p10/p50 = %s / %s mm' % tuple(out['min_gap_mm_p10/p50']))
|
||
if a.out: print('已写', a.out)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|