Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# For each file in new_model_config, save it to its same
# position from the old path in the `out_path`
for f in new_model_config:
out_dir, out_filename = os.path.split(
f.replace(os.path.commonpath([model_config_path, f]), '.')
)
if f == model_config_path:
out_dir_model_config_path = out_dir
out_filename = os.path.basename(model_config_path)
out_file = os.path.join(out_path, out_dir, out_filename)
os.makedirs(os.path.join(out_path, out_dir), exist_ok=True)
new_model_config[f].to_yaml(out_file)
# Read each CSV file in the model data dir and apply index
full_new_config = AttrDict.from_yaml(
os.path.join(out_path, out_dir_model_config_path, os.path.basename(model_config_path))
)
ts_dir = full_new_config.get_key('model.timeseries_data_path')
ts_path_in = os.path.join(
os.path.dirname(model_config_path), ts_dir
)
ts_path_out = os.path.join(
os.path.join(out_path, ts_dir)
)
os.makedirs(ts_path_out, exist_ok=True)
index_t = pd.read_csv(os.path.join(ts_path_in, 'set_t.csv'), index_col=0, header=None)[1]
for f in glob.glob(os.path.join(ts_path_in, '*.csv')):
if 'set_t.csv' not in f:
df = pd.read_csv(f, index_col=0)
import logging
import numpy as np
import pandas as pd
from inspect import signature
import calliope
from calliope._version import __version__
from calliope.core.attrdict import AttrDict
from calliope.core.preprocess.util import get_all_carriers
from calliope.core.util.tools import load_function
logger = logging.getLogger(__name__)
DEFAULTS = AttrDict.from_yaml(os.path.join(os.path.dirname(calliope.__file__), 'config', 'defaults.yaml'))
POSSIBLE_COSTS = [i for i in DEFAULTS.techs.default_tech.costs.default_cost.keys()]
def check_overrides(config_model, override):
"""
Perform checks on the override dict and override file inputs to ensure they
are not doing something silly.
"""
model_warnings = []
info = []
for key in override.as_dict_flat().keys():
if key in config_model.as_dict_flat().keys():
info.append(
'Override applied to {}: {} -> {}'
.format(key, config_model.get_key(key), override.get_key(key))
)
def apply_overrides(config, scenario=None, override_dict=None):
"""
Generate processed Model configuration, applying any scenarios overrides.
Parameters
----------
config : AttrDict
a model configuration AttrDict
scenario : str, optional
override_dict : str or dict or AttrDict, optional
If a YAML string, converted to AttrDict
"""
debug_comments = AttrDict()
config_model = AttrDict.from_yaml(os.path.join(
os.path.dirname(calliope.__file__), 'config', 'defaults.yaml'
))
# Interpret timeseries_data_path as relative
config.model.timeseries_data_path = relative_path(
config.config_path, config.model.timeseries_data_path
)
# FutureWarning: check if config includes an explicit objective cost class.
# Added in 0.6.4-dev, to be removed in v0.7.0-dev.
has_explicit_cost_class = isinstance(config.get_key('run.objective_options.cost_class', None), dict)
# The input files are allowed to override other model defaults
config_model.union(config, allow_override=True)
# First pass of applying override dict before applying scenarios,
def load_with_import_resolution(in_path):
files = {}
top = AttrDict.from_yaml(in_path, resolve_imports=False)
files[in_path] = top
base_path = os.path.dirname(in_path)
for import_path in top.get('import', []):
result = load_with_import_resolution(os.path.join(base_path, import_path))
for k, v in result.items():
files[k] = v
return files
Convert Calliope model configurations from 0.5.x to 0.6.0.
"""
from calliope.core.util.logging import logger
import os
import glob
import pandas as pd
from calliope.core.attrdict import AttrDict, __Missing
_MISSING = __Missing()
_CONVERSIONS = AttrDict.from_yaml(
os.path.join(os.path.dirname(__file__), '..', '..', 'config', 'conversion_0.6.0.yaml')
)
_TECH_GROUPS = [
'supply', 'supply_plus',
'conversion', 'conversion_plus',
'demand', 'transmission', 'storage'
]
def load_with_import_resolution(in_path):
files = {}
top = AttrDict.from_yaml(in_path, resolve_imports=False)
files[in_path] = top
base_path = os.path.dirname(in_path)
for import_path in top.get('import', []):
"""
Generate processed ModelRun configuration from a
YAML model configuration file.
Parameters
----------
model_file : str
Path to YAML file with model configuration.
scenario : str, optional
Name of scenario to apply. Can either be a named scenario, or a
comman-separated list of individual overrides to be combined
ad-hoc, e.g. 'my_scenario_name' or 'override1,override2'.
override_dict : dict or AttrDict, optional
"""
config = AttrDict.from_yaml(model_file)
config.config_path = model_file
config_with_overrides, debug_comments, overrides, scenario = apply_overrides(
config, scenario=scenario, override_dict=override_dict
)
return generate_model_run(
config_with_overrides, debug_comments, overrides, scenario
)
"""
state = {'ensure_feasibility': False}
converted_run_config = AttrDict()
run_config = load_with_import_resolution(run_config_path)
for k, v in run_config.items():
# We consider any files imported in run configuration, but
# disregard file names and simply merge everything together
# into the new model configuration
converted_run_config.update(convert_run_dict(v, _CONVERSIONS))
new_model_config = AttrDict()
model_config = load_with_import_resolution(model_config_path)
# Get all techs from old model that need to be tech_groups in the new one
merged_model_config = AttrDict.from_yaml(model_config_path)
run_config_overrides = AttrDict.from_yaml(run_config_path).get_key('override', None)
if run_config_overrides:
merged_model_config.union(run_config_overrides, allow_override=True)
tech_groups = set()
for tech, tech_dict in merged_model_config.techs.items():
parent = tech_dict.get('parent', None)
if parent and parent not in _TECH_GROUPS:
tech_groups.add(parent)
for k, v in model_config.items():
new_model_config[k] = convert_model_dict(
v, _CONVERSIONS, tech_groups=tech_groups, state=state
)
# Merge run_config into main model config file
new_model_config[model_config_path].union(converted_run_config)