19307136919

Committed 12 Nov 2025 06:03PM UTC coverage: 81.466% (-0.04%) from 81.505%

Build # 19307136919

Build Type

push

github

Committed by

web-flow

Commit Message

update version for rc3

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

264 existing lines in 3 files now uncovered.

13107 of 16089 relevant lines covered (81.47%)

0.81 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

50.0

/wntr/library/msx/_msxlibrary.py

# coding: utf-8
"""
The wntr.msx.library module includes a library of multi-species water 
models

.. rubric:: Environment Variable

.. envvar:: WNTR_LIBRARY_PATH

    This environment variable, if set, will add additional folder(s) to the
    path to search for multi-species water quality model files, 
    (files with an ".msx", ".yaml", or ".json" file extension).
    Multiple folders should be separated using the "``;``" character.
    See :class:`~wntr.msx.library.ReactionLibrary` for more details.    
"""

from __future__ import annotations

import json
import logging
import os, sys
from typing import Any, ItemsView, Iterator, KeysView, List, Tuple, Union, ValuesView

if sys.version_info[0:2] <= (3, 11):
    from pkg_resources import resource_filename
else:
    from importlib.resources import files

from wntr.msx.base import ExpressionType, ReactionType, SpeciesType
from wntr.msx.model import MsxModel

try:
    import yaml

    yaml_err = None
except ImportError as e:
    yaml = None
    yaml_err = e

PIPE = ReactionType.PIPE
TANK = ReactionType.TANK
BULK = SpeciesType.BULK
WALL = SpeciesType.WALL
RATE = ExpressionType.RATE
EQUIL = ExpressionType.EQUIL
FORMULA = ExpressionType.FORMULA

logger = logging.getLogger(__name__)


class MsxLibrary:
    """Library of multi-species water quality models

    This object can be accessed and treated like a dictionary, where keys are
    the model names and the values are the model objects.

    Paths are added/processed in the following order:

    1. the builtin directory of reactions,
    2. any paths specified in the environment variable described below, with
       directories listed first having the highest priority,
    3. any extra paths specified in the constructor, searched in the order
    provided.

    Once created, the library paths cannot be modified. However, a model can
    be added to the library using :meth:`add_model_from_file` or
    :meth:`add_models_from_dir`.

    """

    def __init__(self, extra_paths: List[str] = None, include_builtins=True,
                 include_envvar_paths=True, load=True) -> None:
        """Library of multi-species water quality models

        Parameters
        ----------
        extra_paths : list of str, optional
            User-specified list of reaction library directories, by default
            None
        include_builtins : bool, optional
            Load files built-in with WNTR, by default True
        include_envvar_paths : bool, optional
            Load files from the paths specified in
            :envvar:`WNTR_LIBRARY_PATH`, by default True
        load : bool or str, optional
            Load the files immediately on creation, by default True.
            If a string, then it will be passed as the `duplicates` argument
            to the load function. See :meth:`reset_and_reload` for details.

        Raises
        ------
        TypeError
            If `extra_paths` is not a list
        """
        if extra_paths is None:
            extra_paths = list()
        elif not isinstance(extra_paths, (list, tuple)):
            raise TypeError("Expected a list or tuple, got {}".format(type(extra_paths)))

        self.__library_paths = list()

        self.__data = dict()

        if include_builtins:
            if sys.version_info[0:2] <= (3, 11):
                default_path = os.path.abspath(resource_filename(__name__, '.'))
            else:
                default_path = os.path.abspath(files('wntr.library.msx').joinpath('.'))
            if default_path not in self.__library_paths:
                self.__library_paths.append(default_path)

        if include_envvar_paths:
            environ_path = os.environ.get("WNTR_LIBRARY_PATH", None)
            if environ_path:
                lib_folders = environ_path.split(";")
                for folder in lib_folders:
                    if folder not in self.__library_paths:
                        self.__library_paths.append(os.path.abspath(folder))

        for folder in extra_paths:
            self.__library_paths.append(os.path.abspath(folder))
        if load:
            if isinstance(load, str):
                self.reset_and_reload(duplicates=load)
            else:
                self.reset_and_reload()

    def __repr__(self) -> str:
        if len(self.__library_paths) > 3:
            return "{}(initial_paths=[{}, ..., {}])".format(self.__class__.__name__, repr(self.__library_paths[0]), repr(self.__library_paths[-1]))
        return "{}({})".format(self.__class__.__name__, repr(self.__library_paths))

    def path_list(self) -> List[str]:
        """List of paths used to populate the library

        Returns
        -------
        list of str
            Copy of the paths used to **initially** populate the library
        """
        return self.__library_paths.copy()

    def reset_and_reload(self, duplicates: str = "error") -> List[Tuple[str, str, Any]]:
        """Load data from the configured directories into a library of models

        Note, this function is not recursive and does not 'walk' any of the
        library's directories to look for subfolders.

        The ``duplicates`` argument specifies how models that have the same name,
        or that have the same filename if a name isn't specified in the file,
        are handled. **Warning**, if two files in the same directory have models
        with the same name, there is no guarantee which will be read in first.

        The first directory processed is the builtin library data. Next, any
        paths specified in the environment variable are searched in the order listed
        in the variable. Finally, any directories specified by the user in the 
        constructor are processed in the order listed.

        Parameters
        ----------
        duplicates : {"error" | "skip" | "replace"}, optional
            by default ``"error"``

            - A value of of ``"error"`` raises an exception and stops execution. 
            - A value of ``"skip"`` will skip models with the same `name` as a model that already
              exists in the library.
            - A value of ``"replace"`` will replace any existing model with a model that is read 
              in that has the same `name`.

        Returns
        -------
        (filename, reason, obj) : (str, str, Any)
            the file not read in, the cause of the problem, and the object that was skipped/overwritten
            or the exception raised

        Raises
        ------
        TypeError
            If `duplicates` is not a string
        ValueError
            If `duplicates` is not a valid value
        IOError
            If `path_to_folder` is not a directory
        KeyError
            If `duplicates` is ``"error"`` and two models have the same name
        """
        if duplicates and not isinstance(duplicates, str):
            raise TypeError("The `duplicates` argument must be None or a string")
        elif duplicates.lower() not in ["error", "skip", "replace"]:
            raise ValueError('The `duplicates` argument must be None, "error", "skip", or "replace"')

        load_errors = list()
        for folder in self.__library_paths:
            errs = self.add_models_from_dir(folder, duplicates=duplicates)
            load_errors.extend(errs)
        return load_errors

    def add_models_from_dir(self, path_to_dir: str, duplicates: str = "error") -> List[Tuple[str, str, Union[MsxModel, Exception]]]:
        """Load all valid model files in a folder

        Note, this function is not recursive and does not 'walk' a directory
        tree.

        Parameters
        ----------
        path_to_dir : str
            Path to the folder to search
        duplicates : {"error", "skip", "replace"}, optional
            Specifies how models that have the same name, or that have the same
            filename if a name isn't specified in the file, are handled.
            **Warning**, if two files in the same directory have models with
            the same name, there is no guarantee which will be read in first,
            by default ``"error"``
            - A value of of ``"error"`` raises an exception and stops execution
            - A value of ``"skip"`` will skip models with the same `name` as a
              model that already exists in the library.
            - A value of ``"replace"`` will replace any existing model with a
              model that is read in that has the same `name`.

        Returns
        -------
        (filename, reason, obj) : tuple[str, str, Any]
            File not read in, the cause of the problem, and the object that was
            skipped/overwritten or the exception raised

        Raises
        ------
        TypeError
            If `duplicates` is not a string
        ValueError
            If `duplicates` is not a valid value
        IOError
            If `path_to_folder` is not a directory
        KeyError
            If `duplicates` is ``"error"`` and two models have the same name
        """
        if duplicates and not isinstance(duplicates, str):
            raise TypeError("The `duplicates` argument must be None or a string")
        elif duplicates.lower() not in ["error", "skip", "replace"]:
            raise ValueError('The `duplicates` argument must be None, "error", "skip", or "replace"')
        if not os.path.isdir(path_to_dir):
            raise IOError("The following path is not valid/not a folder, {}".format(path_to_dir))
        load_errors = list()
        folder = path_to_dir
        files = os.listdir(folder)
        for file in files:
            ext = os.path.splitext(file)[1]
            if ext is None or ext.lower() not in [".msx", ".json", ".yaml"]:
                continue
            if ext.lower() == ".msx":
                try:
                    new = MsxModel(file)
                except Exception as e:
                    logger.exception("Error reading file {}".format(os.path.join(folder, file)))
                    load_errors.append((os.path.join(folder, file), "load-failed", e))
                    continue
            elif ext.lower() == ".json":
                with open(os.path.join(folder, file), "r") as fin:
                    try:
                        new = MsxModel.from_dict(json.load(fin))
                    except Exception as e:
                        logger.exception("Error reading file {}".format(os.path.join(folder, file)))
                        load_errors.append((os.path.join(folder, file), "load-failed", e))
                        continue
            elif ext.lower() == ".yaml":
                if yaml is None:
                    logger.exception("Error reading file {}".format(os.path.join(folder, file)), exc_info=yaml_err)
                    load_errors.append((os.path.join(folder, file), "load-failed", yaml_err))
                    continue
                with open(os.path.join(folder, file), "r") as fin:
                    try:
                        new = MsxModel.from_dict(yaml.safe_load(fin))
                    except Exception as e:
                        logger.exception("Error reading file {}".format(os.path.join(folder, file)))
                        load_errors.append((os.path.join(folder, file), "load-failed", e))
                        continue
            else:  # pragma: no cover
                raise RuntimeError("This should be impossible to reach, since `ext` is checked above")
            new._orig_file = os.path.join(folder, file)
            if not new.name:
                new.name = os.path.splitext(os.path.split(file)[1])[0]
            if new.name not in self.__data:
                self.__data[new.name] = new
            else:  # this name exists in the library
                name = new.name
                if not duplicates or duplicates.lower() == "error":
                    raise KeyError('A model named "{}" already exists in the model; failed processing "{}"'.format(new.name, os.path.join(folder, file)))
                elif duplicates.lower() == "skip":
                    load_errors.append((new._orig_file, "skipped", new))
                    continue
                elif duplicates.lower() == "replace":
                    old = self.__data[name]
                    load_errors.append((old._orig_file, "replaced", old))
                    self.__data[name] = new
                else:  # pragma: no cover
                    raise RuntimeError("This should be impossible to get to, since `duplicates` is checked above")
        return load_errors

    def add_model_from_file(self, path_and_filename: str, name: str = None):
        """Load a model from a file and add it to the library

        Note, this **does not check** to see if a model exists with the same
        name, and it will automatically overwrite the existing model if one
        does exist.

        Parameters
        ----------
        path_to_file : str
            The full path and filename where the model is described.
        name : str, optional
            The name to use for the model instead of the name provided in the
            file or the filename, by default None
        """
        if not os.path.isfile(path_and_filename):
            raise IOError("The following path does not identify a file, {}".format(path_and_filename))

        ext = os.path.splitext(path_and_filename)[1]
        if ext is None or ext.lower() not in [".msx", ".json", ".yaml"]:
            raise IOError("The file is in an unknown format, {}".format(ext))
        if ext.lower() == ".msx":
            new = MsxModel(path_and_filename)
        elif ext.lower() == ".json":
            with open(path_and_filename, "r") as fin:
                new = MsxModel.from_dict(json.load(fin))
        elif ext.lower() == ".yaml":
            if yaml is None:
                raise RuntimeError("Unable to import yaml") from yaml_err
            with open(path_and_filename, "r") as fin:
                new = MsxModel.from_dict(yaml.safe_load(fin))
        else:  # pragma: no cover
            raise RuntimeError("This should be impossible to reach, since ext is checked above")
        new._orig_file = path_and_filename
        if not new.name:
            new.name = os.path.splitext(os.path.split(path_and_filename)[1])[0]
        if name is not None:
            new.name = name
        self.__data[new.name] = new

    def get_model(self, name: str) -> MsxModel:
        """Get a model from the library by model name

        Parameters
        ----------
        name : str
            Name of the model

        Returns
        -------
        MsxModel
            Model object
        """
        return self.__data[name]

    def model_name_list(self) -> List[str]:
        """Get the names of all models in the library
        
        Returns
        -------
        list of str
            list of model names
        """
        return list(self.keys())

    def __getitem__(self, __key: Any) -> Any:
        return self.__data.__getitem__(__key)

    def __setitem__(self, __key: Any, __value: Any) -> None:
        return self.__data.__setitem__(__key, __value)

    def __delitem__(self, __key: Any) -> None:
        return self.__data.__delitem__(__key)

    def __contains__(self, __key: object) -> bool:
        return self.__data.__contains__(__key)

    def __iter__(self) -> Iterator:
        return self.__data.__iter__()

    def __len__(self) -> int:
        return self.__data.__len__()

    def keys(self) -> KeysView:
        return self.__data.keys()

    def items(self) -> ItemsView:
        return self.__data.items()

    def values(self) -> ValuesView:
        return self.__data.values()

    def clear(self) -> None:
        return self.__data.clear()

1	# coding: utf-8
2	"""
3	The wntr.msx.library module includes a library of multi-species water
4	models
5
6	.. rubric:: Environment Variable
7
8	.. envvar:: WNTR_LIBRARY_PATH
9
10	This environment variable, if set, will add additional folder(s) to the
11	path to search for multi-species water quality model files,
12	(files with an ".msx", ".yaml", or ".json" file extension).
13	Multiple folders should be separated using the "``;``" character.
14	See :class:`~wntr.msx.library.ReactionLibrary` for more details.
15	"""
16
17	from __future__ import annotations	1✔
18
19	import json	1✔
20	import logging	1✔
21	import os, sys	1✔
22	from typing import Any, ItemsView, Iterator, KeysView, List, Tuple, Union, ValuesView	1✔
23
24	if sys.version_info[0:2] <= (3, 11):	1✔
25	from pkg_resources import resource_filename	1✔
26	else:
27	from importlib.resources import files	1✔
28
29	from wntr.msx.base import ExpressionType, ReactionType, SpeciesType	1✔
30	from wntr.msx.model import MsxModel	1✔
31
32	try:	1✔
33	import yaml	1✔
34
35	yaml_err = None	1✔
UNCOV 36	except ImportError as e:	×
UNCOV 37	yaml = None	×
UNCOV 38	yaml_err = e	×
39
40	PIPE = ReactionType.PIPE	1✔
41	TANK = ReactionType.TANK	1✔
42	BULK = SpeciesType.BULK	1✔
43	WALL = SpeciesType.WALL	1✔
44	RATE = ExpressionType.RATE	1✔
45	EQUIL = ExpressionType.EQUIL	1✔
46	FORMULA = ExpressionType.FORMULA	1✔
47
48	logger = logging.getLogger(__name__)	1✔
49
50
51	class MsxLibrary:	1✔
52	"""Library of multi-species water quality models
53
54	This object can be accessed and treated like a dictionary, where keys are
55	the model names and the values are the model objects.
56
57	Paths are added/processed in the following order:
58
59	1. the builtin directory of reactions,
60	2. any paths specified in the environment variable described below, with
61	directories listed first having the highest priority,
62	3. any extra paths specified in the constructor, searched in the order
63	provided.
64
65	Once created, the library paths cannot be modified. However, a model can
66	be added to the library using :meth:`add_model_from_file` or
67	:meth:`add_models_from_dir`.
68
69	"""
70
71	def __init__(self, extra_paths: List[str] = None, include_builtins=True,	1✔
72	include_envvar_paths=True, load=True) -> None:
73	"""Library of multi-species water quality models
74
75	Parameters
76	----------
77	extra_paths : list of str, optional
78	User-specified list of reaction library directories, by default
79	None
80	include_builtins : bool, optional
81	Load files built-in with WNTR, by default True
82	include_envvar_paths : bool, optional
83	Load files from the paths specified in
84	:envvar:`WNTR_LIBRARY_PATH`, by default True
85	load : bool or str, optional
86	Load the files immediately on creation, by default True.
87	If a string, then it will be passed as the `duplicates` argument
88	to the load function. See :meth:`reset_and_reload` for details.
89
90	Raises
91	------
92	TypeError
93	If `extra_paths` is not a list
94	"""
95	if extra_paths is None:	1✔
96	extra_paths = list()	1✔
UNCOV 97	elif not isinstance(extra_paths, (list, tuple)):	×
UNCOV 98	raise TypeError("Expected a list or tuple, got {}".format(type(extra_paths)))	×
99
100	self.__library_paths = list()	1✔
101
102	self.__data = dict()	1✔
103
104	if include_builtins:	1✔
105	if sys.version_info[0:2] <= (3, 11):	1✔
106	default_path = os.path.abspath(resource_filename(__name__, '.'))	1✔
107	else:
108	default_path = os.path.abspath(files('wntr.library.msx').joinpath('.'))	1✔
109	if default_path not in self.__library_paths:	1✔
110	self.__library_paths.append(default_path)	1✔
111
112	if include_envvar_paths:	1✔
113	environ_path = os.environ.get("WNTR_LIBRARY_PATH", None)	1✔
114	if environ_path:	1✔
115	lib_folders = environ_path.split(";")	×
UNCOV 116	for folder in lib_folders:	×
UNCOV 117	if folder not in self.__library_paths:	×
118	self.__library_paths.append(os.path.abspath(folder))	×
119
120	for folder in extra_paths:	1✔
UNCOV 121	self.__library_paths.append(os.path.abspath(folder))	×
122	if load:	1✔
123	if isinstance(load, str):	1✔
124	self.reset_and_reload(duplicates=load)	×
125	else:
126	self.reset_and_reload()	1✔
127
128	def __repr__(self) -> str:	1✔
UNCOV 129	if len(self.__library_paths) > 3:	×
UNCOV 130	return "{}(initial_paths=[{}, ..., {}])".format(self.__class__.__name__, repr(self.__library_paths[0]), repr(self.__library_paths[-1]))	×
UNCOV 131	return "{}({})".format(self.__class__.__name__, repr(self.__library_paths))	×
132
133	def path_list(self) -> List[str]:	1✔
134	"""List of paths used to populate the library
135
136	Returns
137	-------
138	list of str
139	Copy of the paths used to initially populate the library
140	"""
UNCOV 141	return self.__library_paths.copy()	×
142
143	def reset_and_reload(self, duplicates: str = "error") -> List[Tuple[str, str, Any]]:	1✔
144	"""Load data from the configured directories into a library of models
145
146	Note, this function is not recursive and does not 'walk' any of the
147	library's directories to look for subfolders.
148
149	The ``duplicates`` argument specifies how models that have the same name,
150	or that have the same filename if a name isn't specified in the file,
151	are handled. Warning, if two files in the same directory have models
152	with the same name, there is no guarantee which will be read in first.
153
154	The first directory processed is the builtin library data. Next, any
155	paths specified in the environment variable are searched in the order listed
156	in the variable. Finally, any directories specified by the user in the
157	constructor are processed in the order listed.
158
159	Parameters
160	----------
161	duplicates : {"error" \| "skip" \| "replace"}, optional
162	by default ``"error"``
163
164	- A value of of ``"error"`` raises an exception and stops execution.
165	- A value of ``"skip"`` will skip models with the same `name` as a model that already
166	exists in the library.
167	- A value of ``"replace"`` will replace any existing model with a model that is read
168	in that has the same `name`.
169
170	Returns
171	-------
172	(filename, reason, obj) : (str, str, Any)
173	the file not read in, the cause of the problem, and the object that was skipped/overwritten
174	or the exception raised
175
176	Raises
177	------
178	TypeError
179	If `duplicates` is not a string
180	ValueError
181	If `duplicates` is not a valid value
182	IOError
183	If `path_to_folder` is not a directory
184	KeyError
185	If `duplicates` is ``"error"`` and two models have the same name
186	"""
187	if duplicates and not isinstance(duplicates, str):	1✔
UNCOV 188	raise TypeError("The `duplicates` argument must be None or a string")	×
189	elif duplicates.lower() not in ["error", "skip", "replace"]:	1✔
UNCOV 190	raise ValueError('The `duplicates` argument must be None, "error", "skip", or "replace"')	×
191
192	load_errors = list()	1✔
193	for folder in self.__library_paths:	1✔
194	errs = self.add_models_from_dir(folder, duplicates=duplicates)	1✔
195	load_errors.extend(errs)	1✔
196	return load_errors	1✔
197
198	def add_models_from_dir(self, path_to_dir: str, duplicates: str = "error") -> List[Tuple[str, str, Union[MsxModel, Exception]]]:	1✔
199	"""Load all valid model files in a folder
200
201	Note, this function is not recursive and does not 'walk' a directory
202	tree.
203
204	Parameters
205	----------
206	path_to_dir : str
207	Path to the folder to search
208	duplicates : {"error", "skip", "replace"}, optional
209	Specifies how models that have the same name, or that have the same
210	filename if a name isn't specified in the file, are handled.
211	Warning, if two files in the same directory have models with
212	the same name, there is no guarantee which will be read in first,
213	by default ``"error"``
214	- A value of of ``"error"`` raises an exception and stops execution
215	- A value of ``"skip"`` will skip models with the same `name` as a
216	model that already exists in the library.
217	- A value of ``"replace"`` will replace any existing model with a
218	model that is read in that has the same `name`.
219
220	Returns
221	-------
222	(filename, reason, obj) : tuple[str, str, Any]
223	File not read in, the cause of the problem, and the object that was
224	skipped/overwritten or the exception raised
225
226	Raises
227	------
228	TypeError
229	If `duplicates` is not a string
230	ValueError
231	If `duplicates` is not a valid value
232	IOError
233	If `path_to_folder` is not a directory
234	KeyError
235	If `duplicates` is ``"error"`` and two models have the same name
236	"""
237	if duplicates and not isinstance(duplicates, str):	1✔
UNCOV 238	raise TypeError("The `duplicates` argument must be None or a string")	×
239	elif duplicates.lower() not in ["error", "skip", "replace"]:	1✔
UNCOV 240	raise ValueError('The `duplicates` argument must be None, "error", "skip", or "replace"')	×
241	if not os.path.isdir(path_to_dir):	1✔
UNCOV 242	raise IOError("The following path is not valid/not a folder, {}".format(path_to_dir))	×
243	load_errors = list()	1✔
244	folder = path_to_dir	1✔
245	files = os.listdir(folder)	1✔
246	for file in files:	1✔
247	ext = os.path.splitext(file)[1]	1✔
248	if ext is None or ext.lower() not in [".msx", ".json", ".yaml"]:	1✔
249	continue	1✔
250	if ext.lower() == ".msx":	1✔
UNCOV 251	try:	×
UNCOV 252	new = MsxModel(file)	×
UNCOV 253	except Exception as e:	×
UNCOV 254	logger.exception("Error reading file {}".format(os.path.join(folder, file)))	×
255	load_errors.append((os.path.join(folder, file), "load-failed", e))	×
256	continue	×
257	elif ext.lower() == ".json":	1✔
258	with open(os.path.join(folder, file), "r") as fin:	1✔
259	try:	1✔
260	new = MsxModel.from_dict(json.load(fin))	1✔
261	except Exception as e:	×
262	logger.exception("Error reading file {}".format(os.path.join(folder, file)))	×
263	load_errors.append((os.path.join(folder, file), "load-failed", e))	×
264	continue	×
265	elif ext.lower() == ".yaml":	×
266	if yaml is None:	×
267	logger.exception("Error reading file {}".format(os.path.join(folder, file)), exc_info=yaml_err)	×
268	load_errors.append((os.path.join(folder, file), "load-failed", yaml_err))	×
269	continue	×
270	with open(os.path.join(folder, file), "r") as fin:	×
UNCOV 271	try:	×
UNCOV 272	new = MsxModel.from_dict(yaml.safe_load(fin))	×
UNCOV 273	except Exception as e:	×
UNCOV 274	logger.exception("Error reading file {}".format(os.path.join(folder, file)))	×
275	load_errors.append((os.path.join(folder, file), "load-failed", e))	×
UNCOV 276	continue	×
277	else: # pragma: no cover
278	raise RuntimeError("This should be impossible to reach, since `ext` is checked above")
279	new._orig_file = os.path.join(folder, file)	1✔
280	if not new.name:	1✔
281	new.name = os.path.splitext(os.path.split(file)[1])[0]	×
282	if new.name not in self.__data:	1✔
283	self.__data[new.name] = new	1✔
284	else: # this name exists in the library
285	name = new.name	×
286	if not duplicates or duplicates.lower() == "error":	×
287	raise KeyError('A model named "{}" already exists in the model; failed processing "{}"'.format(new.name, os.path.join(folder, file)))	×
288	elif duplicates.lower() == "skip":	×
UNCOV 289	load_errors.append((new._orig_file, "skipped", new))	×
UNCOV 290	continue	×
UNCOV 291	elif duplicates.lower() == "replace":	×
UNCOV 292	old = self.__data[name]	×
UNCOV 293	load_errors.append((old._orig_file, "replaced", old))	×
UNCOV 294	self.__data[name] = new	×
295	else: # pragma: no cover
296	raise RuntimeError("This should be impossible to get to, since `duplicates` is checked above")
297	return load_errors	1✔
298
299	def add_model_from_file(self, path_and_filename: str, name: str = None):	1✔
300	"""Load a model from a file and add it to the library
301
302	Note, this does not check to see if a model exists with the same
303	name, and it will automatically overwrite the existing model if one
304	does exist.
305
306	Parameters
307	----------
308	path_to_file : str
309	The full path and filename where the model is described.
310	name : str, optional
311	The name to use for the model instead of the name provided in the
312	file or the filename, by default None
313	"""
314	if not os.path.isfile(path_and_filename):	×
315	raise IOError("The following path does not identify a file, {}".format(path_and_filename))	×
316
317	ext = os.path.splitext(path_and_filename)[1]	×
318	if ext is None or ext.lower() not in [".msx", ".json", ".yaml"]:	×
319	raise IOError("The file is in an unknown format, {}".format(ext))	×
320	if ext.lower() == ".msx":	×
321	new = MsxModel(path_and_filename)	×
322	elif ext.lower() == ".json":	×
323	with open(path_and_filename, "r") as fin:	×
UNCOV 324	new = MsxModel.from_dict(json.load(fin))	×
UNCOV 325	elif ext.lower() == ".yaml":	×
326	if yaml is None:	×
327	raise RuntimeError("Unable to import yaml") from yaml_err	×
328	with open(path_and_filename, "r") as fin:	×
329	new = MsxModel.from_dict(yaml.safe_load(fin))	×
330	else: # pragma: no cover
331	raise RuntimeError("This should be impossible to reach, since ext is checked above")
UNCOV 332	new._orig_file = path_and_filename	×
UNCOV 333	if not new.name:	×
UNCOV 334	new.name = os.path.splitext(os.path.split(path_and_filename)[1])[0]	×
UNCOV 335	if name is not None:	×
UNCOV 336	new.name = name	×
UNCOV 337	self.__data[new.name] = new	×
338
339	def get_model(self, name: str) -> MsxModel:	1✔
340	"""Get a model from the library by model name
341
342	Parameters
343	----------
344	name : str
345	Name of the model
346
347	Returns
348	-------
349	MsxModel
350	Model object
351	"""
352	return self.__data[name]	1✔
353
354	def model_name_list(self) -> List[str]:	1✔
355	"""Get the names of all models in the library
356
357	Returns
358	-------
359	list of str
360	list of model names
361	"""
362	return list(self.keys())	×
363
364	def __getitem__(self, __key: Any) -> Any:	1✔
365	return self.__data.__getitem__(__key)	×
366
367	def __setitem__(self, __key: Any, __value: Any) -> None:	1✔
368	return self.__data.__setitem__(__key, __value)	×
369
370	def __delitem__(self, __key: Any) -> None:	1✔
371	return self.__data.__delitem__(__key)	×
372
373	def __contains__(self, __key: object) -> bool:	1✔
374	return self.__data.__contains__(__key)	×
375
376	def __iter__(self) -> Iterator:	1✔
377	return self.__data.__iter__()	×
378
379	def __len__(self) -> int:	1✔
380	return self.__data.__len__()	×
381
382	def keys(self) -> KeysView:	1✔
383	return self.__data.keys()	×
384
385	def items(self) -> ItemsView:	1✔
386	return self.__data.items()	×
387
388	def values(self) -> ValuesView:	1✔
UNCOV 389	return self.__data.values()	×
390
391	def clear(self) -> None:	1✔
UNCOV 392	return self.__data.clear()	×

USEPA / WNTR / 19307136919

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous