Source code for calliope.backend.pyomo.interface

import logging

import xarray as xr
import pyomo.core as po  # pylint: disable=import-error

import calliope
from calliope.backend.pyomo.util import get_var, string_to_datetime
from calliope.backend import run as backend_run
from calliope.backend.pyomo import model as run_pyomo
import calliope.backend.pyomo.interface as pyomo_interface

from calliope.core.util.dataset import reorganise_xarray_dimensions
from calliope.core.util.logging import log_time
from calliope import exceptions
from calliope.core.attrdict import AttrDict
from calliope.postprocess.results import postprocess_model_results

logger = logging.getLogger(__name__)

def access_pyomo_model_inputs(backend_model):
    If the user wishes to inspect the parameter values used as inputs in the backend
    model, they can access a new Dataset of all the backend model inputs, including
    defaults applied where the user did not specify anything for a loc::tech

    all_params = { get_var(backend_model,, sparse=True)
        for i in backend_model.component_objects()
        if isinstance(i, po.base.param.IndexedParam)
    all_param_ds = reorganise_xarray_dimensions(xr.Dataset(all_params))
    all_param_ds = string_to_datetime(backend_model, all_param_ds)
    for var in all_param_ds.data_vars:
        all_param_ds[var].attrs["is_result"] = 0

    return all_param_ds

def update_pyomo_param(backend_model, opt, param, update_dict):
    A Pyomo Param value can be updated without the user directly accessing the
    backend model.

    param : str
        Name of the parameter to update
    update_dict : dict
        keys are parameter indeces (either strings or tuples of strings,
        depending on whether there is one or more than one dimension). Values
        are the new values being assigned to the parameter at the given indeces.

    Value(s) will be updated in-place, requiring the user to run the model again to
    see the effect on results.

    if not hasattr(backend_model, param):
        raise exceptions.ModelError(
            "Parameter `{}` not in the Pyomo Backend. Check that the string "
            "matches the corresponding constraint/cost in the model.inputs "
            "xarray Dataset".format(param)
    elif not isinstance(getattr(backend_model, param), po.base.param.IndexedParam):
        raise exceptions.ModelError(
            "`{}` not a Parameter in the Pyomo Backend. Sets and decision variables "
            "cannot be updated by the user".format(param)
    elif not isinstance(update_dict, dict):
        raise TypeError("`update_dict` must be a dictionary")
        getattr(backend_model, param).store_values(update_dict)

    if opt is not None and "persistent" in
            "Updating the Pyomo parameter won't affect the optimisation run without also "
            "regenerating the relevant constraints or the objective function (see `regenerate_persistent_solver`)."

def regenerate_persistent_pyomo_solver(backend_model, opt, constraints=None, obj=False):
    Having updated a Pyomo Param or several of them, this function can be used
    to regenerate associated constraints in a persistent solver interface, such
    as "gurobi_persistent", before rerunning the model.
    The entire constraint need not be regenerated, it is possible to only point to
    those indexes whose associated parameters have changed.

    constraints : dict of lists or None, default = None
        Names of constraints as keys and list of constraint index items as values,
        e.g. `{"energy_capacity_constraint": ["X1::pv"]}` or `{"cost_constraint": [("monetary", "X1::pv"), ("monetary", "X2::pv")])}`.
        Order of index values can be inferred by inspecting the constraint.
    obj : bool, default = False
        If True, will also regenerate the objective function.
    if opt is None or "persistent" not in
        raise exceptions.ModelError(
            "Can only regenerate persistent solvers. No persistent solver object found for this model run."

    if obj is True:
    if constraints is not None:
        for constraint_name, constraint_idx in constraints.items():
            for idx in constraint_idx:
                opt.remove_constraint(getattr(backend_model, constraint_name)[idx])
                opt.add_constraint(getattr(backend_model, constraint_name)[idx])

    return opt

def activate_pyomo_constraint(backend_model, constraint, active=True):
    Takes a constraint or objective name, finds it in the backend model and sets
    its status to either active or deactive.

    constraint : str
        Name of the constraint/objective to activate/deactivate
        Built-in constraints include '_constraint'
    active: bool, default=True
        status to set the constraint/objective
    if not hasattr(backend_model, constraint):
        raise exceptions.ModelError(
            "constraint/objective `{}` not in the Pyomo Backend.".format(constraint)
    elif not isinstance(getattr(backend_model, constraint), po.base.Constraint):
        raise exceptions.ModelError(
            "`{}` not a constraint in the Pyomo Backend.".format(constraint)
    elif active is True:
        getattr(backend_model, constraint).activate()
    elif active is False:
        getattr(backend_model, constraint).deactivate()
        raise ValueError("Argument `active` must be True or False")

def rerun_pyomo_model(model_data, backend_model, opt):
    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`).

    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(
    timings = {}
    log_time(logger, timings, "model_creation")
    inputs = access_pyomo_model_inputs(backend_model)

    run_mode = backend_model.__calliope_run_config["mode"]
    if run_mode == "plan":
        kwargs = {}
    elif run_mode == "spores":
        kwargs = {"interface": pyomo_interface}
        raise exceptions.ModelError(
            "Cannot rerun the backend in {} run mode. Only `plan` or `spores` modes are "
    run_func = getattr(backend_run, f"run_{run_mode}")
    results, backend_model, opt = run_func(

    # Add additional post-processed result variables to results
    if results.attrs.get("termination_condition", None) in ["optimal", "feasible"]:
        results = postprocess_model_results(results, model_data, timings)

    for var in results.data_vars.values():
        var.attrs["is_result"] = 1

    for var in inputs.data_vars.values():
        var.attrs["is_result"] = 0

    new_model_data = xr.merge((results, inputs), compat="override")

    # Only add coordinates from the original model_data that don't already exist
    new_coords = [
        i for i in model_data.coords.keys() if i not in new_model_data.coords.keys()
    new_model_data = new_model_data.update(model_data[new_coords])

    # Reorganise the coordinates so that model data and new model data share
    # the same order of items in each dimension
    new_model_data = new_model_data.reindex(model_data.coords)

        "The results of rerunning the backend model are only available within "
        "the Calliope model returned by this function call."

    new_calliope_model = calliope.Model(config=None, model_data=new_model_data)
    new_calliope_model._timings = timings

    return opt, new_calliope_model

def add_pyomo_constraint(
    backend_model, constraint_name, constraint_sets, constraint_rule
    Functionality to add a constraint to the Pyomo backend, by pointing it to
    a function which implements the mathematics

    constraint_name : str
        Name for the pyomo.Constraint object
    constraint_sets : list of str
        Sets over which to implement the constraint.
        The list of sets in the Pyomo model corresponds to model dimensions and
        can be found in full by calling the backend interface method `get_all_model_attrs`
    constraint_rule : function with args (backend_model, *args)
        Function corresponding to the format expected by Pyomo,
        i.e. arguments include the Pyomo backend model (`backend_model`)
        and pointers to each set in `constraint_sets`.
        Set arguments need to be in the same order as in constraint_sets.
        The convention we follow is to name the index without the trailing `s`
        E.g. if 'timesteps' is in constraint_sets, then 'timestep' should be one of the function arguments;
        long `loc_techs` set names are abbreviated to `loc_tech`.
        To see what parameters and variables you have to work with,
        call the backend interface method `get_all_model_attrs`

    To limit supply technologies' capacity to 90% of their maximum possible capacity,
    you would do the following:

    constraint_name = 'max_capacity_90_constraint'
    constraint_sets = ['loc_techs_supply']

    def max_capacity_90_constraint_rule(backend_model, loc_tech):

        return backend_model.energy_cap[loc_tech] <= (
            backend_model.energy_cap_max[loc_tech] * 0.9

    # Add the constraint
    model.backend.add_constraint(constraint_name, constraint_sets, max_capacity_90_constraint_rule)

    # Rerun the model with new constraint.
    # Note: will *not* work, since the backend model will be rebuilt, killing any changes you've made.
    new_model = model.backend.rerun()

    Note that we like the convention that constraint names end with 'constraint' and
    constraint rules have the same text, with an appended '_rule',
    but you are not required to follow this convention to have a working constraint.

    assert (
        constraint_rule.__code__.co_varnames[0] == "backend_model"
    ), "First argument of constraint function must be 'backend_model'."
    assert constraint_rule.__code__.co_argcount - 1 == len(
    ), "Number of constraint arguments must equal number of constraint sets + 1."

        sets = [getattr(backend_model, i) for i in constraint_sets]
    except AttributeError as e:
        e.args = (e.args[0].replace("'ConcreteModel'", "Pyomo backend model"),)

        po.Constraint(*sets, **{"rule": constraint_rule}),

    return backend_model

def get_all_pyomo_model_attrs(backend_model):
    Get the name of all sets, parameters, and variables in the generated Pyomo model.

    Dictionary differentiating between variables ('Var'), parameters ('Param'), and sets ('Set').
    variables and parameters are given as a dictionaries of lists, where keys are the item names and
    values are a list of dimensions over which they are indexed. These dimensions correspond to the
    # Indexed objected
    objects = {
        objname: {
   [ for j in i.index_set().subsets()]
            if + "_index" == i.index_set().name
            else [i.index_set().name]
            for i in backend_model.component_objects()
            if isinstance(i, getattr(po.base, objname))
        for objname in ["Var", "Param"]
    # Indices
    objects["Set"] = [ for i in backend_model.component_objects() if isinstance(i, po.base.Set)

    return objects

[docs]class BackendInterfaceMethods: def __init__(self, model): self._backend = model._backend_model self._opt = model._backend_model_opt self._model_data = model._model_data
[docs] def access_model_inputs(self): return access_pyomo_model_inputs(self._backend)
access_model_inputs.__doc__ = access_pyomo_model_inputs.__doc__
[docs] def update_param(self, *args, **kwargs): return update_pyomo_param(self._backend, self._opt, *args, **kwargs)
update_param.__doc__ = update_pyomo_param.__doc__
[docs] def activate_constraint(self, *args, **kwargs): return activate_pyomo_constraint(self._backend, *args, **kwargs)
activate_constraint.__doc__ = activate_pyomo_constraint.__doc__
[docs] def rerun(self, *args, **kwargs): _opt, new_model = rerun_pyomo_model( self._model_data, self._backend, self._opt, *args, **kwargs ) self._opt = _opt return new_model
rerun.__doc__ = rerun_pyomo_model.__doc__ def regenerate_persistent_solver(self, *args, **kwargs): self._opt = regenerate_persistent_pyomo_solver( self._backend, self._opt, *args, **kwargs ) regenerate_persistent_solver.__doc__ = regenerate_persistent_pyomo_solver.__doc__ def add_constraint(self, *args, **kwargs): self._backend = add_pyomo_constraint(self._backend, *args, **kwargs) add_constraint.__doc__ = add_pyomo_constraint.__doc__ def get_all_model_attrs(self, *args, **kwargs): return get_all_pyomo_model_attrs(self._backend, *args, **kwargs) get_all_model_attrs.__doc__ = get_all_pyomo_model_attrs.__doc__