Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def generate_model_run(config_run, config_model, debug_comments=None):
"""
Returns a processed model_run configuration AttrDict and a debug
YAML object with comments attached, ready to write to disk.
Parameters
----------
config_run : AttrDict
config_model : AttrDict
"""
model_run = utils.AttrDict()
if debug_comments is None:
debug_comments = utils.AttrDict()
# README CHANGED: if run_config overrides data_path, it is no longer
# interpreted as relative to the run_config file's path
# 1) Apply any initiall overrides to config_model
# 1.a) Via 'model_override', which is the path to a YAML file
if 'model_override' in config_run:
override_path = utils.relative_path(
config_run.config_run_path, config_run.model_override
)
override_dict = utils.AttrDict.from_yaml(override_path)
config_model.union(
override_dict, allow_override=True, allow_replacement=True
)
for k, v in override_dict.as_dict_flat():
debug_comments.set_key(
all_sets = sets.generate_simple_sets(model_run)
all_sets.union(sets.generate_loc_tech_sets(model_run, all_sets))
all_sets = utils.AttrDict({k: list(v) for k, v in all_sets.items()})
model_run['sets'] = all_sets
# 8) Grab additional relevant bits from run and model config
model_run['run'] = config_run
model_run['model'] = config_model['model']
# 9) Final sense-checking
final_check_comments, warnings, errors = checks.check_final(model_run)
debug_comments.union(final_check_comments)
checks.print_warnings_and_raise_errors(warnings=warnings, errors=errors)
# 10) Build a debug data dict with comments and the original configs
debug_data = utils.AttrDict({
'comments': debug_comments,
'config_model': config_model,
'config_run': config_run
})
return model_run, debug_data
def generate_model_run(config_run, config_model, debug_comments=None):
"""
Returns a processed model_run configuration AttrDict and a debug
YAML object with comments attached, ready to write to disk.
Parameters
----------
config_run : AttrDict
config_model : AttrDict
"""
model_run = utils.AttrDict()
if debug_comments is None:
debug_comments = utils.AttrDict()
# README CHANGED: if run_config overrides data_path, it is no longer
# interpreted as relative to the run_config file's path
# 1) Apply any initiall overrides to config_model
# 1.a) Via 'model_override', which is the path to a YAML file
if 'model_override' in config_run:
override_path = utils.relative_path(
config_run.config_run_path, config_run.model_override
)
override_dict = utils.AttrDict.from_yaml(override_path)
config_model.union(
override_dict, allow_override=True, allow_replacement=True
)
def process_techs(config_model):
default_palette_cycler = itertools.cycle(range(len(_DEFAULT_PALETTE)))
result = utils.AttrDict()
errors = []
debug_comments = utils.AttrDict()
for tech_id, tech_config in config_model.techs.items():
tech_result = utils.AttrDict()
# Add inheritance chain
tech_result.inheritance = get_parents(tech_id, config_model)
# CHECK: A tech's parent must lead to one of the built-in tech_groups
builtin_tech_groups = checks.defaults_model.tech_groups.keys()
if tech_result.inheritance[-1] not in builtin_tech_groups:
errors.append(
'tech {} must inherit from a built-in tech group'.format(tech_id)
)
# Process inheritance
tech_result.essentials = utils.AttrDict()
for parent in reversed(tech_result.inheritance):
# Does the parent group have model-wide settings?
parent_essentials = config_model.tech_groups[parent].essentials
def __init__(self, config_run=None, override=None):
super().__init__()
self.verbose = False
self.debug = utils.AttrDict()
# Populate self.config_run and self.config_model
self.initialize_configuration(config_run, override)
self._get_option = utils.option_getter(self.config_model)
self.get_cost = utils.cost_getter(self._get_option)
# Set random seed if specified in run configuration
random_seed = self.config_run.get('random_seed', None)
if random_seed:
np.random.seed(seed=random_seed)
# Populate config_model with link distances, where metadata is given
# but no distances given in locations.yaml
self.get_distances()
# Initialize sets
# Build up an AttrDict with the specified overrides
override_c = utils.AttrDict()
for k, v in iter_row.to_dict().items():
# NaN values can show in this row if some but not all iterations
# specify a value, so we simply skip them
if not isinstance(v, list) and pd.isnull(v):
# NB the isinstance and pd.isnull checks should cover all cases
# i.e. both not a list (which is definitely not null) or a
# single value that could be null. But this could blow up in
# unexpected edge cases...
continue
# Convert numpy dtypes to python ones, else YAML chokes
if isinstance(v, np.generic):
v = np.asscalar(v)
if isinstance(v, dict):
override_c.set_key(k, utils.AttrDict(v))
else:
override_c.set_key(k, copy.copy(v))
# Finally, add the override AttrDict to the existing configuration
iter_c.union(override_c, allow_override=True, allow_replacement=True)
# Set output dir in configuration object, this is hardcoded
iter_c.set_key('output.path', os.path.join('Output', index_str))
return iter_c
def __init__(self, target_dir, config_run=None):
super(Parallelizer, self).__init__()
if not config_run:
config_run = os.path.join(os.path.dirname(__file__),
'example_models', 'national_scale', 'run.yaml')
self.config_file = config_run
self.config = utils.AttrDict.from_yaml(config_run)
self.target_dir = target_dir
self.f_submit = 'submit_{}.sh'
self.f_run = 'run.sh'
model_run['tech_groups'] = process_tech_groups(config_model, model_run['techs'])
# 5) Fully populate locations
model_run['locations'], debug_locs = locations.process_locations(
config_model, model_run['techs']
)
debug_comments.set_key('model_run.locations', debug_locs)
# 6) Fully populate timeseries data
# Raises ModelErrors if there are problems with timeseries data at this stage
model_run['timeseries_data'] = process_timeseries_data(config_model, model_run)
# 7) Initialize sets
all_sets = sets.generate_simple_sets(model_run)
all_sets.union(sets.generate_loc_tech_sets(model_run, all_sets))
all_sets = utils.AttrDict({k: list(v) for k, v in all_sets.items()})
model_run['sets'] = all_sets
# 8) Grab additional relevant bits from run and model config
model_run['run'] = config_run
model_run['model'] = config_model['model']
# 9) Final sense-checking
final_check_comments, warnings, errors = checks.check_final(model_run)
debug_comments.union(final_check_comments)
checks.print_warnings_and_raise_errors(warnings=warnings, errors=errors)
# 10) Build a debug data dict with comments and the original configs
debug_data = utils.AttrDict({
'comments': debug_comments,
'config_model': config_model,
'config_run': config_run
self.config_run = cr
if override:
assert isinstance(override, utils.AttrDict)
cr.union(override, allow_override=True, allow_replacement=True)
# If manually specify a run_id in debug, overwrite the generated one
if 'debug.run_id' in cr.keys_nested():
self.run_id = cr.debug.run_id
self.config_model = get_model_config(cr, self.config_run_path)
# Override config_model settings if specified in config_run
# 1) Via 'model_override', which is the path to a YAML file
if 'model_override' in cr:
override_path = utils.relative_path(self.config_run_path, cr.model_override)
override_dict = utils.AttrDict.from_yaml(override_path)
self.override_model_config(override_dict)
# 2) Via 'override', which is an AttrDict
if ('override' in cr and isinstance(cr.override, utils.AttrDict)):
self.override_model_config(cr.override)
# Initialize locations
locs = self.config_model.locations
self.config_model.locations = locations.process_locations(locs)
# As a final step, flush the option cache
self.flush_option_cache()
for tech_id, tech_config in config_model.techs.items():
tech_result = utils.AttrDict()
# Add inheritance chain
tech_result.inheritance = get_parents(tech_id, config_model)
# CHECK: A tech's parent must lead to one of the built-in tech_groups
builtin_tech_groups = checks.defaults_model.tech_groups.keys()
if tech_result.inheritance[-1] not in builtin_tech_groups:
errors.append(
'tech {} must inherit from a built-in tech group'.format(tech_id)
)
# Process inheritance
tech_result.essentials = utils.AttrDict()
for parent in reversed(tech_result.inheritance):
# Does the parent group have model-wide settings?
parent_essentials = config_model.tech_groups[parent].essentials
for k in parent_essentials.as_dict_flat():
debug_comments.set_key(
'{}.essentials.{}'.format(tech_id, k),
'From parent tech_group `{}`'.format(parent)
)
tech_result.essentials.union(parent_essentials, allow_override=True)
# Add this tech's essentials, overwriting any essentials from parents
tech_result.essentials.union(tech_config.essentials, allow_override=True)
# Add allowed_constraints and required_constraints from base tech
keys_to_add = ['required_constraints', 'allowed_constraints', 'allowed_costs']
for k in keys_to_add: