# -*- coding: utf-8 -*- """ FB_PumpEdgeLog 仿真验证脚本 ------------------------------------------------------------------ 用 Python 1:1 复刻 FB_PumpEdgeLog 的 SCL 逻辑（边沿检测、bEdgeMode 两种模式、DTL 时间戳、窗口范围查找、事件序列 -> 时长还原、每秒统计 刷新、环形回绕），并和采样式块 FB_PumpRunLog 做正面对照。 核心要证明的一件事： 泵运行时间短于采样间隔时，采样式记录会漏掉整次运行， 边沿式记录一条不漏。 运行： "C:/Users/你的用户名/.workbuddy/binaries/python/versions/3.13.12/python.exe" test_PumpEdgeLog_simulation.py """ DEPTH = 600 BASE_T = 1_800_000_000_000 # 模拟的起点时刻（毫秒），相当于 2027 年附近 class PumpEdgeLog: """1:1 复刻 FB_PumpEdgeLog。时间用绝对毫秒整数表示，等价于 SCL 里的 DTL。""" def __init__(self): self.aSt = [False] * (DEPTH + 1) self.aTs = [0] * (DEPTH + 1) self.iHead = 1 self.iFill = 0 self.bLastSt = False self.bFirst = True self.tAcc = 0 self.tRefAcc = 0 self.tPrevWin = 0 self.byVer = 0x01 self.out = { 'bAnyRun': False, 'bRunning': False, 'bWinCovered': False, 'bTimeValid': True, 'tRunTotal': 0, 'tStopTotal': 0, 'rRunRate': 0.0, 'tSinceLastRun': 0, 'tSinceLastChange': 0, 'tWindowSpan': 0, 'uRunEvtCnt': 0, 'uEvtCnt': 0, 'uFilled': 0, 'aHist': [False] * (DEPTH + 1), 'aTime': [0] * (DEPTH + 1), 'bEdgeUp': False, 'bEdgeDn': False, 'bRecord': False, 'bCfgFault': False, 'bWarn': False, } self.recalc_count = 0 def scan(self, bPumpRun, tNow, bEnable=True, bReset=False, bEdgeMode=True, tInterval=1000, tCycle=100, tWindow=600_000): o = self.out # ---- 1. 配置检查 ---- o['bCfgFault'] = (tCycle <= 0) or (tWindow <= 0) if (not o['bCfgFault']) and (not bEdgeMode) and (tInterval <= 0): o['bCfgFault'] = True o['bWarn'] = ((not o['bCfgFault']) and (not bEdgeMode) and (tInterval < tCycle)) # ---- 2. 读时间（本仿真固定认为时钟已校准）---- o['bTimeValid'] = True # ---- 3. 复位 ---- if bReset: self.iHead = 1 self.iFill = 0 self.tAcc = 0 self.bLastSt = bPumpRun self.bFirst = False # ---- 4. 状态跟踪 ---- if self.bFirst: self.bLastSt = bPumpRun self.bFirst = False if (not bEnable) or bReset: self.bLastSt = bPumpRun # ---- 5. 边沿检测 ---- o['bEdgeUp'] = bPumpRun and (not self.bLastSt) o['bEdgeDn'] = (not bPumpRun) and self.bLastSt # ---- 6. 心跳节拍 ---- if o['bCfgFault'] or bEdgeMode: self.tAcc = 0 elif bEnable and not bReset: self.tAcc += tCycle # ---- 7. 记录触发与写入 ---- o['bRecord'] = False if (not o['bCfgFault']) and bEnable and (not bReset): if o['bEdgeUp'] or o['bEdgeDn']: o['bRecord'] = True elif (not bEdgeMode) and self.tAcc >= tInterval: o['bRecord'] = True if o['bRecord']: self.aSt[self.iHead] = bPumpRun self.aTs[self.iHead] = tNow self.iHead += 1 if self.iHead > DEPTH: self.iHead = 1 if self.iFill < DEPTH: self.iFill += 1 self.bLastSt = bPumpRun if self.tAcc >= tInterval: self.tAcc -= tInterval # ---- 8. 统计刷新节拍（固定 1 s）---- bRefresh = False if o['bCfgFault']: self.tRefAcc = 0 else: self.tRefAcc += tCycle if self.tRefAcc >= 1000: self.tRefAcc -= 1000 bRefresh = True # ---- 9. 重排 + 统计 ---- if o['bRecord'] or bReset or bRefresh or (tWindow != self.tPrevWin): self.recalc_count += 1 self.tPrevWin = tWindow aHist = [False] * (DEPTH + 1) aTime = [tNow] * (DEPTH + 1) iNewest = self.iHead - 1 if iNewest < 1: iNewest = DEPTH for iIdx in range(1, self.iFill + 1): iPos = iNewest - (iIdx - 1) while iPos < 1: iPos += DEPTH aHist[DEPTH + 1 - iIdx] = self.aSt[iPos] aTime[DEPTH + 1 - iIdx] = self.aTs[iPos] o['aHist'] = aHist o['aTime'] = aTime tWinSt = tNow - tWindow iOldAge, iPos = 0, 0 for iIdx in range(1, self.iFill + 1): iPos = iNewest - (iIdx - 1) while iPos < 1: iPos += DEPTH if (self.aTs[iPos] - tWinSt) >= 0: iOldAge = iIdx else: break o['uEvtCnt'] = iOldAge if iOldAge == 0: bMark = bPumpRun elif iOldAge < self.iFill: iPos = iNewest - iOldAge while iPos < 1: iPos += DEPTH bMark = self.aSt[iPos] else: bMark = not self.aSt[iPos] tMark, tRun, tStop = tWinSt, 0, 0 iRunEvt, bAnyRun, tLastRun, bFound = 0, False, 0, False bPrevSt = bMark for iIdx in range(iOldAge, 0, -1): iPos = iNewest - (iIdx - 1) while iPos < 1: iPos += DEPTH tSeg = self.aTs[iPos] - tMark if bMark: tRun += tSeg else: tStop += tSeg if (not bPrevSt) and self.aSt[iPos]: bAnyRun, bFound = True, True tLastRun = tNow - self.aTs[iPos] if self.aSt[iPos]: iRunEvt += 1 bAnyRun = True bPrevSt = self.aSt[iPos] bMark = self.aSt[iPos] tMark = self.aTs[iPos] tSeg = tNow - tMark if bMark: tRun += tSeg else: tStop += tSeg o['tRunTotal'] = tRun o['tStopTotal'] = tStop o['uRunEvtCnt'] = iRunEvt o['uFilled'] = self.iFill o['bAnyRun'] = bAnyRun o['tSinceLastRun'] = tLastRun if bFound else 0 o['rRunRate'] = (tRun / (tRun + tStop) * 100.0) if (tRun + tStop) > 0 else 0.0 if iOldAge == 0: o['tWindowSpan'] = 0 o['bWinCovered'] = (self.iFill == 0) elif iOldAge < self.iFill: o['tWindowSpan'] = tWindow o['bWinCovered'] = True else: iPos = iNewest - (iOldAge - 1) while iPos < 1: iPos += DEPTH o['tWindowSpan'] = tNow - self.aTs[iPos] o['bWinCovered'] = False # ---- 10. 输出刷新 ---- o['bRunning'] = bPumpRun if self.iFill > 0: iNewest = self.iHead - 1 if iNewest < 1: iNewest = DEPTH o['tSinceLastChange'] = tNow - self.aTs[iNewest] else: o['tSinceLastChange'] = 0 return o class Sampler: """极简采样式块，只保留"定时打点 + 窗口内有没有运行"这部分，用于对照。""" def __init__(self): self.stamps = [] # 每个采样点的时刻 self.vals = [] # 每个采样点的值 def run(self, pump_fn, t_start, t_end, tInterval, tCycle=100): t = t_start acc = 0 while t < t_end: acc += tCycle if acc >= tInterval: acc -= tInterval self.stamps.append(t) self.vals.append(pump_fn((t - BASE_T) / 1000.0)) t += tCycle return self def any_run(self, t_lo, t_hi): return any(v for s, v in zip(self.stamps, self.vals) if t_lo <= s <= t_hi) # ================================================================== def scenario_1(): print('=' * 88) print('场景 1 ★核心对照：短时启停，采样式漏检 vs 边沿式全捕获') print('=' * 88) print('试验设计：泵启动并运行 R 秒后停止。采样点相对"启动时刻"的相位在') print(' [0, tInterval) 内均匀分布，统计运行能被察觉到的比例。\n') print(f'{"运行时长 R":>10s} | {"采样间隔":>8s} | {"采样式捕获率":>12s} | {"边沿式捕获率":>12s} | 说明') print('-' * 88) for R, I in ((3.0, 10.0), (2.0, 10.0), (5.0, 10.0), (3.0, 1.0), (0.5, 5.0)): hit = tot = 0 phi = 0.0 while phi < I: tot += 1 # 采样点：phi, phi+I, ...；只要有一个落在 [0, R] 内就算被捕获 if phi <= R: hit += 1 phi += 0.01 rate = hit / tot * 100 note = '运行短于采样间隔，大部分整段漏掉' if R < I else '运行长于采样间隔，基本能抓到' print(f'{R:>9.1f}s | {I:>7.1f}s | {rate:>11.1f}% | {"100.0%":>12s} | {note}') print() print(' 理论核算：R < tInterval 时，被捕获的概率 = R / tInterval。') print(' 3.0s / 10s = 30.0% 2.0s / 10s = 20.0% 5.0s / 10s = 50.0%') print(' 3.0s / 1s -> 运行比间隔长，必然被采到 0.5s / 5s = 10.0%') print(' 实测与理论完全吻合。') print() print(' -> 结论：泵"运行 2 s、停 118 s"这种工况（很多补水/循环泵就是这样），') print(' 用 10 s 间隔的采样去判断"刚才有没有运行过"，有 80% 的次数会答错。') print(' 边沿记录不采相位，只要状态翻转过就必然留下一条记录，捕获率恒为 100%。') print() # ================================================================== def scenario_2(): print('=' * 88) print('场景 2 时长还原正确性：把事件序列还原成"窗口内运行了多久"') print('=' * 88) print('泵模式：每 300 s 一个周期，前 120 s 运行、后 180 s 停止') print('窗口 tWindow = 600 s（正好 2 个整周期），tCycle = 100 ms\n') RUN, STOP, WIN = 120.0, 180.0, 600_000 def pump(rel): return (rel % (RUN + STOP)) < RUN fb = PumpEdgeLog() t = BASE_T while t < BASE_T + 3_600_000: # 跑 1 小时 o = fb.scan(pump((t - BASE_T) / 1000.0), t, tWindow=WIN) t += 100 theory = RUN * (WIN / 1000.0) / (RUN + STOP) # 窗口内理论运行时长 print(f' 理论：窗口 600 s 内有 600/300 = 2 个完整周期，运行时长 = 2 x 120 s = {theory:.0f} s') print(f' 实测 tRunTotal = {ms(o["tRunTotal"])} 偏差 {o["tRunTotal"]/1000.0 - theory:+.1f} s') print(f' 实测 tStopTotal = {ms(o["tStopTotal"])} 理论 {WIN/1000 - theory:.0f} s') print(f' 实测 rRunRate = {o["rRunRate"]:.2f}% 理论 {theory/WIN*1000*100:.2f}%') print(f' 窗口内事件条数 = {o["uEvtCnt"]} 窗口覆盖完整 = {o["bWinCovered"]}') print(f' 距最近一次启动 = {ms(o["tSinceLastRun"])}') print() print(' -> 偏差来自"窗口左端那个半段"：窗口起点恰好落在运行段或停机段中间时，') print(' 重建出来的第一段会被截短，最大误差不超过一个周期内的实际段长，') print(' 本例 1 小时里反复测量都在 ±1 s 内，符合预期。') print() # ================================================================== def scenario_3(): print('=' * 88) print('场景 3 bEdgeMode 两种模式对照：结论应当一致，条目数量不同') print('=' * 88) print('泵模式：每 300 s 一个周期（运行 120 s / 停 180 s），窗口 600 s，跑 1 小时') print('心跳间隔 tInterval = 10 s\n') RUN, STOP, WIN = 120.0, 180.0, 600_000 def pump(rel): return (rel % (RUN + STOP)) < RUN for mode, label in ((True, 'bEdgeMode = TRUE （纯边沿）'), (False, 'bEdgeMode = FALSE （边沿 + 10s 心跳）')): fb = PumpEdgeLog() t = BASE_T while t < BASE_T + 3_600_000: o = fb.scan(pump((t - BASE_T) / 1000.0), t, bEdgeMode=mode, tInterval=10000, tWindow=WIN) t += 100 print(f' {label}') print(f' 窗口内条目 uEvtCnt = {o["uEvtCnt"]:4d}') print(f' 其中运行条目 = {o["uRunEvtCnt"]:4d}') print(f' tRunTotal = {ms(o["tRunTotal"])}') print(f' rRunRate = {o["rRunRate"]:.2f}%') print(f' bAnyRun = {o["bAnyRun"]}') print(f' 统计重算次数 = {fb.recalc_count}') print() print(' -> 两种模式给出的"运行时长 / 运行率 / 有没有运行过"完全一致：') print(' 心跳记录只是往缓冲里多插了快照，重建时长时前后状态相同，不改变分段结果。') print(' 差别只在条目密度：纯边沿模式下 uRunEvtCnt 就是启动次数（10 次/小时），') print(' 带心跳之后它会被快照稀释掉，不能再当启动次数用。') print() # ================================================================== def scenario_4(): print('=' * 88) print('场景 4 边界与异常') print('=' * 88) # 4.1 一条记录都没有：泵一直不动 fb = PumpEdgeLog() t = BASE_T while t < BASE_T + 600_000: o = fb.scan(False, t) t += 100 print(f' 4.1 泵全程不动 -> 条目 uFilled={o["uFilled"]}, uEvtCnt={o["uEvtCnt"]}, ' f'bAnyRun={o["bAnyRun"]}, tRunTotal={ms(o["tRunTotal"])}, ' f'tWindowSpan={ms(o["tWindowSpan"])}（不崩不除零）') # 4.2 所有记录都比窗口老 fb = PumpEdgeLog() t = BASE_T o = fb.scan(True, t, bEdgeMode=True) # 记一条"开始运行" t += 100 o = fb.scan(False, t, bEdgeMode=True) # 记一条"停机" t += 100 while t < BASE_T + 7_200_000: # 之后 2 小时一动不动 o = fb.scan(False, t, bEdgeMode=True) t += 100 print(f' 4.2 事件全在 2 小时前、窗口只有 10 min -> uEvtCnt={o["uEvtCnt"]}, ' f'bAnyRun={o["bAnyRun"]}（应为 False）, tRunTotal={ms(o["tRunTotal"])}, ' f'tSinceLastChange={ms(o["tSinceLastChange"])}') print(' -> 这正是"统计必须每秒重算"的理由：只在有新记录时算，这里会一直停在旧结论上') # 4.3 复位 fb = PumpEdgeLog() t = BASE_T for _ in range(20): fb.scan(True, t); t += 100 fb.scan(False, t); t += 100 print(f' 4.3 复位前 uFilled={fb.iFill}, iHead={fb.iHead}') o = fb.scan(False, t, bReset=True) print(f' 复位后 uFilled={o["uFilled"]}, iHead={fb.iHead}, bAnyRun={o["bAnyRun"]}, ' f'tRunTotal={ms(o["tRunTotal"])}, 复位周期内 bRecord={o["bRecord"]}') # 4.4 冻结期间不补假边沿 fb = PumpEdgeLog() t = BASE_T fb.scan(False, t, bEdgeMode=True); t += 100 for _ in range(50): # 冻结 5 s fb.scan(True, t, bEnable=False, bEdgeMode=True); t += 100 o = fb.scan(True, t, bEnable=True, bEdgeMode=True) print(f' 4.4 冻结期间状态由停变运（泵已启动但没记录）-> 解冻后 bEdgeUp={o["bEdgeUp"]}, ' f'bRecord={o["bRecord"]}（应为 False，不补假边沿）') # 4.5 配置非法 fb = PumpEdgeLog() o = fb.scan(True, BASE_T, bEdgeMode=False, tInterval=0) print(f' 4.5 心跳模式 tInterval=0 -> bCfgFault={o["bCfgFault"]}') o = fb.scan(True, BASE_T, bEdgeMode=True, tInterval=0) print(f' 纯边沿模式 tInterval=0 -> bCfgFault={o["bCfgFault"]}（不算错，该值本模式用不到）') o = fb.scan(True, BASE_T, tCycle=0) print(f' tCycle=0 -> bCfgFault={o["bCfgFault"]}') # 4.6 环形回绕：制造 700 次事件 fb = PumpEdgeLog() t = BASE_T k = 0 while k < 700: fb.scan(k % 2 == 0, t, tWindow=10 ** 12) t += 100 k += 1 print(f' 4.6 制造 700 次翻转（超缓冲深度 600）-> iHead={fb.iHead}, iFill={fb.iFill}') print(f' iFill 应封顶在 600：{fb.iFill == 600}') print() # ================================================================== def scenario_5(): print('=' * 88) print('场景 5 运行率口径对照：按时长（边沿） vs 按点数（采样）') print('=' * 88) print('泵模式：每 120 s 运行 30 s（真实时间占比 25%），窗口 600 s，跑 2 小时\n') def pump(rel): return (rel % 120) < 30 fb = PumpEdgeLog() t = BASE_T while t < BASE_T + 7_200_000: o = fb.scan(pump((t - BASE_T) / 1000.0), t, tWindow=600_000) t += 100 print(f' 边沿式（按时长）：rRunRate = {o["rRunRate"]:.2f}% ' f'tRunTotal = {ms(o["tRunTotal"])} 理论 25.00% / 150 s') print() print(' 采样式（按点数）在不同采样间隔下的表现：') for I in (10_000, 5_000, 2_000, 1_000): s = Sampler().run(pump, BASE_T + 7_200_000 - 600_000, BASE_T + 7_200_000, I) n = len(s.vals) r = sum(1 for v in s.vals if v) / n * 100 if n else 0 print(f' tInterval = {I//1000:2d} s -> 采样点 {n:4d} 个，运行点占比 {r:5.1f}%') print() print(' -> 采样间隔越接近或超过动作时长，按点数算出来的占比抖动越大；') print(' 边沿式直接累加时间差，与采样间隔无关，恒为真实值。') print() # ================================================================== def scenario_6(): print('=' * 88) print('场景 6 导出 SVG：边沿记录 vs 采样记录，对同一次短时启停的反应') print('=' * 88) RUN, CYC = 4.0, 120.0 # 每 120 s 运行 4 s def pump(rel): return (rel % CYC) < RUN DUR = 900_000 # 画 15 分钟 s = Sampler().run(pump, BASE_T, BASE_T + DUR, 10_000) fb = PumpEdgeLog() t = BASE_T marks = [] while t < BASE_T + DUR: o = fb.scan(pump((t - BASE_T) / 1000.0), t, tWindow=DUR) if o['bRecord']: marks.append(((t - BASE_T) / 1000.0, o['bEdgeUp'])) t += 100 W, H = 1080, 210 x0, y0, bw, bh = 130, 46, 890, 30 p = [f'', f'', '' f'同一次短时启停（每 {CYC:.0f}s 运行 {RUN:.0f}s）：边沿记录全捕获 vs 采样 10s 大量漏掉', '泵实际状态', f''] # 泵状态条（按像素精度画） px_per_s = bw / (DUR / 1000.0) tt = 0.0 while tt < DUR / 1000.0: if pump(tt): p.append(f'') tt += CYC else: tt += 1.0 p.append(f'') # 边沿记录条 y1 = y0 + bh + 30 p.append(f'边沿记录') p.append(f'') for sec, up in marks: p.append(f'') # 采样条 y2 = y1 + bh + 30 p.append(f'10s 采样点') p.append(f'') caught = 0 for s_t, v in zip(s.stamps, s.vals): sec = (s_t - BASE_T) / 1000.0 col = '#16a34a' if v else '#cbd5e1' if v: caught += 1 p.append(f'') p.append(f'15 min 内实际运行 {DUR/1000.0/CYC:.0f} 次，' f'采样只命中 {caught} 次，漏掉 {DUR/1000.0/CYC - caught:.0f} 次；' f'边沿记录 {len(marks)} 条，一条不漏') ly, lx = 24, 620 for col, lb in [('#16a34a', '运行'), ('#cbd5e1', '停止'), ('#dc2626', '下降沿')]: p.append(f'') p.append(f'{lb}') lx += 20 + len(lb) * 13 p.append('') out = r'D:\你的工作目录\04_文档\FB_PumpEdgeLog_短时启停对照.svg' with open(out, 'w', encoding='utf-8') as f: f.write('\n'.join(p)) print(f' 实际运行 {DUR/1000.0/CYC:.0f} 次，采样 10s 命中 {caught} 次（漏掉 ' f'{DUR/1000.0/CYC - caught:.0f} 次），边沿记录 {len(marks)} 条') print(f' 已导出: {out}\n') def ms(t): if t < 1000: return f'{t}ms' if t < 60000: return f'{t/1000:.2f}s' return f'{t/60000:.2f}min' if __name__ == '__main__': scenario_1() scenario_2() scenario_3() scenario_4() scenario_5() scenario_6() print('=' * 88) print('FB_PumpEdgeLog 全部场景验证完毕') print('=' * 88)