#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""生成 docs/images/架构总览_端到端链路.svg —— 端到端链路鸟瞰（纯标准库）。
这张图回答什么问题：一个电气量从芯片引脚走到 104 报文 / 显存 / 记录文件，
中间要穿过哪些框；以及「谁在什么时候被谁唤醒、谁独占写、谁只是同一份代码的
另一颗脑袋」——即**数据流与控制握手**，不是层间依赖清单。
与同目录另两张总览图的分工：
gen_arch_svg.py → 架构分层图：六层 + 每层有哪些模块（回答「谁属于哪一层」）
gen_barrier_svg.py → 就绪屏障图：Event Flags 模型细节
本脚本 → 端到端链路：片外硬件 → 驱动 → 算法 → 任务 → 落点 → 对外/同构端
线型三类：实线 = 数据与读写；短虚线 = 命令 / 请求 / 配置；紫点线 = 屏障位、心跳与同源镜像。
为什么不用 Mermaid：本图 42 框 37 边，含跨行长边（遥测快照 → 主机侧第三端独占一条行通道）、
同列相邻框之间的并行双线（定值下行 / YK 上行各占一个出框位置）、绕底槽的握手边
（就绪屏障 → 控制台要跨过整行），还要把五条分层纵带画成等宽对齐的框。
Mermaid 自动布局既不可控（多节点跨层依赖会画成"意大利面"），也表达不了
「一条边独占一条通道」这类判据。
用法（仓库根目录，任意 python3）：
python tools/arch_diagram/gen_arch_endtoend_svg.py
输出：
docs/images/架构总览_端到端链路.svg
改内容：编辑 NODES / EDGES / BANDS / 文案常量后重跑。
调坐标：先改顶部几何常量（CW / CH / PX / PY 与三条独占通道 Y、一条走廊 X），
单边微调走 EDGES 尾部旋钮 f / g（出框、入框位置 0~1）与 my / gx（独占通道坐标）。
产物内容零不确定：**不得**写入时间戳 / 提交号 / 绝对路径（generated-fresh 逐字节比对）。
"""
import html
import os
import sys
import xml.etree.ElementTree as ET
ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
OUT = os.path.join(ROOT, "docs", "images", "架构总览_端到端链路.svg")
# ----------------------------- 画布与几何 -----------------------------
PAD_L = 34
PAD_R = 30
TITLE_H = 74
LEGEND_H = 50
FOOT_GAP = 14
CW = 148
CH = 58
COLS = 6
ROWS = 9
PX = 196
PY = 92
BAND_PAD = 13
BAND_TITLE_H = 30
BAND_BOTTOM_PAD = 24 # 带内底部专用走廊（第 8 行下方绕行的握手边走这里）
GRID_W = (COLS - 1) * PX + CW
W = PAD_L + GRID_W + PAD_R
X0 = PAD_L + BAND_PAD
Y0 = TITLE_H + 30 + BAND_PAD + BAND_TITLE_H
H = Y0 + (ROWS - 1) * PY + CH + BAND_PAD + BAND_BOTTOM_PAD + FOOT_GAP + LEGEND_H
# 独占通道 / 走廊（长边走这里；数值落在通道内，改动后回看有无重叠）
CHAN_B = Y0 + 2 * PY + CH + 11 # 第 2、3 行之间（遥测快照 → 主机侧第三端）
CHAN_B2 = Y0 + 2 * PY + CH + 24 # 同一条走廊的下半股（遥测快照 → iec104Task，两条平行不共线）
CHAN_78 = Y0 + 7 * PY + CH + 12 # 第 7、8 行之间（RS-485 → 屏板链路）
CHAN_SLOTS = Y0 + 8 * PY + CH + 13 # 第 8 行下方底槽（就绪屏障 → 控制台，带内独占）
GUT_C4C5 = X0 + 4 * PX + CW + (PX - CW) * 0.62 # 第 4、5 列之间垂直走廊（配置引擎 → 落点）
FONT = "Microsoft YaHei, Segoe UI, PingFang SC, sans-serif"
TITLE_FONT = 20
SUB_FONT = 12.5
BAND_FONT = 13.5
BAND_SUB_FONT = 10.5
NODE_FONT = 11.5
NODE_SUB_FONT = 9.5
EDGE_FONT = 9.5
FOOT_FONT = 9.5
# 调色板（沿用 docs/架构设计/图形绘制流程.md；紫色整族让给「屏障 / 事件」，纵带改用玫红）
C_HW = ("#475569", "#f1f5f9")
C_DRV = ("#d97706", "#fffbeb")
C_MID = ("#16a34a", "#f0fdf4")
C_TASK = ("#0d9488", "#f0fdfa")
C_SVC = ("#e11d48", "#fff1f2")
C_EXT = ("#2563eb", "#eff6ff")
E_DATA = "#334155"
E_CTRL = "#64748b"
E_EVT = "#7c3aed"
GRID_LINE = "#cbd5e1"
TITLE = "STM32F407ZGTx 工业 DTU 平台 · 端到端链路总览"
SUBTITLE = ("一个电气量从引脚到 104 报文 / 显存 / 记录文件穿过的框，与「谁在什么时候被谁唤醒」；"
"分层归属与层内模块清单见 项目架构总览.svg，本图只画链路与握手")
FOOTNOTE = ("口径：AGENTS.md「数据主线」+ docs/架构设计/项目架构总览.md（行号级证据在那份文档与代码里）。"
"RS-485 采集未接线：rs485_task 有实现但不建任务，接线后数据进语义取数层；"
"其余行为条件（未标定闭锁 / 就绪位 NoClear / CH395F 单连接 / V0.1 双转换器无共同采样触发）见正文，"
"它们只改变判据与端口、不改变本图的框与连线。")
# ----------------------------- 纵带（一列 = 一条带） -----------------------------
BANDS = [
(0, "片外硬件", "板级器件 · 不走软件分层", C_HW),
(1, "驱动层 Drivers/BSP", "手写芯片驱动与总线", C_DRV),
(2, "中间件与算法", "取数 → 相量 → 快照", C_MID),
(3, "任务编排与握手", "App/task · 就绪屏障", C_TASK),
(4, "平台服务与数据落点", "记录 / 配置 / 网络门面", C_SVC),
]
# ----------------------------- 节点：(id, 列, 行, 标题, 副题) -----------------------------
NODES = [
("tpafe", 0, 0, "TPAFE5160 前端", "并行总线 · 8 路"),
("adc1", 0, 1, "片内 ADC1 + DMA", "TIM3 TRGO 乒乓"),
("eth", 0, 4, "CH395F 以太网", "SPI 串行化"),
("nandchip", 0, 5, "GD5F SPI NAND", "2 KB 页 · ECC"),
("rtc", 0, 6, "SD2506 RTC", "I2C1 硬件亚秒"),
("panel", 0, 7, "屏板 160×160", "USART1 700000"),
("rs485hw", 0, 8, "RS-485 总线", "UART4 · 未接线（暂缓）"),
("drv_tpafe", 1, 0, "TPAFE 驱动", "原子快照 snapshot"),
("drv_adc", 1, 1, "ADC · TIM3 · DMA", "手写上电自校准"),
("drv_ch395", 1, 4, "CH395F 驱动", "中断集中读清"),
("drv_nand", 1, 5, "NAND 驱动 + FTL", "ECC · BBT"),
("drv_rtc", 1, 6, "RTC 驱动 + 墙钟", "sys_clock epoch"),
("sb_link", 1, 7, "屏板链路驱动", "借道 RS485"),
("console", 1, 8, "控制台 UART5", "4 KB 环形"),
("adc_input", 2, 0, "语义取数层", "有效性 · 标定"),
("phasor", 2, 1, "全周傅氏相量", "ring → phasor"),
("meas", 2, 2, "遥测快照双区", "测频 200 ms"),
("netmw", 2, 4, "NET 中间件", "socket · 挂起"),
("fs", 2, 5, "FTL + FatFs", "dhara · 8.3 短名"),
("timeproto", 2, 6, "对时协议", "推送写 RTC"),
("lvgl", 2, 7, "LVGL + ui_port", "160×160 1bpp"),
("selfheal", 2, 8, "IWDG + Fault", "心跳掩码"),
("adc_task", 3, 0, "adcTask", "1 kHz 消费者"),
("protect_task", 3, 1, "protectTask", "20 ms 判据"),
("protset", 3, 2, "定值层", "prot_set 校验"),
("iec104_task", 3, 3, "iec104Task", "100 ms · :2404"),
("net_task", 3, 4, "netTask", "MsgQueue 深 8"),
("default_task", 3, 5, "defaultTask", "记录唯一写者"),
("time_sync_task", 3, 6, "timeSyncTask", "等 NET_READY"),
("ui_task", 3, 7, "uiTask", "唯一写侧"),
("barrier", 3, 8, "就绪屏障", "位组 · NoClear"),
("soe", 4, 1, "SOE 事件", "soe.log 定长"),
("setfile", 4, 2, "定值落盘", "/prot_set.bin"),
("cfgplane", 4, 3, "配置引擎", "net_cfg · ftp_cfg"),
("services", 4, 4, "TCP 服务面", ":21 / :2404"),
("files", 4, 5, "文件系统落点", "/log · /wave"),
("vollock", 4, 6, "卷锁 + 账本", "三路 try 协定"),
("disp", 4, 7, "一份显存三出", "屏板 · :7003"),
("ftp_task", 4, 8, "ftpTask", "lftpd 根 :21"),
("hosttest", 5, 2, "主机侧第三端", "直打生产 .c"),
("peer", 5, 3, "上位侧对端", "104 主站"),
("sim", 5, 7, "PC 模拟器", "SDL2 同源码"),
]
# ----------------------------- 边 -----------------------------
# (起点, 终点, 标签, 类别, 出框, 入框, 旋钮)
# 类别 d=数据/读写 c=命令/请求/配置 e=屏障·心跳·同源
# 出/入框 t b l r；出框轴与入框轴同向时自动加一段（水平用 gx，垂直用 my）
# 旋钮 f/g = 该边上的位置 0~1（默认 0.5）；my/gx = 独占通道坐标；seg/t = 标签落点
EDGES = [
("tpafe", "drv_tpafe", "8 路帧", "d", "r", "l", {}),
("drv_adc", "adc1", "TRGO", "c", "l", "r", {}),
("drv_tpafe", "adc_input", "原子快照", "d", "r", "l", {}),
("drv_adc", "adc_input", "DMA 双缓冲", "d", "t", "b", {"f": 0.72, "g": 0.28}),
("adc_input", "phasor", "语义通道", "d", "b", "t", {}),
("phasor", "meas", "相量", "d", "b", "t", {}),
("adc_task", "protect_task", "1 kHz → 20 ms", "c", "b", "t", {}),
("meas", "iec104_task", "同一上游", "d", "b", "t", {"f": 0.3, "g": 0.5, "my": CHAN_B2}),
("protset", "protect_task", "定值", "d", "t", "b", {"f": 0.34, "g": 0.34}),
("iec104_task", "protset", "YK 请求", "c", "t", "b", {"f": 0.68, "g": 0.68}),
("iec104_task", "cfgplane", "脏段", "c", "r", "l", {}),
("protect_task", "soe", "边沿投递", "d", "r", "l", {}),
("default_task", "files", "唯一写者", "d", "r", "l", {}),
("default_task", "vollock", "try 语义", "c", "b", "t", {"f": 0.86, "g": 0.26}),
("eth", "drv_ch395", "SPI 事务", "d", "r", "l", {}),
("drv_ch395", "netmw", "集中读清", "d", "r", "l", {}),
("netmw", "net_task", "串行执行", "d", "r", "l", {}),
("net_task", "services", "单连接", "d", "r", "l", {}),
("services", "peer", "TCP 对端", "c", "r", "b", {"f": 0.26, "g": 0.6}),
("nandchip", "drv_nand", "页读写", "d", "r", "l", {}),
("drv_nand", "fs", "逻辑块", "d", "r", "l", {}),
("fs", "default_task", "挂载", "d", "r", "l", {}),
("rtc", "drv_rtc", "I2C1", "d", "r", "l", {}),
("drv_rtc", "timeproto", "墙钟", "d", "r", "l", {}),
("time_sync_task", "timeproto", "收包对时", "d", "l", "r", {}),
("lvgl", "sb_link", "帧 B", "d", "l", "r", {}),
("ui_task", "lvgl", "灌数", "d", "l", "r", {}),
("ui_task", "disp", "同一显存", "d", "r", "l", {}),
("sb_link", "panel", "700000", "d", "l", "r", {}),
("ui_task", "selfheal", "100 ms 喂心跳", "c", "b", "t", {"f": 0.26, "g": 0.32, "my": CHAN_78}),
("ui_task", "barrier", "等就绪", "c", "b", "t", {"f": 0.62}),
("barrier", "ftp_task", "WaitAll", "d", "r", "l", {}),
("disp", "sim", "同源像素", "e", "r", "l", {}),
("meas", "hosttest", "纯逻辑直跑", "e", "b", "b", {"f": 0.82, "g": 0.62, "my": CHAN_B}),
("barrier", "console", "唤醒发送", "e", "b", "b", {"my": CHAN_SLOTS}),
("cfgplane", "files", "唯一落盘者", "d", "r", "r", {"f": 0.72, "g": 0.5, "gx": GUT_C4C5}),
("rs485hw", "sb_link", "回调宿主", "d", "t", "b", {"f": 0.28, "g": 0.28, "my": CHAN_78}),
]
OVERFLOW = []
def emit(s):
"""控制台打印按目标编码替换，绝不因 GBK 终端遇上 → / × 之类字符而崩
（generated-fresh 钩子会把本脚本拷进临时目录重跑，崩了就是一片假红）。"""
enc = getattr(sys.stdout, "encoding", None) or "ascii"
sys.stdout.write(s.encode(enc, "replace").decode(enc, "replace") + "\n")
def esc(s):
return html.escape(s, quote=True)
def text_w(s, fs):
"""估宽：CJK / 全角 / 箭头 ≈ 1em，其余 ≈ 0.58em（宁大勿小）。"""
w = 0.0
for ch in s:
o = ord(ch)
if o > 0x2E80 or ch in "→×·…—":
w += 1.0
else:
w += 0.58
return w * fs
def check_fit(s, fs, avail, where):
got = text_w(s, fs)
if got > avail:
OVERFLOW.append("%s: %.0f > %.0f [%s]" % (where, got, avail, s))
def wrap(s, fs, avail, max_lines, where):
"""按估宽折行；超行数截断并加 … 记一条 OVERFLOW。"""
lines, cur = [], ""
for ch in s:
if cur and text_w(cur + ch, fs) > avail:
lines.append(cur)
cur = ch
else:
cur += ch
if cur:
lines.append(cur)
if len(lines) > max_lines:
lines = lines[:max_lines]
lines[-1] = lines[-1].rstrip()[:-1] + "…"
OVERFLOW.append("%s: 折行超 %d 行 [%s]" % (where, max_lines, s))
return lines
def rect(x, y, w, h, fill, stroke, rx=7, sw=1.2, opacity=1.0):
o = ' fill-opacity="%.2f"' % opacity if opacity != 1.0 else ""
return ('' % (x, y, w, h, rx, fill, stroke, sw, o))
def txt(x, y, s, fs, fill, anchor="start", weight="normal", opacity=1.0):
o = ' fill-opacity="%.2f"' % opacity if opacity != 1.0 else ""
return ('%s'
% (x, y, FONT, fs, fill, anchor, weight, o, esc(s)))
ECOL = {"d": E_DATA, "c": E_CTRL, "e": E_EVT}
EDASH = {"d": None, "c": "5 3", "e": "1.5 3"}
def poly(points, cls):
d = " ".join(("M" if i == 0 else "L") + "%.1f %.1f" % p for i, p in enumerate(points))
dash = EDASH[cls]
dd = ' stroke-dasharray="%s"' % dash if dash else ""
w = 1.35 if cls == "d" else 1.25
return ('' % (d, ECOL[cls], w, dd, cls))
def elabel(x, y, s, cls):
"""边标签：先铺白色底片，避免与线或其他标签叠字。"""
w = text_w(s, EDGE_FONT) + 7
h = EDGE_FONT + 5
return ('%s'
% (x - w / 2, y - h / 2, w, h, x, y + EDGE_FONT * 0.36, FONT, EDGE_FONT,
ECOL[cls], esc(s)))
class Box(object):
def __init__(self, nid, col, row, label, sub):
self.id = nid
self.col = col
self.row = row
self.label = label
self.sub = sub
self.x = float(X0 + col * PX)
self.y = float(Y0 + row * PY)
self.w = float(CW)
self.h = float(CH)
def side(self, sd, f):
"""出框点：sd ∈ t b l r，f 沿该边位置（0 靠左/上，1 靠右/下）。"""
if sd == "t":
return (self.x + self.w * f, self.y)
if sd == "b":
return (self.x + self.w * f, self.y + self.h)
if sd == "l":
return (self.x, self.y + self.h * f)
return (self.x + self.w, self.y + self.h * f)
def route(a, b, ex, en, k):
"""折线点集：出框轴与入框轴垂直 → 3 点；同轴 → 4 点（加一段 gx 或 my）。"""
f = k.get("f", 0.5)
g = k.get("g", 0.5)
pa = a.side(ex, f)
pb = b.side(en, g)
ax = ex in "lr"
bx = en in "lr"
if ax != bx:
if ax: # 水平出、垂直入
return [pa, (pb[0], pa[1]), pb]
return [pa, (pa[0], pb[1]), pb]
if ax: # 都水平
if abs(pa[1] - pb[1]) < 0.6:
return [pa, pb] # 同一行：直连（别插退化折点，否则标签落点偏半格）
gx = k.get("gx", (pa[0] + pb[0]) / 2)
return [pa, (gx, pa[1]), (gx, pb[1]), pb]
if abs(pa[0] - pb[0]) < 0.6: # 同一列：直连
return [pa, pb]
my = k.get("my", (pa[1] + pb[1]) / 2) # 都垂直：走 my 水平通道
return [pa, (pa[0], my), (pb[0], my), pb]
def label_at(points, k):
"""标签落点：默认整条折线的弧长中点（不会退化成「第一半的中点」）；
seg = 指定段序号、t = 该段（或整条线）上的比例。"""
if "seg" in k:
i = k["seg"]
t = k.get("t", 0.5)
x0, y0 = points[i]
x1, y1 = points[i + 1]
return (x0 + (x1 - x0) * t, y0 + (y1 - y0) * t)
lens = [abs(points[i + 1][0] - points[i][0]) + abs(points[i + 1][1] - points[i][1])
for i in range(len(points) - 1)]
total = sum(lens) or 1.0
want = total * k.get("t", 0.5)
acc = 0.0
for i, L in enumerate(lens):
if acc + L >= want or i == len(lens) - 1:
u = 0.5 if L <= 0 else min(1.0, max(0.0, (want - acc) / L))
x0, y0 = points[i]
x1, y1 = points[i + 1]
return (x0 + (x1 - x0) * u, y0 + (y1 - y0) * u)
acc += L
return points[0]
def build():
boxes = [Box(*n) for n in NODES]
nodes = dict((b.id, b) for b in boxes)
col_band = dict((c, cb) for c, _t, _s, cb in BANDS)
check_fit(TITLE, TITLE_FONT, W - PAD_L - PAD_R, "title")
p = ['")
svg = "\n".join(p) + "\n"
ET.fromstring(svg)
d = os.path.dirname(OUT)
if not os.path.isdir(d):
os.makedirs(d)
with open(OUT, "wb") as f:
f.write(svg.encode("utf-8"))
emit("OK %d bytes, canvas %dx%d -> %s" % (len(svg.encode("utf-8")), W, H, OUT))
if OVERFLOW:
for w in OVERFLOW:
emit("OVERFLOW WARN " + w)
emit("!! %d overflow warnings" % len(OVERFLOW))
if __name__ == "__main__":
build()