拓冰网站定制 
AFT二次开发教程17实战一——管网水力批量分析端到端项目版本与事实声明产品与版本AFT Fathom 15 / AFT Impulse 12当前各通道机制引用官方Fathom 13帮助页正文当前版以官方文档为准。语言/环境Python 3.xpandas openpyxl sqlite3。本文目标读完能跑起一个从工况设计到包络报告的端到端批量水力分析项目并理解为什么AFT 侧是 GUI、Python 侧是自动这条分界必须写在流程里。所有模型、编号与数值为示例性建模不代表任何标准规定。一句话结论端到端批量水力分析项目 5 个 Python 阶段设计/生成/取数/落库/报告 3 个 AFT GUI 动作Import Excel Change Data / Start Batch Run / Excel Export Manager 配置其骨架是一本断点账本记录每个工况的状态planned → integrated → run → collected让跑到一半崩了变成只补跑缺口。〇、本篇要解决的认知问题Q1一个完整的批量水力分析项目哪些步骤能自动化、哪些必须在 GUIQ2为什么必须有一本账本它记录什么、怎么保证幂等Q3目录怎么组织才能让一批跑坏不至于污染上一批的好结果Q4断点续跑在 AFT 里到底续在哪里Q5整条流水线怎么验收一、机制解析17.1 自动化边界3 个 GUI 闸门这是全系列最重要的一张分工图。AFT 侧的三个动作永远是 GUI因为官方无公开 CLI/API铁律 2┌─────────────────────────── Python 全自动 ───────────────────────────┐ │ ① 设计矩阵 ② 生成 AFT Transfer ③ 取数清洗 ④ 落库 ⑤ 报告 │ └───────────────────────────────────────────────────────────────────┘ │ ▲ ▼ │ ┌──────────────────── AFT GUI 三闸门 ────────────────────┐ │ G1: File Import Excel Change Data 把变更灌进去 │ │ G2: File Start Batch Run 把工况跑出来 │ │ G3: File Excel Export Manager 配置 把出口定下来 │ └───────────────────────────────────────────────────────┘关键认识自动化不是把 AFT 也自动化掉而是把 AFT 前后的所有苦活自动化掉把 AFT 变成流水线上的三个稳定卡口。这也解释了为什么本系列反复讲数据面——因为你能自动化的部分恰恰是数据面。G3 的顺序陷阱再强调一次G3 必须在 G2 之前——批跑对话框里的 Excel 导出开关只有在已配置导出项时才可用第 03/06 篇。17.2 账本项目的中枢账本是一张表就用 sqlite每行一个工况字段含义case_id工况编号第 09 篇生成scenario全限定 Scenario Path Nameparams_json该工况的参数水平join 回设计statusplanned → integrated → run → collectedattempts尝试次数重试用updated_at时间戳状态机planned --导入成功-- integrated --批跑完成-- run --取数落库-- collected ^ | | | └──── 失败/回退 ────────┴────────────────────┘为什么账本是中枢幂等只有status ! collected的工况才需要处理可续跑崩溃后重跑跳过已完成可审计每个工况到哪一步了有据可查可重试attempts记录尝试次数超限转人工。17.3 目录规范一次作业一个总目录内含四类子目录 一本账本job_20260926/ ├── ledger.sqlite # 账本中枢 ├── design/ # ① 设计场景规划、变量定义 │ ├── variables.json │ └── ledger_design.json ├── transfer/ # ② 生成分批 AFT Transfer 工作簿 │ ├── transfer_block_01.xlsx │ └── ... ├── export/ # ③ 取数Excel Export 产物AFT 落到这里 │ └── results.xlsx ├── store/ # ④ 落库sqlite parquet 快照 │ ├── results.sqlite │ └── snapshots/ └── report/ # ⑤ 报告包络/差值/仪表盘 ├── envelope.xlsx └── summary.md铁律 7 落地整个 job 目录在本地磁盘不要放在网络盘/云同步目录里——因为里面装着.fth工作副本而官方明示网络/云同步会损坏模型第 04 篇。17.4 断点续跑续在哪AFT 侧有两个天然断点批跑可 Cancel已跑输出保留官方明示——所以跑到一半崩了时前 N 个场景的结果还在磁盘上Export Only (Do Not Run)可复用已有输出重导第 10 篇。于是续跑逻辑是读账本 → 找出 status ! collected 的工况 → 其中 status run 的跳过 G2只做取数G3 Export Only 或直接读已有导出件 → status in (planned,integrated) 的重新生成/导入G1 批跑G2注意取数本身是幂等的第 11 篇的 upsert所以重复取数没有副作用——这让续跑实现可以宁可多做、不可漏做。17.5 验收标准一个端到端项目跑完验收看四条验收项判据完整性账本里所有工况status collected守恒落库总行数 工况数 × 对象数 × 参数数可追溯每个 case 能 join 回参数水平每个结果能指回场景路径物理合理抽查若干工况的方向性合理第 07/11 篇的抽查17.6 幂等性为什么让续跑变简单端到端项目里“能不能重跑比跑得快不快重要得多。而幂等idempotent——即同一个操作做一次和做多次结果相同”——是让续跑变简单的根本原因。本项目里的幂等来自两处设计账本幂等case_id是主键重复design只会更新同一工况不会新增。所以我不确定上次设计有没有成功时再跑一遍就对了。落库幂等结果表以(scenario, object, parameter)为主键做 upsert重复collect只会覆盖旧值不会翻倍。所以我不确定这次取数漏没漏时再取一遍就对了。幂等带来的操作自由因为这两个动作都幂等续跑逻辑可以采取**“宁可多做、不可漏做”**的保守策略——凡是状态不明、状态可疑的工况一律重做。代价可控重复是免费的收益巨大不会漏。反例依赖 append 的后果如果落库用 append那么再取一遍会让数据翻倍于是你必须精确知道上次取到哪——这就把续跑从再跑一遍变成了精确记账难度陡增。幂等是把复杂问题变简单的一招。经验法则凡是可能被重复执行的动作都要设计成幂等。批跑链路里设计、生成、取数、落库、出报告——只有批跑本身不幂等跑两次是两倍时间其余的都可以做成幂等。这也解释了为什么账本要用状态位而非计数器来驱动。二、完整代码与逐行剖析代码 17-1aft_sweep_cli.py端到端 CLI# -*- coding: utf-8 -*- aft_sweep_cli.py —— 管网水力批量分析端到端 CLI幂等、可续跑 子命令 design 生成设计矩阵与账本 generate 生成分批 AFT Transfer 工作簿 → 之后执行 GUI 闸门 G1 status 打印账本状态 collect 取数清洗落库幂等 ← 在 GUI 闸门 G3 之后 report 输出包络与差值报告 运行python aft_sweep_cli.py --selftest importargparseimportitertoolsimportjsonimportosimportsqlite3importsysfromopenpyxlimportWorkbook SHEETAFT TransferCOLS[Apply,Object Type,Object Number,Parameter,Change Code,Value,Scenario Path Name]STATES(planned,integrated,run,collected)classLedger:断点账本sqlite 实现天然幂等同一 case_id 覆盖更新。DDL(CREATE TABLE IF NOT EXISTS ledger(case_id TEXT PRIMARY KEY, scenario TEXT, params_json TEXT,status TEXT, attempts INTEGER DEFAULT 0, updated_at TEXT))def__init__(self,path):self.consqlite3.connect(path)self.con.execute(self.DDL)defupsert(self,cid,scn,params,statusplanned):curself.con.execute(SELECT status FROM ledger WHERE case_id?,(cid,))rowcur.fetchone()ifrowisNone:self.con.execute(INSERT INTO ledger VALUES(?,?,?,?,0,datetime(now)),(cid,scn,json.dumps(params,ensure_asciiFalse),status))else:self.con.execute(UPDATE ledger SET scenario?, params_json?, updated_atdatetime(now) WHERE case_id?,(scn,json.dumps(params,ensure_asciiFalse),cid))self.con.commit()defset_status(self,cid,status):assertstatusinSTATES,status self.con.execute(UPDATE ledger SET status?, attemptsattempts1, updated_atdatetime(now) WHERE case_id?,(status,cid))self.con.commit()defpending(self,target(planned,integrated)):qSELECT case_id, scenario FROM ledger WHERE status IN (%s)%\,.join(?*len(target))returnself.con.execute(q,target).fetchall()defsummary(self):returndict(self.con.execute(SELECT status, COUNT(*) FROM ledger GROUP BY status).fetchall())defcmd_design(variables,base,jobdir):os.makedirs(jobdir,exist_okTrue)ledLedger(os.path.join(jobdir,ledger.sqlite))n0forcomboinitertools.product(*[v[levels]forvinvariables]):cidC_.join(str(c).replace(.,p)forcincombo)scnf{base}\\{cid}led.upsert(cid,scn,{v[name]:lvlforv,lvlinzip(variables,combo)})n1returnndefcmd_generate(variables,jobdir,block20):ledLedger(os.path.join(jobdir,ledger.sqlite))rowsled.pending()outdiros.path.join(jobdir,transfer)os.makedirs(outdir,exist_okTrue)# 按场景分块第 09 篇保证同一场景的行不跨块by_scn{}forcid,scninrows:paramsjson.loads(led.con.execute(SELECT params_json FROM ledger WHERE case_id?,(cid,)).fetchone()[0])by_scn[scn][]forvinvariables:by_scn[scn].append({Apply:Yes,Object Type:v[object_type],Object Number:v[object_number],Parameter:v[parameter],Change Code:v[change_code],Value:params[v[name]],Scenario Path Name:scn})sceneslist(by_scn)files[]foriinrange(0,len(scenes),block):wbWorkbook();wswb.active;ws.titleSHEET;ws.append(COLS)forscninscenes[i:iblock]:forrinby_scn[scn]:ws.append([r[c]forcinCOLS])pathos.path.join(outdir,ftransfer_block_{i//block1:02d}.xlsx)wb.save(path);files.append(path)forscninscenes:cidled.con.execute(SELECT case_id FROM ledger WHERE scenario?,(scn,)).fetchone()[0]led.set_status(cid,integrated)returnfilesdefcmd_collect(jobdir,long_csvNone):取数落库幂等 upsert语义同第 11 篇。importpandasaspd ledLedger(os.path.join(jobdir,ledger.sqlite))dbos.path.join(jobdir,store,results.sqlite)os.makedirs(os.path.dirname(db),exist_okTrue)consqlite3.connect(db)try:con.execute(CREATE TABLE IF NOT EXISTS results(scenario TEXT, object TEXT, parameter TEXT, unit TEXT, value REAL,PRIMARY KEY(scenario,object,parameter)))iflong_csvandos.path.exists(long_csv):dfpd.read_csv(long_csv,encodingutf-8-sig)con.executemany(INSERT INTO results VALUES(?,?,?,?,?) ON CONFLICT(scenario,object,parameter) DO UPDATE SET unitexcluded.unit, valueexcluded.value,df[[scenario,object,parameter,unit,value]].itertuples(indexFalse,nameNone))con.commit()forscnindf[scenario].unique():rled.con.execute(SELECT case_id FROM ledger WHERE scenario?,(scn,)).fetchone()ifr:led.set_status(r[0],collected)returncon.execute(SELECT COUNT(*) FROM results).fetchone()[0]finally:con.close()defselftest():jobselftest_jobvars_[{name:pump_speed,object_type:Pump,object_number:3,parameter:Fixed Speed (%),change_code:Set equal to value,levels:[80.0,90.0,100.0]}]ifos.path.isdir(job):importshutil;shutil.rmtree(job)assertcmd_design(vars_,rB\U,job)3filescmd_generate(vars_,job,block2)assertlen(files)2,files# 3 场景 / 每块2 - 2 块ledLedger(os.path.join(job,ledger.sqlite))assertled.summary(){integrated:3},led.summary()print(SELFTEST OK设计 3 工况生成 2 块账本全部为 integrated。)print(后续动作G1 导入 → G2 批跑 → 取数 CSV → collect → report)defmain():apargparse.ArgumentParser()ap.add_argument(cmd,choices[design,generate,status,collect,report])ap.add_argument(--job,defaultjob)ap.add_argument(--base,defaultrBase Scenario\US Units)ap.add_argument(--long,helpcollect 用的长表 CSV)ap.add_argument(--selftest,actionstore_true)aap.parse_args()ifa.selftest:selftest()return0ifa.cmdstatus:print(json.dumps(Ledger(os.path.join(a.job,ledger.sqlite)).summary(),ensure_asciiFalse))else:print(f执行{a.cmd}完整实现见正文逐行剖析示例变量定义需按项目替换)return0if__name____main__:sys.exit(main())逐行剖析Ledger类是整个项目的中枢。upsert()用 sqlite 主键实现重复 design 不产生重复工况set_status()强制状态必须是STATES之一assert防止状态机被写歪。pending()默认返回planned/integrated的工况——续跑的核心查询只处理还没跑完的。cmd_design()与cmd_generate()分开是为了让 G1人工导入插在中间design 出账本 → generate 出工作簿 →人去 GUI 导→ 回来继续。cmd_generate()按场景分块第 09 篇的正确性点生成后把状态推到integrated表示变更表已备好、等你导入。cmd_collect()用upsert第 11 篇保证幂等并把对应场景状态推到collected——这是重复取数没副作用的实现。selftest()断言设计 3 工况、3 工况按每块 2 个分成 2 块、账本全为integrated——三个可手算的数字。代码 17-2README_job.md作业目录的自述模板# -*- coding: utf-8 -*-make_job_readme.py —— 为每个作业目录生成自述文件人接手时不用问TMPL# 作业 {job} ## 状态 - 账本ledger.sqlite状态planned/integrated/run/collected - 产品与版本AFT Fathom 15 / AFT Impulse 12机制依据 Fathom 13 官方帮助 ## 执行顺序严格 1. python aft_sweep_cli.py design --job {job} 2. python aft_sweep_cli.py generate --job {job} 3. **GUI 闸门 G1**File Import Excel Change Data逐个导入 transfer/*.xlsx确认 Object Change Log 零错误 4. **GUI 闸门 G3**File Excel Export Manager配好导出项先于 G2 5. **GUI 闸门 G2**File Start Batch RunScenarios in Current Model后台运行勾 Save Using Excel Export Manager 6. python aft_sweep_cli.py collect --job {job} --long export/长表.csv 7. python aft_sweep_cli.py report --job {job} ## 纪律 - 本地磁盘操作模型文件禁止放网络盘/云同步铁律 7 - 批跑期间不要打开目标工作簿每场景保存会撞锁 - 批跑前确认导出项已配置否则 G2 里的 Excel 导出开关不可用 defrender(job:str)-str:returnTMPL.format(jobjob)if__name____main__:print(render(job_20260926))为什么把流程写成 README 模板端到端项目的失败多半不是代码错而是步骤顺序错G3 排在 G2 后面、忘了确认日志。把顺序固化成作业自述文件接手的人照做即可。三、常见报错与排查报错 17-1续跑时把已经跑好的工况又跑了一遍白等两小时。现象重复批跑。根因续跑没读账本状态。解法用pending()只取未完成工况对statusrun的跳过 G2 只取数。报错 17-2落库行数比预期多/少。现象数据对不上。根因多了→用了 append 而非 upsert少了→有工况没取到导出件缺该场景或该场景没输出。解法以主键 upsert取数后断言总行数 工况数 × 对象数 × 参数数。报错 17-3作业目录放在云盘跑到一半模型损坏。现象Cant find [TOOLBOX PREFERENCES] heading。根因违反铁律 7云同步/网络盘保存时断连损坏模型。解法作业目录放本地磁盘需要共享时下载→本地算→上传。报错 17-4G2 批跑时 Excel 导出选项是灰的。现象无法勾 Save Using Excel Export Manager。根因G3 没先做导出项未配置。解法先 G3 后 G2这是流水线顺序的硬性点。报错 17-5账本状态混乱不知道某工况到底跑到哪。现象状态与事实不符。根因手工改了状态但没改数据或反之。解法状态只能由脚本推进set_status手工干预要同步改账本。报错 17-6批量跑到第 N 个场景报错之后全部失败。现象一个工况的问题连累整批。根因某工况的变更表带了非法值如Special Condition写错或某场景 Analysis Setup 不完整。解法导入前用第 05/15 篇的预检拦非法行批跑前做场景完整性自检不完整者先从清单剔除单工况失败不影响其余批跑是顺序的已跑完的输出保留。四、动手练习练习 1跑通骨架跑python aft_sweep_cli.py --selftest。判定输出SELFTEST OKselftest_job/下含ledger.sqlite与transfer/里 2 个工作簿。练习 2真机三闸门用你自己定义的变量走一遍 design → generate →G1/G3/G2→ collect → report。判定账本最终全部collected落库总行数 工况数 × 对象数 × 参数数写出乘法式。练习 3断点续跑在 G2 中途中止只对剩余工况生成清单并补跑。判定账本里已跑的仍是run/collected未跑的补到collected最终无planned/integrated残留。练习 4作业自述用make_job_readme.py为你的作业生成 README。判定README 含 7 步执行顺序与 3 条纪律把实际执行中与 README 不符之处记下来并修订 README。练习 5幂等验证对同一作业连续执行两次design与两次collect。判定ledger.sqlite的工况总数两次相同不翻倍results.sqlite的COUNT(*)两次相同写出幂等来源的两条机制。五、小结与下一篇预告本篇把前 16 篇缝成了端到端项目5 个 Python 阶段 3 个 AFT GUI 闸门中枢是一本状态机账本planned → integrated → run → collected续跑靠只处理未完成工况 取数幂等目录规范让一批跑坏不污染上一批验收看完整性/守恒/可追溯/物理合理四条。你现在拥有一台能跑几百个工况的批量水力分析机器。第 18 篇《实战二Fathom → Impulse 水锤分析自动化流水线》我们在这台机器上加装瞬态段——稳态批跑 → 转 Impulse → 阀关闭/泵停机工况 → 瞬态批跑 → Force File 交付强调瞬态的时间步与收敛陷阱并给出surge_pipeline.py。FAQ与第〇节一一对应Q1批量水力分析项目哪些步骤能自动化、哪些必须在 GUIAPython 侧可全自动的是设计矩阵、生成 AFT Transfer、取数清洗、落库、出报告五个阶段必须人工在 GUI 完成的是三个闸门——File Import Excel Change Data、File Excel Export Manager 配置、File Start Batch Run因为 AFT 无公开 CLI/API。Q2为什么必须有账本它记录什么A账本是项目中枢记录每个工况的 case_id、全限定场景路径、参数水平、状态planned/integrated/run/collected、尝试次数与时间戳它让续跑只处理未完成工况、让重跑幂等、让流程可审计、让失败可重试。Q3目录怎么组织才安全A一次作业一个总目录含 ledger.sqlite 与 design/transfer/export/store/report 五个子目录整个作业目录必须放在本地磁盘因为里面含模型工作副本而官方明示网络盘或云同步目录保存会损坏模型铁律 7。Q4断点续跑在 AFT 里续在哪里A续在两个天然断点上——Start Batch Run 可 Cancel 且已跑完的场景输出仍保留以及 Export Only (Do Not Run) 可复用已有输出重导配合账本只处理 status 不为 collected 的工况status 为 run 的跳过批跑只做取数。Q5整条流水线怎么验收A四条判据完整性账本所有工况 status 为 collected、守恒落库总行数等于工况数×对象数×参数数、可追溯每个 case 能 join 回参数水平、结果能指回场景路径、物理合理抽查若干工况方向性合理。 AFT二次开发教程（17）：实战一——管网水力批量分析端到端项目
AFT二次开发教程17实战一——管网水力批量分析端到端项目版本与事实声明产品与版本AFT Fathom 15 / AFT Impulse 12当前各通道机制引用官方Fathom 13帮助页正文当前版以官方文档为准。语言/环境Python 3.xpandas openpyxl sqlite3。本文目标读完能跑起一个从工况设计到包络报告的端到端批量水力分析项目并理解为什么AFT 侧是 GUI、Python 侧是自动这条分界必须写在流程里。所有模型、编号与数值为示例性建模不代表任何标准规定。一句话结论端到端批量水力分析项目 5 个 Python 阶段设计/生成/取数/落库/报告 3 个 AFT GUI 动作Import Excel Change Data / Start Batch Run / Excel Export Manager 配置其骨架是一本断点账本记录每个工况的状态planned → integrated → run → collected让跑到一半崩了变成只补跑缺口。〇、本篇要解决的认知问题Q1一个完整的批量水力分析项目哪些步骤能自动化、哪些必须在 GUIQ2为什么必须有一本账本它记录什么、怎么保证幂等Q3目录怎么组织才能让一批跑坏不至于污染上一批的好结果Q4断点续跑在 AFT 里到底续在哪里Q5整条流水线怎么验收一、机制解析17.1 自动化边界3 个 GUI 闸门这是全系列最重要的一张分工图。AFT 侧的三个动作永远是 GUI因为官方无公开 CLI/API铁律 2┌─────────────────────────── Python 全自动 ───────────────────────────┐ │ ① 设计矩阵 ② 生成 AFT Transfer ③ 取数清洗 ④ 落库 ⑤ 报告 │ └───────────────────────────────────────────────────────────────────┘ │ ▲ ▼ │ ┌──────────────────── AFT GUI 三闸门 ────────────────────┐ │ G1: File Import Excel Change Data 把变更灌进去 │ │ G2: File Start Batch Run 把工况跑出来 │ │ G3: File Excel Export Manager 配置 把出口定下来 │ └───────────────────────────────────────────────────────┘关键认识自动化不是把 AFT 也自动化掉而是把 AFT 前后的所有苦活自动化掉把 AFT 变成流水线上的三个稳定卡口。这也解释了为什么本系列反复讲数据面——因为你能自动化的部分恰恰是数据面。G3 的顺序陷阱再强调一次G3 必须在 G2 之前——批跑对话框里的 Excel 导出开关只有在已配置导出项时才可用第 03/06 篇。17.2 账本项目的中枢账本是一张表就用 sqlite每行一个工况字段含义case_id工况编号第 09 篇生成scenario全限定 Scenario Path Nameparams_json该工况的参数水平join 回设计statusplanned → integrated → run → collectedattempts尝试次数重试用updated_at时间戳状态机planned --导入成功-- integrated --批跑完成-- run --取数落库-- collected ^ | | | └──── 失败/回退 ────────┴────────────────────┘为什么账本是中枢幂等只有status ! collected的工况才需要处理可续跑崩溃后重跑跳过已完成可审计每个工况到哪一步了有据可查可重试attempts记录尝试次数超限转人工。17.3 目录规范一次作业一个总目录内含四类子目录 一本账本job_20260926/ ├── ledger.sqlite # 账本中枢 ├── design/ # ① 设计场景规划、变量定义 │ ├── variables.json │ └── ledger_design.json ├── transfer/ # ② 生成分批 AFT Transfer 工作簿 │ ├── transfer_block_01.xlsx │ └── ... ├── export/ # ③ 取数Excel Export 产物AFT 落到这里 │ └── results.xlsx ├── store/ # ④ 落库sqlite parquet 快照 │ ├── results.sqlite │ └── snapshots/ └── report/ # ⑤ 报告包络/差值/仪表盘 ├── envelope.xlsx └── summary.md铁律 7 落地整个 job 目录在本地磁盘不要放在网络盘/云同步目录里——因为里面装着.fth工作副本而官方明示网络/云同步会损坏模型第 04 篇。17.4 断点续跑续在哪AFT 侧有两个天然断点批跑可 Cancel已跑输出保留官方明示——所以跑到一半崩了时前 N 个场景的结果还在磁盘上Export Only (Do Not Run)可复用已有输出重导第 10 篇。于是续跑逻辑是读账本 → 找出 status ! collected 的工况 → 其中 status run 的跳过 G2只做取数G3 Export Only 或直接读已有导出件 → status in (planned,integrated) 的重新生成/导入G1 批跑G2注意取数本身是幂等的第 11 篇的 upsert所以重复取数没有副作用——这让续跑实现可以宁可多做、不可漏做。17.5 验收标准一个端到端项目跑完验收看四条验收项判据完整性账本里所有工况status collected守恒落库总行数 工况数 × 对象数 × 参数数可追溯每个 case 能 join 回参数水平每个结果能指回场景路径物理合理抽查若干工况的方向性合理第 07/11 篇的抽查17.6 幂等性为什么让续跑变简单端到端项目里“能不能重跑比跑得快不快重要得多。而幂等idempotent——即同一个操作做一次和做多次结果相同”——是让续跑变简单的根本原因。本项目里的幂等来自两处设计账本幂等case_id是主键重复design只会更新同一工况不会新增。所以我不确定上次设计有没有成功时再跑一遍就对了。落库幂等结果表以(scenario, object, parameter)为主键做 upsert重复collect只会覆盖旧值不会翻倍。所以我不确定这次取数漏没漏时再取一遍就对了。幂等带来的操作自由因为这两个动作都幂等续跑逻辑可以采取**“宁可多做、不可漏做”**的保守策略——凡是状态不明、状态可疑的工况一律重做。代价可控重复是免费的收益巨大不会漏。反例依赖 append 的后果如果落库用 append那么再取一遍会让数据翻倍于是你必须精确知道上次取到哪——这就把续跑从再跑一遍变成了精确记账难度陡增。幂等是把复杂问题变简单的一招。经验法则凡是可能被重复执行的动作都要设计成幂等。批跑链路里设计、生成、取数、落库、出报告——只有批跑本身不幂等跑两次是两倍时间其余的都可以做成幂等。这也解释了为什么账本要用状态位而非计数器来驱动。二、完整代码与逐行剖析代码 17-1aft_sweep_cli.py端到端 CLI# -*- coding: utf-8 -*- aft_sweep_cli.py —— 管网水力批量分析端到端 CLI幂等、可续跑 子命令 design 生成设计矩阵与账本 generate 生成分批 AFT Transfer 工作簿 → 之后执行 GUI 闸门 G1 status 打印账本状态 collect 取数清洗落库幂等 ← 在 GUI 闸门 G3 之后 report 输出包络与差值报告 运行python aft_sweep_cli.py --selftest importargparseimportitertoolsimportjsonimportosimportsqlite3importsysfromopenpyxlimportWorkbook SHEETAFT TransferCOLS[Apply,Object Type,Object Number,Parameter,Change Code,Value,Scenario Path Name]STATES(planned,integrated,run,collected)classLedger:断点账本sqlite 实现天然幂等同一 case_id 覆盖更新。DDL(CREATE TABLE IF NOT EXISTS ledger(case_id TEXT PRIMARY KEY, scenario TEXT, params_json TEXT,status TEXT, attempts INTEGER DEFAULT 0, updated_at TEXT))def__init__(self,path):self.consqlite3.connect(path)self.con.execute(self.DDL)defupsert(self,cid,scn,params,statusplanned):curself.con.execute(SELECT status FROM ledger WHERE case_id?,(cid,))rowcur.fetchone()ifrowisNone:self.con.execute(INSERT INTO ledger VALUES(?,?,?,?,0,datetime(now)),(cid,scn,json.dumps(params,ensure_asciiFalse),status))else:self.con.execute(UPDATE ledger SET scenario?, params_json?, updated_atdatetime(now) WHERE case_id?,(scn,json.dumps(params,ensure_asciiFalse),cid))self.con.commit()defset_status(self,cid,status):assertstatusinSTATES,status self.con.execute(UPDATE ledger SET status?, attemptsattempts1, updated_atdatetime(now) WHERE case_id?,(status,cid))self.con.commit()defpending(self,target(planned,integrated)):qSELECT case_id, scenario FROM ledger WHERE status IN (%s)%\,.join(?*len(target))returnself.con.execute(q,target).fetchall()defsummary(self):returndict(self.con.execute(SELECT status, COUNT(*) FROM ledger GROUP BY status).fetchall())defcmd_design(variables,base,jobdir):os.makedirs(jobdir,exist_okTrue)ledLedger(os.path.join(jobdir,ledger.sqlite))n0forcomboinitertools.product(*[v[levels]forvinvariables]):cidC_.join(str(c).replace(.,p)forcincombo)scnf{base}\\{cid}led.upsert(cid,scn,{v[name]:lvlforv,lvlinzip(variables,combo)})n1returnndefcmd_generate(variables,jobdir,block20):ledLedger(os.path.join(jobdir,ledger.sqlite))rowsled.pending()outdiros.path.join(jobdir,transfer)os.makedirs(outdir,exist_okTrue)# 按场景分块第 09 篇保证同一场景的行不跨块by_scn{}forcid,scninrows:paramsjson.loads(led.con.execute(SELECT params_json FROM ledger WHERE case_id?,(cid,)).fetchone()[0])by_scn[scn][]forvinvariables:by_scn[scn].append({Apply:Yes,Object Type:v[object_type],Object Number:v[object_number],Parameter:v[parameter],Change Code:v[change_code],Value:params[v[name]],Scenario Path Name:scn})sceneslist(by_scn)files[]foriinrange(0,len(scenes),block):wbWorkbook();wswb.active;ws.titleSHEET;ws.append(COLS)forscninscenes[i:iblock]:forrinby_scn[scn]:ws.append([r[c]forcinCOLS])pathos.path.join(outdir,ftransfer_block_{i//block1:02d}.xlsx)wb.save(path);files.append(path)forscninscenes:cidled.con.execute(SELECT case_id FROM ledger WHERE scenario?,(scn,)).fetchone()[0]led.set_status(cid,integrated)returnfilesdefcmd_collect(jobdir,long_csvNone):取数落库幂等 upsert语义同第 11 篇。importpandasaspd ledLedger(os.path.join(jobdir,ledger.sqlite))dbos.path.join(jobdir,store,results.sqlite)os.makedirs(os.path.dirname(db),exist_okTrue)consqlite3.connect(db)try:con.execute(CREATE TABLE IF NOT EXISTS results(scenario TEXT, object TEXT, parameter TEXT, unit TEXT, value REAL,PRIMARY KEY(scenario,object,parameter)))iflong_csvandos.path.exists(long_csv):dfpd.read_csv(long_csv,encodingutf-8-sig)con.executemany(INSERT INTO results VALUES(?,?,?,?,?) ON CONFLICT(scenario,object,parameter) DO UPDATE SET unitexcluded.unit, valueexcluded.value,df[[scenario,object,parameter,unit,value]].itertuples(indexFalse,nameNone))con.commit()forscnindf[scenario].unique():rled.con.execute(SELECT case_id FROM ledger WHERE scenario?,(scn,)).fetchone()ifr:led.set_status(r[0],collected)returncon.execute(SELECT COUNT(*) FROM results).fetchone()[0]finally:con.close()defselftest():jobselftest_jobvars_[{name:pump_speed,object_type:Pump,object_number:3,parameter:Fixed Speed (%),change_code:Set equal to value,levels:[80.0,90.0,100.0]}]ifos.path.isdir(job):importshutil;shutil.rmtree(job)assertcmd_design(vars_,rB\U,job)3filescmd_generate(vars_,job,block2)assertlen(files)2,files# 3 场景 / 每块2 - 2 块ledLedger(os.path.join(job,ledger.sqlite))assertled.summary(){integrated:3},led.summary()print(SELFTEST OK设计 3 工况生成 2 块账本全部为 integrated。)print(后续动作G1 导入 → G2 批跑 → 取数 CSV → collect → report)defmain():apargparse.ArgumentParser()ap.add_argument(cmd,choices[design,generate,status,collect,report])ap.add_argument(--job,defaultjob)ap.add_argument(--base,defaultrBase Scenario\US Units)ap.add_argument(--long,helpcollect 用的长表 CSV)ap.add_argument(--selftest,actionstore_true)aap.parse_args()ifa.selftest:selftest()return0ifa.cmdstatus:print(json.dumps(Ledger(os.path.join(a.job,ledger.sqlite)).summary(),ensure_asciiFalse))else:print(f执行{a.cmd}完整实现见正文逐行剖析示例变量定义需按项目替换)return0if__name____main__:sys.exit(main())逐行剖析Ledger类是整个项目的中枢。upsert()用 sqlite 主键实现重复 design 不产生重复工况set_status()强制状态必须是STATES之一assert防止状态机被写歪。pending()默认返回planned/integrated的工况——续跑的核心查询只处理还没跑完的。cmd_design()与cmd_generate()分开是为了让 G1人工导入插在中间design 出账本 → generate 出工作簿 →人去 GUI 导→ 回来继续。cmd_generate()按场景分块第 09 篇的正确性点生成后把状态推到integrated表示变更表已备好、等你导入。cmd_collect()用upsert第 11 篇保证幂等并把对应场景状态推到collected——这是重复取数没副作用的实现。selftest()断言设计 3 工况、3 工况按每块 2 个分成 2 块、账本全为integrated——三个可手算的数字。代码 17-2README_job.md作业目录的自述模板# -*- coding: utf-8 -*-make_job_readme.py —— 为每个作业目录生成自述文件人接手时不用问TMPL# 作业 {job} ## 状态 - 账本ledger.sqlite状态planned/integrated/run/collected - 产品与版本AFT Fathom 15 / AFT Impulse 12机制依据 Fathom 13 官方帮助 ## 执行顺序严格 1. python aft_sweep_cli.py design --job {job} 2. python aft_sweep_cli.py generate --job {job} 3. **GUI 闸门 G1**File Import Excel Change Data逐个导入 transfer/*.xlsx确认 Object Change Log 零错误 4. **GUI 闸门 G3**File Excel Export Manager配好导出项先于 G2 5. **GUI 闸门 G2**File Start Batch RunScenarios in Current Model后台运行勾 Save Using Excel Export Manager 6. python aft_sweep_cli.py collect --job {job} --long export/长表.csv 7. python aft_sweep_cli.py report --job {job} ## 纪律 - 本地磁盘操作模型文件禁止放网络盘/云同步铁律 7 - 批跑期间不要打开目标工作簿每场景保存会撞锁 - 批跑前确认导出项已配置否则 G2 里的 Excel 导出开关不可用 defrender(job:str)-str:returnTMPL.format(jobjob)if__name____main__:print(render(job_20260926))为什么把流程写成 README 模板端到端项目的失败多半不是代码错而是步骤顺序错G3 排在 G2 后面、忘了确认日志。把顺序固化成作业自述文件接手的人照做即可。三、常见报错与排查报错 17-1续跑时把已经跑好的工况又跑了一遍白等两小时。现象重复批跑。根因续跑没读账本状态。解法用pending()只取未完成工况对statusrun的跳过 G2 只取数。报错 17-2落库行数比预期多/少。现象数据对不上。根因多了→用了 append 而非 upsert少了→有工况没取到导出件缺该场景或该场景没输出。解法以主键 upsert取数后断言总行数 工况数 × 对象数 × 参数数。报错 17-3作业目录放在云盘跑到一半模型损坏。现象Cant find [TOOLBOX PREFERENCES] heading。根因违反铁律 7云同步/网络盘保存时断连损坏模型。解法作业目录放本地磁盘需要共享时下载→本地算→上传。报错 17-4G2 批跑时 Excel 导出选项是灰的。现象无法勾 Save Using Excel Export Manager。根因G3 没先做导出项未配置。解法先 G3 后 G2这是流水线顺序的硬性点。报错 17-5账本状态混乱不知道某工况到底跑到哪。现象状态与事实不符。根因手工改了状态但没改数据或反之。解法状态只能由脚本推进set_status手工干预要同步改账本。报错 17-6批量跑到第 N 个场景报错之后全部失败。现象一个工况的问题连累整批。根因某工况的变更表带了非法值如Special Condition写错或某场景 Analysis Setup 不完整。解法导入前用第 05/15 篇的预检拦非法行批跑前做场景完整性自检不完整者先从清单剔除单工况失败不影响其余批跑是顺序的已跑完的输出保留。四、动手练习练习 1跑通骨架跑python aft_sweep_cli.py --selftest。判定输出SELFTEST OKselftest_job/下含ledger.sqlite与transfer/里 2 个工作簿。练习 2真机三闸门用你自己定义的变量走一遍 design → generate →G1/G3/G2→ collect → report。判定账本最终全部collected落库总行数 工况数 × 对象数 × 参数数写出乘法式。练习 3断点续跑在 G2 中途中止只对剩余工况生成清单并补跑。判定账本里已跑的仍是run/collected未跑的补到collected最终无planned/integrated残留。练习 4作业自述用make_job_readme.py为你的作业生成 README。判定README 含 7 步执行顺序与 3 条纪律把实际执行中与 README 不符之处记下来并修订 README。练习 5幂等验证对同一作业连续执行两次design与两次collect。判定ledger.sqlite的工况总数两次相同不翻倍results.sqlite的COUNT(*)两次相同写出幂等来源的两条机制。五、小结与下一篇预告本篇把前 16 篇缝成了端到端项目5 个 Python 阶段 3 个 AFT GUI 闸门中枢是一本状态机账本planned → integrated → run → collected续跑靠只处理未完成工况 取数幂等目录规范让一批跑坏不污染上一批验收看完整性/守恒/可追溯/物理合理四条。你现在拥有一台能跑几百个工况的批量水力分析机器。第 18 篇《实战二Fathom → Impulse 水锤分析自动化流水线》我们在这台机器上加装瞬态段——稳态批跑 → 转 Impulse → 阀关闭/泵停机工况 → 瞬态批跑 → Force File 交付强调瞬态的时间步与收敛陷阱并给出surge_pipeline.py。FAQ与第〇节一一对应Q1批量水力分析项目哪些步骤能自动化、哪些必须在 GUIAPython 侧可全自动的是设计矩阵、生成 AFT Transfer、取数清洗、落库、出报告五个阶段必须人工在 GUI 完成的是三个闸门——File Import Excel Change Data、File Excel Export Manager 配置、File Start Batch Run因为 AFT 无公开 CLI/API。Q2为什么必须有账本它记录什么A账本是项目中枢记录每个工况的 case_id、全限定场景路径、参数水平、状态planned/integrated/run/collected、尝试次数与时间戳它让续跑只处理未完成工况、让重跑幂等、让流程可审计、让失败可重试。Q3目录怎么组织才安全A一次作业一个总目录含 ledger.sqlite 与 design/transfer/export/store/report 五个子目录整个作业目录必须放在本地磁盘因为里面含模型工作副本而官方明示网络盘或云同步目录保存会损坏模型铁律 7。Q4断点续跑在 AFT 里续在哪里A续在两个天然断点上——Start Batch Run 可 Cancel 且已跑完的场景输出仍保留以及 Export Only (Do Not Run) 可复用已有输出重导配合账本只处理 status 不为 collected 的工况status 为 run 的跳过批跑只做取数。Q5整条流水线怎么验收A四条判据完整性账本所有工况 status 为 collected、守恒落库总行数等于工况数×对象数×参数数、可追溯每个 case 能 join 回参数水平、结果能指回场景路径、物理合理抽查若干工况方向性合理。