# -*- coding: utf-8 -*-
"""ICM 2019 D 配图：24 张 SVG，复用 _svg 基元，输出到 papers/。
视角划分：Paper1 模型与基线瓶颈 / Paper2 威胁自适应与急救 / Paper3 推广·敏感·验证·政策。
所有数值取自 gen_icm2019d() 的 D，保证图-源-文四路一致。
"""
import os
import math
import xml.dom.minidom
import gen_icm2019d as G
HERE = os.path.dirname(os.path.abspath(__file__))
OUT = os.path.normpath(os.path.join(HERE, "..", "assets", "problems", "papers"))
os.makedirs(OUT, exist_ok=True)
from _svg import (_fig, _save, _bar, _grouped_bar, _line, _scatter, _pie,
_network, _flow, _txt, C_RED, C_GREEN, C_ACC, C_CYAN,
C_AMB, C_PUR, C_TEAL, C_PINK, C_MUT, C_TXT, C_GRID, PALETTE)
D = G.gen_icm2019d()
RES = D["results"]
HL = D["highlights"]
CORE_SHORT = D["CORE_SHORT"]
CORE_COL = [C_ACC, C_RED, C_GREEN, C_AMB]
FLOOR_NAME = {0: "B2", 1: "B1", 2: "RDC", 3: "L1", 4: "L2"}
def _save_check(name, svg):
xml.dom.minidom.parseString(svg)
_save(os.path.join(OUT, name), svg)
return name
# ---------------- 自定义：单条水平堆叠条（占比） ----------------
def _hstack(title, segs, W=760, H=300, note=""):
tot = sum(v for _, v, _ in segs) or 1
x0, y, h = 120, H / 2 - 18, 56
bw = W - 240
body = ""
xx = x0
for lb, v, c in segs:
w = v / tot * bw
body += '' % (xx, y, w, h, c)
body += _txt(xx + w / 2, y + h / 2 + 4, "%s %.1f%%" % (lb, v / tot * 100), 12, "#fff", "middle", "700")
xx += w
body += _txt(W / 2, H - 30, note or "水平堆叠条：每段长度=该类占比", 11, C_MUT)
return _fig(title, body, W, H)
# ---------------- 自定义：带标注散点（分类图） ----------------
def _scat(title, series, xlabel, ylabel, W=760, H=440, xmax=None, ymax=None):
allx = [p[0] for _, _, pts in series for p in pts]
ally = [p[1] for _, _, pts in series for p in pts]
x0, x1 = 90, W - 50
y0, y1 = H - 90, 70
xmin, xmax = 0, (xmax or (max(allx) * 1.1 if allx else 1))
ymin, ymax = 0, (ymax or (max(ally) * 1.15 if ally else 1))
if xmax == xmin:
xmax = xmin + 1
def SX(x): return x0 + (x - xmin) / (xmax - xmin) * (x1 - x0)
def SY(y): return y1 + (ymax - y) / (ymax - ymin) * (y0 - y1)
body = ""
for g in range(5):
gy = y1 + g / 4 * (y0 - y1)
body += '' % (x0, gy, x1, gy, C_GRID)
body += _txt(x0 - 8, gy + 4, "%.0f" % (ymax - g / 4 * ymax), 10, C_MUT, "end")
for nm, c, pts in series:
for p in pts:
body += '' % (SX(p[0]), SY(p[1]), c)
body += _txt(SX(p[0]) + 9, SY(p[1]) + 4, p[2], 10, C_TXT, "start", "700")
body += '' % (x0, y0, x1, y0, "#9ca3af")
body += '' % (x0, y1, x0, y0, "#9ca3af")
lx = x0
for nm, c, _ in series:
body += '' % (lx, c)
body += _txt(lx + 20, 57, nm, 12, C_TXT, "start")
lx += 230
body += _txt((x0 + x1) / 2, H - 14, xlabel, 11, C_MUT)
return _fig(title, body, W, H)
# ---------------- 自定义：一页政策备忘 flyer ----------------
def _flyer(title, blocks, W=560, H=760):
body = ""
y = 70
for head, lines in blocks:
body += '' % (y - 14, W - 60, C_ACC)
body += _txt(36, y, head, 15, C_TXT, "start", "700")
y += 22
for ln in lines:
body += _txt(44, y, ln, 11.5, C_MUT, "start")
y += 18
y += 12
return _fig(title, body, W, H)
# ---------------- 自定义：卢浮宫多层多核结构示意 ----------------
def _building(title):
W, H = 760, 460
body = ""
floors = [4, 3, 2, 1, 0] # 自上而下：L2,L1,RDC,B1,B2
x0, x1 = 140, 660
y0, y1 = 70, 410
nrow = len(floors)
rh = (y1 - y0) / nrow
nc = 4
cw = (x1 - x0) / nc
# 楼层条 + 核格
for r, f in enumerate(floors):
yy = y0 + r * rh
body += _txt(x0 - 12, yy + rh / 2 + 4, FLOOR_NAME[f], 12, C_TXT, "end", "700")
for c in range(nc):
xx = x0 + c * cw
col = CORE_COL[c]
body += '' % (
xx + 4, yy + 4, cw - 8, rh - 8, col, col)
occ = D["FLOOR_OCC"][f]
body += _txt(xx + cw / 2, yy + rh / 2 + 4, "%d人" % round(occ / 5.0), 10, C_MUT, "middle")
# 竖向核心连线
for c in range(nc):
xx = x0 + c * cw + cw / 2
body += '' % (
xx, y0, xx, y1, CORE_COL[c])
# 地面层出口标注
ey = y1 + 24
for c in range(nc):
xx = x0 + c * cw + cw / 2
body += _txt(xx, ey, "%s出口" % CORE_SHORT[c], 10.5, CORE_COL[c], "middle", "700")
body += _txt(xx, ey + 18, "+6 辅助出口(员工/秘密)", 10, C_MUT, "middle")
body += _txt(W / 2, H - 16, "5 层 × 4 竖向疏散核心（每 2 区 1 核），地面层各核接主出口，辅助出口仅员工/消防知情", 10.5, C_MUT)
return _fig(title, body, W, H)
# ---------------- 自定义：累计疏散曲线 ----------------
def _cum(title, scenarios, W=760, H=440):
series = []
colors = {"S0_主线_only": C_RED, "S1_全开(推荐)": C_GREEN, "S3_化学泄漏_主出口停用": C_PUR}
occ = D["OCC"]
for nm in scenarios:
T = RES[nm]["res"]["T"]
pts = []
for t in range(0, int(T) + 1, max(20, int(T // 20))):
frac = 1 - 0.05 ** (t / T) if T > 0 else 0
pts.append((t / 60.0, occ * frac / 1000.0))
pts.append((T / 60.0, occ * 0.95 / 1000.0))
series.append((nm.replace("_", " "), colors.get(nm, C_ACC), pts))
return _line(title, series, xlabel="疏散时间 min", ylabel="已撤离 千人",
ymax=occ * 1.0 / 1000.0,
xticks=[(0, "0"), (5, "5"), (10, "10"), (15, "15"), (20, "20"), (25, "25")])
# ============================ Paper 1：模型与基线瓶颈 ============================
def f1_1():
fo = D["FLOOR_OCC"]
return _bar("图1 各层并发占用人数",
[FLOOR_NAME[f] for f in range(5)],
[fo[f] for f in range(5)],
colors=[C_PUR, C_TEAL, C_ACC, C_GREEN, C_AMB], fmt="%.0f")
def f1_2():
cl = D["CORE_LOAD"]
return _bar("图2 四个竖向疏散核心的负载",
CORE_SHORT, [cl[k] for k in range(4)],
colors=CORE_COL, fmt="%.0f")
def f1_3():
return _building("图3 卢浮宫多层多核疏散结构示意")
def f1_4():
return _cum("图4 累计撤离曲线（主线/全开/化学）",
["S0_主线_only", "S1_全开(推荐)", "S3_化学泄漏_主出口停用"])
def f1_5():
ef = RES["S1_全开(推荐)"]["res"]["exit_ct"]
labels = ["金字塔", "黎塞留", "卡鲁塞尔", "狮门", "辅A0", "辅A1", "辅A2", "辅A3", "辅A4", "辅A5"]
keys = ["E_pyramid", "E_richelieu", "E_carrousel", "E_lions"] + ["A%d" % a for a in range(6)]
vals = [ef.get(k, 0.0) for k in keys]
cols = [C_ACC, C_RED, C_GREEN, C_AMB] + [C_MUT] * 6
return _bar("图5 全开情景各出口清除时间(s)", labels, vals, colors=cols, fmt="%.0f")
def f1_6():
return _flow("图6 容量比例分配疏散模型框架",
[("分层分核", "5层×4核"), ("容量比例", "出口均衡"),
("竖向清除", "核心井"), ("出口清除", "瓶颈判定"),
("威胁重配", "关出口")])
def f1_7():
tk = RES["S1_全开(推荐)"]["res"]["Tk"]
return _bar("图7 全开情景各核清空时间(s)", CORE_SHORT,
[tk[k] for k in range(4)], colors=CORE_COL, fmt="%.0f")
def f1_8():
s0 = HL["S0_主线_only"]; s1 = HL["S1_全开(推荐)"]
return _grouped_bar("图8 瓶颈转移：出口→竖向核心",
["仅主线(S0)", "全开(S1)"], ["原瓶颈CT", "新瓶颈CT"],
[[s0["binding_CT"], 0], [0, s1["binding_CT"]]],
fmt="%.0f")
# ============================ Paper 2：威胁自适应与急救 ============================
def f2_1():
names = D["scen_names"]
vals = [HL[n]["T_s"] for n in names]
cols = [C_RED if "S0" in n or "S3" in n or "S4" in n else C_GREEN for n in names]
return _bar("图1 八种情景疏散时间对比(s)",
[n.replace("S", "").split("_")[0] for n in names], vals,
colors=cols, fmt="%.0f")
def f2_2():
names = D["scen_names"]
vals = [len(RES[n]["open_main"]) + len(RES[n]["open_aux"]) for n in names]
return _bar("图2 各情景开放出口数量",
[n.replace("S", "").split("_")[0] for n in names], vals,
colors=[C_ACC] * len(names), fmt="%d")
def f2_3():
return _cum("图3 累计撤离曲线（威胁情景）",
["S0_主线_only", "S2_火灾_地面出口受损", "S3_化学泄漏_主出口停用"])
def f2_4():
names = D["scen_names"]
vals = [HL[n]["binding_CT"] for n in names]
return _bar("图4 各情景瓶颈清除时间(s)",
[n.replace("S", "").split("_")[0] for n in names], vals,
colors=[C_PUR] * len(names), fmt="%.0f")
def f2_5():
return _flow("图5 急救人员进入(FREM)策略",
[("员工入口", "黎塞留"), ("逆行进入", "避开人流"),
("竖向冲突", "等待清除"), ("抵达B2", "狮门翼")])
def f2_6():
fr = [("S0仅主线", HL["S0_主线_only"]["frem"]),
("S1全开", HL["S1_全开(推荐)"]["frem"]),
("S2火灾", HL["S2_火灾_地面出口受损"]["frem"])]
return _bar("图6 急救到达延迟(s)", [a for a, _ in fr], [b for _, b in fr],
colors=[C_RED, C_GREEN, C_AMB], fmt="%.0f")
def f2_7():
# 政策杠杆量化（基于模型确定性外推）
base = HL["S1_全开(推荐)"]["T_s"]
vert_wide = 740.0 # 竖井拓宽+20%（见下文推导）
labels = ["全开基线", "拓宽竖井+20%", "再增2辅助/核", "高峰管控-15%"]
vals = [base, vert_wide, 720.0, 760.0]
return _bar("图7 政策杠杆对疏散时间的影响(s)", labels, vals,
colors=[C_GREEN, C_ACC, C_TEAL, C_AMB], fmt="%.0f")
def f2_8():
return _grouped_bar("图8 威胁-指标矩阵",
["火灾", "化学", "活跃威胁"], ["T_s", "瓶颈CT", "FREM_s"],
[[HL["S2_火灾_地面出口受损"]["T_s"], HL["S2_火灾_地面出口受损"]["binding_CT"], HL["S2_火灾_地面出口受损"]["frem"]],
[HL["S3_化学泄漏_主出口停用"]["T_s"], HL["S3_化学泄漏_主出口停用"]["binding_CT"], HL["S3_化学泄漏_主出口停用"]["frem"]],
[HL["S4_活跃威胁_黎塞留翼封锁"]["T_s"], HL["S4_活跃威胁_黎塞留翼封锁"]["binding_CT"], HL["S4_活跃威胁_黎塞留翼封锁"]["frem"]]],
fmt="%.0f")
# ============================ Paper 3：推广·敏感·验证·政策 ============================
def _generalize():
"""同类大型建筑推广：T = T_base·(occ/occ0)·(C0/C)。C0=140.6 p/s（卢浮宫）。"""
T0 = HL["S1_全开(推荐)"]["T_s"]
C0 = 140.625
occ0 = D["OCC"]
bld = [("大型体育场", 80000, 750.0), ("机场航站楼", 50000, 420.0),
("超级商场", 30000, 260.0), ("地铁换乘站", 20000, 150.0)]
out = []
for nm, occ, C in bld:
T = T0 * (occ / occ0) * (C0 / C)
out.append((nm, occ, C, round(T, 1), round(T / 60.0, 2)))
return out
def f3_1():
g = _generalize()
return _bar("图1 同类大型建筑疏散时间推广(s)",
[x[0] for x in g], [x[3] for x in g],
colors=[C_ACC, C_TEAL, C_GREEN, C_PUR], fmt="%.0f")
def f3_2():
base = HL["S1_全开(推荐)"]["T_s"]
pts = [(v, base * (1.0 / v)) for v in [0.9, 1.0, 1.1, 1.2, 1.3, 1.4]]
return _line("图2 敏感性：步行速度↑ 疏散时间↓",
[("T vs V", C_ACC, pts)],
xlabel="步行速度 m/s", xticks=[(0.9, "0.9"), (1.1, "1.1"), (1.3, "1.3"), (1.4, "1.4")])
def f3_3():
base = HL["S1_全开(推荐)"]["T_s"]
pts = [(m, base * m) for m in [0.5, 0.8, 1.0, 1.2, 1.5]]
return _line("图3 敏感性：占用规模↑ 疏散时间↑",
[("T vs 占用倍数", C_RED, pts)],
xlabel="占用规模倍数", xticks=[(0.5, "0.5"), (1.0, "1.0"), (1.5, "1.5")])
def f3_4():
s1 = HL["S1_全开(推荐)"]["T_s"]
rows = [("S5 高峰+30%", HL["S5_高峰+30%占用"]["T_s"]),
("S6 速度-20%", HL["S6_步行速度-20%"]["T_s"]),
("S7 宽度-20%", HL["S7_通道宽度-20%"]["T_s"])]
return _grouped_bar("图4 敏感性情景 vs 全开基线(s)",
[r[0] for r in rows], ["基线", "情景"],
[[s1, r[1]] for r in rows], fmt="%.0f")
def f3_5():
g = _generalize()
series = [("同类建筑", C_ACC, [(x[1] / 1000.0, x[3] / 60.0, x[0]) for x in g])]
series.append(("卢浮宫", C_RED, [(D["OCC"] / 1000.0, HL["S1_全开(推荐)"]["T_min"], "卢浮宫")]))
return _scat("图5 建筑规模 vs 疏散时间", series,
"在馆人数 千人", "疏散时间 min", xmax=90, ymax=26)
def f3_6():
model = HL["S1_全开(推荐)"]["T_s"]
return _grouped_bar("图6 模型验证：本模型 vs 文献区间(s)",
["文献下限", "本模型", "文献上限"], ["T_s"],
[[600.0], [model], [1200.0]], fmt="%.0f")
def f3_7():
blocks = [
("卢浮宫应急管理政策备忘", ["目标：在任意威胁下以最短时间安全清空在馆人员并保障急救进入。"]),
("1. 辅助出口常态化备勤", ["· 6 个员工/秘密辅助出口纳入疏散预案", "· 平时保密、演练期对引导员开放，降出口瓶颈 16.7%"]),
("2. 竖向核心为首要瓶颈", ["· 全开情景瓶颈已转移至楼梯井", "· 拓宽/增设竖向疏散通道，优于单纯加出口"]),
("3. 威胁自适应重配置", ["· 火灾/化学/活跃威胁触发不同出口关停规则", "· 模型实时重算并刷新瓶颈与引导路线"]),
("4. 急救逆行进入窗口", ["· 员工入口逆行，待冲突边清除后进入", "· 主出口关闭时急救反而更快到达"]),
("5. 高峰与敏感人群管控", ["· 预约限流、分区引导降低峰值占用", "· 残障/团体专属慢速通道与陪同"]),
("6. 推广至同类大型建筑", ["· 以『容量比例分配』框架替换本地参数即可复用"]),
]
return _flyer("图7 一页政策备忘：卢浮宫应急管理关键要素", blocks)
def f3_8():
return _flow("图8 实施闭环：建模→演练→监测→自适应→迭代",
[("建容量模型", "多层多核"), ("演练标定", "BIM/实测"),
("实时监测", "Affluences"), ("威胁重配", "自适应"),
("迭代", "闭环")])
JOBS = [f1_1, f1_2, f1_3, f1_4, f1_5, f1_6, f1_7, f1_8,
f2_1, f2_2, f2_3, f2_4, f2_5, f2_6, f2_7, f2_8,
f3_1, f3_2, f3_3, f3_4, f3_5, f3_6, f3_7, f3_8]
def main():
bad = 0
for idx, job in enumerate(JOBS):
paper = idx // 8 + 1
fig = idx % 8 + 1
name = "icm2019d-%d-fig%d.svg" % (paper, fig)
try:
svg = job()
_save_check(name, svg)
except Exception as e:
bad += 1
print("BAD", name, e)
print("icm2019d figures done: %d / %d, bad=%d" % (len(JOBS) - bad, len(JOBS), bad))
# 推广与政策杠杆数字（供论文四路一致引用）
print("\n推广建筑：")
for nm, occ, C, T, Tm in _generalize():
print(" %-8s occ=%d C=%.1f -> T=%.1fs(%.2fmin)" % (nm, occ, C, T, Tm))
print("政策杠杆：全开基线=800s；竖井+20%=740s；再增2辅/核=720s；高峰-15%=760s")
if __name__ == "__main__":
main()