from __future__ import annotations import csv import hashlib import json from collections import defaultdict from decimal import Decimal, ROUND_HALF_UP from pathlib import Path import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt from matplotlib import font_manager from matplotlib.ticker import PercentFormatter ROOT = Path(__file__).resolve().parent DATA = ROOT / "data" OUTPUTS = ROOT / "outputs" FIGURES = OUTPUTS / "figures" FONT = ROOT / "assets" / "fonts" / "NotoSansTC-Regular.ttf" FONT_SHA256 = "40452f54ea36b5c1f1ec8bd92dc18d5cd19fafb73b8abf9c48500a00d09ec762" ENERGY_FIELDS = { "total": "total_million_kwh", "pumped_storage": "pumped_storage_million_kwh", "thermal_total": "thermal_total_million_kwh", "coal": "coal_million_kwh", "oil": "oil_million_kwh", "gas": "gas_million_kwh", "nuclear": "nuclear_million_kwh", "renewable_total": "renewable_total_million_kwh", "conventional_hydro": "conventional_hydro_million_kwh", "geothermal": "geothermal_million_kwh", "solar": "solar_million_kwh", "wind": "wind_million_kwh", "biomass": "biomass_million_kwh", "waste": "waste_million_kwh", } RENEWABLE_TAIPOWER_TYPES = { "geothermal", "conventional_hydro", "contracted_hydro", "hydro", "wind", "solar", "biomass", "waste", "waste_biogas", } def rounded(value: float, digits: int = 4) -> float: quantum = Decimal(1).scaleb(-digits) return float(Decimal(str(value)).quantize(quantum, rounding=ROUND_HALF_UP)) def read_csv(path: Path) -> list[dict[str, str]]: with path.open("r", encoding="utf-8", newline="") as stream: return list(csv.DictReader(stream)) def write_csv(path: Path, rows: list[dict[str, object]]) -> None: if not rows: raise ValueError(f"Refusing to write empty output: {path}") with path.open("w", encoding="utf-8", newline="") as stream: writer = csv.DictWriter(stream, fieldnames=list(rows[0])) writer.writeheader() writer.writerows(rows) def choose_font() -> Path: if not FONT.is_file(): raise FileNotFoundError(f"Packaged Noto Sans TC font is missing: {FONT}") actual = hashlib.sha256(FONT.read_bytes()).hexdigest() if actual != FONT_SHA256: raise ValueError(f"Packaged font hash mismatch: {actual}") return FONT def main() -> None: OUTPUTS.mkdir(parents=True, exist_ok=True) FIGURES.mkdir(parents=True, exist_ok=True) monthly = read_csv(DATA / "energy-monthly-generation-2016-2025.csv") if len(monthly) != 120: raise ValueError(f"Expected 120 monthly observations, found {len(monthly)}") if monthly[0]["month"] != "201601" or monthly[-1]["month"] != "202512": raise ValueError("Monthly observation window is not exactly 2016-01 through 2025-12") grouped: dict[int, dict[str, float]] = defaultdict(lambda: defaultdict(float)) max_component_gap = 0.0 for row in monthly: year = int(row["month"][:4]) values = {name: float(row[field]) for name, field in ENERGY_FIELDS.items()} component_total = values["pumped_storage"] + values["thermal_total"] + values["nuclear"] + values["renewable_total"] max_component_gap = max(max_component_gap, abs(values["total"] - component_total)) for name, value in values.items(): grouped[year][name] += value annual_rows: list[dict[str, object]] = [] for year in sorted(grouped): values = grouped[year] total = values["total"] row: dict[str, object] = {"year": year, "total_generation_million_kwh": f"{total:.6f}"} for name in ( "coal", "gas", "oil", "nuclear", "renewable_total", "pumped_storage", "conventional_hydro", "solar", "wind", "geothermal", "biomass", "waste", ): row[f"{name}_million_kwh"] = f"{values[name]:.6f}" row[f"{name}_share_pct"] = f"{values[name] / total * 100:.4f}" annual_rows.append(row) write_csv(OUTPUTS / "annual-generation-mix.csv", annual_rows) monthly_2025: list[dict[str, object]] = [] for raw in monthly: if not raw["month"].startswith("2025"): continue total = float(raw["total_million_kwh"]) monthly_2025.append( { "month": raw["month"], "total_generation_million_kwh": f"{total:.6f}", "coal_share_pct": f"{float(raw['coal_million_kwh']) / total * 100:.4f}", "gas_share_pct": f"{float(raw['gas_million_kwh']) / total * 100:.4f}", "nuclear_share_pct": f"{float(raw['nuclear_million_kwh']) / total * 100:.4f}", "renewable_share_pct": f"{float(raw['renewable_total_million_kwh']) / total * 100:.4f}", "solar_share_pct": f"{float(raw['solar_million_kwh']) / total * 100:.4f}", "wind_share_pct": f"{float(raw['wind_million_kwh']) / total * 100:.4f}", } ) if len(monthly_2025) != 12: raise ValueError("2025 is not a complete 12-month period") write_csv(OUTPUTS / "monthly-generation-shares-2025.csv", monthly_2025) taipower = read_csv(DATA / "taipower-net-generation-purchases-2016-2025.csv") taipower_grouped: dict[int, dict[str, float]] = defaultdict(lambda: defaultdict(float)) for row in taipower: if row["net_generation_or_purchase_kwh"] == "": continue year = int(row["year"]) value = float(row["net_generation_or_purchase_kwh"]) taipower_grouped[year]["total"] += value taipower_grouped[year][row["energy_type"]] += value if row["energy_type"] in RENEWABLE_TAIPOWER_TYPES: taipower_grouped[year]["renewable_group"] += value cross_rows: list[dict[str, object]] = [] for year in sorted(taipower_grouped): values = taipower_grouped[year] total = values["total"] cross_rows.append( { "year": year, "net_generation_and_purchases_kwh": f"{total:.3f}", "coal_share_pct": f"{values['coal'] / total * 100:.4f}", "gas_share_pct": f"{values['gas'] / total * 100:.4f}", "nuclear_share_pct": f"{values['nuclear'] / total * 100:.4f}", "renewable_group_share_pct": f"{values['renewable_group'] / total * 100:.4f}", "scope_note": "Taipower-owned net generation plus purchased electricity; not equal to nationwide gross generation", } ) write_csv(OUTPUTS / "taipower-scope-cross-check.csv", cross_rows) factors = read_csv(DATA / "factor-observations.csv") factor_rows: list[dict[str, object]] = [] for row in factors: factor_rows.append( { "year": row["year"], "factor_scope": row["factor_scope"], "value_kg_co2e_per_kwh": row["value_kg_co2e_per_kwh"], "status": row["status"], "difference_from_2025_public_g_per_kwh": "" if row["year"] != "2025" else f"{(float(row['value_kg_co2e_per_kwh']) - 0.467) * 1000:.1f}", "applicable_to": row["applicable_to"], } ) write_csv(OUTPUTS / "electricity-factor-comparison.csv", factor_rows) annual = {int(row["year"]): row for row in annual_rows} def share(year: int, name: str) -> float: return grouped[year][name] / grouped[year]["total"] * 100 total_2024 = grouped[2024]["total"] total_2025 = grouped[2025]["total"] changes_2025 = { name: rounded(share(2025, name) - share(2024, name), 4) for name in ("coal", "gas", "oil", "nuclear", "renewable_total", "pumped_storage") } changes_since_2016 = { name: rounded(share(2025, name) - share(2016, name), 4) for name in ("coal", "gas", "oil", "nuclear", "renewable_total", "pumped_storage") } factor_change_2024 = (0.467 / 0.474 - 1) * 100 factor_change_2016 = (0.467 / 0.530 - 1) * 100 # Cross-source direction check is deliberately limited to broad directional movements. cross = {int(row["year"]): row for row in cross_rows} def cross_share(year: int, name: str) -> float: field = "renewable_group" if name == "renewable_total" else name return taipower_grouped[year][field] / taipower_grouped[year]["total"] * 100 cross_direction = { name: cross_share(2025, name) - cross_share(2024, name) for name in ("coal", "gas", "nuclear") } cross_direction["renewable_total"] = cross_share(2025, "renewable_total") - cross_share(2024, "renewable_total") nationwide_direction = {name: changes_2025[name] for name in cross_direction} direction_agreement = { name: (cross_direction[name] == 0 and nationwide_direction[name] == 0) or (cross_direction[name] * nationwide_direction[name] > 0) for name in cross_direction } font_path = choose_font() font_manager.fontManager.addfont(str(font_path)) font_prop = font_manager.FontProperties(fname=str(font_path)) plt.rcParams.update({ "font.family": "sans-serif", "font.sans-serif": [font_prop.get_name()], "axes.unicode_minus": False, "figure.dpi": 160, "savefig.dpi": 180, }) colors = { "coal": "#5b4b44", "gas": "#d47a3a", "oil": "#8b6c5c", "nuclear": "#7b6fa8", "renewable_total": "#4f9d69", "pumped_storage": "#4f86a6", } labels = { "coal": "燃煤", "gas": "燃氣", "oil": "燃油", "nuclear": "核能", "renewable_total": "再生能源", "pumped_storage": "抽蓄水力", } years = sorted(annual) fig, ax = plt.subplots(figsize=(9.4, 5.4)) stack_names = ["coal", "gas", "oil", "nuclear", "renewable_total", "pumped_storage"] ax.stackplot( years, [[share(year, name) for year in years] for name in stack_names], labels=[labels[name] for name in stack_names], colors=[colors[name] for name in stack_names], alpha=0.92, ) ax.set_title("2016-2025年臺灣全國毛發電結構", fontproperties=font_prop, fontsize=15) ax.set_ylabel("占全國毛發電量（%）", fontproperties=font_prop) ax.yaxis.set_major_formatter(PercentFormatter(100)) ax.set_xlim(2016, 2025) ax.set_ylim(0, 100) ax.grid(axis="y", alpha=0.18) ax.legend(loc="upper center", bbox_to_anchor=(0.5, -0.10), ncol=3, frameon=False, prop=font_prop) fig.tight_layout() fig.savefig(FIGURES / "figure-1-generation-mix-2016-2025.png", bbox_inches="tight") plt.close(fig) fig, ax = plt.subplots(figsize=(8.8, 5.1)) names = ["coal", "gas", "oil", "nuclear", "renewable_total", "pumped_storage"] values = [changes_2025[name] for name in names] bars = ax.bar([labels[name] for name in names], values, color=[colors[name] for name in names]) ax.axhline(0, color="#342b26", linewidth=0.8) ax.set_title("2024至2025年全國毛發電占比變化", fontproperties=font_prop, fontsize=15) ax.set_ylabel("百分點變化", fontproperties=font_prop) ax.grid(axis="y", alpha=0.18) for bar, value in zip(bars, values): ax.text(bar.get_x() + bar.get_width() / 2, value + (0.16 if value >= 0 else -0.28), f"{value:+.2f}", ha="center", va="bottom" if value >= 0 else "top", fontsize=9) fig.tight_layout() fig.savefig(FIGURES / "figure-2-mix-change-2024-2025.png", bbox_inches="tight") plt.close(fig) fig, ax = plt.subplots(figsize=(9.4, 5.3)) month_numbers = list(range(1, 13)) series = { "燃煤": [float(row["coal_share_pct"]) for row in monthly_2025], "燃氣": [float(row["gas_share_pct"]) for row in monthly_2025], "再生能源": [float(row["renewable_share_pct"]) for row in monthly_2025], "太陽光電": [float(row["solar_share_pct"]) for row in monthly_2025], "風力": [float(row["wind_share_pct"]) for row in monthly_2025], } line_colors = ["#5b4b44", "#d47a3a", "#4f9d69", "#d6aa26", "#4f86a6"] for (label, values), color in zip(series.items(), line_colors): ax.plot(month_numbers, values, marker="o", linewidth=2, markersize=3.5, label=label, color=color) ax.set_title("2025年各月全國毛發電占比", fontproperties=font_prop, fontsize=15) ax.set_xlabel("月份", fontproperties=font_prop) ax.set_ylabel("占全國毛發電量（%）", fontproperties=font_prop) ax.set_xticks(month_numbers) ax.grid(alpha=0.18) ax.legend(ncol=5, loc="upper center", bbox_to_anchor=(0.5, -0.12), frameon=False, prop=font_prop) fig.tight_layout() fig.savefig(FIGURES / "figure-3-monthly-generation-shares-2025.png", bbox_inches="tight") plt.close(fig) fig, ax = plt.subplots(figsize=(8.8, 5.2)) factor_labels = ["全國電力\n（初估）", "產業盤查", "公用售電業", "民生住宅"] factor_values = [0.456, 0.466, 0.467, 0.471] bars = ax.bar(factor_labels, factor_values, color=["#4f9d69", "#4f86a6", "#7a6b60", "#d47a3a"]) ax.set_ylim(0.44, 0.478) ax.set_ylabel("kg CO2e／度", fontproperties=font_prop) ax.set_title("2025年官方電力排碳係數：不同口徑不可互換", fontproperties=font_prop, fontsize=15) ax.grid(axis="y", alpha=0.18) for bar, value in zip(bars, factor_values): ax.text(bar.get_x() + bar.get_width() / 2, value + 0.0006, f"{value:.3f}", ha="center", fontsize=10) fig.tight_layout() fig.savefig(FIGURES / "figure-4-factor-scope-comparison-2025.png", bbox_inches="tight") plt.close(fig) results = { "reportNumber": "SHWRP-2026-030", "analysisPeriod": {"start": "2016-01", "end": "2025-12"}, "sampleCounts": { "monthlyGenerationRows": len(monthly), "completeYears": len(annual_rows), "taipowerInputRows": len(taipower), "taipowerCrossCheckAnnualRows": len(cross_rows), "factorObservations": len(factors), "figures": 4, }, "dataQuality": { "maxMonthlyComponentIdentityGapMillionKwh": rounded(max_component_gap, 9), "allYearsHave12Months": all(sum(1 for row in monthly if row["month"].startswith(str(year))) == 12 for year in years), "directionAgreement2024to2025": direction_agreement, "scopeWarning": "Energy Administration data are nationwide gross generation; Taipower data are owned net generation plus purchased electricity. Levels are not treated as identical.", }, "year2025": { "totalGenerationMillionKwh": rounded(total_2025, 6), "totalGrowthFrom2024Pct": rounded((total_2025 / total_2024 - 1) * 100, 4), "sharesPct": {name: rounded(share(2025, name), 4) for name in ("coal", "gas", "oil", "nuclear", "renewable_total", "pumped_storage", "solar", "wind", "conventional_hydro")}, "changeFrom2024PercentagePoints": changes_2025, "changeFrom2016PercentagePoints": changes_since_2016, }, "emissionFactors": { "publicSales2025": 0.467, "industrialInventory2025": 0.466, "residentialInventory2025": 0.471, "nationalPreliminary2025": 0.456, "publicSalesChangeFrom2024Pct": rounded(factor_change_2024, 4), "publicSalesChangeFrom2016Pct": rounded(factor_change_2016, 4), "residentialMinusIndustrialGramsPerKwh": 5.0, "residentialMinusIndustrialKgPer1000Kwh": 5.0, "nationalPreliminaryMinusPublicSalesGramsPerKwh": -11.0, }, "interpretiveLimits": [ "The factor differences are accounting allocations for defined tariff and market scopes, not evidence that named users receive physically separate electricity.", "Generation shares and emission factors are descriptive aggregates; the study does not identify causal contributions of any technology or policy.", "The nationwide 2025 factor of 0.456 kg CO2e/kWh is explicitly preliminary in the official release.", "Gross generation and net generation-plus-purchases use different system boundaries and are compared only for broad direction.", ], } (OUTPUTS / "analysis-results.json").write_text( json.dumps(results, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" ) print(json.dumps(results, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()