Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def rerun_pyomo_model(model_data, backend_model):
"""
Rerun the Pyomo backend, perhaps after updating a parameter value,
(de)activating a constraint/objective or updating run options in the model
model_data object (e.g. `run.solver`).
Returns
-------
new_model : calliope.Model
New calliope model, including both inputs and results, but no backend interface.
"""
backend_model.__calliope_run_config = AttrDict.from_yaml_string(model_data.attrs['run_config'])
if backend_model.__calliope_run_config['mode'] != 'plan':
raise exceptions.ModelError(
'Cannot rerun the backend in {} run mode. Only `plan` mode is '
'possible.'.format(backend_model.__calliope_run_config['mode'])
)
timings = {}
log_time(logger, timings, 'model_creation')
results, backend_model = backend_run.run_plan(
model_data, timings, run_pyomo,
build_only=False, backend_rerun=backend_model
)
inputs = access_pyomo_model_inputs(backend_model)
if v.attrs['is_result'] == 0 or v.attrs.get('operate_param', 0) == 1
},
'dims': {
k: v.dims
for k, v in model_data.data_vars.items()
if v.attrs['is_result'] == 0 or v.attrs.get('operate_param', 0) == 1
},
'sets': list(model_data.coords),
'attrs': {k: v for k, v in model_data.attrs.items() if k is not 'defaults'}
}
# Dims in the dict's keys are ordered as in model_data, which is enforced
# in model_data generation such that timesteps are always last and the
# remainder of dims are in alphabetic order
backend_model.__calliope_model_data = model_data_dict
backend_model.__calliope_defaults = AttrDict.from_yaml_string(model_data.attrs['defaults'])
backend_model.__calliope_run_config = AttrDict.from_yaml_string(model_data.attrs['run_config'])
for k, v in model_data_dict['data'].items():
if k in backend_model.__calliope_defaults.keys():
setattr(
backend_model, k,
po.Param(*[getattr(backend_model, i)
for i in model_data_dict['dims'][k]],
initialize=v, mutable=True,
default=backend_model.__calliope_defaults[k])
)
# In operate mode, e.g. energy_cap is a parameter, not a decision variable,
# so add those in.
elif (backend_model.__calliope_run_config['mode'] == 'operate' and
model_data[k].attrs.get('operate_param') == 1):
setattr(
backend_model, k,
def combine_overrides(config_model, overrides):
override_dict = AttrDict()
for override in overrides:
try:
yaml_string = config_model.overrides[override].to_yaml()
override_with_imports = AttrDict.from_yaml_string(yaml_string)
except KeyError:
raise exceptions.ModelError(
'Override `{}` is not defined.'.format(override)
)
try:
override_dict.union(override_with_imports, allow_override=False)
except KeyError as e:
raise exceptions.ModelError(
str(e)[1:-1] + '. Already specified but defined again in '
'override `{}`.'.format(override)
)
return override_dict
def _init_from_model_data(self, model_data):
if '_model_run' in model_data.attrs:
self._model_run = AttrDict.from_yaml_string(
model_data.attrs['_model_run'])
del model_data.attrs['_model_run']
if '_debug_data' in model_data.attrs:
self._debug_data = AttrDict.from_yaml_string(
model_data.attrs['_debug_data'])
del model_data.attrs['_debug_data']
self._model_data = model_data
self.inputs = self._model_data.filter_by_attrs(is_result=0)
self.model_config = UpdateObserverDict(
initial_yaml_string=model_data.attrs.get('model_config', '{}'),
name='model_config', observer=self._model_data
)
self.run_config = UpdateObserverDict(
initial_yaml_string=model_data.attrs.get('run_config', '{}'),
build_only : bool, optional
If True, the backend only constructs its in-memory representation
of the problem rather than solving it. Used for debugging and
testing.
"""
BACKEND = {
'pyomo': run_pyomo
}
INTERFACE = {
'pyomo': pyomo_interface
}
run_config = AttrDict.from_yaml_string(model_data.attrs['run_config'])
if run_config['mode'] == 'plan':
results, backend = run_plan(
model_data, timings,
backend=BACKEND[run_config.backend], build_only=build_only
)
elif run_config['mode'] == 'operate':
results, backend = run_operate(
model_data, timings,
backend=BACKEND[run_config.backend], build_only=build_only
)
return results, backend, INTERFACE[run_config.backend].BackendInterfaceMethods
def run_operate(model_data, timings, backend, build_only):
"""
For use when mode is 'operate', to allow the model to be built, edited, and
iteratively run within Pyomo.
"""
log_time(logger, timings, 'run_start',
comment='Backend: starting model run in operational mode')
defaults = AttrDict.from_yaml_string(model_data.attrs['defaults'])
run_config = AttrDict.from_yaml_string(model_data.attrs['run_config'])
operate_params = ['purchased'] + [
i.replace('_max', '') for i in defaults if i[-4:] == '_max'
]
# Capacity results (from plan mode) can be used as the input to operate mode
if (any(model_data.filter_by_attrs(is_result=1).data_vars) and
run_config.get('operation.use_cap_results', False)):
# Anything with is_result = 1 will be ignored in the Pyomo model
for varname, varvals in model_data.data_vars.items():
if varname in operate_params:
varvals.attrs['is_result'] = 1
varvals.attrs['operate_param'] = 1
else:
def run_operate(model_data, timings, backend, build_only):
"""
For use when mode is 'operate', to allow the model to be built, edited, and
iteratively run within Pyomo.
"""
log_time(logger, timings, 'run_start',
comment='Backend: starting model run in operational mode')
defaults = AttrDict.from_yaml_string(model_data.attrs['defaults'])
run_config = AttrDict.from_yaml_string(model_data.attrs['run_config'])
operate_params = ['purchased'] + [
i.replace('_max', '') for i in defaults if i[-4:] == '_max'
]
# Capacity results (from plan mode) can be used as the input to operate mode
if (any(model_data.filter_by_attrs(is_result=1).data_vars) and
run_config.get('operation.use_cap_results', False)):
# Anything with is_result = 1 will be ignored in the Pyomo model
for varname, varvals in model_data.data_vars.items():
if varname in operate_params:
varvals.attrs['is_result'] = 1
varvals.attrs['operate_param'] = 1
else:
cap_max = xr.merge([
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,
# so that can override scenario definitions by override_dict
if override_dict:
if isinstance(override_dict, str):
override_dict = AttrDict.from_yaml_string(override_dict)
elif not isinstance(override_dict, AttrDict):
override_dict = AttrDict(override_dict)
warning_messages = checks.check_overrides(config_model, override_dict)
exceptions.print_warnings_and_raise_errors(warnings=warning_messages)
# FutureWarning: If config does not include an explicit objective cost class, check override dict.
# Added in 0.6.4-dev, to be removed in v0.7.0-dev.
if has_explicit_cost_class is False:
has_explicit_cost_class = isinstance(override_dict.get_key('run.objective_options.cost_class', None), dict)
config_model.union(
override_dict, allow_override=True, allow_replacement=True
)
if scenario: