#!/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()