def postprocess_model_results(results, model_data, timings):
def capacity_factor(results, model_data, systemwide=False):
def systemwide_levelised_cost(results, model_data, total=False):
def clean_results(results, zero_threshold, timings): def read_netcdf(path):
def save_netcdf(model_data, path, model=None):
def save_csv(model_data, path, dropna=True):
def save_lp(model, path): def model_run_from_yaml(
    model_file, timeseries_dataframes=None, scenario=None, override_dict=None
):
    """
    Generate processed ModelRun configuration from a
    YAML model configuration file.

    Parameters
    ----------
    model_file : str
        Path to YAML file with model configuration.
    timeseries_dataframes : dict, optional
        Dictionary of timeseries dataframes. The keys are strings
        corresponding to the dataframe names given in the yaml files and
        the values are dataframes with time series data.
    scenario : str, optional
        Name of scenario to apply. Can either be a named scenario, or a
        comma-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
    )
    subsets = AttrDict.from_yaml(
        os.path.join(os.path.dirname(calliope.__file__), "config", "subsets.yaml")
    )

    return generate_model_run(
        config_with_overrides,
        timeseries_dataframes,
        debug_comments,
        overrides,
        scenario,
        subsets,
    ) def model_run_from_dict(
    config_dict, timeseries_dataframes=None, scenario=None, override_dict=None
):
    """
    Generate processed ModelRun configuration from a
    model configuration dictionary.

    Parameters
    ----------
    config_dict : dict or AttrDict
    timeseries_dataframes : dict, optional
        Dictionary of timeseries dataframes. The keys are strings
        corresponding to the dataframe names given in the yaml files and
        the values are dataframes with time series data.
    scenario : str, optional
        Name of scenario to apply. Can either be a named scenario, or a
        comma-separated list of individual overrides to be combined
        ad-hoc, e.g. 'my_scenario_name' or 'override1,override2'.
    override_dict : dict or AttrDict, optional

    """
    if not isinstance(config_dict, AttrDict):
        config = AttrDict(config_dict)
    else:
        config = config_dict
    config.config_path = None

    config_with_overrides, debug_comments, overrides, scenario = apply_overrides(
        config, scenario=scenario, override_dict=override_dict
    )
    subsets = AttrDict.from_yaml(
        os.path.join(os.path.dirname(calliope.__file__), "config", "subsets.yaml")
    )

    return generate_model_run(
        config_with_overrides,
        timeseries_dataframes,
        debug_comments,
        overrides,
        scenario,
        subsets,
    ) class ModelDataFactory:

    UNWANTED_TECH_KEYS = [
        "allowed_constraints",
        "required_constraints",
        "allowed_costs",
        "allowed_switches",
        "constraints",
        "essentials.carrier",
    ]

    LOOKUP_STR = "[\\w\\-]*"  # all alphanumerics + `_` and `-`

    def __init__(self, model_run_dict):
        """
        Take a Calliope model_run and convert it into an xarray Dataset, ready for
        constraint generation. Timeseries data is also extracted from file at this
        point, and the time dimension added to the data

        Parameters
        ----------
        model_run_dict : AttrDict
            preprocessed model_run dictionary, as produced by
            Calliope.preprocess.preprocess_model

        Returns
        -------
        data : xarray Dataset
            Dataset with optimisation param_dict as variables, optimisation sets as
            coordinates, and other information in attributes.
        data_pre_time : xarray Dataset, only returned if debug = True
            Dataset, prior to time dimension addition, with optimisation param_dict
            as variables, optimisation sets as coordinates, and other information
            in attributes.

        """
        self.node_dict = model_run_dict.nodes.as_dict_flat()
        self.tech_dict = model_run_dict.techs.as_dict_flat()
        self.model_run = model_run_dict
        self.model_data = xr.Dataset(
            coords={"timesteps": model_run_dict.timeseries_data.index}
        )
        self._add_attributes(model_run_dict)
        self.template_config = AttrDict.from_yaml(
            os.path.join(
                os.path.dirname(calliope.__file__), "config", "model_data_lookup.yaml"
            )
        )
        self._strip_unwanted_keys()
        self._add_node_tech_sets()

    def __call__(self):
        self._extract_node_tech_data()
        self._add_time_dimension()
        self._clean_model_data()

        return (
            self.model_data_pre_clustering,
            self.model_data,
            self.data_pre_time,
            self.stripped_keys,
        )

    def _extract_node_tech_data(self):
        self._add_param_from_template()
        self._clean_unused_techs_nodes_and_carriers()

    def _add_time_dimension(self):
        self.data_pre_time = self.model_data.copy(deep=True)
        self.model_data = time.add_time_dimension(self.model_data, self.model_run)
        self._update_dtypes()

        if self.model_run.get_key("model.random_seed", None):
            np.random.seed(seed=self.model_run.model.random_seed)
        self.model_data_pre_clustering = self.model_data.copy(deep=True)
        if self.model_run.get_key("model.time", None):
            self.model_data = time.apply_time_clustering(
                self.model_data, self.model_run
            )

        self.model_data = time.add_max_demand_timesteps(self.model_data)

    def _clean_model_data(self):
        self.model_data = dataset.reorganise_xarray_dimensions(self.model_data)
        self._add_var_attrs()
        self._update_dtypes()
        self._check_data()

    @staticmethod
    def _empty_or_invalid(var):
        if isinstance(var, str):
            return False
        elif isinstance(var, list):
            return not var or pd.isnull(var).any()
        elif isinstance(var, (tuple, set, dict)):
            return not var
        else:
            return pd.isnull(var)

    def _strip_unwanted_keys(self):
        """
        These are keys in `model_run` that we don't need in `model_data`.
        Removing them now ensures they don't end up in model_data by mistake
        and that the final `tech_dict` and `node_dicts` should be empty if all
        relevant data *has* made it through to model_data.
        """
        self.stripped_keys = list(
            self._reformat_model_run_dict(
                self.tech_dict,
                [],
                get_method="pop",
                end="({})".format("|".join(self.UNWANTED_TECH_KEYS)),
            ).keys()
        )

        for subdict in ["tech", "node"]:
            for key, val in list(getattr(self, f"{subdict}_dict").items()):
                if self._empty_or_invalid(val):
                    self.stripped_keys.append(key)
                    getattr(self, f"{subdict}_dict").pop(key)

    def _add_node_tech_sets(self):
        """
        Run through the whole `model_run` and extract all the valid combinations of techs
        at nodes.
        """
        kwargs = {"get_method": "get", "end": "\\.({0}).*"}
        df = self._dict_to_df(
            data_dict=self._reformat_model_run_dict(
                self.node_dict, ["techs"], **kwargs
            ),
            data_dimensions=["nodes", "techs"],
            is_link=False,
            var_name="node_tech",
        )

        link_data_dict = self._reformat_model_run_dict(
            self.node_dict, ["links", "techs"], **kwargs
        )
        if not link_data_dict:
            self.link_techs = pd.DataFrame([None])
        else:
            df_link = self._dict_to_df(
                link_data_dict,
                data_dimensions=["nodes", "node_to", "techs"],
                is_link=True,
                var_name="node_tech",
            )
            self._get_link_remotes(link_data_dict)
            df = df.append(df_link)

        df = self._all_df_to_true(df)

        self.model_data = self.model_data.merge(xr.Dataset.from_dataframe(df))

    def _get_link_remotes(self, link_data_dict):
        df = self._dict_to_df(
            data_dict=link_data_dict,
            data_dimensions=["nodes", "node_to", "techs"],
            is_link=False,
            var_name="node_tech",
        )
        df = df.assign(
            link_remote_techs=df.index.get_level_values("techs")
            + ":"
            + df.index.get_level_values("nodes"),
            link_remote_nodes=df.index.get_level_values("node_to"),
            base_techs=df.index.get_level_values("techs"),
        )
        df_all_link_techs = self._update_link_idx_levels(df)
        self.link_techs = df_all_link_techs["base_techs"].groupby(level="techs").first()

        self.model_data = self.model_data.merge(
            xr.Dataset.from_dataframe(
                df_all_link_techs.drop(["node_tech", "base_techs"], axis=1)
            )
        )

    def _format_lookup(self, string_to_format):
        return string_to_format.format(self.LOOKUP_STR)

    def _get_key_matching_nesting(
        self, nesting, key_to_check, start="({0})\\.", end="\\.({0})", **kwargs
    ):
        nesting_string = "\\.({0})\\.".join(nesting)
        search_string = self._format_lookup(f"^{start}{nesting_string}{end}$")
        return re.search(search_string, key_to_check)

    def _reformat_model_run_dict(
        self,
        model_run_subdict,
        expected_nesting,
        get_method="pop",
        values_as_dimension=False,
        **kwargs,
    ):
        """
        Extract key:value pairs from `model_run` which match the expected dictionary
        nesting. If value is a list, return it as a `.` concatenated string.
        """
        data_dict = {}

        for key in list(model_run_subdict.keys()):
            key_match = self._get_key_matching_nesting(expected_nesting, key, **kwargs)
            if key_match is None:
                continue

            groups = [tuple(key_match.groups())]
            val = getattr(model_run_subdict, get_method)(key)
            if self._empty_or_invalid(val):
                continue
            if values_as_dimension:
                # this if/else is for multiple carriers defined under one carrier tier
                if isinstance(val, list):
                    groups = [groups[0] + (v,) for v in val]
                else:
                    groups[0] += (val,)
                val = 1
            if isinstance(val, list):
                val = ".".join(val)
            for group in groups:
                data_dict[group] = val

        if not data_dict:
            return None
        else:
            return data_dict

    def _dict_to_df(
        self,
        data_dict,
        data_dimensions,
        var_name=None,
        var_name_prefix=None,
        is_link=False,
        **kwargs,
    ):
        """
        Take in a dictionary with tuple keys and turn it into a pandas multi-index dataframe.
        Index levels are data dimensions; columns are data variables.
        """
        df = pd.Series(data_dict)
        if len(data_dimensions) < len(df.index.names):
            df = df.unstack(-1)
        elif var_name is not None:
            df = df.to_frame(var_name)
        df = df.rename_axis(index=data_dimensions)

        if "var_name" in data_dimensions:
            df = df.unstack(data_dimensions.index("var_name"))

        if var_name_prefix is not None:
            df = df.rename(columns=lambda x: var_name_prefix + "_" + x)

        if is_link:
            df = self._update_link_idx_levels(df)

        return df

    def _model_run_dict_to_dataset(
        self,
        group_name,
        model_run_subdict_name,
        expected_nesting,
        data_dimensions,
        **kwargs,
    ):
        """
        Pop out key:value pairs from `model_run` nodes or techs subdicts,
        and turn them into beautifully tabulated, multi-dimensional data in an
        xarray dataset.
        """
        model_run_subdict = getattr(self, f"{model_run_subdict_name}_dict")
        data_dict = self._reformat_model_run_dict(
            model_run_subdict, expected_nesting, **kwargs
        )
        if not data_dict:
            logging.info(
                f"No relevant data found for `{group_name}` group of parameters"
            )
            return None
        df = self._dict_to_df(data_dict, data_dimensions, **kwargs)
        if model_run_subdict_name == "tech":
            df = self._update_link_tech_names(df)
        new_model_data_vars = xr.Dataset.from_dataframe(df)

        self.model_data = self.model_data.combine_first(new_model_data_vars)

    def _update_link_tech_names(self, df):
        """
        tech-specific information will only have info on link techs by their base name,
        but the data needs to be duplicated across all link techs, i.e. for every node
        that tech is linking: (`tech_name` -> [`tech_name:node1`, `tech_name:node2`, ...])
        """
        if isinstance(df.index, pd.MultiIndex):
            idx_to_stack = df.index.names.difference(["techs"])
            df = df.unstack(idx_to_stack)
        else:
            idx_to_stack = []
        if df.index.intersection(self.link_techs.values).empty:
            return df.stack(idx_to_stack)
        df_link_tech_data = df.reindex(self.link_techs.values)
        df_link_tech_data.index = self.link_techs.index
        df = (
            df.append(df_link_tech_data)
            .drop(self.link_techs.unique(), errors="ignore")
            .dropna(how="all")
            .stack(idx_to_stack)
        )

        return df

    def _add_var_attrs(self):
        for var_data in self.model_data.data_vars.values():
            var_data.attrs["parameters"] = 1
            var_data.attrs["is_result"] = 0

    @staticmethod
    def _update_link_idx_levels(df):
        """
        ([(`tech_name`, `node1`), (`tech_name`, `node2`)] -> [`tech_name:node1`, `tech_name:node2`])
        """
        new_tech = (
            df.index.get_level_values("techs")
            + ":"
            + df.index.get_level_values("node_to")
        )
        df = (
            df.assign(techs=new_tech)
            .droplevel(["techs", "node_to"])
            .set_index("techs", append=True)
        )
        return df

    @staticmethod
    def _all_df_to_true(df):
        return df == df

    def _add_attributes(self, model_run):
        attr_dict = AttrDict()

        attr_dict["calliope_version"] = __version__
        attr_dict["applied_overrides"] = model_run["applied_overrides"]
        attr_dict["scenario"] = model_run["scenario"]

        default_tech_dict = checks.DEFAULTS.techs.default_tech
        default_cost_dict = {
            "cost_{}".format(k): v
            for k, v in default_tech_dict.costs.default_cost.items()
        }
        default_node_dict = checks.DEFAULTS.nodes.default_node

        attr_dict["defaults"] = AttrDict(
            {
                **default_tech_dict.constraints.as_dict(),
                **default_tech_dict.switches.as_dict(),
                **default_cost_dict,
                **default_node_dict.as_dict(),
            }
        ).to_yaml()

        self.model_data.attrs = attr_dict

    def _clean_unused_techs_nodes_and_carriers(self):
        """
        Remove techs not assigned to nodes, nodes with no associated techs, and carriers associated with removed techs
        """
        for dim in ["nodes", "techs"]:
            self.model_data = self.model_data.dropna(
                dim, how="all", subset=["node_tech"]
            )
        for dim in ["carriers", "carrier_tiers"]:
            self.model_data = self.model_data.dropna(dim, how="all")

        self.model_data = self.model_data.drop_vars(
            [
                var_name
                for var_name, var in self.model_data.data_vars.items()
                if var.isnull().all()
            ]
        )

    def _add_param_from_template(self):
        for group, group_config in self.template_config.items():
            self._model_run_dict_to_dataset(group_name=group, **group_config)

    def _update_dtypes(self):
        """
        Update dtypes to not be 'Object', if possible.
        Order of preference is: bool, int, float
        """
        # TODO: this should be redundant once typedconfig is in (params will have predefined dtypes)
        for var_name, var in self.model_data.data_vars.items():
            if var.dtype.kind == "O":
                no_nans = var.where(var != "nan", drop=True)
                self.model_data[var_name] = var.where(var != "nan")
                if no_nans.isin(["True", 0, 1, "False", "0", "1"]).all():
                    # Turn to bool
                    self.model_data[var_name] = var.isin(["True", 1, "1"])
                else:
                    try:
                        self.model_data[var_name] = var.astype(np.int, copy=False)
                    except (ValueError, OverflowError):
                        try:
                            self.model_data[var_name] = var.astype(np.float, copy=False)
                        except ValueError:
                            None

    def _check_data(self):
        if self.node_dict or self.tech_dict:
            raise exceptions.ModelError(
                "Some data not extracted from inputs into model dataset:\n"
                f"{self.node_dict}"
            )
        self.model_data, final_check_comments, warns, errors = checks.check_model_data(
            self.model_data
        )
        exceptions.print_warnings_and_raise_errors(warnings=warns, errors=errors) class AttrDict(dict):
    """
    A subclass of ``dict`` with key access by attributes::

        d = AttrDict({'a': 1, 'b': 2})
        d.a == 1  # True

    Includes a range of additional methods to read and write to YAML,
    and to deal with nested keys.

    """

    __name__ = "AttrDict"

    __getattr__ = dict.__getitem__
    __setattr__ = dict.__setitem__

    def __init__(self, source_dict=None):
        super().__init__()

        if source_dict is not None:
            if isinstance(source_dict, dict):
                self.init_from_dict(source_dict)
            else:
                raise ValueError("Must pass a dict to AttrDict")

    def copy(self):
        """Override copy method so that it returns an AttrDict"""
        return AttrDict(self.as_dict().copy())

    def init_from_dict(self, d):
        """
        Initialize a new AttrDict from the given dict. Handles any
        nested dicts by turning them into AttrDicts too::

            d = AttrDict({'a': 1, 'b': {'x': 1, 'y': 2}})
            d.b.x == 1  # True

        """
        for k, v in d.items():
            # First, keys must be strings, not ints
            if isinstance(k, int):
                k = str(k)

            # Now, assign to the key, handling nested AttrDicts properly
            if isinstance(v, dict):
                self.set_key(k, AttrDict(v))
            elif isinstance(v, list):
                # Modifying the list in-place so that if it is a modified
                # list subclass, e.g. CommentedSeq, it is not killed
                for i in range(len(v)):
                    if isinstance(v[i], dict):
                        v[i] = AttrDict(v[i])
                self.set_key(k, v)
            else:
                self.set_key(k, v)

    @classmethod
    def _resolve_imports(cls, loaded, resolve_imports, base_path=None):
        if (
            isinstance(resolve_imports, bool)
            and resolve_imports is True
            and "import" in loaded
        ):
            loaded_dict = loaded
        elif (
            isinstance(resolve_imports, str)
            and resolve_imports + ".import" in loaded.keys_nested()
        ):
            loaded_dict = loaded.get_key(resolve_imports)
        else:  # Return right away if no importing to be done
            return loaded

        # If we end up here, we have something to import
        imports = loaded_dict.get_key("import")
        if not isinstance(imports, list):
            raise ValueError("`import` must be a list.")

        for k in imports:
            if base_path:
                path = relative_path(base_path, k)
            else:
                path = k
            imported = cls.from_yaml(path)
            # loaded is added to imported (i.e. it takes precedence)
            imported.union(loaded_dict)
            loaded_dict = imported
        # 'import' key itself is no longer needed
        loaded_dict.del_key("import")

        if isinstance(resolve_imports, str):
            loaded.set_key(resolve_imports, loaded_dict)
        else:
            loaded = loaded_dict

        return loaded

    @classmethod
    def from_yaml(cls, f, resolve_imports=True):
        """
        Returns an AttrDict initialized from the given path or
        file object ``f``, which must point to a YAML file. The path can
        be a string or a pathlib.Path.

        Parameters
        ----------

        f : str or pathlib.Path
        resolve_imports : bool or str, optional
            If ``resolve_imports`` is True, top-level ``import:`` statements
            are resolved recursively.
            If ``resolve_imports is False, top-level ``import:`` statements
            are treated like any other key and not further processed.
            If ``resolve_imports`` is a string, such as ``foobar``, import
            statements underneath that key are resolved, i.e. ``foobar.import:``.
            When resolving import statements, anything defined locally
            overrides definitions in the imported file.

        """
        if isinstance(f, str) or isinstance(f, Path):
            with open(
                f,
                "r",
                encoding="utf-8",
            ) as src:
                loaded = cls(_yaml_load(src))
        else:
            loaded = cls(_yaml_load(f))
        loaded = cls._resolve_imports(loaded, resolve_imports, base_path=f)
        return loaded

    @classmethod
    def from_yaml_string(cls, string, resolve_imports=True):
        """
        Returns an AttrDict initialized from the given string, which
        must be valid YAML.

        """
        loaded = cls(_yaml_load(string))
        loaded = cls._resolve_imports(loaded, resolve_imports)
        return loaded

    def set_key(self, key, value):
        """
        Set the given ``key`` to the given ``value``. Handles nested
        keys, e.g.::

            d = AttrDict()
            d.set_key('foo.bar', 1)
            d.foo.bar == 1  # True

        """
        if isinstance(value, dict) and not isinstance(value, AttrDict):
            value = AttrDict(value)
        if "." in key:
            key, remainder = key.split(".", 1)
            try:
                self[key].set_key(remainder, value)
            except KeyError:
                self[key] = AttrDict()
                self[key].set_key(remainder, value)
            except AttributeError:
                if self[key] is None:  # If the value is None, we replace it
                    self[key] = AttrDict()
                    self[key].set_key(remainder, value)
                # Else there is probably something there, and we don't just
                # want to overwrite so stop and warn the user
                else:
                    raise KeyError("Cannot set nested key on non-dict key.")
        else:
            if key in self and isinstance(value, AttrDict):
                for k, v in value.items():
                    self[key].set_key(k, v)
            else:
                self[key] = value

    def get_key(self, key, default=_MISSING):
        """
        Looks up the given ``key``. Like set_key(), deals with nested
        keys.

        If default is anything but ``_MISSING``, the given default is
        returned if the key does not exist.

        """
        if "." in key:
            # Nested key of form "foo.bar"
            key, remainder = key.split(".", 1)
            if default != _MISSING:
                try:
                    value = self[key].get_key(remainder, default)
                except KeyError:
                    # subdict exists, but doesn't contain key
                    return default
                except AttributeError:
                    # key points to non-dict thing, so no get_key attribute
                    return default
            else:
                value = self[key].get_key(remainder)
        else:
            # Single, non-nested key of form "foo"
            if default != _MISSING:
                return self.get(key, default)
            else:
                return self[key]
        return value

    def del_key(self, key):
        """Delete the given key. Properly deals with nested keys."""
        if "." in key:
            key, remainder = key.split(".", 1)
            try:
                del self[key][remainder]
            except KeyError:
                self[key].del_key(remainder)

            # If we removed the last subkey, delete the parent key too
            if len(self[key].keys()) == 0:
                del self[key]

        else:
            del self[key]

    def as_dict(self, flat=False):
        """
        Return the AttrDict as a pure dict (with nested dicts if
        necessary).

        """
        if flat:
            return self.as_dict_flat()
        else:
            return self.as_dict_nested()

    def as_dict_nested(self):
        d = {}
        for k, v in self.items():
            if isinstance(v, AttrDict):
                d[k] = v.as_dict()
            elif isinstance(v, list):
                d[k] = [i if not isinstance(i, AttrDict) else i.as_dict() for i in v]
            else:
                d[k] = v
        return d

    def as_dict_flat(self):
        d = {}
        keys = self.keys_nested()
        for k in keys:
            d[k] = self.get_key(k)
        return d

    def to_yaml(self, path=None):
        """
        Saves the AttrDict to the ``path`` as a YAML file, or returns
        a YAML string if ``path`` is None.

        """
        result = self.copy()
        yaml_ = ruamel_yaml.YAML()
        yaml_.indent = 2
        yaml_.block_seq_indent = 0

        # Numpy objects should be converted to regular Python objects,
        # so that they are properly displayed in the resulting YAML output
        for k in result.keys_nested():
            # Convert numpy numbers to regular python ones
            v = result.get_key(k)
            if isinstance(v, np.floating):
                result.set_key(k, float(v))
            elif isinstance(v, np.integer):
                result.set_key(k, int(v))
            # Lists are turned into seqs so that they are formatted nicely
            elif isinstance(v, list):
                result.set_key(k, yaml_.seq(v))

        result = result.as_dict()

        if path is not None:
            with open(path, "w") as f:
                yaml_.dump(result, f)
        else:
            stream = io.StringIO()
            yaml_.dump(result, stream)
            return stream.getvalue()

    def keys_nested(self, subkeys_as="list"):
        """
        Returns all keys in the AttrDict, sorted, including the keys of
        nested subdicts (which may be either regular dicts or AttrDicts).

        If ``subkeys_as='list'`` (default), then a list of
        all keys is returned, in the form ``['a', 'b.b1', 'b.b2']``.

        If ``subkeys_as='dict'``, a list containing keys and dicts of
        subkeys is returned, in the form ``['a', {'b': ['b1', 'b2']}]``.

        """
        keys = []
        for k, v in sorted(self.items()):
            # Check if dict instance (which AttrDict is too),
            # and for non-emptyness of the dict
            if isinstance(v, dict) and v:
                if subkeys_as == "list":
                    keys.extend([k + "." + kk for kk in v.keys_nested()])
                elif subkeys_as == "dict":
                    keys.append({k: v.keys_nested(subkeys_as=subkeys_as)})
            else:
                keys.append(k)
        return keys

    def union(
        self,
        other,
        allow_override=False,
        allow_replacement=False,
        allow_subdict_override_with_none=False,
    ):
        """
        Merges the AttrDict in-place with the passed ``other``
        AttrDict. Keys in ``other`` take precedence, and nested keys
        are properly handled.

        If ``allow_override`` is False, a KeyError is raised if
        other tries to redefine an already defined key.

        If ``allow_replacement``, allow "_REPLACE_" key to replace an
        entire sub-dict.

        If ``allow_subdict_override_with_none`` is False (default),
        a key of the form ``this.that: None`` in other will be ignored
        if subdicts exist in self like ``this.that.foo: 1``, rather
        than wiping them.

        """
        self_keys = self.keys_nested()
        other_keys = other.keys_nested()
        if allow_replacement:
            WIPE_KEY = "_REPLACE_"
            override_keys = [k for k in other_keys if WIPE_KEY not in k]
            wipe_keys = [
                k.split("." + WIPE_KEY)[0] for k in other_keys if WIPE_KEY in k
            ]
        else:
            override_keys = other_keys
            wipe_keys = []
        for k in override_keys:
            if not allow_override and k in self_keys:
                raise KeyError("Key defined twice: {}".format(k))
            else:
                other_value = other.get_key(k)
                # If other value is None, and would overwrite an entire subdict,
                # we skip it
                if not (
                    other_value is None and isinstance(self.get_key(k, None), AttrDict)
                ):
                    self.set_key(k, other_value)
        for k in wipe_keys:
            self.set_key(k, other.get_key(k + "." + WIPE_KEY)) def log_time(
    logger, timings, identifier, comment=None, level="info", time_since_run_start=False
):
    if comment is None:
        comment = identifier

    timings[identifier] = now = datetime.datetime.now()

    if time_since_run_start and "run_start" in timings:
        time_diff = now - timings["run_start"]
        comment += ". Time since start of model run: {}".format(time_diff)

    getattr(logger, level)(comment) class UpdateObserverDict(ObservedDict):
    """
    Dictionary subclass which observes a dictionary and updates an attribute of
    the model_data xarray dataset with a YAML string of that dictionary.
    This update takes place whenever a value is updated in the dictionary.

    Parameters
    ----------
    name : str
        The model_data attribute key to update.
    observer : xarray Dataset (e.g. calliope model_data)
        The Dataset whose attribute dictionary will contain the key `name`
        and value = a YAML string of the initial_dict.
    initial_dict : dict, optional, default = None
        An initial dictionary to copy for observing.
        One of initial_dict or initial_yaml_string must be defined.
    initial_yaml_string : str, optional, default = None
        A YAML string of an initial dictionary to copy for observing.
        One of initial_dict or initial_yaml_string must be defined.

    Returns
    -------
    Observed dictionary, which acts as a dictionary in every sense *except* that
    on changing or adding any key:value pairs in that dictionary, the YAML string
    stored at `observer.attrs[name]` will be updated to reflect the new Observed
    dicitonary.
    """

    def __init__(
        self,
        name,
        observer,
        initial_dict=None,
        initial_yaml_string=None,
        *args,
        **kwargs,
    ):
        self.observer = observer
        self.name = name

        check_input_args = [i is None for i in [initial_dict, initial_yaml_string]]
        if all(check_input_args) or not any(check_input_args):
            raise ValueError(
                "must supply one, and only one, of initial_dict or initial_yaml_string"
            )

        super().__init__(initial_dict, initial_yaml_string, *args, **kwargs)

    def notify(self, updated=None):
        temp_dict = {
            k: v
            for k, v in self.items()
            if (not isinstance(v, dict) and v is not None)
            or (isinstance(v, dict) and len(v.keys()) > 0)
        }
        self.observer.attrs[self.name] = AttrDict(temp_dict).to_yaml() def _formatwarning(message, category, filename, lineno, line=None):
def warn(message, _class=ModelWarning):
def print_warnings_and_raise_errors(warnings=None, errors=None):
class ModelError(Exception):
class BackendError(Exception):
class ModelWarning(Warning):
class BackendWarning(Warning): def run(model_data, timings, build_only=False):
    """
    Parameters
    ----------
    model_data : xarray.Dataset
        Pre-processed dataset of Calliope model data.
    timings : dict
        Stores timings of various stages of model processing.
    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, opt = run_plan(
            model_data,
            run_config,
            timings,
            backend=BACKEND[run_config.backend],
            build_only=build_only,
        )

    elif run_config["mode"] == "operate":
        results, backend, opt = run_operate(
            model_data,
            run_config,
            timings,
            backend=BACKEND[run_config.backend],
            build_only=build_only,
        )

    elif run_config["mode"] == "spores":
        results, backend, opt = run_spores(
            model_data,
            run_config,
            timings,
            interface=INTERFACE[run_config.backend],
            backend=BACKEND[run_config.backend],
            build_only=build_only,
        )

    return results, backend, opt, INTERFACE[run_config.backend].BackendInterfaceMethods
import logging
import warnings
import xarray as xr
from calliope.postprocess import results as postprocess_results
from calliope.core import io
from calliope.preprocess import (
    model_run_from_yaml,
    model_run_from_dict,
)
from calliope.preprocess.model_data import ModelDataFactory
from calliope.core.attrdict import AttrDict
from calliope.core.util.logging import log_time
from calliope.core.util.observed_dict import UpdateObserverDict
from calliope import exceptions
from calliope.backend.run import run as run_backend
"""
Copyright (C) since 2013 Calliope contributors listed in AUTHORS.
Licensed under the Apache 2.0 License (see LICENSE file).

model.py
~~~~~~~~

Implements the core Model class.

"""



logger = logging.getLogger(__name__)


def read_netcdf(path):
    """
    Return a Model object reconstructed from model data in a NetCDF file.

    """

    model_data = io.read_netcdf(path)