17441743838

Committed 03 Sep 2025 05:51PM UTC coverage: 83.305% (+0.01%) from 83.291%

Build # 17441743838

Build Type

Pull #405

github

Committed by

quaquel

Commit Message

Update ci.yml

Pull Request Pull Request #405: WIP

Run Details

50 of 53 new or added lines in 8 files covered. (94.34%)

13 existing lines in 1 file now uncovered.

7245 of 8697 relevant lines covered (83.3%)

0.83 hits per line

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

85.92

/ema_workbench/em_framework/model.py

"""This module specifies the abstract base class for interfacing with models.

Any model that is to be controlled from the workbench is controlled via
an instance of an extension of this abstract base class.

"""

import operator
import os
from collections import defaultdict

from ..util import EMAError
from ..util.ema_logging import get_module_logger, method_logger
from .outcomes import AbstractOutcome
from .parameters import CategoricalParameter, Constant, Parameter
from .points import ExperimentReplication
from .util import NamedObject, NamedObjectMapDescriptor, combine

# Created on 23 dec. 2010
#
# .. codeauthor:: jhkwakkel <j.h.kwakkel (at) tudelft (dot) nl>
#

__all__ = [
    "AbstractModel",
    "FileModel",
    "Model",
    "Replicator",
    "ReplicatorModel",
    "SingleReplication",
]
_logger = get_module_logger(__name__)


class AbstractModel(NamedObject):
    """Abstract base class for models.

    This is an abstract base class
    and cannot be used directly.

    Attributes:
    ----------
    uncertainties : list
                    list of parameter instances
    levers : list
             list of parameter instances
    outcomes : list
               list of outcome instances
    name : str
           alphanumerical name of model structure interface
    output : dict
             this should be a dict with the names of the outcomes as
             key

    When extending this class :meth:`model_init` and
    :meth:`run_model` have to be implemented.

    """

    @property
    def outcomes_output(self):
        """Getter for outcomes output."""
        return self._outcomes_output

    @outcomes_output.setter
    def outcomes_output(self, outputs):
        """Setter for outcomes output."""
        for outcome in self.outcomes:
            data = [outputs[var] for var in outcome.variable_name]
            self._outcomes_output[outcome.name] = outcome.process(data)

    @property
    def output_variables(self):
        """Getter for output variables."""
        if self._output_variables is None:
            self._output_variables = [
                var for o in self.outcomes for var in o.variable_name
            ]

        return self._output_variables

    uncertainties = NamedObjectMapDescriptor(Parameter)
    levers = NamedObjectMapDescriptor(Parameter)
    outcomes = NamedObjectMapDescriptor(AbstractOutcome)
    constants = NamedObjectMapDescriptor(Constant)

    def __init__(self, name):
        """Interface to the model.

        Parameters
        ----------
        name : str
               name of the model. The name should be a valid python identifier

        Raises:
        ------
        EMAError if name is not a valid python identifier

        """
        if not name.isidentifier():
            raise EMAError(
                f"'{name}' is not a valid Python identifier. Starting from version 3.0 of the EMAworkbench,"
                 " the name of model must be valid python identifiers"
            )

        super().__init__(name)

        self._output_variables = None
        self._outcomes_output = {}
        self._constraints_output = {}
        self.policy = None

    @method_logger(__name__)
    def model_init(self, policy):
        """Method called to initialize the model.

        Parameters
        ----------
        policy : dict
                 policy to be run.

        """
        self.policy = policy

        remove = []
        for key, value in policy.items():
            if hasattr(self, key):
                setattr(self, key, value)
                remove.append(key)

        for k in remove:
            del policy[k]

    @method_logger(__name__)
    def _transform(self, sampled_parameters, parameters):
        """Helper method to transform the sampled parameters."""
        if not parameters:
            # no parameters defined, so nothing to transform, mainly
            # useful for manual specification of scenario /  policy
            # without having to define uncertainties / levers
            return

        temp = {}
        for par in parameters:
            # only keep uncertainties that exist in this model
            try:
                value = sampled_parameters[par.name]
            except KeyError:
                if par.default is not None:
                    value = par.default
                else:
                    _logger.debug(
                        f"parameter {par.name} not found in sampled_parameters"
                    )
                    continue

            multivalue = False
            if isinstance(par, CategoricalParameter) and par.multivalue:
                multivalue = True
                values = value

            # translate uncertainty name to variable name
            for i, varname in enumerate(par.variable_name):
                # a bit hacky implementation, investigate some kind of
                # zipping of variable_names and values
                if multivalue:
                    value = values[i]

                temp[varname] = value

        sampled_parameters.data = temp

    @method_logger(__name__)
    def run_model(self, scenario, policy, constants):
        """Method for running an instantiated model structure.

        Parameters
        ----------
        scenario : Scenario instance
        policy : Policy instance

        """
        if not self.initialized(policy):
            self.model_init(policy)

        self._transform(scenario, self.uncertainties)
        self._transform(policy, self.levers)
        self._transform(constants, self.constants)

    @method_logger(__name__)
    def initialized(self, policy):
        """Check if model has been initialized.

        Parameters
        ----------
        policy : a Policy instance

        """
        try:
            return self.policy.name == policy.name
        except AttributeError:
            return False

    @method_logger(__name__)
    def reset_model(self):
        """Method for resetting the model to its initial state.

        The default implementation only sets the outputs to an empty dict.

        """
        self._outcome_output = {}
        self._constraints_output = {}

    @method_logger(__name__)
    def cleanup(self):
        """Hook for performing cleanup after all experiments have completed.

        This model is called after finishing all the experiments, but
        just prior to returning the results. This method gives a hook for
        doing any cleanup, such as closing applications.

        In case of running in parallel, this method is called during
        the cleanup of the pool, just prior to removing the temporary
        directories.

        """

    def as_dict(self):
        """Returns a dict representation of the model."""

        def join_attr(field):
            joined = ", ".join(
                [
                    repr(entry)
                    for entry in sorted(field, key=operator.attrgetter("name"))
                ]
            )
            return f"[{joined}]"

        model_spec = {}

        klass = self.__class__.__name__
        name = self.name

        uncs = ""
        for uncertainty in self.uncertainties:
            uncs += "\n" + repr(uncertainty)

        model_spec["class"] = klass
        model_spec["name"] = name
        model_spec["uncertainties"] = join_attr(self.uncertainties)
        model_spec["outcomes"] = join_attr(self.outcomes)
        model_spec["constants"] = join_attr(self.constants)

        return model_spec


class MyDict(dict):
    """Hacky helper class."""
    # bit of a dirty hack to be able to assign attributes to a dict
    # in a replication context


class Replicator(AbstractModel):
    """Base class for a model where experiments are run for multiple replications."""
    @property
    def replications(self):
        """Getter for replications."""
        return self._replications

    @replications.setter
    def replications(self, replications):
        """Setter for replications."""
        # int
        if isinstance(replications, int):
            # TODO:: use a repeating generator instead

            self._replications = [MyDict() for _ in range(replications)]
            self.nreplications = replications
        elif isinstance(replications, list):
            # should we check if all are dict?
            # TODO:: this needs testing
            self._replications = [MyDict(**entry) for entry in replications]
            self.nreplications = len(replications)
        else:
            raise TypeError(
                f"Replications should be int or list, not {type(replications)}"
            )

    @method_logger(__name__)
    def run_model(self, scenario, policy, constants):
        """Method for running an instantiated model structure.

        Parameters
        ----------
        scenario : Scenario instance
        policy : Policy instance
        constants : ??

        """
        super().run_model(scenario, policy, constants)

        constants = {c.name: c.value for c in self.constants}
        outputs = defaultdict(list)
        partial_experiment = combine(scenario, self.policy, constants)

        for i, rep in enumerate(self.replications):
            _logger.debug(f"replication {i} of {self.nreplications}")
            rep.id = i
            experiment = ExperimentReplication(scenario, self.policy, constants, rep)
            output = self.run_experiment(experiment)
            for key, value in output.items():
                outputs[key].append(value)

        self.outcomes_output = outputs

        # perhaps set constraints with the outcomes instead
        # this avoids double processing, it also means that
        # each constraint needs to apply to an actual outcome
        self.constraints_output = (partial_experiment, self.outcomes_output)


class SingleReplication(AbstractModel):
    """Base class for models that require only a single replication."""

    @method_logger(__name__)
    def run_model(self, scenario, policy, constants):
        """Method for running an instantiated model structure.

        Parameters
        ----------
        scenario : Scenario instance
        policy : Policy instance
        constants : ??

        """
        super().run_model(scenario, policy, constants)
        experiment = ExperimentReplication(scenario, self.policy, constants)

        outputs = self.run_experiment(experiment)

        self.outcomes_output = outputs
        self.constraints_output = (experiment, self.outcomes_output)


class BaseModel(AbstractModel):
    """Generic class for working with models implemented as a Python callable.

    Parameters
    ----------
    name : str
    function : callable
               a function with each of the uncertain parameters as a
               keyword argument

    Attributes:
    ----------
    uncertainties : listlike
                    list of parameter
    levers : listlike
             list of parameter instances
    outcomes : listlike
               list of outcome instances
    name : str
           alphanumerical name of model structure interface
    output : dict
             this should be a dict with the names of the outcomes as key

    """

    def __init__(self, name, function=None):
        """Init."""
        super().__init__(name)

        if not callable(function):
            raise ValueError("Function should be callable")

        self.function = function

    @method_logger(__name__)
    def run_experiment(self, experiment):
        """Method for running an instantiated model structure.

        Parameters
        ----------
        experiment : dict like

        """
        model_output = self.function(**experiment)

        # TODO: might it be possible to somehow abstract this
        # perhaps expose a get_data on modelInterface?
        # different connectors can than implement only this
        # get method
        results = {}
        for i, variable in enumerate(self.output_variables):
            try:
                value = model_output[variable]
            except KeyError:
                _logger.warning(f"variable {variable} not found in model output")
                value = None
            except TypeError:
                value = model_output[i]
            results[variable] = value
        return results

    def as_dict(self):
        """Returns a dict representation of the model."""
        model_specs = super().as_dict()
        model_specs["function"] = self.function
        return model_specs


class WorkingDirectoryModel(AbstractModel):
    """Base class for a model that needs its dedicated working directory."""

    @property
    def working_directory(self):
        """Getter for working directory."""
        return self._working_directory

    @working_directory.setter
    def working_directory(self, path):
        """Setter for working directory."""
        wd = os.path.abspath(path)
        _logger.debug(f"setting working directory to {wd}")
        self._working_directory = wd

    def __init__(self, name, wd=None):
        """Interface to the model.

        Parameters
        ----------
        name : str
               name of the modelInterface. The name should contain only
               alpha-numerical characters.
        working_directory : str
                            working_directory for the model.

        Raises:
        ------
        EMAError
            if name contains non alpha-numerical characters
        ValueError
            if working_directory does not exist
        """
        super().__init__(name)
        self.working_directory = wd

        if not os.path.exists(self.working_directory):
            raise ValueError(
                f"Working directory {self.working_directory} does not exist"
            )

    def as_dict(self):
        """Returns a dict representation of the model."""
        model_specs = super().as_dict()
        model_specs["working_directory"] = self.working_directory
        return model_specs


class FileModel(WorkingDirectoryModel):
    """Base class for a model that uses underlying files."""

    @property
    def working_directory(self):
        """Getter for working directory."""
        return self._working_directory

    @working_directory.setter
    def working_directory(self, path):
        """Setter for working directory."""
        wd = os.path.abspath(path)
        _logger.debug(f"setting working directory to {wd}")
        self._working_directory = wd

    def __init__(self, name, wd=None, model_file=None):
        """Interface to the model.

        Parameters
        ----------
        name : str
               name of the modelInterface. The name should contain only
               alpha-numerical characters.
        working_directory : str
                            working_directory for the model.
        model_file  : str
                     the name of the model file

        Raises:
        ------
        EMAError
            if name contains non alpha-numerical characters
        ValueError
            if model_file cannot be found
            if the current working directory is the same as the working
            directory of the model


        The parallelization code offered by the workbench implicitly assumes
        that the current working directory (i.e., os.getwcd()) is different
        from the directory in which the model resides.

        It is best practice to place the model files in a subdirectory of
        the folder within which the file resides used for performing
        experiments.

        """
        super().__init__(name, wd=wd)

        path_to_file = os.path.join(self.working_directory, model_file)
        if not os.path.isfile(path_to_file):
            raise ValueError(
                f"Cannot find model file: {model_file},\nPath to file: {path_to_file}"
            )

        if os.getcwd() == self.working_directory:
            raise ValueError(
                f"The working directory of the model cannot be the same as the current working directory\nBoth are: {self.working_directory}"
            )

        self.model_file = model_file

    def as_dict(self):
        """Returns a dict representation of the model."""
        model_specs = super().as_dict()
        model_specs["model_file"] = self.model_file
        return model_specs


class Model(SingleReplication, BaseModel):
    """Default model class for python callables that are run once per experiment."""


class ReplicatorModel(Replicator, BaseModel):
    """Default model class for python callables that are run for multiple replications per experiment."""

1	"""This module specifies the abstract base class for interfacing with models.
2
3	Any model that is to be controlled from the workbench is controlled via
4	an instance of an extension of this abstract base class.
5
6	"""
7
8	import operator	1✔
9	import os	1✔
10	from collections import defaultdict	1✔
11
12	from ..util import EMAError	1✔
13	from ..util.ema_logging import get_module_logger, method_logger	1✔
14	from .outcomes import AbstractOutcome	1✔
15	from .parameters import CategoricalParameter, Constant, Parameter	1✔
16	from .points import ExperimentReplication	1✔
17	from .util import NamedObject, NamedObjectMapDescriptor, combine	1✔
18
19	# Created on 23 dec. 2010
20	#
21	# .. codeauthor:: jhkwakkel <j.h.kwakkel (at) tudelft (dot) nl>
22	#
23
24	__all__ = [	1✔
25	"AbstractModel",
26	"FileModel",
27	"Model",
28	"Replicator",
29	"ReplicatorModel",
30	"SingleReplication",
31	]
32	_logger = get_module_logger(__name__)	1✔
33
34
35	class AbstractModel(NamedObject):	1✔
36	"""Abstract base class for models.
37
38	This is an abstract base class
39	and cannot be used directly.
40
41	Attributes:
42	----------
43	uncertainties : list
44	list of parameter instances
45	levers : list
46	list of parameter instances
47	outcomes : list
48	list of outcome instances
49	name : str
50	alphanumerical name of model structure interface
51	output : dict
52	this should be a dict with the names of the outcomes as
53	key
54
55	When extending this class :meth:`model_init` and
56	:meth:`run_model` have to be implemented.
57
58	"""
59
60	@property	1✔
61	def outcomes_output(self):	1✔
62	"""Getter for outcomes output."""
63	return self._outcomes_output	1✔
64
65	@outcomes_output.setter	1✔
66	def outcomes_output(self, outputs):	1✔
67	"""Setter for outcomes output."""
68	for outcome in self.outcomes:	1✔
69	data = [outputs[var] for var in outcome.variable_name]	1✔
70	self._outcomes_output[outcome.name] = outcome.process(data)	1✔
71
72	@property	1✔
73	def output_variables(self):	1✔
74	"""Getter for output variables."""
75	if self._output_variables is None:	1✔
76	self._output_variables = [	1✔
77	var for o in self.outcomes for var in o.variable_name
78	]
79
80	return self._output_variables	1✔
81
82	uncertainties = NamedObjectMapDescriptor(Parameter)	1✔
83	levers = NamedObjectMapDescriptor(Parameter)	1✔
84	outcomes = NamedObjectMapDescriptor(AbstractOutcome)	1✔
85	constants = NamedObjectMapDescriptor(Constant)	1✔
86
87	def __init__(self, name):	1✔
88	"""Interface to the model.
89
90	Parameters
91	----------
92	name : str
93	name of the model. The name should be a valid python identifier
94
95	Raises:
96	------
97	EMAError if name is not a valid python identifier
98
99	"""
100	if not name.isidentifier():	1✔
101	raise EMAError(	1✔
102	f"'{name}' is not a valid Python identifier. Starting from version 3.0 of the EMAworkbench,"
103	" the name of model must be valid python identifiers"
104	)
105
106	super().__init__(name)	1✔
107
108	self._output_variables = None	1✔
109	self._outcomes_output = {}	1✔
110	self._constraints_output = {}	1✔
111	self.policy = None	1✔
112
113	@method_logger(__name__)	1✔
114	def model_init(self, policy):	1✔
115	"""Method called to initialize the model.
116
117	Parameters
118	----------
119	policy : dict
120	policy to be run.
121
122	"""
123	self.policy = policy	1✔
124
125	remove = []	1✔
126	for key, value in policy.items():	1✔
127	if hasattr(self, key):	1✔
128	setattr(self, key, value)	1✔
129	remove.append(key)	1✔
130
131	for k in remove:	1✔
132	del policy[k]	1✔
133
134	@method_logger(__name__)	1✔
135	def _transform(self, sampled_parameters, parameters):	1✔
136	"""Helper method to transform the sampled parameters."""
137	if not parameters:	1✔
138	# no parameters defined, so nothing to transform, mainly
139	# useful for manual specification of scenario / policy
140	# without having to define uncertainties / levers
141	return	1✔
142
143	temp = {}	1✔
144	for par in parameters:	1✔
145	# only keep uncertainties that exist in this model
146	try:	1✔
147	value = sampled_parameters[par.name]	1✔
UNCOV 148	except KeyError:	×
149	if par.default is not None:	×
150	value = par.default	×
151	else:
UNCOV 152	_logger.debug(	×
153	f"parameter {par.name} not found in sampled_parameters"
154	)
UNCOV 155	continue	×
156
157	multivalue = False	1✔
158	if isinstance(par, CategoricalParameter) and par.multivalue:	1✔
159	multivalue = True	1✔
160	values = value	1✔
161
162	# translate uncertainty name to variable name
163	for i, varname in enumerate(par.variable_name):	1✔
164	# a bit hacky implementation, investigate some kind of
165	# zipping of variable_names and values
166	if multivalue:	1✔
167	value = values[i]	1✔
168
169	temp[varname] = value	1✔
170
171	sampled_parameters.data = temp	1✔
172
173	@method_logger(__name__)	1✔
174	def run_model(self, scenario, policy, constants):	1✔
175	"""Method for running an instantiated model structure.
176
177	Parameters
178	----------
179	scenario : Scenario instance
180	policy : Policy instance
181
182	"""
183	if not self.initialized(policy):	1✔
184	self.model_init(policy)	1✔
185
186	self._transform(scenario, self.uncertainties)	1✔
187	self._transform(policy, self.levers)	1✔
188	self._transform(constants, self.constants)	1✔
189
190	@method_logger(__name__)	1✔
191	def initialized(self, policy):	1✔
192	"""Check if model has been initialized.
193
194	Parameters
195	----------
196	policy : a Policy instance
197
198	"""
199	try:	1✔
200	return self.policy.name == policy.name	1✔
201	except AttributeError:	1✔
202	return False	1✔
203
204	@method_logger(__name__)	1✔
205	def reset_model(self):	1✔
206	"""Method for resetting the model to its initial state.
207
208	The default implementation only sets the outputs to an empty dict.
209
210	"""
UNCOV 211	self._outcome_output = {}	×
212	self._constraints_output = {}	×
213
214	@method_logger(__name__)	1✔
215	def cleanup(self):	1✔
216	"""Hook for performing cleanup after all experiments have completed.
217
218	This model is called after finishing all the experiments, but
219	just prior to returning the results. This method gives a hook for
220	doing any cleanup, such as closing applications.
221
222	In case of running in parallel, this method is called during
223	the cleanup of the pool, just prior to removing the temporary
224	directories.
225
226	"""
227
228	def as_dict(self):	1✔
229	"""Returns a dict representation of the model."""
230
231	def join_attr(field):	1✔
232	joined = ", ".join(	1✔
233	[
234	repr(entry)
235	for entry in sorted(field, key=operator.attrgetter("name"))
236	]
237	)
238	return f"[{joined}]"	1✔
239
240	model_spec = {}	1✔
241
242	klass = self.__class__.__name__	1✔
243	name = self.name	1✔
244
245	uncs = ""	1✔
246	for uncertainty in self.uncertainties:	1✔
247	uncs += "\n" + repr(uncertainty)	1✔
248
249	model_spec["class"] = klass	1✔
250	model_spec["name"] = name	1✔
251	model_spec["uncertainties"] = join_attr(self.uncertainties)	1✔
252	model_spec["outcomes"] = join_attr(self.outcomes)	1✔
253	model_spec["constants"] = join_attr(self.constants)	1✔
254
255	return model_spec	1✔
256
257
258	class MyDict(dict):	1✔
259	"""Hacky helper class."""
260	# bit of a dirty hack to be able to assign attributes to a dict
261	# in a replication context
262
263
264	class Replicator(AbstractModel):	1✔
265	"""Base class for a model where experiments are run for multiple replications."""
266	@property	1✔
267	def replications(self):	1✔
268	"""Getter for replications."""
269	return self._replications	1✔
270
271	@replications.setter	1✔
272	def replications(self, replications):	1✔
273	"""Setter for replications."""
274	# int
275	if isinstance(replications, int):	1✔
276	# TODO:: use a repeating generator instead
277
278	self._replications = [MyDict() for _ in range(replications)]	1✔
279	self.nreplications = replications	1✔
UNCOV 280	elif isinstance(replications, list):	×
281	# should we check if all are dict?
282	# TODO:: this needs testing
UNCOV 283	self._replications = [MyDict(**entry) for entry in replications]	×
UNCOV 284	self.nreplications = len(replications)	×
285	else:
286	raise TypeError(	×
287	f"Replications should be int or list, not {type(replications)}"
288	)
289
290	@method_logger(__name__)	1✔
291	def run_model(self, scenario, policy, constants):	1✔
292	"""Method for running an instantiated model structure.
293
294	Parameters
295	----------
296	scenario : Scenario instance
297	policy : Policy instance
298	constants : ??
299
300	"""
301	super().run_model(scenario, policy, constants)	1✔
302
303	constants = {c.name: c.value for c in self.constants}	1✔
304	outputs = defaultdict(list)	1✔
305	partial_experiment = combine(scenario, self.policy, constants)	1✔
306
307	for i, rep in enumerate(self.replications):	1✔
308	_logger.debug(f"replication {i} of {self.nreplications}")	1✔
309	rep.id = i	1✔
310	experiment = ExperimentReplication(scenario, self.policy, constants, rep)	1✔
311	output = self.run_experiment(experiment)	1✔
312	for key, value in output.items():	1✔
313	outputs[key].append(value)	1✔
314
315	self.outcomes_output = outputs	1✔
316
317	# perhaps set constraints with the outcomes instead
318	# this avoids double processing, it also means that
319	# each constraint needs to apply to an actual outcome
320	self.constraints_output = (partial_experiment, self.outcomes_output)	1✔
321
322
323	class SingleReplication(AbstractModel):	1✔
324	"""Base class for models that require only a single replication."""
325
326	@method_logger(__name__)	1✔
327	def run_model(self, scenario, policy, constants):	1✔
328	"""Method for running an instantiated model structure.
329
330	Parameters
331	----------
332	scenario : Scenario instance
333	policy : Policy instance
334	constants : ??
335
336	"""
337	super().run_model(scenario, policy, constants)	1✔
338	experiment = ExperimentReplication(scenario, self.policy, constants)	1✔
339
340	outputs = self.run_experiment(experiment)	1✔
341
342	self.outcomes_output = outputs	1✔
343	self.constraints_output = (experiment, self.outcomes_output)	1✔
344
345
346	class BaseModel(AbstractModel):	1✔
347	"""Generic class for working with models implemented as a Python callable.
348
349	Parameters
350	----------
351	name : str
352	function : callable
353	a function with each of the uncertain parameters as a
354	keyword argument
355
356	Attributes:
357	----------
358	uncertainties : listlike
359	list of parameter
360	levers : listlike
361	list of parameter instances
362	outcomes : listlike
363	list of outcome instances
364	name : str
365	alphanumerical name of model structure interface
366	output : dict
367	this should be a dict with the names of the outcomes as key
368
369	"""
370
371	def __init__(self, name, function=None):	1✔
372	"""Init."""
373	super().__init__(name)	1✔
374
375	if not callable(function):	1✔
UNCOV 376	raise ValueError("Function should be callable")	×
377
378	self.function = function	1✔
379
380	@method_logger(__name__)	1✔
381	def run_experiment(self, experiment):	1✔
382	"""Method for running an instantiated model structure.
383
384	Parameters
385	----------
386	experiment : dict like
387
388	"""
389	model_output = self.function(**experiment)	1✔
390
391	# TODO: might it be possible to somehow abstract this
392	# perhaps expose a get_data on modelInterface?
393	# different connectors can than implement only this
394	# get method
395	results = {}	1✔
396	for i, variable in enumerate(self.output_variables):	1✔
397	try:	1✔
398	value = model_output[variable]	1✔
UNCOV 399	except KeyError:	×
UNCOV 400	_logger.warning(f"variable {variable} not found in model output")	×
401	value = None	×
402	except TypeError:	×
403	value = model_output[i]	×
404	results[variable] = value	1✔
405	return results	1✔
406
407	def as_dict(self):	1✔
408	"""Returns a dict representation of the model."""
409	model_specs = super().as_dict()	1✔
410	model_specs["function"] = self.function	1✔
411	return model_specs	1✔
412
413
414	class WorkingDirectoryModel(AbstractModel):	1✔
415	"""Base class for a model that needs its dedicated working directory."""
416
417	@property	1✔
418	def working_directory(self):	1✔
419	"""Getter for working directory."""
420	return self._working_directory	×
421
422	@working_directory.setter	1✔
423	def working_directory(self, path):	1✔
424	"""Setter for working directory."""
425	wd = os.path.abspath(path)	×
UNCOV 426	_logger.debug(f"setting working directory to {wd}")	×
427	self._working_directory = wd	×
428
429	def __init__(self, name, wd=None):	1✔
430	"""Interface to the model.
431
432	Parameters
433	----------
434	name : str
435	name of the modelInterface. The name should contain only
436	alpha-numerical characters.
437	working_directory : str
438	working_directory for the model.
439
440	Raises:
441	------
442	EMAError
443	if name contains non alpha-numerical characters
444	ValueError
445	if working_directory does not exist
446	"""
447	super().__init__(name)	1✔
448	self.working_directory = wd	1✔
449
450	if not os.path.exists(self.working_directory):	1✔
UNCOV 451	raise ValueError(	×
452	f"Working directory {self.working_directory} does not exist"
453	)
454
455	def as_dict(self):	1✔
456	"""Returns a dict representation of the model."""
457	model_specs = super().as_dict()	×
458	model_specs["working_directory"] = self.working_directory	×
459	return model_specs	×
460
461
462	class FileModel(WorkingDirectoryModel):	1✔
463	"""Base class for a model that uses underlying files."""
464
465	@property	1✔
466	def working_directory(self):	1✔
467	"""Getter for working directory."""
468	return self._working_directory	1✔
469
470	@working_directory.setter	1✔
471	def working_directory(self, path):	1✔
472	"""Setter for working directory."""
473	wd = os.path.abspath(path)	1✔
474	_logger.debug(f"setting working directory to {wd}")	1✔
475	self._working_directory = wd	1✔
476
477	def __init__(self, name, wd=None, model_file=None):	1✔
478	"""Interface to the model.
479
480	Parameters
481	----------
482	name : str
483	name of the modelInterface. The name should contain only
484	alpha-numerical characters.
485	working_directory : str
486	working_directory for the model.
487	model_file : str
488	the name of the model file
489
490	Raises:
491	------
492	EMAError
493	if name contains non alpha-numerical characters
494	ValueError
495	if model_file cannot be found
496	if the current working directory is the same as the working
497	directory of the model
498
499
500	The parallelization code offered by the workbench implicitly assumes
501	that the current working directory (i.e., os.getwcd()) is different
502	from the directory in which the model resides.
503
504	It is best practice to place the model files in a subdirectory of
505	the folder within which the file resides used for performing
506	experiments.
507
508	"""
509	super().__init__(name, wd=wd)	1✔
510
511	path_to_file = os.path.join(self.working_directory, model_file)	1✔
512	if not os.path.isfile(path_to_file):	1✔
513	raise ValueError(	1✔
514	f"Cannot find model file: {model_file},\nPath to file: {path_to_file}"
515	)
516
517	if os.getcwd() == self.working_directory:	1✔
UNCOV 518	raise ValueError(	×
519	f"The working directory of the model cannot be the same as the current working directory\nBoth are: {self.working_directory}"
520	)
521
522	self.model_file = model_file	1✔
523
524	def as_dict(self):	1✔
525	"""Returns a dict representation of the model."""
526	model_specs = super().as_dict()	×
527	model_specs["model_file"] = self.model_file	×
528	return model_specs	×
529
530
531	class Model(SingleReplication, BaseModel):	1✔
532	"""Default model class for python callables that are run once per experiment."""
533
534
535	class ReplicatorModel(Replicator, BaseModel):	1✔
536	"""Default model class for python callables that are run for multiple replications per experiment."""

quaquel / EMAworkbench / 17441743838

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