221 lines
8.2 KiB
Python
221 lines
8.2 KiB
Python
#!/usr/bin/env python3
|
||
"""生成试点数据分析图表。
|
||
|
||
输出：
|
||
plots/pilot_run_data.png — 各地市预警数据与接入时间线（多面板）
|
||
plots/vendor_effort.png — 供应商人天投入饼图
|
||
plots/access_cost.png — 接入链路耗时对比图
|
||
"""
|
||
|
||
import csv
|
||
import os
|
||
import sys
|
||
from datetime import datetime
|
||
|
||
import matplotlib
|
||
matplotlib.use("Agg")
|
||
import matplotlib.pyplot as plt
|
||
from matplotlib.patches import Patch
|
||
import matplotlib.dates as mdates
|
||
|
||
PLOT_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
|
||
_CJK_FONT = "Noto Sans CJK SC"
|
||
_STAGE_COLORS = {
|
||
"稳定运行": "#2E7D32",
|
||
"实际产出": "#1565C0",
|
||
"推广接入": "#BDBDBD",
|
||
}
|
||
|
||
|
||
def load_csv(name):
|
||
path = os.path.join(PLOT_DIR, name)
|
||
with open(path, encoding="utf-8") as f:
|
||
return list(csv.DictReader(f))
|
||
|
||
|
||
def _parse_date(s):
|
||
s = (s or "").strip()
|
||
if not s:
|
||
return None
|
||
for fmt in ("%Y-%m-%d", "%Y-%m"):
|
||
try:
|
||
return datetime.strptime(s, fmt)
|
||
except ValueError:
|
||
continue
|
||
return None
|
||
|
||
|
||
def plot_run_data():
|
||
rows = load_csv("pilot_run_data.csv")
|
||
plt.rcParams["font.sans-serif"] = [_CJK_FONT]
|
||
plt.rcParams["axes.unicode_minus"] = False
|
||
|
||
cities = [r["city"] for r in rows]
|
||
alerts = [int(r["alert_count"]) if r.get("alert_count") else 0 for r in rows]
|
||
stages = [r.get("stage", "") for r in rows]
|
||
bar_colors = [_STAGE_COLORS.get(s, "#999") for s in stages]
|
||
|
||
fig, (ax1, ax2) = plt.subplots(
|
||
2, 1, figsize=(13, 9), gridspec_kw={"height_ratios": [3, 2]}
|
||
)
|
||
fig.suptitle(
|
||
"图 1 电量突增能力应用运行数据可视化（截至 2026-06-25）",
|
||
fontsize=14, fontweight="bold", y=0.98,
|
||
)
|
||
|
||
# ── Panel A: 横向柱状图 ──
|
||
y_pos = list(range(len(cities)))
|
||
bars = ax1.barh(y_pos, alerts, color=bar_colors, edgecolor="white", linewidth=0.6, height=0.65)
|
||
ax1.set_yticks(y_pos)
|
||
ax1.set_yticklabels(cities, fontsize=10)
|
||
ax1.set_xlabel("累计预警数据（条）", fontsize=10)
|
||
ax1.invert_yaxis()
|
||
ax1.grid(axis="x", alpha=0.25, linestyle="--")
|
||
ax1.spines["top"].set_visible(False)
|
||
ax1.spines["right"].set_visible(False)
|
||
|
||
x_max = max(alerts) if max(alerts) > 0 else 10000
|
||
ax1.set_xlim(0, x_max * 1.28)
|
||
|
||
for i, (val, stage) in enumerate(zip(alerts, stages)):
|
||
if val > 0:
|
||
ax1.text(val + x_max * 0.01, i, f"{val:,}",
|
||
va="center", fontsize=9, fontweight="bold", color="#333")
|
||
else:
|
||
ax1.text(x_max * 0.01, i, "接入中（尚无产出）",
|
||
va="center", fontsize=8, color="#888")
|
||
ax1.text(x_max * 1.22, i, stage,
|
||
va="center", fontsize=8, color=_STAGE_COLORS.get(stage, "#666"))
|
||
|
||
legend_elems = [Patch(facecolor=_STAGE_COLORS[s], label=s) for s in ["稳定运行", "实际产出", "推广接入"]]
|
||
ax1.legend(handles=legend_elems, loc="lower right", fontsize=9, framealpha=0.9, title="运行阶段")
|
||
|
||
stats_text = (
|
||
"汇总指标（表 2）\n"
|
||
"─────────────\n"
|
||
"异常命中 144,853 条\n"
|
||
"预警推送 476 次\n"
|
||
"监测规则 10 条\n"
|
||
"覆盖人员 6 名\n"
|
||
"推广培训 7 场\n"
|
||
"对接地市 8 个"
|
||
)
|
||
ax1.text(0.98, 0.02, stats_text, transform=ax1.transAxes,
|
||
fontsize=8, va="bottom", ha="right", family="Noto Sans Mono CJK SC",
|
||
bbox=dict(boxstyle="round,pad=0.5", facecolor="#F5F5F5", edgecolor="#CCC", alpha=0.92))
|
||
|
||
# ── Panel B: 接入时间线 ──
|
||
data_end = datetime(2026, 6, 25)
|
||
for i, r in enumerate(rows):
|
||
start = _parse_date(r.get("start_date"))
|
||
output = _parse_date(r.get("output_date"))
|
||
stage = r.get("stage", "")
|
||
color = _STAGE_COLORS.get(stage, "#BDBDBD")
|
||
if start:
|
||
ax2.barh(i, (data_end - start).days, left=mdates.date2num(start),
|
||
height=0.5, color=color, edgecolor="white", linewidth=0.5, alpha=0.85)
|
||
if output:
|
||
ax2.plot(output, i, marker="D", markersize=7, color="#E65100",
|
||
markeredgecolor="white", markeredgewidth=0.8, zorder=5)
|
||
ax2.annotate(f"产出\n{output.strftime('%m-%d')}",
|
||
xy=(output, i), xytext=(8, 0), textcoords="offset points",
|
||
fontsize=7, color="#E65100", va="center")
|
||
|
||
ax2.set_yticks(y_pos)
|
||
ax2.set_yticklabels(cities, fontsize=10)
|
||
ax2.invert_yaxis()
|
||
ax2.set_title("各地市接入时间线与产出里程碑", fontsize=11, pad=8)
|
||
ax2.xaxis.set_major_locator(mdates.MonthLocator())
|
||
ax2.xaxis.set_minor_locator(mdates.DayLocator(bymonthday=[10, 20]))
|
||
ax2.xaxis.set_major_formatter(mdates.DateFormatter("%Y-%m"))
|
||
ax2.grid(axis="x", alpha=0.2, linestyle="--")
|
||
ax2.spines["top"].set_visible(False)
|
||
ax2.spines["right"].set_visible(False)
|
||
|
||
from matplotlib.lines import Line2D
|
||
diamond = Line2D([0], [0], marker="D", color="w", markerfacecolor="#E65100",
|
||
markersize=7, label="产出起始")
|
||
ax2.legend(handles=[diamond], loc="lower right", fontsize=9, framealpha=0.9)
|
||
|
||
plt.tight_layout(rect=[0, 0, 1, 0.96])
|
||
plt.savefig(os.path.join(PLOT_DIR, "pilot_run_data.png"), dpi=150, bbox_inches="tight")
|
||
plt.close()
|
||
|
||
|
||
def plot_vendor_effort():
|
||
rows = load_csv("vendor_effort.csv")
|
||
labels = [r["category"] for r in rows]
|
||
sizes = [int(r["person_days"]) for r in rows]
|
||
|
||
fig, ax = plt.subplots(figsize=(8, 6))
|
||
wedges, texts, autotexts = ax.pie(
|
||
sizes, labels=labels, autopct="%1.0f%%", startangle=140,
|
||
textprops={"fontsize": 9},
|
||
)
|
||
ax.set_title("供应商人天投入结构（合计 100 人天）")
|
||
plt.tight_layout()
|
||
plt.savefig(os.path.join(PLOT_DIR, "vendor_effort.png"), dpi=150)
|
||
plt.close()
|
||
|
||
|
||
def plot_access_cost():
|
||
rows = load_csv("access_cost.csv")
|
||
steps = [r["step"] for r in rows if r["step"] != "端到端合计"]
|
||
ideal = [int(r["ideal_days"]) for r in rows if r["step"] != "端到端合计"]
|
||
actual = [int(r["actual_days"]) for r in rows if r["step"] != "端到端合计"]
|
||
|
||
x = range(len(steps))
|
||
width = 0.35
|
||
fig, ax = plt.subplots(figsize=(10, 5))
|
||
ax.bar([i - width / 2 for i in x], ideal, width, label="理想耗时", color="#4CAF50")
|
||
ax.bar([i + width / 2 for i in x], actual, width, label="实际耗时", color="#FF5722")
|
||
ax.set_ylabel("耗时（工作日）")
|
||
ax.set_title("单地市接入链路耗时：理想 vs 实际")
|
||
ax.set_xticks(x)
|
||
ax.set_xticklabels(steps, rotation=20, ha="right")
|
||
ax.legend()
|
||
plt.tight_layout()
|
||
plt.savefig(os.path.join(PLOT_DIR, "access_cost.png"), dpi=150)
|
||
plt.close()
|
||
|
||
|
||
def plot_city_comparison():
|
||
rows = load_csv("city_comparison.csv")
|
||
cities = [r["city"] for r in rows]
|
||
alerts = [int(r["alert_count"]) for r in rows]
|
||
teams = [r["team"] for r in rows]
|
||
colors = {"第一梯队": "#4CAF50", "第二梯队": "#2196F3", "第三梯队": "#BDBDBD"}
|
||
bar_colors = [colors.get(t, "#999") for t in teams]
|
||
|
||
fig, ax = plt.subplots(figsize=(10, 5))
|
||
bars = ax.bar(cities, alerts, color=bar_colors)
|
||
ax.set_ylabel("累计预警数据（条）")
|
||
ax.set_title("各地市接入阶段与产出对比（截至 2026-06-25）")
|
||
for bar, val, team in zip(bars, alerts, teams):
|
||
if val > 0:
|
||
ax.text(bar.get_x() + bar.get_width() / 2, val + 200,
|
||
f"{val:,}", ha="center", fontsize=9)
|
||
else:
|
||
ax.text(bar.get_x() + bar.get_width() / 2, 200,
|
||
team, ha="center", fontsize=8, color="#666")
|
||
from matplotlib.patches import Patch
|
||
legend = [Patch(facecolor=colors[t], label=t) for t in ["第一梯队", "第二梯队", "第三梯队"]]
|
||
ax.legend(handles=legend, loc="upper right")
|
||
plt.xticks(rotation=30, ha="right")
|
||
plt.tight_layout()
|
||
plt.savefig(os.path.join(PLOT_DIR, "city_comparison.png"), dpi=150)
|
||
plt.close()
|
||
|
||
|
||
def main():
|
||
plot_run_data()
|
||
plot_vendor_effort()
|
||
plot_access_cost()
|
||
plot_city_comparison()
|
||
print("图表已生成：pilot_run_data.png, vendor_effort.png, access_cost.png, city_comparison.png")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|