bjoern/pypsa-eur
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

snakefile: move copy_default_files and process_run_config to helpers

Browse Source
This commit is contained in:
Fabian 2024-03-19 08:20:23 +01:00
parent f6f18e1868
commit 2565a7db4f
2 changed files with 35 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
Snakefile
View File

@ -2,26 +2,18 @@
# #
# SPDX-License-Identifier: MIT # SPDX-License-Identifier: MIT
from os.path import normpath, exists
from shutil import copyfile, move, rmtree
from pathlib import Path from pathlib import Path
import yaml import yaml
from os.path import normpath
from shutil import move, rmtree
from snakemake.utils import min_version from snakemake.utils import min_version
min_version("8.5") min_version("8.5")
from scripts._helpers import path_provider from scripts._helpers import path_provider, copy_default_files, process_run_config
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" configfile: "config/config.default.yaml"
@ -29,17 +21,7 @@ configfile: "config/config.yaml"
run = config["run"] run = config["run"]
scenarios = run.get("scenarios", {}) RDIR = process_run_config(run)
if run["name"] and scenarios.get("enable"):
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"]) logs = path_provider("logs/", RDIR, run["shared_resources"])
benchmarks = path_provider("benchmarks/", RDIR, run["shared_resources"]) benchmarks = path_provider("benchmarks/", RDIR, run["shared_resources"])

29
scripts/_helpers.py
View File

@ -11,7 +11,9 @@ import os
import re import re
import urllib import urllib
from functools import partial from functools import partial
from os.path import exists
from pathlib import Path from pathlib import Path
from shutil import copyfile
import pandas as pd import pandas as pd
import pytz import pytz
@ -25,6 +27,33 @@ logger = logging.getLogger(__name__)
REGION_COLS = ["geometry", "name", "x", "y", "country"] 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 process_run_config(run):
scenarios = run.get("scenarios", {})
if run["name"] and scenarios.get("enable"):
fn = Path(scenarios["file"])
scenarios = yaml.safe_load(fn.read_text())
RDIR = "{run}/"
if run["name"] == "all":
run["name"] = list(scenarios.keys())
elif run["name"]:
RDIR = run["name"] + "/"
else:
RDIR = ""
return RDIR
def get_run_path(fn, dir, rdir, shared_resources): def get_run_path(fn, dir, rdir, shared_resources):
""" """
Dynamically provide paths based on shared resources and filename. Dynamically provide paths based on shared resources and filename.