calliope.Model(model_definition, debug=False, scenario=None, override_dict=None, data_source_dfs=None, **kwargs)

Bases: object

A Calliope Model.

Returns a new Model from either the path to a YAML model configuration file or a dict fully specifying the model.

Parameters:

Name	Type	Description	Default
model_definition	str | Path | dict | Dataset	If str or Path, must be the path to a model configuration file. If dict or AttrDict, must fully specify the model. If an xarray dataset, must be a valid calliope model.	required
debug	bool	If True, additional debug data will be included in the built model. Defaults to False.	False
scenario	str	Comma delimited string of pre-defined scenarios to apply to the model,	None
override_dict	dict	Additional overrides to apply to config. These will be applied after applying any defined scenario overrides.	None
data_source_dfs	dict[str, DataFrame]	Model definition data_source entries can reference in-memory pandas DataFrames. The referenced data must be supplied here as a dictionary of those DataFrames. Defaults to None.	None

Source code in src/calliope/model.py

def __init__(
    self,
    model_definition: str | Path | dict | xr.Dataset,
    debug: bool = False,
    scenario: Optional[str] = None,
    override_dict: Optional[dict] = None,
    data_source_dfs: Optional[dict[str, pd.DataFrame]] = None,
    **kwargs,
):
    """
    Returns a new Model from either the path to a YAML model
    configuration file or a dict fully specifying the model.

    Args:
        model_definition (str | Path | dict | xr.Dataset):
            If str or Path, must be the path to a model configuration file.
            If dict or AttrDict, must fully specify the model.
            If an xarray dataset, must be a valid calliope model.
        debug (bool, optional):
            If True, additional debug data will be included in the built model.
            Defaults to False.
        scenario (str):
            Comma delimited string of pre-defined `scenarios` to apply to the model,
        override_dict (dict):
            Additional overrides to apply to `config`.
            These will be applied *after* applying any defined `scenario` overrides.
        data_source_dfs (dict[str, pd.DataFrame], optional):
            Model definition `data_source` entries can reference in-memory pandas DataFrames.
            The referenced data must be supplied here as a dictionary of those DataFrames.
            Defaults to None.
    """
    self._timings: dict = {}
    self.config: AttrDict
    self.defaults: AttrDict
    self.math: AttrDict
    self._model_def_path: Optional[Path]
    self.backend: BackendModel
    self.math_documentation = MathDocumentation()
    self._is_built: bool = False
    self._is_solved: bool = False

    # try to set logging output format assuming python interactive. Will
    # use CLI logging format if model called from CLI
    timestamp_model_creation = log_time(
        LOGGER, self._timings, "model_creation", comment="Model: initialising"
    )
    if isinstance(model_definition, xr.Dataset):
        self._init_from_model_data(model_definition)
    else:
        (model_def, self._model_def_path, applied_overrides) = (
            load.load_model_definition(
                model_definition, scenario, override_dict, **kwargs
            )
        )
        self._init_from_model_def_dict(
            model_def, applied_overrides, scenario, debug, data_source_dfs
        )

    self._model_data.attrs["timestamp_model_creation"] = timestamp_model_creation
    version_def = self._model_data.attrs["calliope_version_defined"]
    version_init = self._model_data.attrs["calliope_version_initialised"]
    if version_def is not None and not version_init.startswith(version_def):
        exceptions.warn(
            f"Model configuration specifies calliope version {version_def}, "
            f"but you are running {version_init}. Proceed with caution!"
        )

    self.math_documentation.inputs = self._model_data

backend: BackendModel instance-attribute

config: AttrDict instance-attribute

defaults: AttrDict instance-attribute

inputs property

is_built property

is_solved property

math: AttrDict instance-attribute

math_documentation = MathDocumentation() instance-attribute

name property

results property

build(force=False, **kwargs)

Build description of the optimisation problem in the chosen backend interface.

Parameters:

Name	Type	Description	Default
force	bool	If force is True, any existing results will be overwritten. Defaults to False.	False

Source code in src/calliope/model.py

def build(self, force: bool = False, **kwargs) -> None:
    """Build description of the optimisation problem in the chosen backend interface.

    Args:
        force (bool, optional):
            If ``force`` is True, any existing results will be overwritten.
            Defaults to False.
    """

    if self._is_built and not force:
        raise exceptions.ModelError(
            "This model object already has a built optimisation problem. Use model.build(force=True) "
            "to force the existing optimisation problem to be overwritten with a new one."
        )
    self._model_data.attrs["timestamp_build_start"] = log_time(
        LOGGER,
        self._timings,
        "build_start",
        comment="Model: backend build starting",
    )

    updated_build_config = {**self.config["build"], **kwargs}
    if updated_build_config["mode"] == "operate":
        if not self._model_data.attrs["allow_operate_mode"]:
            raise exceptions.ModelError(
                "Unable to run this model in operate (i.e. dispatch) mode, probably because "
                "there exist non-uniform timesteps (e.g. from time clustering)"
            )
        start_window_idx = updated_build_config.pop("start_window_idx", 0)
        input = self._prepare_operate_mode_inputs(
            start_window_idx, **updated_build_config
        )
    else:
        input = self._model_data
    backend_name = updated_build_config["backend"]
    backend = self._BACKENDS[backend_name](input, **updated_build_config)
    backend._build()
    self.backend = backend

    self._model_data.attrs["timestamp_build_complete"] = log_time(
        LOGGER,
        self._timings,
        "build_complete",
        comment="Model: backend build complete",
    )
    self._is_built = True

info()

Generate basic description of the model, combining its name and a rough indication of the model size.

Returns:

Name	Type	Description
str	str	Basic description of the model.

Source code in src/calliope/model.py

def info(self) -> str:
    """Generate basic description of the model, combining its name and a rough indication of the model size.

    Returns:
        str: Basic description of the model.
    """
    info_strings = []
    model_name = self.name
    info_strings.append(f"Model name:   {model_name}")
    msize = dict(self._model_data.dims)
    msize_exists = self._model_data.definition_matrix.sum()
    info_strings.append(
        f"Model size:   {msize} ({msize_exists.item()} valid node:tech:carrier combinations)"
    )
    return "\n".join(info_strings)

run(force_rerun=False, **kwargs)

Run the model. If force_rerun is True, any existing results will be overwritten.

Additional kwargs are passed to the backend.

Source code in src/calliope/model.py

def run(self, force_rerun=False, **kwargs):
    """
    Run the model. If ``force_rerun`` is True, any existing results
    will be overwritten.

    Additional kwargs are passed to the backend.

    """
    exceptions.warn(
        "`run()` is deprecated and will be removed in a "
        "future version. Use `model.build()` followed by `model.solve()`.",
        FutureWarning,
    )
    self.build(force=force_rerun)
    self.solve(force=force_rerun)

solve(force=False, warmstart=False, **kwargs)

Solve the built optimisation problem.

Parameters:

Name	Type	Description	Default
force	bool	If force is True, any existing results will be overwritten. Defaults to False.	False
warmstart	bool	If True and the optimisation problem has already been run in this session (i.e., force is not True), the next optimisation will be run with decision variables initially set to their previously optimal values. If the optimisation problem is similar to the previous run, this can decrease the solution time. Warmstart will not work with some solvers (e.g., CBC, GLPK). Defaults to False.	False

Raises:

Type	Description
ModelError	Optimisation problem must already be built.
ModelError	Cannot run the model if there are already results loaded, unless force is True.
ModelError	Some preprocessing steps will stop a run mode of "operate" from being possible.

Source code in src/calliope/model.py

def solve(self, force: bool = False, warmstart: bool = False, **kwargs) -> None:
    """
    Solve the built optimisation problem.

    Args:
        force (bool, optional):
            If ``force`` is True, any existing results will be overwritten.
            Defaults to False.
        warmstart (bool, optional):
            If True and the optimisation problem has already been run in this session
            (i.e., `force` is not True), the next optimisation will be run with
            decision variables initially set to their previously optimal values.
            If the optimisation problem is similar to the previous run, this can
            decrease the solution time.
            Warmstart will not work with some solvers (e.g., CBC, GLPK).
            Defaults to False.

    Raises:
        exceptions.ModelError: Optimisation problem must already be built.
        exceptions.ModelError: Cannot run the model if there are already results loaded, unless `force` is True.
        exceptions.ModelError: Some preprocessing steps will stop a run mode of "operate" from being possible.
    """

    # Check that results exist and are non-empty
    if not self._is_built:
        raise exceptions.ModelError(
            "You must build the optimisation problem (`.build()`) "
            "before you can run it."
        )

    if hasattr(self, "results"):
        if self.results.data_vars and not force:
            raise exceptions.ModelError(
                "This model object already has results. "
                "Use model.solve(force=True) to force"
                "the results to be overwritten with a new run."
            )
        else:
            to_drop = self.results.data_vars
    else:
        to_drop = []

    run_mode = self.backend.inputs.attrs["config"]["build"]["mode"]
    self._model_data.attrs["timestamp_solve_start"] = log_time(
        LOGGER,
        self._timings,
        "solve_start",
        comment=f"Optimisation model | starting model in {run_mode} mode.",
    )

    solver_config = update_then_validate_config("solve", self.config, **kwargs)

    if run_mode == "operate":
        if not self._model_data.attrs["allow_operate_mode"]:
            raise exceptions.ModelError(
                "Unable to run this model in operate (i.e. dispatch) mode, probably because "
                "there exist non-uniform timesteps (e.g. from time clustering)"
            )
        results = self._solve_operate(**solver_config)
    else:
        results = self.backend._solve(warmstart=warmstart, **solver_config)

    log_time(
        LOGGER,
        self._timings,
        "solver_exit",
        time_since_solve_start=True,
        comment="Backend: solver finished running",
    )

    # Add additional post-processed result variables to results
    if results.attrs["termination_condition"] in ["optimal", "feasible"]:
        results = postprocess_results.postprocess_model_results(
            results, self._model_data
        )

    log_time(
        LOGGER,
        self._timings,
        "postprocess_complete",
        time_since_solve_start=True,
        comment="Postprocessing: ended",
    )

    self._model_data = self._model_data.drop_vars(to_drop)

    self._model_data.attrs.update(results.attrs)
    self._model_data = xr.merge(
        [results, self._model_data], compat="override", combine_attrs="no_conflicts"
    )
    self._add_model_data_methods()

    self._model_data.attrs["timestamp_solve_complete"] = log_time(
        LOGGER,
        self._timings,
        "solve_complete",
        time_since_solve_start=True,
        comment="Backend: model solve completed",
    )

    self._is_solved = True

to_csv(path, dropna=True, allow_overwrite=False)

Save complete model data (inputs and, if available, results) as a set of CSV files to the given path.

Parameters:

Name	Type	Description	Default
path	str | Path		required
dropna	bool	If True, NaN values are dropped when saving, resulting in significantly smaller CSV files. Defaults to True	True
allow_overwrite	bool	If True, allow the option to overwrite the directory contents if it already exists. This will overwrite CSV files one at a time, so if the dataset has different arrays to the previous saved models, you will get a mix of old and new files. Defaults to False.	False

Source code in src/calliope/model.py

def to_csv(
    self, path: str | Path, dropna: bool = True, allow_overwrite: bool = False
):
    """
    Save complete model data (inputs and, if available, results)
    as a set of CSV files to the given ``path``.

    Args:
        path (str | Path):
        dropna (bool, optional):
            If True, NaN values are dropped when saving, resulting in significantly smaller CSV files.
            Defaults to True
        allow_overwrite (bool, optional):
            If True, allow the option to overwrite the directory contents if it already exists.
            This will overwrite CSV files one at a time, so if the dataset has different arrays to the previous saved models, you will get a mix of old and new files.
            Defaults to False.

    """
    io.save_csv(self._model_data, path, dropna, allow_overwrite)

to_netcdf(path)

Save complete model data (inputs and, if available, results) to a NetCDF file at the given path.

Source code in src/calliope/model.py

def to_netcdf(self, path):
    """
    Save complete model data (inputs and, if available, results)
    to a NetCDF file at the given ``path``.

    """
    io.save_netcdf(self._model_data, path, model=self)

validate_math_strings(math_dict)

Validate that expression and where strings of a dictionary containing string mathematical formulations can be successfully parsed.

This function can be used to test user-defined math before attempting to build the optimisation problem.

NOTE: strings are not checked for evaluation validity. Evaluation issues will be raised only on calling Model.build().

Parameters:

Name	Type	Description	Default
math_dict	dict	Math formulation dictionary to validate. Top level keys must be one or more of ["variables", "global_expressions", "constraints", "objectives"], e.g.: { "constraints": { "my_constraint_name": { "foreach": ["nodes"], "where": "base_tech=supply", "equations": [{"expression": "sum(flow_cap, over=techs) >= 10"}] } } }	required

Returns: If all components of the dictionary are parsed successfully, this function will log a success message to the INFO logging level and return None. Otherwise, a calliope.ModelError will be raised with parsing issues listed.

Source code in src/calliope/model.py

def validate_math_strings(self, math_dict: dict) -> None:
    """Validate that `expression` and `where` strings of a dictionary containing string mathematical formulations can be successfully parsed.

    This function can be used to test user-defined math before attempting to build the optimisation problem.

    NOTE: strings are not checked for evaluation validity. Evaluation issues will be raised only on calling `Model.build()`.

    Args:
        math_dict (dict): Math formulation dictionary to validate. Top level keys must be one or more of ["variables", "global_expressions", "constraints", "objectives"], e.g.:
            ```python
            {
                "constraints": {
                    "my_constraint_name":
                        {
                            "foreach": ["nodes"],
                            "where": "base_tech=supply",
                            "equations": [{"expression": "sum(flow_cap, over=techs) >= 10"}]
                        }

                    }
            }
            ```
    Returns:
        If all components of the dictionary are parsed successfully, this function will log a success message to the INFO logging level and return None.
        Otherwise, a calliope.ModelError will be raised with parsing issues listed.
    """
    validate_dict(math_dict, MATH_SCHEMA, "math")
    valid_component_names = [
        *self.math["variables"].keys(),
        *self.math["global_expressions"].keys(),
        *math_dict.get("variables", {}).keys(),
        *math_dict.get("global_expressions", {}).keys(),
        *self.inputs.data_vars.keys(),
        *self.inputs.attrs["defaults"].keys(),
    ]
    collected_errors: dict = dict()
    for component_group, component_dicts in math_dict.items():
        for name, component_dict in component_dicts.items():
            parsed = parsing.ParsedBackendComponent(
                component_group, name, component_dict
            )
            parsed.parse_top_level_where(errors="ignore")
            parsed.parse_equations(set(valid_component_names), errors="ignore")
            if not parsed._is_valid:
                collected_errors[f"{component_group}:{name}"] = parsed._errors

    if collected_errors:
        exceptions.print_warnings_and_raise_errors(
            during="math string parsing (marker indicates where parsing stopped, which might not be the root cause of the issue; sorry...)",
            errors=collected_errors,
        )

    LOGGER.info("Model: validated math strings")