Merge pull request #978 from PyPSA/refactor-copy-config-get-rdir

snakefile: move copy_default_files and process_run_config to helpers

Snakefile

@@ -2,26 +2,18 @@
 #
 # SPDX-License-Identifier: MIT
 
-from os.path import normpath, exists
-from shutil import copyfile, move, rmtree
 from pathlib import Path
 import yaml
-
+from os.path import normpath, exists
+from shutil import copyfile, move, rmtree
 from snakemake.utils import min_version
 
 min_version("8.5")
 
-from scripts._helpers import path_provider
+from scripts._helpers import path_provider, copy_default_files, get_scenarios, get_rdir
 
-default_files = {
+
-    "config/config.default.yaml": "config/config.yaml",
+copy_default_files(workflow)
-    "config/scenarios.template.yaml": "config/scenarios.yaml",
-}
-for template, target in default_files.items():
-    target = os.path.join(workflow.current_basedir, target)
-    template = os.path.join(workflow.current_basedir, template)
-    if not exists(target) and exists(template):
-        copyfile(template, target)
 
 
 configfile: "config/config.default.yaml"
@@ -29,17 +21,8 @@ configfile: "config/config.yaml"
 
 
 run = config["run"]
-scenarios = run.get("scenarios", {})
+scenarios = get_scenarios(run)
-if run["name"] and scenarios.get("enable"):
+RDIR = get_rdir(run)
-    fn = Path(scenarios["file"])
-    scenarios = yaml.safe_load(fn.read_text())
-    RDIR = "{run}/"
-    if run["name"] == "all":
-        config["run"]["name"] = list(scenarios.keys())
-elif run["name"]:
-    RDIR = run["name"] + "/"
-else:
-    RDIR = ""
 
 logs = path_provider("logs/", RDIR, run["shared_resources"])
 benchmarks = path_provider("benchmarks/", RDIR, run["shared_resources"])

rules/solve_perfect.smk

@@ -110,7 +110,8 @@ rule solve_sector_network_perfect:
     output:
         network=RESULTS
         + "postnetworks/elec_s{simpl}_{clusters}_l{ll}_{opts}_{sector_opts}_brownfield_all_years.nc",
-        config="configs/config.elec_s{simpl}_{clusters}_l{ll}_{opts}_{sector_opts}_brownfield_all_years.yaml",
+        config=RESULTS
+        + "configs/config.elec_s{simpl}_{clusters}_l{ll}_{opts}_{sector_opts}_brownfield_all_years.yaml",
     threads: solver_threads
     resources:
         mem_mb=config_provider("solving", "mem"),

scripts/_helpers.py

@@ -11,7 +11,9 @@ import os
 import re
 import urllib
 from functools import partial
+from os.path import exists
 from pathlib import Path
+from shutil import copyfile
 
 import pandas as pd
 import pytz
@@ -25,6 +27,40 @@ logger = logging.getLogger(__name__)
 REGION_COLS = ["geometry", "name", "x", "y", "country"]
 
 
+def copy_default_files(workflow):
+    default_files = {
+        "config/config.default.yaml": "config/config.yaml",
+        "config/scenarios.template.yaml": "config/scenarios.yaml",
+    }
+    for template, target in default_files.items():
+        target = os.path.join(workflow.current_basedir, target)
+        template = os.path.join(workflow.current_basedir, template)
+        if not exists(target) and exists(template):
+            copyfile(template, target)
+
+
+def get_scenarios(run):
+    scenario_config = run.get("scenarios", {})
+    if run["name"] and scenario_config.get("enable"):
+        fn = Path(scenario_config["file"])
+        scenarios = yaml.safe_load(fn.read_text())
+        if run["name"] == "all":
+            run["name"] = list(scenarios.keys())
+        return scenarios
+    return {}
+
+
+def get_rdir(run):
+    scenario_config = run.get("scenarios", {})
+    if run["name"] and scenario_config.get("enable"):
+        RDIR = "{run}/"
+    elif run["name"]:
+        RDIR = run["name"] + "/"
+    else:
+        RDIR = ""
+    return RDIR
+
+
 def get_run_path(fn, dir, rdir, shared_resources):
     """
     Dynamically provide paths based on shared resources and filename.