Files
SystemSimulationApp/tests/manual/summarize_output_semantics.py
ljz 7611f13208 修复循环信号与事件采样并接入 LSTP 接触定位,补充八路验证及复用实验
相较上一版 Jacobian 确定性复用更新,本次补齐事件边界一致性、结果两侧采样及接触事件定位;保留已有物性复用和组件力学公式。

- 统一 UD00 信号求值与下一事件查询的绝对时间边界,修复循环边界浮点舍入导致的阶段错位、重复或漏报,并覆盖零时长、多阶段及长周期场景。
- 引入原生输出语义 v2:保留规则网格真实时间,补充内部时间事件和状态事件的左邻及事件后采样,按保存时间、状态和离散模式重放结果。
- 两条代码生成路径均发出 LSTP 接触描述,默认定位间隙过零及非负力模式的力截断;仅在接受事件时更新防重复记录,增加 contactEvents 诊断计数。
- 补充 MASS/LSTP 独立事件实验、八路全曲线与驱动阶段配对评估,以及 Amesim 不连续点输出对照和力差定位报告;MASS 新增释放机制仍保留为独立实验。
- 保存局部 probe、context 访问与回退、shadow replay、R288 real skip/typed replay 及阀门数值尾部诊断工具和报告;未证明净收益的实验不启用为生产默认优化。
- 更新原生运行说明和元件建模规范,补充信号边界、输出语义、接触事件和实验依赖回归测试。

验证:五组专项回归共 34 项全部通过;37 个待提交 Python 文件语法检查通过;git diff --cached --check 通过。
2026-09-17 23:50:13 +08:00

72 lines
4.1 KiB
Python

"""Audit the controlled output-only eight-branch experiment without hiding raw errors."""
import argparse
import json
from pathlib import Path
import numpy as np
def load(path):
return json.loads(path.read_bytes())
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('directory', type=Path)
args = parser.parse_args()
out, result = args.directory, {}
for name in ('full', 'noncyclic'):
before, after = (out / stage / name for stage in ('before', 'after'))
a, b = (load(p/'native/result.json') for p in (before, after))
at, bt = (np.asarray(r['series']['time']) for r in (a, b))
indices = np.searchsorted(bt, at)
assert np.array_equal(bt[indices], at), 'Every old sample must remain at its real timestamp'
differences = []
for key, values in a['series'].items():
old, new = np.asarray(values), np.asarray(b['series'][key])[indices]
if not np.array_equal(old, new):
differences.append(dict(key=key, maxAbsoluteDifference=float(np.max(np.abs(new-old)))))
counters = ('acceptedSteps', 'rejectedSteps', 'nfev', 'njev', 'nlu', 'solverStarts', 'stateTransitions')
counter_pairs = {key: [a[key], b[key]] for key in counters}
paired = np.load(after/'curves.npz')
contact_phase = {}
for key in paired.files:
if key.startswith('platform|') and key.endswith('.gap'):
other = 'amesim|' + key.split('|', 1)[1]
mismatch = paired['phaseMatched'] & ((paired[key]<0) != (paired[other]<0))
contact_phase[key.split('|', 1)[1]] = dict(mismatchCount=int(mismatch.sum()),
times=paired['time'][mismatch].tolist())
old_metrics = load(before/'raw-time-comparison.json')
new_raw = load(after/'raw-time-comparison.json')
aligned = load(after/'comparison.json')
quantities = {}
for quantity in ('signal', 'force', 'pressure', 'temperature', 'mass_flow', 'enthalpy_flow', 'velocity'):
groups = []
for rows in (old_metrics['curves'], new_raw['curves'], aligned['curves']):
rows = [r for r in rows if r['quantity'] == quantity]
worst = max(rows, key=lambda r: r['maxAbsoluteError'])
groups.append(dict(maxAbsoluteError=worst['maxAbsoluteError'], key=worst['key'],
time=worst['worstTime'], above5PercentCount=sum(r['above5PercentCount'] for r in rows)))
quantities[quantity] = dict(beforeRaw=groups[0], afterRaw=groups[1], afterSamePhase=groups[2])
result[name] = dict(finalStateExactlyEqual=a['finalState']==b['finalState'],
solverCountersEqual=all(x==y for x,y in counter_pairs.values()), counters=counter_pairs,
originalSampleCount=len(at), newSampleCount=len(bt), extraSamples=len(bt)-len(at),
allOriginalSamplesPreserved=True, changedOriginalOutputs=differences,
solveSeconds=[a['solveSeconds'], b['solveSeconds']],
processWallSeconds=[load(before/'native-summary.json')['processWallSeconds'],
load(after/'native-summary.json')['processWallSeconds']],
phasePairing=load(after/'phase-pairing.json'), contactPhase=contact_phase,
quantities=quantities,
above5PercentCount=[old_metrics['above5PercentCount'], new_raw['above5PercentCount'], aligned['above5PercentCount']],
above5PercentOutsideEvents=aligned['above5PercentOutsideEvents'],
rawPeakContactForce=[load(before/'comparison.json')['extraEventPointMaxContactForce'], aligned['extraEventPointMaxContactForce']],
sourceVerification={s: load(out/s/'source-verification.json') for s in ('before','after')})
(out/'effect-summary.json').write_text(json.dumps(result, ensure_ascii=False, indent=2, allow_nan=False)+'\n', encoding='utf-8')
for name, row in result.items():
print(name, json.dumps({k:v for k,v in row.items() if k not in ('phasePairing','quantities','sourceVerification')}, ensure_ascii=False))
print(json.dumps(row['quantities'], ensure_ascii=False))
if __name__ == '__main__':
main()