Use shared constant for hours per year

workflow/rules/common.smk

@@ -11,6 +11,7 @@ path = workflow.source_path("../scripts/_helpers.py")
 sys.path.insert(0, os.path.dirname(path))
 
 from _helpers import validate_checksum, update_config_from_wildcards
+from constants import HOURS_PER_YEAR
 from snakemake.utils import update_config
 
 
@@ -95,7 +96,7 @@ def memory(w):
     for o in w.opts.split("-"):
         m = re.match(r"^(\d+)seg$", o, re.IGNORECASE)
         if m is not None:
-            factor *= int(m.group(1)) / 8760
+            factor *= int(m.group(1)) / HOURS_PER_YEAR
             break
     if w.clusters.endswith("m") or w.clusters.endswith("c"):
         val = int(factor * (50000 + 30 * int(w.simpl) + 195 * int(w.clusters[:-1])))

workflow/scripts/_helpers.py

@@ -12,6 +12,7 @@
 import pypsa
 import requests
 import yaml
+from constants import HOURS_PER_YEAR
 from snakemake.utils import update_config
 
 REGION_COLS = ["geometry", "name", "x", "y", "country"]
@@ -184,7 +185,7 @@ def load_network_for_plots(fn, tech_costs, config, combine_hydro_ps=True):
     # bus_carrier = n.storage_units.bus.map(n.buses.carrier)
     # n.storage_units.loc[bus_carrier == "heat","carrier"] = "water tanks"
 
-    num_years = n.snapshot_weightings.loc[n.investment_periods[0]].objective.sum() / 8760.0
+    num_years = n.snapshot_weightings.loc[n.investment_periods[0]].objective.sum() / HOURS_PER_YEAR
     costs = load_costs(tech_costs, config["costs"], config["electricity"], num_years)
     update_transmission_costs(n, costs)
 

workflow/scripts/add_extra_components.py

@@ -8,6 +8,7 @@
 import pypsa
 from _helpers import calculate_annuity, configure_logging
 from add_electricity import add_missing_carriers
+from constants import HOURS_PER_YEAR
 from eia import FuelCosts
 from opts._helpers import get_region_buses
 from pypsa.descriptors import get_switchable_as_dense as get_as_dense
@@ -154,7 +155,7 @@ def attach_phs_storageunits(n: pypsa.Network, elec_opts, costs: pd.DataFrame):
             * region_onshore_psh_grp["cost_kw_round"]
             * 1e3
             * n.snapshot_weightings.objective.sum()
-            / 8760.0
+            / HOURS_PER_YEAR
         )
 
         region_onshore_psh_grp["marginal_cost"] = psh_vom
@@ -1342,7 +1343,7 @@ def add_co2_network(n: pypsa.Network, config: dict):
         connections = n.lines
 
     # calculate annualized capital cost
-    number_years = n.snapshot_weightings.generators.sum() / 8760
+    number_years = n.snapshot_weightings.generators.sum() / HOURS_PER_YEAR
     cost = (
         config["co2"]["network"]["capital_cost"]
         * calculate_annuity(config["co2"]["network"]["lifetime"], config["co2"]["network"]["discount_rate"])
@@ -1459,7 +1460,7 @@ def add_dac(n: pypsa.Network, config: dict, sector: bool):
     )
 
     # calculate annualized capital cost
-    number_years = n.snapshot_weightings.generators.sum() / 8760
+    number_years = n.snapshot_weightings.generators.sum() / HOURS_PER_YEAR
     cost = (
         config["dac"]["capital_cost"]
         * calculate_annuity(config["dac"]["lifetime"], config["dac"]["discount_rate"])

workflow/scripts/build_cost_data.py

@@ -166,7 +166,7 @@ def calculate_capex(df: pd.DataFrame, discount_rate: float) -> pd.DataFrame:
         capex = capex.value.unstack().fillna(0)
 
         # n years should be
-        # n.snapshot_weightings.loc[n.investment_periods[x]].objective.sum() / 8760.0
+        # n.snapshot_weightings.loc[n.investment_periods[x]].objective.sum() / const.HOURS_PER_YEAR
 
         capex["capital_cost"] = (
             (

workflow/scripts/build_demand.py

@@ -382,7 +382,7 @@ def apply_timezone_shift(timezone: str) -> int:
 
         df["utc_shift"] = df.State.map(utc_shift)
         df["UtcHourID"] = df.LocalHourID + df.utc_shift
-        df["UtcHourID"] = df.UtcHourID.map(lambda x: x if x < 8761 else x - 8760)
+        df["UtcHourID"] = df.UtcHourID.map(lambda x: x if x < const.HOURS_PER_YEAR + 1 else x - const.HOURS_PER_YEAR)
         df = df.drop(columns=["utc_shift"])
         return df
 
@@ -497,7 +497,7 @@ def _format_data(self, data: dict[str, pd.DataFrame]) -> pd.DataFrame:
             aggfunc="sum",
         )
         df = df.rename(columns=CODE_2_STATE)
-        assert len(df.index.get_level_values("snapshot").unique()) == 8760
+        assert len(df.index.get_level_values("snapshot").unique()) == const.HOURS_PER_YEAR
         assert not df.empty
         return df
 
@@ -680,7 +680,7 @@ def _apply_profiles(
         """
         Applies profile data to annual demand data.
 
-        This is quite a heavy operation, as it can be up to 8760hrs,
+        This is quite a heavy operation, as it can be up to a full year of hourly records,
         3000+ counties, 50+ subsectors and 4 fuels.
         """
 
@@ -1416,7 +1416,7 @@ def _format_data(self, data: pd.DataFrame) -> pd.DataFrame:
 
             df = df.div(100)  # convert from percentage to decimal
             df = df.mul(demand_national.loc[(self.vehicle, year), "value"])
-            df = df.mul(1 / 8760)  # uniform load over the year
+            df = df.mul(1 / const.HOURS_PER_YEAR)  # uniform load over the year
 
             dfs.append(df)
 

workflow/scripts/constants.py

@@ -17,6 +17,9 @@
 # convert euros to USD
 EUR_2_USD = 1.07  # taken on 12-12-2023
 
+# number of hours in a non-leap year
+HOURS_PER_YEAR = 8760
+
 # energy content of natural gas
 # Assumes national averages for the conversion
 # https://www.eia.gov/naturalgas/monthly/pdf/table_25.pdf

workflow/scripts/eulp.py

@@ -9,6 +9,7 @@
 from typing import ClassVar
 
 import pandas as pd
+from constants import HOURS_PER_YEAR
 from matplotlib.axes import Axes
 
 
@@ -222,7 +223,7 @@ def _resample_data(df: pd.DataFrame) -> pd.DataFrame:
             df.index = pd.to_datetime(df.index)
         df.index = df.index.map(lambda x: x.replace(year=2018))
         resampled = df.resample("1h").sum()
-        assert len(resampled == 8760), "Length of resampled != 8760 :("
+        assert len(resampled) == HOURS_PER_YEAR, f"Length of resampled != {HOURS_PER_YEAR} :("
         return resampled.sort_index()
 
     def _aggregate_data(self, df: pd.DataFrame) -> pd.DataFrame:
@@ -344,7 +345,7 @@ def _resample_data(df: pd.DataFrame) -> pd.DataFrame:
             df.index = pd.to_datetime(df.index)
         df.index = df.index.map(lambda x: x.replace(year=2018))
         resampled = df.resample("1h").sum()
-        assert len(resampled == 8760), "Length of resampled != 8760 :("
+        assert len(resampled) == HOURS_PER_YEAR, f"Length of resampled != {HOURS_PER_YEAR} :("
         return resampled.sort_index()
 
     def _aggregate_data(self, df: pd.DataFrame) -> pd.DataFrame:

workflow/scripts/prepare_network.py

@@ -22,6 +22,7 @@
     set_scenario_config,
     update_config_from_wildcards,
 )
+from constants import HOURS_PER_YEAR
 
 idx = pd.IndexSlice
 
@@ -287,7 +288,7 @@ def set_line_nom_max(
     transport_model = is_transport_model(params.transmission_network)
 
     n = pypsa.Network(snakemake.input[0])
-    num_years = n.snapshot_weightings.loc[n.investment_periods[0]].objective.sum() / 8760.0
+    num_years = n.snapshot_weightings.loc[n.investment_periods[0]].objective.sum() / HOURS_PER_YEAR
     costs = pd.read_csv(snakemake.input.tech_costs)
     costs = costs.pivot(index="pypsa-name", columns="parameter", values="value")
     # Set Investment Period Year Weightings

workflow/scripts/solve_network.py

@@ -34,6 +34,7 @@
     configure_logging,
     update_config_from_wildcards,
 )
+from constants import HOURS_PER_YEAR
 from opts.bidirectional_link import add_bidirectional_link_constraints
 from opts.interchange import add_interchange_constraints
 from opts.land import add_land_use_constraints
@@ -124,7 +125,7 @@ def prepare_network(n, solve_opts=None):
             names=n.snapshots.names,
         )
         n.set_snapshots(first_nhours)
-        n.snapshot_weightings[:] = 8760.0 / nhours
+        n.snapshot_weightings[:] = HOURS_PER_YEAR / nhours
 
     return n
 

workflow/scripts/test/test_constants.py

@@ -0,0 +1,26 @@
+"""Tests for shared constant usage."""
+
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[3]
+
+
+def test_hours_per_year_uses_shared_constant():
+    """Avoid reintroducing bare 8760 literals in workflow source."""
+    source_files = [
+        *REPO_ROOT.glob("workflow/scripts/**/*.py"),
+        *REPO_ROOT.glob("workflow/rules/**/*.smk"),
+    ]
+
+    offenders = []
+    for path in source_files:
+        if path.name == "constants.py" or "/test/" in path.as_posix():
+            continue
+
+        for line_number, line in enumerate(path.read_text().splitlines(), start=1):
+            if "8760" in line:
+                offenders.append(f"{path.relative_to(REPO_ROOT)}:{line_number}: {line.strip()}")
+
+    assert not offenders, "Use the shared HOURS_PER_YEAR constant instead of bare 8760 literals:\n" + "\n".join(
+        offenders,
+    )