11245969734

Committed 09 Oct 2024 12:06AM UTC coverage: 83.919% (-0.004%) from 83.923%

Build # 11245969734

Build Type

push

github

Committed by

kmantel

Commit Message

conditions: explicitly specify pnl Conditions in __all__

psyneulink-wrapped versions of graph_scheduler Conditions are
dynamically generated. This prevented some IDEs from recognizing
their names.

Run Details

11711 of 15067 branches covered (77.73%)

Branch coverage included in aggregate %.

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

19 existing lines in 2 files now uncovered.

32172 of 37225 relevant lines covered (86.43%)

0.86 hits per line

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

86.97

/psyneulink/core/scheduling/condition.py

# Princeton University licenses this file to You under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.  You may obtain a copy of the License at:
#     http://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software distributed under the License is distributed
# on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and limitations under the License.


# ********************************************* Condition **************************************************************

import collections
import inspect
import numbers
import warnings

import dill
import graph_scheduler
import numpy as np

from psyneulink.core.globals.context import handle_external_context
from psyneulink.core.globals.mdf import MDFSerializable
from psyneulink.core.globals.keywords import MODEL_SPEC_ID_TYPE, comparison_operators
from psyneulink.core.globals.parameters import parse_context
from psyneulink.core.globals.utilities import parse_valid_identifier, toposort_key


__all__ = [  # noqa: F822  (dynamically generated)
    'AbsoluteCondition', 'AddEdgeTo', 'AfterCall',
    'AfterConsiderationSetExecution', 'AfterEnvironmentSequence',
    'AfterEnvironmentStateUpdate', 'AfterNCalls', 'AfterNCallsCombined',
    'AfterNConsiderationSetExecutions', 'AfterNEnvironmentSequences',
    'AfterNEnvironmentStateUpdates', 'AfterNPasses', 'AfterNRuns',
    'AfterNTimeSteps', 'AfterNTrials', 'AfterNode', 'AfterNodes',
    'AfterPass', 'AfterRun', 'AfterTimeStep', 'AfterTrial', 'All',
    'AllHaveRun', 'Always', 'And', 'Any', 'AtConsiderationSetExecution',
    'AtEnvironmentSequence', 'AtEnvironmentSequenceNStart',
    'AtEnvironmentSequenceStart', 'AtEnvironmentStateUpdate',
    'AtEnvironmentStateUpdateNStart', 'AtEnvironmentStateUpdateStart',
    'AtNCalls', 'AtPass', 'AtRun', 'AtRunNStart', 'AtRunStart',
    'AtTimeStep', 'AtTrial', 'AtTrialNStart', 'AtTrialStart',
    'BeforeConsiderationSetExecution', 'BeforeEnvironmentStateUpdate',
    'BeforeNCalls', 'BeforeNode', 'BeforeNodes', 'BeforePass',
    'BeforeTimeStep', 'BeforeTrial', 'CompositeCondition', 'Condition',
    'ConditionBase', 'ConditionError', 'ConditionSet',
    'CustomGraphStructureCondition', 'EveryNCalls', 'EveryNPasses',
    'GraphStructureCondition', 'JustRan', 'NWhen', 'Never', 'Not',
    'Operation', 'Or', 'RemoveEdgeFrom', 'Threshold', 'TimeInterval',
    'TimeTermination', 'WhenFinished', 'WhenFinishedAll',
    'WhenFinishedAny', 'When', 'While', 'WhileNot', 'WithNode',
]


# avoid restricting graph_scheduler versions for this code
# ConditionBase was introduced with graph structure conditions
gs_condition_base_class = graph_scheduler.condition.Condition
condition_class_parents = [graph_scheduler.condition.Condition]


try:
    gs_condition_base_class = graph_scheduler.condition.ConditionBase
except AttributeError:
    pass
else:
    class ConditionBase(graph_scheduler.condition.ConditionBase, MDFSerializable):
        def as_mdf_model(self):
            raise graph_scheduler.ConditionError(
                f'MDF support not yet implemented for {type(self)}'
            )
    condition_class_parents.append(ConditionBase)


try:
    graph_scheduler.condition.GraphStructureCondition
except AttributeError:
    graph_structure_conditions_available = False
    gsc_unavailable_message = (
        'Graph structure conditions are not available'
        f'in your installed graph-scheduler v{graph_scheduler.__version__}'
    )
else:
    graph_structure_conditions_available = True
    gsc_unavailable_message = ''


def _create_as_pnl_condition(condition):
    import psyneulink as pnl

    try:
        pnl_class = getattr(pnl.core.scheduling.condition, type(condition).__name__)
    except (AttributeError, TypeError):
        return condition

    # already a pnl Condition
    if isinstance(condition, pnl_condition_base_class):
        return condition

    if not issubclass(pnl_class, gs_condition_base_class):
        return None

    if (
        graph_structure_conditions_available
        and isinstance(condition, graph_scheduler.condition.GraphStructureCondition)
    ):
        try:
            return pnl_class(
                *condition.nodes,
                **{k: v for k, v in condition.kwargs.items() if k != 'nodes'}
            )
        except AttributeError:
            return pnl_class(**condition.kwargs)

    new_args = [_create_as_pnl_condition(a) or a for a in condition.args]
    new_kwargs = {k: _create_as_pnl_condition(v) or v for k, v in condition.kwargs.items()}
    sig = inspect.signature(pnl_class)

    if 'func' in sig.parameters or 'function' in sig.parameters:
        # Condition takes a function as an argument
        res = pnl_class(condition.func, *new_args, **new_kwargs)
    else:
        res = pnl_class(*new_args, **new_kwargs)

    res.owner = condition.owner
    return res


class Condition(*condition_class_parents, MDFSerializable):
    @handle_external_context()
    def is_satisfied(self, *args, context=None, execution_id=None, **kwargs):
        if execution_id is None:
            try:
                execution_id = parse_context(context).execution_id
            except AttributeError:
                pass

        return super().is_satisfied(
            *args,
            context=context,
            execution_id=execution_id,
            **kwargs
        )

    def as_mdf_model(self):
        import modeci_mdf.mdf as mdf
        from psyneulink.core.components.component import Component

        def _parse_condition_arg(arg):
            if isinstance(arg, Component):
                return parse_valid_identifier(arg.name)
            elif isinstance(arg, Condition):
                return arg.as_mdf_model()
            elif (
                isinstance(arg, np.number)
                or (isinstance(arg, np.ndarray) and arg.ndim == 0)
            ):
                return arg.item()
            elif arg is None or isinstance(arg, numbers.Number):
                return arg
            else:
                try:
                    iter(arg)
                except TypeError:
                    return str(arg)
                else:
                    return arg

        if type(self) in {graph_scheduler.Condition, Condition}:
            try:
                func_val = inspect.getsource(self.func)
            except OSError:
                func_val = dill.dumps(self.func)
            func_dict = {'function': func_val}
        else:
            func_dict = {}

        extra_args = {MODEL_SPEC_ID_TYPE: getattr(graph_scheduler.condition, type(self).__name__).__name__}

        sig = inspect.signature(self.__init__)

        for name, param in sig.parameters.items():
            if param.kind is inspect.Parameter.VAR_POSITIONAL:
                args_list = []
                for a in self.args:
                    if isinstance(a, Component):
                        a = parse_valid_identifier(a.name)
                    elif isinstance(a, Condition):
                        a = a.as_mdf_model()
                    args_list.append(a)
                extra_args[name] = args_list

        for i, (name, param) in enumerate(filter(
            lambda item: item[1].kind is inspect.Parameter.POSITIONAL_OR_KEYWORD and item[0] not in self.kwargs,
            sig.parameters.items()
        )):
            try:
                extra_args[name] = self.args[i]
            except IndexError:
                # was specified with keyword not as positional arg
                extra_args[name] = param.default

        return mdf.Condition(
            **func_dict,
            **{
                k: _parse_condition_arg(v)
                for k, v in (*self.kwargs.items(), *extra_args.items())
            }
        )


# below produces psyneulink versions of each Condition class so that
# they are compatible with the extra changes made in Condition above
# (the scheduler does not handle Context objects or mdf/json export)
gs_class_dependencies = {}
gs_classes_to_copy_as_pnl = []
pnl_conditions_module = locals()  # inserting into locals defines the classes
pnl_condition_base_class = pnl_conditions_module[gs_condition_base_class.__name__]

for class_name in graph_scheduler.condition.__dict__:
    cls_ = getattr(graph_scheduler.condition, class_name)
    if inspect.isclass(cls_):
        # don't substitute classes explicitly defined above
        if class_name not in pnl_conditions_module:
            if issubclass(cls_, gs_condition_base_class):
                gs_classes_to_copy_as_pnl.append(class_name)
            else:
                pnl_conditions_module[class_name] = cls_

        gs_class_dependencies[class_name] = {
            c.__name__ for c in cls_.__mro__ if c.__name__ != class_name
        }

# iterate in order such that superclass types are before subclass types
for cond_name in sorted(
    gs_classes_to_copy_as_pnl,
    key=toposort_key(gs_class_dependencies)
):
    sched_module_cond_obj = getattr(graph_scheduler.condition, cond_name)
    new_bases = []
    for cls_ in sched_module_cond_obj.__mro__:
        try:
            new_bases.append(pnl_conditions_module[cls_.__name__])
        except KeyError:
            new_bases.append(cls_)
        if cls_ is gs_condition_base_class:
            break

    new_meta = type(new_bases[0])
    if new_meta is not type:
        pnl_conditions_module[cond_name] = new_meta(
            cond_name, tuple(new_bases), {'__module__': Condition.__module__}
        )
    else:
        pnl_conditions_module[cond_name] = type(cond_name, tuple(new_bases), {})

    pnl_conditions_module[cond_name].__doc__ = sched_module_cond_obj.__doc__


_doc_subs = {
    None: [
        (
            r'def converge\(node, thresh\):',
            'def converge(node, thresh, context):'
        ),
        (
            r'node\.delta',
            'node.parameters.value.get_delta(context)'
        ),
        (
            r'the ``delta`` attribute of\na node \(which reports the change in its ``value``\)',
            "the change in a node's `value`"
        )
    ]
}


class Threshold(graph_scheduler.condition._DependencyValidation, Condition):
    """Threshold

    Attributes:

        dependency
            the node on which the Condition depends

        parameter
            the name of the parameter of **dependency** whose value is
            to be compared to **threshold**

        threshold
            the fixed value compared to the value of the **parameter**

        comparator
            the string comparison operator determining the direction or
            type of comparison of the value of the **parameter**
            relative to **threshold**

        indices
            if specified, a series of indices that reach the desired
            number given an iterable value for **parameter**

        atol
            absolute tolerance for the comparison

        rtol
            relative tolerance (to **threshold**) for the comparison

    Satisfied when:

        The comparison between the value of the **parameter** and
        **threshold** using **comparator** is true. If **comparator** is
        an equality (==, !=), the comparison will be considered equal
        within tolerances **atol** and **rtol**.

    Notes:

        The comparison must be done with scalars. If the value of
        **parameter** contains more than one item, **indices** must be
        specified.
    """

    def __init__(self, dependency, parameter, threshold, comparator, indices=None, atol=0, rtol=0):
        if comparator not in comparison_operators:
            raise graph_scheduler.ConditionError(
                f'Operator must be one of {list(comparison_operators.keys())}'
            )

        if parameter not in dependency.parameters:
            raise graph_scheduler.ConditionError(
                f'{dependency} has no {parameter} parameter'
            )

        if (atol != 0 or rtol != 0) and comparator in {'<', '<=', '>', '>='}:
            warnings.warn('Tolerances for inequality comparators are ignored')

        if (
            indices is not None
            and not isinstance(indices, graph_scheduler.TimeScale)
            and not isinstance(indices, collections.abc.Iterable)
        ):
            indices = [indices]

        def func(threshold, comparator, indices, atol, rtol, execution_id):
            param_value = getattr(self.dependency.parameters, self.parameter).get(execution_id)

            if isinstance(indices, graph_scheduler.TimeScale):
                param_value = param_value._get_by_time_scale(indices)
            elif indices is not None:
                for i in indices:
                    param_value = param_value[i]

            param_value = float(param_value)

            if comparator == '==':
                return np.isclose(param_value, threshold, atol=atol, rtol=rtol)
            elif comparator == '!=':
                return not np.isclose(param_value, threshold, atol=atol, rtol=rtol)
            else:
                return comparison_operators[comparator](param_value, threshold)

        super().__init__(
            func,
            dependency=dependency,
            parameter=parameter,
            threshold=threshold,
            comparator=comparator,
            indices=indices,
            atol=atol,
            rtol=rtol,
        )

    def as_mdf_model(self):
        m = super().as_mdf_model()

        if self.parameter == 'value':
            m.kwargs['parameter'] = f'{self.dependency.name}_OutputPort_0'

        return m


When = Condition

1	# Princeton University licenses this file to You under the Apache License, Version 2.0 (the "License");
2	# you may not use this file except in compliance with the License. You may obtain a copy of the License at:
3	# http://www.apache.org/licenses/LICENSE-2.0
4	# Unless required by applicable law or agreed to in writing, software distributed under the License is distributed
5	# on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
6	# See the License for the specific language governing permissions and limitations under the License.
7
8
9	# ******************************************* Condition ************************************************************
10
11	import collections	1✔
12	import inspect	1✔
13	import numbers	1✔
14	import warnings	1✔
15
16	import dill	1✔
17	import graph_scheduler	1✔
18	import numpy as np	1✔
19
20	from psyneulink.core.globals.context import handle_external_context	1✔
21	from psyneulink.core.globals.mdf import MDFSerializable	1✔
22	from psyneulink.core.globals.keywords import MODEL_SPEC_ID_TYPE, comparison_operators	1✔
23	from psyneulink.core.globals.parameters import parse_context	1✔
24	from psyneulink.core.globals.utilities import parse_valid_identifier, toposort_key	1✔
25
26
27	__all__ = [ # noqa: F822 (dynamically generated)	1✔
28	'AbsoluteCondition', 'AddEdgeTo', 'AfterCall',
29	'AfterConsiderationSetExecution', 'AfterEnvironmentSequence',
30	'AfterEnvironmentStateUpdate', 'AfterNCalls', 'AfterNCallsCombined',
31	'AfterNConsiderationSetExecutions', 'AfterNEnvironmentSequences',
32	'AfterNEnvironmentStateUpdates', 'AfterNPasses', 'AfterNRuns',
33	'AfterNTimeSteps', 'AfterNTrials', 'AfterNode', 'AfterNodes',
34	'AfterPass', 'AfterRun', 'AfterTimeStep', 'AfterTrial', 'All',
35	'AllHaveRun', 'Always', 'And', 'Any', 'AtConsiderationSetExecution',
36	'AtEnvironmentSequence', 'AtEnvironmentSequenceNStart',
37	'AtEnvironmentSequenceStart', 'AtEnvironmentStateUpdate',
38	'AtEnvironmentStateUpdateNStart', 'AtEnvironmentStateUpdateStart',
39	'AtNCalls', 'AtPass', 'AtRun', 'AtRunNStart', 'AtRunStart',
40	'AtTimeStep', 'AtTrial', 'AtTrialNStart', 'AtTrialStart',
41	'BeforeConsiderationSetExecution', 'BeforeEnvironmentStateUpdate',
42	'BeforeNCalls', 'BeforeNode', 'BeforeNodes', 'BeforePass',
43	'BeforeTimeStep', 'BeforeTrial', 'CompositeCondition', 'Condition',
44	'ConditionBase', 'ConditionError', 'ConditionSet',
45	'CustomGraphStructureCondition', 'EveryNCalls', 'EveryNPasses',
46	'GraphStructureCondition', 'JustRan', 'NWhen', 'Never', 'Not',
47	'Operation', 'Or', 'RemoveEdgeFrom', 'Threshold', 'TimeInterval',
48	'TimeTermination', 'WhenFinished', 'WhenFinishedAll',
49	'WhenFinishedAny', 'When', 'While', 'WhileNot', 'WithNode',
50	]
51
52
53	# avoid restricting graph_scheduler versions for this code
54	# ConditionBase was introduced with graph structure conditions
55	gs_condition_base_class = graph_scheduler.condition.Condition	1✔
56	condition_class_parents = [graph_scheduler.condition.Condition]	1✔
57
58
59	try:	1✔
60	gs_condition_base_class = graph_scheduler.condition.ConditionBase	1✔
UNCOV 61	except AttributeError:	×
UNCOV 62	pass	×
63	else:
64	class ConditionBase(graph_scheduler.condition.ConditionBase, MDFSerializable):	1✔
65	def as_mdf_model(self):	1✔
66	raise graph_scheduler.ConditionError(
67	f'MDF support not yet implemented for {type(self)}'
68	)
69	condition_class_parents.append(ConditionBase)	1✔
70
71
72	try:	1✔
73	graph_scheduler.condition.GraphStructureCondition	1✔
UNCOV 74	except AttributeError:	×
UNCOV 75	graph_structure_conditions_available = False	×
UNCOV 76	gsc_unavailable_message = (	×
77	'Graph structure conditions are not available'
78	f'in your installed graph-scheduler v{graph_scheduler.__version__}'
79	)
80	else:
81	graph_structure_conditions_available = True	1✔
82	gsc_unavailable_message = ''	1✔
83
84
85	def _create_as_pnl_condition(condition):	1✔
86	import psyneulink as pnl	1✔
87
88	try:	1✔
89	pnl_class = getattr(pnl.core.scheduling.condition, type(condition).__name__)	1✔
90	except (AttributeError, TypeError):	1✔
91	return condition	1✔
92
93	# already a pnl Condition
94	if isinstance(condition, pnl_condition_base_class):	1✔
95	return condition	1✔
96
97	if not issubclass(pnl_class, gs_condition_base_class):	1!
98	return None	×
99
100	if (	1!
101	graph_structure_conditions_available
102	and isinstance(condition, graph_scheduler.condition.GraphStructureCondition)
103	):
UNCOV 104	try:	×
UNCOV 105	return pnl_class(	×
106	*condition.nodes,
107	**{k: v for k, v in condition.kwargs.items() if k != 'nodes'}
108	)
UNCOV 109	except AttributeError:	×
UNCOV 110	return pnl_class(**condition.kwargs)	×
111
112	new_args = [_create_as_pnl_condition(a) or a for a in condition.args]	1✔
113	new_kwargs = {k: _create_as_pnl_condition(v) or v for k, v in condition.kwargs.items()}	1✔
114	sig = inspect.signature(pnl_class)	1✔
115
116	if 'func' in sig.parameters or 'function' in sig.parameters:	1!
117	# Condition takes a function as an argument
UNCOV 118	res = pnl_class(condition.func, new_args, *new_kwargs)	×
119	else:
120	res = pnl_class(new_args, *new_kwargs)	1✔
121
122	res.owner = condition.owner	1✔
123	return res	1✔
124
125
126	class Condition(*condition_class_parents, MDFSerializable):	1✔
127	@handle_external_context()	1✔
128	def is_satisfied(self, args, context=None, execution_id=None, *kwargs):	1✔
129	if execution_id is None:	1✔
130	try:	1✔
131	execution_id = parse_context(context).execution_id	1✔
132	except AttributeError:	×
UNCOV 133	pass	×
134
135	return super().is_satisfied(	1✔
136	*args,
137	context=context,
138	execution_id=execution_id,
139	**kwargs
140	)
141
142	def as_mdf_model(self):	1✔
143	import modeci_mdf.mdf as mdf	1✔
144	from psyneulink.core.components.component import Component	1✔
145
146	def _parse_condition_arg(arg):	1✔
147	if isinstance(arg, Component):	1✔
148	return parse_valid_identifier(arg.name)	1✔
149	elif isinstance(arg, Condition):	1✔
150	return arg.as_mdf_model()	1✔
151	elif (	1✔
152	isinstance(arg, np.number)
153	or (isinstance(arg, np.ndarray) and arg.ndim == 0)
154	):
155	return arg.item()	1✔
156	elif arg is None or isinstance(arg, numbers.Number):	1✔
157	return arg	1✔
158	else:
159	try:	1✔
160	iter(arg)	1✔
161	except TypeError:	1✔
162	return str(arg)	1✔
163	else:
164	return arg	1✔
165
166	if type(self) in {graph_scheduler.Condition, Condition}:	1!
UNCOV 167	try:	×
UNCOV 168	func_val = inspect.getsource(self.func)	×
UNCOV 169	except OSError:	×
UNCOV 170	func_val = dill.dumps(self.func)	×
UNCOV 171	func_dict = {'function': func_val}	×
172	else:
173	func_dict = {}	1✔
174
175	extra_args = {MODEL_SPEC_ID_TYPE: getattr(graph_scheduler.condition, type(self).__name__).__name__}	1✔
176
177	sig = inspect.signature(self.__init__)	1✔
178
179	for name, param in sig.parameters.items():	1✔
180	if param.kind is inspect.Parameter.VAR_POSITIONAL:	1✔
181	args_list = []	1✔
182	for a in self.args:	1✔
183	if isinstance(a, Component):	1!
UNCOV 184	a = parse_valid_identifier(a.name)	×
185	elif isinstance(a, Condition):	1!
186	a = a.as_mdf_model()	1✔
187	args_list.append(a)	1✔
188	extra_args[name] = args_list	1✔
189
190	for i, (name, param) in enumerate(filter(	1✔
191	lambda item: item[1].kind is inspect.Parameter.POSITIONAL_OR_KEYWORD and item[0] not in self.kwargs,
192	sig.parameters.items()
193	)):
194	try:	1✔
195	extra_args[name] = self.args[i]	1✔
196	except IndexError:	1✔
197	# was specified with keyword not as positional arg
198	extra_args[name] = param.default	1✔
199
200	return mdf.Condition(	1✔
201	**func_dict,
202	**{
203	k: _parse_condition_arg(v)
204	for k, v in (self.kwargs.items(), extra_args.items())
205	}
206	)
207
208
209	# below produces psyneulink versions of each Condition class so that
210	# they are compatible with the extra changes made in Condition above
211	# (the scheduler does not handle Context objects or mdf/json export)
212	gs_class_dependencies = {}	1✔
213	gs_classes_to_copy_as_pnl = []	1✔
214	pnl_conditions_module = locals() # inserting into locals defines the classes	1✔
215	pnl_condition_base_class = pnl_conditions_module[gs_condition_base_class.__name__]	1✔
216
217	for class_name in graph_scheduler.condition.__dict__:	1✔
218	cls_ = getattr(graph_scheduler.condition, class_name)	1✔
219	if inspect.isclass(cls_):	1✔
220	# don't substitute classes explicitly defined above
221	if class_name not in pnl_conditions_module:	1✔
222	if issubclass(cls_, gs_condition_base_class):	1✔
223	gs_classes_to_copy_as_pnl.append(class_name)	1✔
224	else:
225	pnl_conditions_module[class_name] = cls_	1✔
226
227	gs_class_dependencies[class_name] = {	1✔
228	c.__name__ for c in cls_.__mro__ if c.__name__ != class_name
229	}
230
231	# iterate in order such that superclass types are before subclass types
232	for cond_name in sorted(	1✔
233	gs_classes_to_copy_as_pnl,
234	key=toposort_key(gs_class_dependencies)
235	):
236	sched_module_cond_obj = getattr(graph_scheduler.condition, cond_name)	1✔
237	new_bases = []	1✔
238	for cls_ in sched_module_cond_obj.__mro__:	1!
239	try:	1✔
240	new_bases.append(pnl_conditions_module[cls_.__name__])	1✔
241	except KeyError:	1✔
242	new_bases.append(cls_)	1✔
243	if cls_ is gs_condition_base_class:	1✔
244	break	1✔
245
246	new_meta = type(new_bases[0])	1✔
247	if new_meta is not type:	1✔
248	pnl_conditions_module[cond_name] = new_meta(	1✔
249	cond_name, tuple(new_bases), {'__module__': Condition.__module__}
250	)
251	else:
252	pnl_conditions_module[cond_name] = type(cond_name, tuple(new_bases), {})	1✔
253
254	pnl_conditions_module[cond_name].__doc__ = sched_module_cond_obj.__doc__	1✔
255
256
257	_doc_subs = {	1✔
258	None: [
259	(
260	r'def converge\(node, thresh\):',
261	'def converge(node, thresh, context):'
262	),
263	(
264	r'node\.delta',
265	'node.parameters.value.get_delta(context)'
266	),
267	(
268	r'the ``delta`` attribute of\na node \(which reports the change in its ``value``\)',
269	"the change in a node's `value`"
270	)
271	]
272	}
273
274
275	class Threshold(graph_scheduler.condition._DependencyValidation, Condition):	1✔
276	"""Threshold
277
278	Attributes:
279
280	dependency
281	the node on which the Condition depends
282
283	parameter
284	the name of the parameter of dependency whose value is
285	to be compared to threshold
286
287	threshold
288	the fixed value compared to the value of the parameter
289
290	comparator
291	the string comparison operator determining the direction or
292	type of comparison of the value of the parameter
293	relative to threshold
294
295	indices
296	if specified, a series of indices that reach the desired
297	number given an iterable value for parameter
298
299	atol
300	absolute tolerance for the comparison
301
302	rtol
303	relative tolerance (to threshold) for the comparison
304
305	Satisfied when:
306
307	The comparison between the value of the parameter and
308	threshold using comparator is true. If comparator is
309	an equality (==, !=), the comparison will be considered equal
310	within tolerances atol and rtol.
311
312	Notes:
313
314	The comparison must be done with scalars. If the value of
315	parameter contains more than one item, indices must be
316	specified.
317	"""
318
319	def __init__(self, dependency, parameter, threshold, comparator, indices=None, atol=0, rtol=0):	1✔
320	if comparator not in comparison_operators:	1✔
321	raise graph_scheduler.ConditionError(
322	f'Operator must be one of {list(comparison_operators.keys())}'
323	)
324
325	if parameter not in dependency.parameters:	1✔
326	raise graph_scheduler.ConditionError(
327	f'{dependency} has no {parameter} parameter'
328	)
329
330	if (atol != 0 or rtol != 0) and comparator in {'<', '<=', '>', '>='}:	1!
UNCOV 331	warnings.warn('Tolerances for inequality comparators are ignored')	×
332
333	if (	1✔
334	indices is not None
335	and not isinstance(indices, graph_scheduler.TimeScale)
336	and not isinstance(indices, collections.abc.Iterable)
337	):
338	indices = [indices]	1✔
339
340	def func(threshold, comparator, indices, atol, rtol, execution_id):	1✔
341	param_value = getattr(self.dependency.parameters, self.parameter).get(execution_id)	1✔
342
343	if isinstance(indices, graph_scheduler.TimeScale):	1✔
344	param_value = param_value._get_by_time_scale(indices)	1✔
345	elif indices is not None:	1✔
346	for i in indices:	1✔
347	param_value = param_value[i]	1✔
348
349	param_value = float(param_value)	1✔
350
351	if comparator == '==':	1✔
352	return np.isclose(param_value, threshold, atol=atol, rtol=rtol)	1✔
353	elif comparator == '!=':	1✔
354	return not np.isclose(param_value, threshold, atol=atol, rtol=rtol)	1✔
355	else:
356	return comparison_operators[comparator](param_value, threshold)	1✔
357
358	super().__init__(	1✔
359	func,
360	dependency=dependency,
361	parameter=parameter,
362	threshold=threshold,
363	comparator=comparator,
364	indices=indices,
365	atol=atol,
366	rtol=rtol,
367	)
368
369	def as_mdf_model(self):	1✔
370	m = super().as_mdf_model()	1✔
371
372	if self.parameter == 'value':	1!
373	m.kwargs['parameter'] = f'{self.dependency.name}_OutputPort_0'	1✔
374
375	return m	1✔
376
377
378	When = Condition	1✔

PrincetonUniversity / PsyNeuLink / 11245969734

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