OrgTJWater/TJWaterAgent
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

重新整理提示词和工具说明。

Browse Source
This commit is contained in:
Huarch
2026-06-02 17:42:02 +08:00
parent 5b285ad7a5
commit a825c3c31d
14 changed files with 126 additions and 1246 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.opencode/skills/workflow/SKILL.md
-70
View File
@@ -1,70 +0,0 @@
---
name: tjwater-workflow
description: 分析类工作流技能。所有工作流遵循 "CLI 获取 → 本地分析 → 报告输出" 模式。
version: 3.0.0
---
# Workflow Domain Skill
## 简介
负责分析场景下的工作流组织与调用入口。前端工具仅用于渲染。
## 使用策略
1. **优先匹配固定工作流** — 用户问题属于已有覆盖场景时，直接走 workflow
2. **按需升级** — 浅层工作流满足不了时，沿升级路径进入深层工作流
3. **缺少 workflow 时** — Agent 自行组合 CLI 命令 + Python 脚本
---
## 工作流决策树
```
用户请求
│
├─ 含"服务范围/供水分区/水源范围/分区渲染"？
│ └─ 是 → 走 source-service-area-analysis
│
├─ 含"改管径/瓶颈/扩容"？
│ └─ 是 → 直接走 bottleneck-analysis
│
└─ 否 → 走 simulation-diagnosis（快速诊断）
│
├─ 全局异常？（断流/未收敛/系统性低压）
│ └─ 直接输出针对性诊断，结束
│
├─ 少量告警、无紧急项？
│ └─ 输出报告，结束
│
└─ ≥5 条紧急超速 或 存在多维度交叉异常？
└─ 输出快速诊断报告 + 建议升级 bottleneck-analysis
+ 传递 TOP 问题管段 ID 列表
```
---
## 工作流清单
| 工作流 | 数据需求 | 典型步骤 |
|--------|---------|---------|
| **simulation-diagnosis** | `simulation run` + `data timeseries realtime links/nodes` | 1-2 次 CLI + 诊断脚本 |
| **bottleneck-analysis** | 模拟结果 + `network get-link-properties` 逐管道查询 | 多次 CLI + 合并分析脚本 |
| **source-service-area-analysis** | 模拟结果 + 管道拓扑 + 水库列表 | 多次 CLI + 图遍历脚本 |
---
## 衔接规则
`simulation-diagnosis` → `bottleneck-analysis` 升级条件：
| 信号 | 阈值 |
|------|------|
| 紧急超速管段数 | ≥ 5 条流速 > 1.5 m/s |
| 多维度交叉异常 | 同一管段同时满足高速 + 高水损 + 大流量 |
| 高水损地理聚集 | TOP 水损管段集中在同一区域 |
## 子模块索引
- **source-service-area-analysis**: `./source-service-area-analysis/SKILL.md`
- **simulation-diagnosis**: `./simulation-diagnosis/SKILL.md`
- **bottleneck-analysis**: `./bottleneck-analysis/SKILL.md`
.opencode/skills/workflow/bottleneck-analysis/SKILL.md
-106
View File
@@ -1,106 +0,0 @@
---
name: tjwater-workflow-bottleneck-analysis
description: 水力瓶颈分析工作流。
version: 2.0.0
---
# bottleneck-analysis Workflow Skill
## 简介
结合管道属性与水力模拟结果，识别管网中超负荷、高流速、高水头损失的瓶颈管道，计算综合评分并给出分级改造建议。
## 前置依赖
### 依赖 1：管道属性数据
```
tjwater_cli:
command: "network get-link-properties --link P1"
```
逐条查询所需管道 ID 的属性（id、diameter、length、roughness、node1、node2）。
> 首批 CLI 仅支持按 ID 单项查询，需对目标管段逐个调用。
### 依赖 2：水力模拟结果
```
tjwater_cli:
command: "data timeseries realtime links --start-time TIME --end-time TIME"
timeout: 300
```
获取全部管段的 flow(LPS)、velocity(m/s)、headloss(m)、status 等。
---
## 工作流步骤
### 第 1 步：获取管道属性
从 `simulation-diagnosis` 输出的 TOP 问题管段 ID 列表出发，逐条调用 `network get-link-properties --link ID` 获取管径/长度/粗糙度。
### 第 2 步：获取模拟结果
调用 `data timeseries realtime links` 获取对应时间窗口的全部管道水力结果。
### 第 3 步：合并数据
Python 脚本将管道属性 `id` 与模拟结果的 `link_id` 关联，构建合并数据集：
| 字段 | 来源 | 说明 |
|------|------|------|
| id | 关联键 | 管道 ID |
| flow | 模拟 | 流量 (LPS) |
| velocity | 模拟 | 流速 (m/s) |
| headloss | 模拟 | 水头损失 (m) |
| diameter | 管道属性 | 管径 (mm) |
| length | 管道属性 | 长度 (m) |
| roughness | 管道属性 | 粗糙度系数 |
| node1 / node2 | 管道属性 | 起端/终端节点 |
| unit_headloss | 计算 | headloss / length (m/m) |
| capacity_ratio | 计算 | \|flow\| / (π×(d/2000)²×1000) |
### 第 4 步：多维度瓶颈识别
| 维度 | 筛选条件 | 指示含义 |
|------|---------|---------|
| 高流速 | velocity > 1.2 m/s | 管径不足 |
| 主干管高流量 | diameter ≥ 300mm 且 velocity > 0.5 m/s | 传输瓶颈 |
| 高水头损失 | headloss > 5m 且 0.3 < velocity < 1.5 | 能耗瓶颈 |
| 高单位水头损失 | unit_headloss > 1.0 m/m | 严重局部瓶颈 |
| 超负荷 | capacity_ratio > 1.0 | 超过设计能力 |
排除极短管道（length < 0.5m）。
### 第 5 步：综合评分
```
composite_score = (velocity / max_velocity) × 0.4
+ (headloss / max_headloss) × 0.3
+ (capacity_ratio / max_capacity_ratio) × 0.3
```
取 TOP 10~20 作为最严重瓶颈管道。
### 第 6 步：分级改造建议
| 级别 | 评分范围 | 行动 |
|------|---------|------|
| 🚨 紧急 | > 0.3 | 立即安排管径升级 |
| ⚡ 重点 | 0.15~0.3 | 纳入近期改造计划 |
| 📋 关注 | 0.05~0.15 | 持续监测 |
建议管径公式：
```
建议管径(mm) = 2 × 1000 × sqrt(|flow| / (π × target_velocity × 1000))
```
DN<300 目标流速 1.0 m/s，DN≥300 目标流速 1.2 m/s。
### 第 7 步：可视化
- `show_chart`：流速分布柱状图
- `locate_features`：地图定位 TOP 瓶颈管道
---
## 证据约束
- 关键数据不完整时不得输出最终瓶颈结论
- 改造建议必须区分"数据直接支持"和"工程经验推断"
## Learned Patterns
- 先按"属性数据获取 → 模拟结果获取 → 本地关联 → 多指标筛选 → 分级建议"拆解工作流。
- 关联前统一字段：`link_id`、`diameter(mm)`、`length(m)`、`flow(LPS)`、`pressure(KPA)`。
- `unit_headloss`、`capacity_ratio` 等衍生指标应在过滤异常数据后再计算。
- 常见坑点：短管导致单位水头损失虚高、节点链路映射缺失、模拟结果不完整误当全量结论。
.opencode/skills/workflow/simulation-diagnosis/SKILL.md
-210
View File
@@ -1,210 +0,0 @@
---
name: tjwater-workflow-simulation-diagnosis
description: 单步长水力计算与管网问题诊断工作流。
version: 3.0.0
---
# Simulation Diagnosis Workflow Skill
## 简介
`simulation run` → 查询模拟结果 → 快速综合诊断 → 结构化报告。覆盖超流速、低压区、高水损、死水管段、Closed 阀门等常见运行问题。
**仅依赖模拟结果**（无需管径/长度/粗糙度等属性），适用于快速巡检、初筛、事故后评估。如需结合管径做详细瓶颈评分与改造建议，走 `bottleneck-analysis` 工作流。
---
## 执行速览（Golden Path）
"跑一下水力计算看看管网有什么问题" 的标准执行路径：
```
┌──────────────────────────────────────────────────────────────┐
│ 1. tjwater_cli(command="simulation run --start-time ... │
│ --duration 30") │
│ ↓ 模拟结果写入服务端时序库 │
│ 2. tjwater_cli(command="data timeseries realtime links │
│ --start-time ... --end-time ...") │
│ ↓ (写为本地 JSON) │
│ 3. python diagnose_simulation.py result.json │
│ ↓ (stdout 结构化报告) │
│ 4. 解读脚本输出 → 按严重级别组织 Markdown 报告 │
│ ↓ (可选) │
│ 5. show_chart / locate_features / view_scada │
└──────────────────────────────────────────────────────────────┘
```
> **关键原则**：排序/筛选/聚合全部在 Python 脚本或本地完成。前端工具仅用于渲染。
---
## 适用场景
| 场景 | 说明 |
|------|------|
| "跑一次水力计算看看管网有什么问题" | 直接适用，走 Golden Path |
| 管道属性接口不可用或数据不全 | 仅依赖模拟结果，无需属性数据 |
| 快速巡检 / 定期健康检查 | 低开销，脚本一秒给出问题概览 |
| 爆管/事故后的快速评估 | 单步稳态模拟，快速定位异常区域 |
| 作为 bottleneck-analysis 的前置筛查 | 先定位问题区域，再深入分析 |
---
## 阶段一：触发模拟运行
### CLI 调用
```
tjwater_cli:
command: "simulation run --start-time 2025-01-02T03:04:05+08:00 --duration 30"
```
- `--start-time`：模拟起始绝对时间，必须显式带时区（UTC+8）
- `--duration`：模拟持续分钟数
- 结果写入服务端时序库，不直接返回
### 查询模拟结果
```
tjwater_cli:
command: "data timeseries realtime links --start-time 2025-01-02T03:04:05+08:00 --end-time 2025-01-02T03:34:05+08:00"
timeout: 300
```
```
tjwater_cli:
command: "data timeseries realtime nodes --start-time 2025-01-02T03:04:05+08:00 --end-time 2025-01-02T03:34:05+08:00"
timeout: 300
```
### 返回数据结构
输出为 JSON，`data` 字段包含结果数组。
**links** — 管段水力状态：
| 字段 | 单位 | 说明 |
|------|------|------|
| link_id | — | 管段 ID |
| flow | LPS | 流量（正负表示流向） |
| velocity | m/s | 流速 |
| headloss | m | 水头损失 |
| status | — | Open / Closed / Critical |
**nodes** — 节点水力状态：
| 字段 | 单位 | 说明 |
|------|------|------|
| node_id | — | 节点 ID |
| demand | LPS | 需水量 |
| head | m | 水头 |
| pressure | KPA | 压力 |
---
## 阶段二：全局异常检查（优先！）
**在进入局部分析之前**，必须检查以下全局信号——如果命中则直接输出针对性诊断：
| 全局信号 | 判定条件 | 含义 | 处理 |
|---------|---------|------|------|
| **模拟断流** | 所有 link velocity ≈ 0 | 阀门全关 / 水源缺失 / 模型断连 | 检查模型配置，不做局部分析 |
| **模拟未收敛** | 所有 node pressure = 0 | 水力方程未收敛 | 检查 INP 设定，不做局部分析 |
| **系统性低压** | >70% 节点 pressure < 28 KPA | 水源水头不足 / 泵站扬程不够 | 输出"全局性低压"结论，排查水源/泵站 |
| **结果不完整** | links 或 nodes 结果为空 | 接口异常或模型损坏 | 报告数据不可用，提示重试 |
---
## 阶段三：自动诊断
### 使用 `diagnose_simulation.py`
脚本路径：`scripts/diagnose_simulation.py`
```bash
# 基本用法
python scripts/diagnose_simulation.py /tmp/opencode/sim_result.json
# 自定义阈值
python scripts/diagnose_simulation.py /tmp/opencode/sim_result.json \
--velocity-warn 1.2 \
--velocity-critical 2.0 \
--pressure-warn 40 \
--pressure-critical 28 \
--headloss-top 30 \
--flow-top 20
# 仅输出摘要
python scripts/diagnose_simulation.py /tmp/opencode/sim_result.json --summary-only
```
脚本自动完成：
- 全局统计（管段/节点总数，流速/压力/流量/水损的 max/mean/分位数）
- 超流速管段 TOP 20
- 高水损管段 TOP 30
- 低压节点统计
- 零流量管段、Closed 管段
- 高流量主干管 TOP 20
---
## 阶段四：报告生成与可视化
### 报告结构
```
---
## 管网运行诊断报告
### 一、模拟概况
- 时间窗口 / 管网规模 / 单位
### 二、主要发现（按严重程度排序）
🚨 紧急 / ⚡ 重点 / 📋 关注
每项含：问题标题 + 受影响数 + TOP 示例（3-5条含数据） + 可能原因 + 建议
### 三、数据局限
- 区分"数据直接支持的结论"和"工程经验的推测"
- 标注管径数据是否缺失及其对结论的影响
```
### 可视化（可选）
- `show_chart`：流速/压力分布柱状图
- `locate_features`：地图定位 TOP 问题管段
- `view_scada`：问题区域实时监测
- `render_junctions` + `store_render_ref`：按压力区间着色
---
## 诊断指标参考阈值
| 指标 | 阈值 | 依据 |
|------|------|------|
| 流速安全上限 | 1.2 m/s | 行业管道冲刷防护上限 |
| 流速紧急上限 | 2.0 m/s | 严重超速临界值 |
| 最低供水压力 | 28 KPA | 国标 GB 50268 |
| 正常压力下限 | 40 KPA | 居民供水压力下限 |
> 所有阈值为可调启发式，脚本支持参数覆写。
---
## 下游衔接：何时升级到 bottleneck-analysis
| 诊断发现 | 行动 |
|---------|------|
| 系统性低压（>70% 节点 <28 KPA） | **不**进入 bottleneck-analysis；排查水源/泵站 |
| TOP 20 超速管段中 ≥5 条流速 >1.5 m/s | 升级到 bottleneck-analysis |
| 存在多维度交叉异常管段 | 升级分析 |
| 仅少量告警、无紧急项 | 结束，不升级 |
**数据移交**：从 diagnosis 结果提取 TOP 问题管段 ID 列表，传入 bottleneck-analysis。
---
## Learned Patterns
- [d1f826184d8709db1988b00a] **渐进式 Top-K 分析**：大结果集先算全局统计量定位异常范围，再对 Top-K 做明细检查。
- [a19477bebe442e5d87bc78bc] **多准则交叉印证**：同时分析 velocity / headloss / flow / pressure 四个维度，紧急管段需满足多项异常条件。
- [cf50cc317b9bfe254426482d] **API 数据降级**：管道属性不可用时不停流程，用 velocity / headloss 作为代理指标。
- [1cd621a60d7481a1c072a54f] **系统性低压优先诊断**：全网 >70% 节点 pressure<28KPA 时定位为全局性问题。
- [a7c30102cf5c2a5c0e8cc8dd] **诊断脚本标准化**：`diagnose_simulation.py` 是分析主力，支持自定义阈值。
- [8b533e5bfc43c014ebcc90cf] **压力单位验证**：验证 pressure 字段实际单位（KPA vs m H₂O），阈值随单位调整。
.opencode/skills/workflow/simulation-diagnosis/scripts/diagnose_simulation.py
-341
View File
@@ -1,341 +0,0 @@
#!/usr/bin/env python3
"""
diagnose_simulation.py — 水力模拟结果快速诊断脚本
用途：对 runprojectreturndict 返回的 link_results + node_results 做多维度分析。
输出排序后的问题清单与统计摘要，供仿真诊断工作流使用。
用法：
python diagnose_simulation.py <input_json> [options]
输入 JSON 格式（runprojectreturndict 返回结构）：
{
"link_results": [
{"link_id": "...", "flow": float, "velocity": float,
"headloss": float, "status": "...", "friction": float},
...
],
"node_results": [
{"node_id": "...", "demand": float, "head": float, "pressure": float},
...
},
... // 其他字段（如 tanks, pumps, valves 等）可选
}
Options:
--velocity-warn FLOAT 超速预警阈值 (默认 1.2 m/s)
--velocity-critical FLOAT 超速严重阈值 (默认 2.0 m/s)
--pressure-warn FLOAT 低压预警阈值 (默认 40 KPA)
--pressure-critical FLOAT 低压严重阈值 (默认 28 KPA)
--headloss-top INT 高水损 TOP N (默认 30)
--flow-top INT 高流量主干管 TOP N (默认 20)
--output FILE 输出结果到 JSON 文件 (可选)
--summary-only 只输出统计摘要，不输出明细列表
输出：
stdout 打印结构化诊断报告，含各维度问题清单和全局统计。
"""
import json
import sys
import argparse
from collections import Counter
# ── 默认阈值 ──────────────────────────────────────────────
DEFAULT_VELOCITY_WARN = 1.2 # m/s
DEFAULT_VELOCITY_CRITICAL = 2.0 # m/s
DEFAULT_PRESSURE_WARN = 40.0 # KPA
DEFAULT_PRESSURE_CRITICAL = 28.0 # KPA
DEFAULT_HEADLOSS_TOP = 30
DEFAULT_FLOW_TOP = 20
def load_input(path: str) -> dict:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
def analyze_links(link_results: list, args):
"""分析管段维度：超流速、高水损、零流量、Closed 状态。"""
issues = {}
# ── 超流速管段 ──
over_velocity_warn = []
over_velocity_critical = []
zero_flow = []
closed_links = []
for link in link_results:
v = link.get("velocity", 0) or 0
q = abs(link.get("flow", 0) or 0)
status = (link.get("status") or "").strip().lower()
hl = link.get("headloss", 0) or 0
# 流速分级
if v > args.velocity_critical:
over_velocity_critical.append(link)
elif v > args.velocity_warn:
over_velocity_warn.append(link)
# 零流量（仅统计 Open 状态的）
if q < 0.001 and status == "open":
zero_flow.append(link)
# Closed 管段
if status != "open":
closed_links.append(link)
# 按流速降序排序
over_velocity_critical.sort(key=lambda x: x.get("velocity", 0), reverse=True)
over_velocity_warn.sort(key=lambda x: x.get("velocity", 0), reverse=True)
issues["over_velocity_critical"] = {
"count": len(over_velocity_critical),
"threshold": f">{args.velocity_critical} m/s",
"items": over_velocity_critical[:20],
"top_velocity": over_velocity_critical[0]["velocity"] if over_velocity_critical else 0,
}
issues["over_velocity_warn"] = {
"count": len(over_velocity_warn),
"threshold": f">{args.velocity_warn} m/s",
"items": over_velocity_warn[:20],
}
# ── 高水损管段（按绝对值降序） ──
high_headloss = sorted(link_results, key=lambda x: abs(x.get("headloss", 0) or 0), reverse=True)
issues["high_headloss"] = {
"count": len([x for x in link_results if abs(x.get("headloss", 0) or 0) > 0.5]),
"items": high_headloss[:args.headloss_top],
}
# ── 零流量管段 ──
issues["zero_flow_open"] = {
"count": len(zero_flow),
}
# ── Closed 管段 ──
issues["closed_links"] = {
"count": len(closed_links),
"items": closed_links[:20],
}
# ── 高流量主干管 ──
high_flow = sorted(link_results, key=lambda x: abs(x.get("flow", 0) or 0), reverse=True)
issues["high_flow_trunk"] = {
"items": high_flow[:args.flow_top],
}
return issues
def analyze_nodes(node_results: list, args):
"""分析节点维度：低压、高压。"""
issues = {}
pressure_critical = []
pressure_warn = []
high_pressure = []
for node in node_results:
p = node.get("pressure", 0) or 0
if p < args.pressure_critical:
pressure_critical.append(node)
elif p < args.pressure_warn:
pressure_warn.append(node)
if p > 200:
high_pressure.append(node)
pressure_critical.sort(key=lambda x: x.get("pressure", 0))
pressure_warn.sort(key=lambda x: x.get("pressure", 0))
total = len(node_results)
issues["pressure_critical"] = {
"count": len(pressure_critical),
"pct": round(len(pressure_critical) / total * 100, 1) if total else 0,
"threshold": f"<{args.pressure_critical} KPA",
"items": pressure_critical[:20],
}
issues["pressure_warn"] = {
"count": len(pressure_warn),
"pct": round(len(pressure_warn) / total * 100, 1) if total else 0,
"threshold": f"<{args.pressure_warn} KPA (>= {args.pressure_critical})",
"items": pressure_warn[:20],
}
issues["high_pressure"] = {
"count": len(high_pressure),
"items": high_pressure[:20],
}
return issues, total
def global_stats(link_results, node_results):
"""计算全局统计量。"""
velocities = [x.get("velocity", 0) or 0 for x in link_results]
flows = [abs(x.get("flow", 0) or 0) for x in link_results]
headlosses = [abs(x.get("headloss", 0) or 0) for x in link_results]
pressures = [x.get("pressure", 0) or 0 for x in node_results]
demands = [x.get("demand", 0) or 0 for x in node_results]
# 状态分布
status_counter = Counter((x.get("status") or "").strip().lower() for x in link_results)
stats = {
"link_count": len(link_results),
"node_count": len(node_results),
"velocity": {
"max": round(max(velocities), 4),
"mean": round(sum(velocities) / len(velocities), 4) if velocities else 0,
"p95": sorted(velocities)[int(len(velocities) * 0.95)] if velocities else 0,
},
"pressure": {
"max": round(max(pressures), 2),
"min": round(min(pressures), 2),
"mean": round(sum(pressures) / len(pressures), 2) if pressures else 0,
"p5": sorted(pressures)[int(len(pressures) * 0.05)] if pressures else 0,
},
"flow": {
"max": round(max(flows), 4),
},
"headloss": {
"max": round(max(headlosses), 4),
},
"status_distribution": dict(status_counter.most_common()),
}
return stats
def format_report(stats, link_issues, node_issues, node_total, summary_only=False):
"""输出格式化的诊断报告。"""
lines = []
lines.append("=" * 60)
lines.append(" 水力模拟诊断报告")
lines.append("=" * 60)
lines.append("")
# ── 全局统计 ──
lines.append("【一、全局统计】")
lines.append(f" 管段总数: {stats['link_count']}")
lines.append(f" 节点总数: {stats['node_count']}")
lines.append(f" 流速: 最大 {stats['velocity']['max']} m/s, "
f"平均 {stats['velocity']['mean']} m/s, P95 {stats['velocity']['p95']} m/s")
lines.append(f" 压力: 最大 {stats['pressure']['max']} KPA, "
f"最小 {stats['pressure']['min']} KPA, "
f"平均 {stats['pressure']['mean']} KPA, P5 {stats['pressure']['p5']} KPA")
lines.append(f" 最大流量: {stats['flow']['max']} LPS")
lines.append(f" 最大水损: {stats['headloss']['max']} m")
lines.append(f" 状态分布: {stats['status_distribution']}")
lines.append("")
# ── 超流速 ──
crit_v = link_issues["over_velocity_critical"]
warn_v = link_issues["over_velocity_warn"]
lines.append("【二、超流速管段】")
lines.append(f" 🔴 严重超速 (> {crit_v['threshold']}): {crit_v['count']} 条, "
f"最高 {crit_v['top_velocity']} m/s")
if crit_v["count"] > 0 and not summary_only:
for item in crit_v["items"][:10]:
lines.append(f" - {item['link_id']}: {item['velocity']} m/s, "
f"flow={item.get('flow', 'N/A')} LPS, status={item.get('status', 'N/A')}")
lines.append(f" 🟡 流速预警 (> {warn_v['threshold']}): {warn_v['count']} 条")
lines.append("")
# ── 高水损 ──
hl = link_issues["high_headloss"]
lines.append("【三、高水损管段 TOP】")
lines.append(f" 水损 > 0.5m 的管段: {hl['count']} 条")
if hl["items"] and not summary_only:
for item in hl["items"][:10]:
lines.append(f" - {item['link_id']}: headloss={item.get('headloss', 0)} m, "
f"velocity={item.get('velocity', 0)} m/s")
lines.append("")
# ── 低压节点 ──
pc = node_issues["pressure_critical"]
pw = node_issues["pressure_warn"]
lines.append("【四、低压节点】")
lines.append(f" 🔴 严重低压 ({pc['threshold']}): {pc['count']} 个 "
f"({pc['pct']}%)")
if pc["count"] > 0 and not summary_only:
for item in pc["items"][:10]:
lines.append(f" - {item['node_id']}: {item['pressure']} KPA, "
f"demand={item.get('demand', 0)} LPS, head={item.get('head', 0)} m")
lines.append(f" 🟡 低压预警 ({pw['threshold']}): {pw['count']} 个 "
f"({pw['pct']}%)")
lines.append("")
# ── 其他发现 ──
lines.append("【五、其他发现】")
zf = link_issues["zero_flow_open"]
cl = link_issues["closed_links"]
hp = node_issues["high_pressure"]
lines.append(f" - 零流量 (Open 状态但 flow≈0): {zf['count']} 条")
lines.append(f" - Closed 管段/阀门: {cl['count']} 条")
if hp["count"] > 0:
lines.append(f" - 超压节点 (> 200 KPA): {hp['count']} 个")
lines.append("")
# ── 高流量主干 ──
hft = link_issues["high_flow_trunk"]
lines.append("【六、高流量主干管 TOP】")
if hft["items"] and not summary_only:
for item in hft["items"][:10]:
lines.append(f" - {item['link_id']}: |flow|={abs(item.get('flow', 0))} LPS, "
f"velocity={item.get('velocity', 0)} m/s")
lines.append("")
lines.append("=" * 60)
return "\n".join(lines)
def main():
parser = argparse.ArgumentParser(description="水力模拟结果快速诊断")
parser.add_argument("input", help="输入 JSON 文件路径 (runprojectreturndict 结果)")
parser.add_argument("--velocity-warn", type=float, default=DEFAULT_VELOCITY_WARN)
parser.add_argument("--velocity-critical", type=float, default=DEFAULT_VELOCITY_CRITICAL)
parser.add_argument("--pressure-warn", type=float, default=DEFAULT_PRESSURE_WARN)
parser.add_argument("--pressure-critical", type=float, default=DEFAULT_PRESSURE_CRITICAL)
parser.add_argument("--headloss-top", type=int, default=DEFAULT_HEADLOSS_TOP)
parser.add_argument("--flow-top", type=int, default=DEFAULT_FLOW_TOP)
parser.add_argument("--output", help="输出 JSON 结果到文件")
parser.add_argument("--summary-only", action="store_true", help="仅输出统计摘要")
args = parser.parse_args()
data = load_input(args.input)
link_results = data.get("link_results", [])
node_results = data.get("node_results", [])
if not link_results and not node_results:
print("错误: 输入数据中既无 link_results 也无 node_results", file=sys.stderr)
sys.exit(1)
# 执行分析
stats = global_stats(link_results, node_results)
link_issues = analyze_links(link_results, args)
node_issues, node_total = analyze_nodes(node_results, args)
# 输出报告
report = format_report(stats, link_issues, node_issues, node_total, args.summary_only)
print(report)
# 输出 JSON（可选）
if args.output:
output_data = {
"stats": stats,
"link_issues": {
k: {kk: vv for kk, vv in v.items() if kk != "items"}
for k, v in link_issues.items()
},
"node_issues": {
k: {kk: vv for kk, vv in v.items() if kk != "items"}
for k, v in node_issues.items()
},
}
with open(args.output, "w", encoding="utf-8") as f:
json.dump(output_data, f, ensure_ascii=False, indent=2)
print(f"详细结果已输出到: {args.output}")
if __name__ == "__main__":
main()
.opencode/skills/workflow/source-service-area-analysis/SKILL.md
-113
View File
@@ -1,113 +0,0 @@
---
name: tjwater-workflow-source-service-area-analysis
description: 水源服务范围分区分析工作流。
version: 2.0.0
---
# Source Service Area Analysis Workflow Skill
## 简介
基于水力模拟结果的流向数据构建有向图，通过下游遍历确定每个节点的供水水源归属，生成分区渲染数据并在地图上以不同颜色展示各水源的服务范围。
## 前置依赖
### 依赖 1：水库（水源）列表
从模拟结果的 node_results 中识别 type=reservoir 的节点，或逐条查询已知水库节点。
### 依赖 2：管道拓扑数据
```
tjwater_cli:
command: "network get-link-properties --link P1"
```
逐条获取管道 ID 的 node1（起端）、node2（终端），构建管网拓扑。
### 依赖 3：水力模拟结果
```
tjwater_cli:
command: "data timeseries realtime links --start-time TIME --end-time TIME"
timeout: 300
```
获取各管段的 `flow`(LPS) 和 `status`，确定水流方向。
- `flow > 0` → node1 → node2
- `flow < 0` → node2 → node1
- `flow ≈ 0` 或 `status != "Open"` → 忽略
---
## 工作流步骤
### 第 1 步：获取三大数据源
并行调用 CLI 命令获取模拟结果、管道拓扑和水库信息，写入本地临时 JSON 文件。
### 第 2 步：构建有向图并计算服务范围
```bash
python scripts/compute_service_areas.py \
/tmp/opencode/sim_result.json \
/tmp/opencode/pipes.json \
/tmp/opencode/reservoirs.json \
--output /tmp/opencode/service_area_result.json
```
脚本核心逻辑：
1. **构建无向拓扑图**：从管道属性提取 `(node1, node2)`，构建邻接表
2. **确定流向**：从模拟结果读取 flow 和 status
- flow > 0 → node1 → node2
- flow < 0 → node2 → node1
- flow ≈ 0 或 status != "Open" → 忽略
3. **获取水源集合**：从水库列表提取所有水库 ID 作为 source nodes
4. **多源 BFS 下游遍历**：从所有水源节点同时出发，首次访问归属策略
5. **统计输出**：各水源服务节点数、占比、未分配节点数
6. **生成渲染数据**：`node_area_map: { node_id: area_id }`，每个水源分配唯一颜色
### 第 3 步：结果验证
- **服务覆盖率**：正常管网应 > 80%
- **水源孤立**：某水源服务节点数为 0，检查出口管道是否 Closed
- **大区未分配**：检查区域与水源之间管道是否 Closed
### 第 4 步：前端可视化
1. `store_render_ref` 将 `node_area_map` 迁移为受控 `render_ref`
2. `render_junctions` 地图着色
3. `locate_features` 高亮水源位置
4. `show_chart` 各水源服务节点数柱状图（可选）
### 第 5 步：报告输出
```
---
## 水源服务范围分析报告
### 一、管网概况
- 总节点数 / 总管段数 / 水源数 / 覆盖率
### 二、各水源服务范围
| 水源 ID | 服务节点数 | 占比 | 颜色 |
### 三、未分配节点分析
- 数量、可能原因
### 四、假设与局限
- 流向基于稳态模拟，不代表所有工况
- 有向图忽略零流量管段
```
---
## 颜色调色板
脚本内置 20 色：
```
#1f77b4, #ff7f0e, #2ca02c, #d62728, #9467bd, #8c564b, #e377c2,
#7f7f7f, #bcbd22, #17becf, #aec7e8, #ffbb78, #98df8a, #ff9896,
#c5b0d5, #c49c94, #f7b6d2, #c7c7c7, #dbdb8d, #9edae5
```
## 证据约束
- 计算脚本的输出是唯一数据真相源
- 未完成数据回读时不得执行分析
- 零流量管段视为断连，需在报告中说明
## Learned Patterns
- 核心数据流：模拟结果(流向) + 管道属性(拓扑) + 水库列表(起点) → 有向图 BFS → 分区渲染
- 有向图遍历前务必过滤 Closed 和零流量管段
- 渲染数据必须通过 `store_render_ref` 迁移再传给 `render_junctions`
- 多水源 BFS 首次访问归属策略隐含"距离最近水源"语义
.opencode/skills/workflow/source-service-area-analysis/scripts/compute_service_areas.py
-281
View File
@@ -1,281 +0,0 @@
#!/usr/bin/env python3
"""
compute_service_areas.py — 水源服务范围分析核心脚本
输入：
sim_result.json — runprojectreturndict 的水力模拟结果
pipes.json — getallpipeproperties 的管道属性列表
reservoirs.json — getallreservoirproperties 的水库属性列表
输出：
service_area_result.json — 包含 node_area_map 和统计信息
算法：
1. 从 pipes 构建无向拓扑图 (node1 <-> node2)
2. 从 sim link_results 确定流向，生成有向图
3. 从 reservoirs 获取所有水源节点
4. 多源 BFS 下游遍历，首次访问即归属性
5. 生成渲染数据与统计
用法:
python compute_service_areas.py sim_result.json pipes.json reservoirs.json \
--output service_area_result.json --network tjwater
"""
import argparse
import json
import os
import sys
from collections import deque
# ── 高区分度 20 色调色板 ──────────────────────────────────────────
PALETTE = [
"#1f77b4", "#ff7f0e", "#2ca02c", "#d62728", "#9467bd",
"#8c564b", "#e377c2", "#7f7f7f", "#bcbd22", "#17becf",
"#aec7e8", "#ffbb78", "#98df8a", "#ff9896", "#c5b0d5",
"#c49c94", "#f7b6d2", "#c7c7c7", "#dbdb8d", "#9edae5",
]
def load_json(path: str) -> dict | list:
"""加载 JSON 文件，支持引用格式（含 result_ref 字段）或列表。"""
with open(path, "r", encoding="utf-8") as f:
data = json.load(f)
# 如果顶层是引用包装，展开到内部的列表
if isinstance(data, dict) and "data" in data:
data = data["data"]
if isinstance(data, dict) and "items" in data:
data = data["items"]
return data
def build_undirected_graph(pipes: list) -> tuple[dict, dict]:
"""
从管道列表构建无向拓扑图。
返回:
adj: { node_id: set(neighbor_node_ids) }
edge_map: { (node1, node2): pipe_id } 用于查找管道 ID
"""
adj = {}
edge_map = {}
for p in pipes:
pipe_id = str(p.get("id", p.get("pipe_id", p.get("name", ""))))
n1 = str(p.get("node1", ""))
n2 = str(p.get("node2", ""))
if not n1 or not n2:
continue
adj.setdefault(n1, set()).add(n2)
adj.setdefault(n2, set()).add(n1)
edge_map[(n1, n2)] = pipe_id
edge_map[(n2, n1)] = pipe_id
return adj, edge_map
def build_directed_graph(adj: dict, edge_map: dict, link_results: list) -> dict:
"""
基于模拟流向构建有向图。
link_results 中每条记录含 link_id, flow, status。
流向判定:
flow > 0 → node1 → node2
flow < 0 → node2 → node1
flow ≈ 0 或 status != "Open" → 忽略（视为断连）
注意: link_results 中没有 node1/node2，需要通过 link_id 反查 edge_map。
由于 edge_map 只有 (node1,node2) → pipe_id 的映射，
这里需要先建立 pipe_id → (node1, node2) 的反向索引。
"""
# 建立 pipe_id → (node1, node2) 反向索引
# 注意：edge_map 含有 (n1,n2) 和 (n2,n1) 两个方向，只保留第一个以避免反向覆写
pipe_to_nodes = {}
for (n1, n2), pid in edge_map.items():
if pid not in pipe_to_nodes:
pipe_to_nodes[pid] = (n1, n2)
digraph = {}
missing = 0
closed_or_zero = 0
for link in link_results:
lid = str(link.get("link_id", link.get("id", "")))
flow = link.get("flow", 0)
status = str(link.get("status", "Open"))
# 忽略非 Open 管段和零流量管段（大小写不敏感）
if status.upper() != "OPEN":
closed_or_zero += 1
continue
if abs(flow) < 1e-6:
closed_or_zero += 1
continue
# 查找管段对应的节点对
pair = pipe_to_nodes.get(lid)
if pair is None:
missing += 1
continue
n1, n2 = pair
if flow > 0:
digraph.setdefault(n1, set()).add(n2)
else:
digraph.setdefault(n2, set()).add(n1)
if missing > 0:
print(f"[WARN] {missing} 条模拟管段未在管道拓扑中找到对应节点对", file=sys.stderr)
if closed_or_zero > 0:
print(f"[INFO] 忽略 {closed_or_zero} 条 Closed/零流量管段", file=sys.stderr)
return digraph
def bfs_from_sources(sources: list, digraph: dict) -> dict:
"""
多水源 BFS 下游遍历。首次访问的节点归属到对应水源。
返回: { node_id: source_id }
"""
node_area = {}
queue = deque()
# 初始化：所有水源入队，归属自身
for src in sources:
node_area[src] = src
queue.append(src)
while queue:
cur = queue.popleft()
cur_source = node_area[cur]
for nb in digraph.get(cur, set()):
if nb not in node_area:
node_area[nb] = cur_source
queue.append(nb)
return node_area
def generate_render_data(node_area: dict, sources: list) -> dict:
"""
生成 render_junctions 所需的渲染数据。
返回:
node_area_map: { node_id: area_id }
area_ids: 水源 ID 列表
area_colors: { area_id: color_hex }
"""
# 为每个水源分配颜色
area_colors = {}
for i, src in enumerate(sources):
area_colors[src] = PALETTE[i % len(PALETTE)]
return {
"node_area_map": node_area,
"area_ids": sources,
"area_colors": area_colors,
}
def compute_stats(node_area: dict, sources: list, all_nodes: set) -> dict:
"""计算各水源服务节点数和覆盖率统计。"""
source_counts = {}
for src in sources:
source_counts[src] = sum(1 for v in node_area.values() if v == src)
total = len(all_nodes)
assigned = len(node_area)
unassigned = total - assigned
coverage = (assigned / total * 100) if total > 0 else 0
return {
"total_nodes": total,
"assigned_nodes": assigned,
"unassigned_nodes": unassigned,
"coverage_pct": round(coverage, 2),
"source_counts": source_counts,
}
def main():
parser = argparse.ArgumentParser(
description="水源服务范围分析 — 基于水力模拟流向的多源 BFS 分区"
)
parser.add_argument("sim_file", help="runprojectreturndict 模拟结果 JSON")
parser.add_argument("pipes_file", help="getallpipeproperties 管道属性 JSON")
parser.add_argument("reservoirs_file", help="getallreservoirproperties 水库属性 JSON")
parser.add_argument("--output", "-o", default="/tmp/opencode/service_area_result.json",
help="输出 JSON 文件路径")
parser.add_argument("--network", "-n", default="tjwater",
help="管网名称（仅用于报告标注）")
args = parser.parse_args()
# ── 1. 加载数据 ──
print(f"[INFO] 加载模拟结果: {args.sim_file}")
sim_data = load_json(args.sim_file)
link_results = sim_data.get("link_results", [])
print(f"[INFO] 加载管道属性: {args.pipes_file}")
pipes = load_json(args.pipes_file)
print(f"[INFO] 加载水库列表: {args.reservoirs_file}")
reservoirs = load_json(args.reservoirs_file)
# 提取水源 ID 列表
sources = []
if isinstance(reservoirs, list):
sources = [str(r.get("id", r.get("name", ""))) for r in reservoirs if r.get("id") or r.get("name")]
elif isinstance(reservoirs, dict):
# 可能是 { data: [...] } 格式
for key in ["data", "items", "reservoirs"]:
if key in reservoirs and isinstance(reservoirs[key], list):
sources = [str(r.get("id", r.get("name", ""))) for r in reservoirs[key] if r.get("id") or r.get("name")]
break
if not sources:
# 也可能是 { "551656": {...}, ... } 格式
sources = [str(k) for k in reservoirs if isinstance(reservoirs[k], dict)]
if not sources:
print("[ERROR] 未能从水库数据中提取水源 ID 列表", file=sys.stderr)
sys.exit(1)
print(f"[INFO] 水源数: {len(sources)}")
print(f"[INFO] 水源列表: {sources}")
# ── 2. 构建无向图 ──
adj, edge_map = build_undirected_graph(pipes)
all_nodes = set(adj.keys())
# ── 3. 构建有向图 ──
digraph = build_directed_graph(adj, edge_map, link_results)
# ── 4. BFS 下游遍历 ──
print("[INFO] 执行多源 BFS 下游遍历...")
node_area = bfs_from_sources(sources, digraph)
# ── 5. 统计 ──
stats = compute_stats(node_area, sources, all_nodes)
print(f"[INFO] 总节点数: {stats['total_nodes']}")
print(f"[INFO] 已分配: {stats['assigned_nodes']} ({stats['coverage_pct']:.1f}%)")
print(f"[INFO] 未分配: {stats['unassigned_nodes']}")
print("\n各水源服务节点数 (按数量降序):")
for src, count in sorted(stats["source_counts"].items(), key=lambda x: -x[1]):
pct = (count / stats["assigned_nodes"] * 100) if stats["assigned_nodes"] > 0 else 0
print(f" {src}: {count} ({pct:.1f}%)")
# ── 6. 生成渲染数据 ──
render = generate_render_data(node_area, sources)
# ── 7. 输出 ──
output = {
"network": args.network,
"stats": stats,
"render": render,
"sources": sources,
}
os.makedirs(os.path.dirname(args.output) or ".", exist_ok=True)
with open(args.output, "w", encoding="utf-8") as f:
json.dump(output, f, ensure_ascii=False, indent=2)
print(f"\n[OK] 结果已写入: {args.output}")
if __name__ == "__main__":
main()