diff --git a/tests/scheduling/test_condition.py b/tests/scheduling/test_condition.py index a9f2d318..e5bf222f 100644 --- a/tests/scheduling/test_condition.py +++ b/tests/scheduling/test_condition.py @@ -4,21 +4,7 @@ import numpy as np import psyneulink as pnl import pytest -from psyneulink import _unit_registry -from psyneulink.core.components.functions.nonstateful.transferfunctions import Linear -from psyneulink.core.components.mechanisms.processing.transfermechanism import TransferMechanism -from psyneulink.core.components.projections.pathway.mappingprojection import MappingProjection -from psyneulink.core.compositions.composition import Composition -from psyneulink.core.scheduling.condition import ( - AfterCall, AfterNCalls, AfterNCallsCombined, AfterNPasses, AfterNEnvironmentStateUpdates, - AfterPass, AfterEnvironmentStateUpdate, All, AllHaveRun, Always, Any, AtPass, AtConsiderationSetExecution, - AtEnvironmentStateUpdate, AtEnvironmentStateUpdateStart, BeforeNCalls, BeforePass, BeforeConsiderationSetExecution, - BeforeEnvironmentStateUpdate, Condition, ConditionError, EveryNCalls, EveryNPasses, Not, - NWhen, TimeInterval, TimeTermination, WhenFinished, WhenFinishedAll, - WhenFinishedAny, WhileNot, -) -from psyneulink.core.scheduling.scheduler import Scheduler -from psyneulink.core.scheduling.time import TimeScale + logger = logging.getLogger(__name__) @@ -26,27 +12,27 @@ class TestCondition: def test_invalid_input_WhenFinished(self): - with pytest.raises(ConditionError): - WhenFinished(None).is_satisfied() + with pytest.raises(pnl.ConditionError): + pnl.WhenFinished(None).is_satisfied() def test_invalid_input_WhenFinishedAny_1(self): - with pytest.raises(ConditionError): - WhenFinished(None).is_satisfied() + with pytest.raises(pnl.ConditionError): + pnl.WhenFinished(None).is_satisfied() def test_invalid_input_WhenFinishedAny_2(self): - with pytest.raises(ConditionError): - WhenFinished({None}).is_satisfied() + with pytest.raises(pnl.ConditionError): + pnl.WhenFinished({None}).is_satisfied() def test_invalid_input_WhenFinishedAll_1(self): - with pytest.raises(ConditionError): - WhenFinished(None).is_satisfied() + with pytest.raises(pnl.ConditionError): + pnl.WhenFinished(None).is_satisfied() def test_invalid_input_WhenFinishedAll_2(self): - with pytest.raises(ConditionError): - WhenFinished({None}).is_satisfied() + with pytest.raises(pnl.ConditionError): + pnl.WhenFinished({None}).is_satisfied() def test_additional_args(self): - class OneSatisfied(Condition): + class OneSatisfied(pnl.Condition): def __init__(self, a): def func(a, b): return a or b @@ -61,7 +47,7 @@ def func(a, b): assert not cond.is_satisfied(False) def test_additional_kwargs(self): - class OneSatisfied(Condition): + class OneSatisfied(pnl.Condition): def __init__(self, a, c=True): def func(a, b, c=True): return a or b or c @@ -86,32 +72,32 @@ def func(a, b, c=True): @pytest.mark.psyneulink class TestGeneric: def test_WhileNot_AtPass(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') comp.add_node(A) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, WhileNot(lambda sched: sched.get_clock(sched.default_execution_id).get_total_times_relative(TimeScale.PASS, TimeScale.ENVIRONMENT_STATE_UPDATE) == 0, sched)) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(A, pnl.WhileNot(lambda sched: sched.get_clock(sched.default_execution_id).get_total_times_relative(pnl.TimeScale.PASS, pnl.TimeScale.ENVIRONMENT_STATE_UPDATE) == 0, sched)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(5) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(5) output = list(sched.run(termination_conds=termination_conds)) expected_output = [set(), A, A, A, A] assert output == pytest.helpers.setify_expected_output(expected_output) def test_WhileNot_AtPass_in_middle(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') comp.add_node(A) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, WhileNot(lambda sched: sched.get_clock(sched.default_execution_id).get_total_times_relative(TimeScale.PASS, TimeScale.ENVIRONMENT_STATE_UPDATE) == 2, sched)) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(A, pnl.WhileNot(lambda sched: sched.get_clock(sched.default_execution_id).get_total_times_relative(pnl.TimeScale.PASS, pnl.TimeScale.ENVIRONMENT_STATE_UPDATE) == 2, sched)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(5) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(5) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A, A, set(), A, A] @@ -121,66 +107,66 @@ def test_WhileNot_AtPass_in_middle(self): class TestRelative: def test_Any_end_before_one_finished(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') for m in [A]: comp.add_node(m) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) + sched.add_condition(A, pnl.EveryNPasses(1)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = Any(AfterNCalls(A, 10), AtPass(5)) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.Any(pnl.AfterNCalls(A, 10), pnl.AtPass(5)) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A for _ in range(5)] assert output == pytest.helpers.setify_expected_output(expected_output) def test_All_end_after_one_finished(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') for m in [A]: comp.add_node(m) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) + sched.add_condition(A, pnl.EveryNPasses(1)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = Any(AfterNCalls(A, 5), AtPass(10)) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.Any(pnl.AfterNCalls(A, 5), pnl.AtPass(10)) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A for _ in range(5)] assert output == pytest.helpers.setify_expected_output(expected_output) def test_Not_AtPass(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') comp.add_node(A) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, Not(AtPass(0))) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(A, pnl.Not(pnl.AtPass(0))) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(5) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(5) output = list(sched.run(termination_conds=termination_conds)) expected_output = [set(), A, A, A, A] assert output == pytest.helpers.setify_expected_output(expected_output) def test_Not_AtPass_in_middle(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') comp.add_node(A) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, Not(AtPass(2))) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(A, pnl.Not(pnl.AtPass(2))) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(5) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(5) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A, A, set(), A, A] @@ -194,20 +180,20 @@ def test_Not_AtPass_in_middle(self): ] ) def test_NWhen_AfterNCalls(self, n, expected_output): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') - B = TransferMechanism(function=Linear(intercept=4.0), name='B') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='B') for m in [A, B]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), A, B) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, Always()) - sched.add_condition(B, NWhen(AfterNCalls(A, 3), n)) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(A, pnl.Always()) + sched.add_condition(B, pnl.NWhen(pnl.AfterNCalls(A, 3), n)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(A, 6) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(A, 6) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A if x == 'A' else B for x in expected_output] @@ -218,204 +204,204 @@ def test_NWhen_AfterNCalls(self, n, expected_output): class TestTimePNL: def test_BeforeConsiderationSetExecution(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') comp.add_node(A) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, BeforeConsiderationSetExecution(2)) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(A, pnl.BeforeConsiderationSetExecution(2)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(5) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(5) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A, A, set(), set(), set()] assert output == pytest.helpers.setify_expected_output(expected_output) def test_BeforeConsiderationSetExecution_2(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') - B = TransferMechanism(name='B') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') + B = pnl.TransferMechanism(name='B') comp.add_node(A) comp.add_node(B) - comp.add_projection(MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), A, B) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, BeforeConsiderationSetExecution(2)) - sched.add_condition(B, Always()) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(A, pnl.BeforeConsiderationSetExecution(2)) + sched.add_condition(B, pnl.Always()) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(5) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(5) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A, B, B, B, B, B] assert output == pytest.helpers.setify_expected_output(expected_output) def test_AtConsiderationSetExecution(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') comp.add_node(A) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, AtConsiderationSetExecution(0)) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(A, pnl.AtConsiderationSetExecution(0)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(5) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(5) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A, set(), set(), set(), set()] assert output == pytest.helpers.setify_expected_output(expected_output) def test_BeforePass(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') comp.add_node(A) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, BeforePass(2)) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(A, pnl.BeforePass(2)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(5) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(5) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A, A, set(), set(), set()] assert output == pytest.helpers.setify_expected_output(expected_output) def test_AtPass(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') comp.add_node(A) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, AtPass(0)) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(A, pnl.AtPass(0)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(5) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(5) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A, set(), set(), set(), set()] assert output == pytest.helpers.setify_expected_output(expected_output) def test_AtPass_underconstrained(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') - B = TransferMechanism(function=Linear(intercept=4.0), name='B') - C = TransferMechanism(function=Linear(intercept=1.5), name='C') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='B') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='C') for m in [A, B, C]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) - comp.add_projection(MappingProjection(), B, C) + comp.add_projection(pnl.MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), B, C) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, AtPass(0)) - sched.add_condition(B, Always()) - sched.add_condition(C, Always()) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(A, pnl.AtPass(0)) + sched.add_condition(B, pnl.Always()) + sched.add_condition(C, pnl.Always()) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(C, 2) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(C, 2) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A, B, C, B, C] assert output == pytest.helpers.setify_expected_output(expected_output) def test_AtPass_in_middle(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') comp.add_node(A) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, AtPass(2)) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(A, pnl.AtPass(2)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(5) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(5) output = list(sched.run(termination_conds=termination_conds)) expected_output = [set(), set(), A, set(), set()] assert output == pytest.helpers.setify_expected_output(expected_output) def test_AtPass_at_end(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') comp.add_node(A) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, AtPass(5)) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(A, pnl.AtPass(5)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(5) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(5) output = list(sched.run(termination_conds=termination_conds)) expected_output = [set(), set(), set(), set(), set()] assert output == pytest.helpers.setify_expected_output(expected_output) def test_AtPass_after_end(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') comp.add_node(A) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, AtPass(6)) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(A, pnl.AtPass(6)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(5) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(5) output = list(sched.run(termination_conds=termination_conds)) expected_output = [set(), set(), set(), set(), set()] assert output == pytest.helpers.setify_expected_output(expected_output) def test_AfterPass(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') comp.add_node(A) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, AfterPass(0)) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(A, pnl.AfterPass(0)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(5) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(5) output = list(sched.run(termination_conds=termination_conds)) expected_output = [set(), A, A, A, A] assert output == pytest.helpers.setify_expected_output(expected_output) def test_AfterNPasses(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') comp.add_node(A) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, AfterNPasses(1)) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(A, pnl.AfterNPasses(1)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(5) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(5) output = list(sched.run(termination_conds=termination_conds)) expected_output = [set(), A, A, A, A] assert output == pytest.helpers.setify_expected_output(expected_output) def test_BeforeEnvironmentStateUpdate(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') comp.add_node(A) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, BeforeEnvironmentStateUpdate(4)) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(A, pnl.BeforeEnvironmentStateUpdate(4)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(5) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(5) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(1) comp.run( inputs={A: range(6)}, scheduler=sched, @@ -427,16 +413,16 @@ def test_BeforeEnvironmentStateUpdate(self): assert output == pytest.helpers.setify_expected_output(expected_output) def test_AtEnvironmentStateUpdate(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') comp.add_node(A) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, Always()) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(A, pnl.Always()) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AtEnvironmentStateUpdate(4) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AtEnvironmentStateUpdate(4) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(1) comp.run( inputs={A: range(6)}, scheduler=sched, @@ -448,16 +434,16 @@ def test_AtEnvironmentStateUpdate(self): assert output == pytest.helpers.setify_expected_output(expected_output) def test_AfterEnvironmentStateUpdate(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') comp.add_node(A) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, Always()) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(A, pnl.Always()) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterEnvironmentStateUpdate(4) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterEnvironmentStateUpdate(4) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(1) comp.run( inputs={A: range(6)}, scheduler=sched, @@ -469,16 +455,16 @@ def test_AfterEnvironmentStateUpdate(self): assert output == pytest.helpers.setify_expected_output(expected_output) def test_AfterNEnvironmentStateUpdates(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') comp.add_node(A) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, AfterNPasses(1)) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(A, pnl.AfterNPasses(1)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(5) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(5) output = list(sched.run(termination_conds=termination_conds)) expected_output = [set(), A, A, A, A] @@ -492,8 +478,8 @@ class TestTime: pytest.param( gs.AfterNPasses(1), { - TimeScale.ENVIRONMENT_SEQUENCE: gs.AfterNEnvironmentStateUpdates(1), - TimeScale.ENVIRONMENT_STATE_UPDATE: gs.AfterConsiderationSetExecution(4) + pnl.TimeScale.ENVIRONMENT_SEQUENCE: gs.AfterNEnvironmentStateUpdates(1), + pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: gs.AfterConsiderationSetExecution(4) }, [set(), 'A', 'A', 'A', 'A'], id='AfterConsiderationSetExecution' @@ -501,8 +487,8 @@ class TestTime: pytest.param( gs.AfterNPasses(1), { - TimeScale.ENVIRONMENT_SEQUENCE: gs.AfterNEnvironmentStateUpdates(1), - TimeScale.ENVIRONMENT_STATE_UPDATE: gs.AfterNConsiderationSetExecutions(5) + pnl.TimeScale.ENVIRONMENT_SEQUENCE: gs.AfterNEnvironmentStateUpdates(1), + pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: gs.AfterNConsiderationSetExecutions(5) }, [set(), 'A', 'A', 'A', 'A'], id='AfterNConsiderationSetExecutions' @@ -525,7 +511,7 @@ def test_single_node( [ pytest.param( gs.AtEnvironmentStateUpdateNStart(2), - {TimeScale.ENVIRONMENT_STATE_UPDATE: gs.AfterNPasses(1)}, + {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: gs.AfterNPasses(1)}, [[set(), set(), 'A', set()]], 1, 4, @@ -533,7 +519,7 @@ def test_single_node( ), pytest.param( gs.AtEnvironmentSequence(4), - {TimeScale.ENVIRONMENT_STATE_UPDATE: gs.AfterNPasses(1)}, + {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: gs.AfterNPasses(1)}, [[set()], [set()], [set()], [set()], ['A'], [set()]], 6, 1, @@ -541,7 +527,7 @@ def test_single_node( ), pytest.param( gs.AfterEnvironmentSequence(3), - {TimeScale.ENVIRONMENT_STATE_UPDATE: gs.AfterNPasses(1)}, + {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: gs.AfterNPasses(1)}, [[set()], [set()], [set()], [set()], ['A'], ['A']], 6, 1, @@ -549,7 +535,7 @@ def test_single_node( ), pytest.param( gs.AfterNEnvironmentSequences(4), - {TimeScale.ENVIRONMENT_STATE_UPDATE: gs.AfterNPasses(1)}, + {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: gs.AfterNPasses(1)}, [[set()], [set()], [set()], [set()], ['A'], ['A']], 6, 1, @@ -557,7 +543,7 @@ def test_single_node( ), pytest.param( gs.AtEnvironmentSequenceStart(), - {TimeScale.ENVIRONMENT_STATE_UPDATE: gs.AfterNPasses(1)}, + {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: gs.AfterNPasses(1)}, [['A', set()], ['A', set()]], 2, 2, @@ -565,7 +551,7 @@ def test_single_node( ), pytest.param( gs.AtEnvironmentSequenceNStart(1), - {TimeScale.ENVIRONMENT_STATE_UPDATE: gs.AfterNPasses(1)}, + {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: gs.AfterNPasses(1)}, [[set(), set()], ['A', set()], [set(), set()]], 3, 2, @@ -603,16 +589,16 @@ def test_single_node_n_sequences( class TestComponentBased: def test_BeforeNCalls(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') comp.add_node(A) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, BeforeNCalls(A, 3)) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(A, pnl.BeforeNCalls(A, 3)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(5) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(5) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A, A, A, set(), set()] @@ -624,57 +610,57 @@ def test_BeforeNCalls(self): # A fix could invalidate key assumptions and affect many other conditions # Since this condition is unlikely to be used, it's best to leave it for now # def test_AtCall(self): - # comp = Composition() - # A = TransferMechanism(function = Linear(slope=5.0, intercept = 2.0), name = 'A') - # B = TransferMechanism(function = Linear(intercept = 4.0), name = 'B') - # C = TransferMechanism(function = Linear(intercept = 1.5), name = 'C') + # comp = pnl.Composition() + # A = pnl.TransferMechanism(function = pnl.Linear(slope=5.0, intercept = 2.0), name = 'A') + # B = pnl.TransferMechanism(function = pnl.Linear(intercept = 4.0), name = 'B') + # C = pnl.TransferMechanism(function = pnl.Linear(intercept = 1.5), name = 'C') # for m in [A,B]: # comp.add_node(m) - # sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - # sched.add_condition(A, Always()) + # sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + # sched.add_condition(A, pnl.Always()) # sched.add_condition(B, AtCall(A, 3)) # termination_conds = {} - # termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - # termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(5) + # termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + # termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(5) # output = list(sched.run(termination_conds=termination_conds)) # expected_output = [A, A, set([A, B]), A, A] # assert output == pytest.helpers.setify_expected_output(expected_output) def test_AfterCall(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') - B = TransferMechanism(function=Linear(intercept=4.0), name='B') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='B') for m in [A, B]: comp.add_node(m) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(B, AfterCall(A, 3)) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(B, pnl.AfterCall(A, 3)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(5) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(5) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A, A, A, set([A, B]), set([A, B])] assert output == pytest.helpers.setify_expected_output(expected_output) def test_AfterNCalls(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') - B = TransferMechanism(function=Linear(intercept=4.0), name='B') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='B') for m in [A, B]: comp.add_node(m) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, Always()) - sched.add_condition(B, AfterNCalls(A, 3)) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(A, pnl.Always()) + sched.add_condition(B, pnl.AfterNCalls(A, 3)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(5) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(5) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A, A, set([A, B]), set([A, B]), set([A, B])] @@ -684,16 +670,16 @@ def test_AfterNCalls(self): class TestConvenience: def test_AtEnvironmentStateUpdateStart(self): - comp = Composition() - A = TransferMechanism(name='A') - B = TransferMechanism(name='B') + comp = pnl.Composition() + A = pnl.TransferMechanism(name='A') + B = pnl.TransferMechanism(name='B') comp.add_linear_processing_pathway([A, B]) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(B, AtEnvironmentStateUpdateStart()) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + sched.add_condition(B, pnl.AtEnvironmentStateUpdateStart()) termination_conds = { - TimeScale.ENVIRONMENT_STATE_UPDATE: AtPass(3) + pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.AtPass(3) } output = list(sched.run(termination_conds=termination_conds)) @@ -702,33 +688,33 @@ def test_AtEnvironmentStateUpdateStart(self): @pytest.mark.psyneulink def test_composite_condition_multi(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') - B = TransferMechanism(function=Linear(intercept=4.0), name='B') - C = TransferMechanism(function=Linear(intercept=1.5), name='C') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='B') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='C') for m in [A, B, C]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) - comp.add_projection(MappingProjection(), B, C) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, EveryNCalls(A, 2)) - sched.add_condition(C, All( - Any( - AfterPass(6), - AfterNCalls(B, 2) + comp.add_projection(pnl.MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), B, C) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.EveryNCalls(A, 2)) + sched.add_condition(C, pnl.All( + pnl.Any( + pnl.AfterPass(6), + pnl.AfterNCalls(B, 2) ), - Any( - AfterPass(2), - AfterNCalls(B, 3) + pnl.Any( + pnl.AfterPass(2), + pnl.AfterNCalls(B, 3) ) ) ) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(C, 3) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(C, 3) output = list(sched.run(termination_conds=termination_conds)) expected_output = [ A, A, B, A, A, B, C, A, C, A, B, C @@ -737,25 +723,25 @@ def test_composite_condition_multi(self): @pytest.mark.psyneulink def test_AfterNCallsCombined(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') A.is_finished_flag = True - B = TransferMechanism(function=Linear(intercept=4.0), name='B') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='B') B.is_finished_flag = True - C = TransferMechanism(function=Linear(intercept=1.5), name='C') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='C') for m in [A, B, C]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) - comp.add_projection(MappingProjection(), B, C) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + comp.add_projection(pnl.MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), B, C) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, EveryNCalls(A, 2)) - sched.add_condition(C, EveryNCalls(B, 2)) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.EveryNCalls(A, 2)) + sched.add_condition(C, pnl.EveryNCalls(B, 2)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCallsCombined(B, C, n=4) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCallsCombined(B, C, n=4) output = list(sched.run(termination_conds=termination_conds)) expected_output = [ @@ -765,25 +751,25 @@ def test_AfterNCallsCombined(self): @pytest.mark.psyneulink def test_AllHaveRun(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') A.is_finished_flag = False - B = TransferMechanism(function=Linear(intercept=4.0), name='B') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='B') B.is_finished_flag = True - C = TransferMechanism(function=Linear(intercept=1.5), name='C') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='C') for m in [A, B, C]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) - comp.add_projection(MappingProjection(), B, C) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + comp.add_projection(pnl.MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), B, C) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, EveryNCalls(A, 2)) - sched.add_condition(C, EveryNCalls(B, 2)) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.EveryNCalls(A, 2)) + sched.add_condition(C, pnl.EveryNCalls(B, 2)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AllHaveRun(A, B, C) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AllHaveRun(A, B, C) output = list(sched.run(termination_conds=termination_conds)) expected_output = [ @@ -793,23 +779,23 @@ def test_AllHaveRun(self): @pytest.mark.psyneulink def test_AllHaveRun_2(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') - B = TransferMechanism(function=Linear(intercept=4.0), name='B') - C = TransferMechanism(function=Linear(intercept=1.5), name='C') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='B') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='C') for m in [A, B, C]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) - comp.add_projection(MappingProjection(), B, C) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + comp.add_projection(pnl.MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), B, C) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, EveryNCalls(A, 2)) - sched.add_condition(C, EveryNCalls(B, 2)) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.EveryNCalls(A, 2)) + sched.add_condition(C, pnl.EveryNCalls(B, 2)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AllHaveRun() + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AllHaveRun() output = list(sched.run(termination_conds=termination_conds)) expected_output = [ @@ -824,7 +810,7 @@ def test_AllHaveRun_2(self): ('value', None, None, 1, [[[10]]]), ('value', (0, 0), [[0, 0]], [1, 2], [[[10, 20]]]), ('value', (0, 1), [[0, 0]], [1, 2], [[[5, 10]]]), - ('num_executions', TimeScale.TRIAL, None, 1, [[[10]]]), + ('num_executions', pnl.TimeScale.TRIAL, None, 1, [[[10]]]), ] ) @pytest.mark.parametrize('threshold', [10, 10.0]) @@ -832,16 +818,16 @@ def test_Threshold_parameters( self, parameter, indices, default_variable, integration_rate, expected_results, threshold, ): - A = TransferMechanism( + A = pnl.TransferMechanism( default_variable=default_variable, integrator_mode=True, integrator_function=pnl.SimpleIntegrator, integration_rate=integration_rate, ) - comp = Composition(pathways=[A]) + comp = pnl.Composition(pathways=[A]) comp.termination_processing = { - TimeScale.TRIAL: gs.Threshold(A, parameter, threshold, '>=', indices=indices) + pnl.TimeScale.TRIAL: gs.Threshold(A, parameter, threshold, '>=', indices=indices) } comp.run(inputs={A: np.ones(A.defaults.variable.shape)}) @@ -864,14 +850,14 @@ def test_Threshold_parameters( def test_Threshold_comparators( self, comparator, increment, threshold, expected_results ): - A = TransferMechanism( + A = pnl.TransferMechanism( integrator_mode=True, integrator_function=pnl.AccumulatorIntegrator(rate=1, increment=increment), ) - comp = Composition(pathways=[A]) + comp = pnl.Composition(pathways=[A]) comp.termination_processing = { - TimeScale.TRIAL: gs.Threshold(A, 'value', threshold, comparator) + pnl.TimeScale.TRIAL: gs.Threshold(A, 'value', threshold, comparator) } comp.run(inputs={A: np.ones(A.defaults.variable.shape)}) @@ -918,14 +904,14 @@ def has_parameter(self, parameter): def test_Threshold_tolerances( self, comparator, increment, threshold, atol, rtol, expected_results ): - A = TransferMechanism( + A = pnl.TransferMechanism( integrator_mode=True, integrator_function=pnl.AccumulatorIntegrator(rate=1, increment=increment), ) - comp = Composition(pathways=[A]) + comp = pnl.Composition(pathways=[A]) comp.termination_processing = { - TimeScale.TRIAL: gs.Threshold(A, 'value', threshold, comparator, atol=atol, rtol=rtol) + pnl.TimeScale.TRIAL: gs.Threshold(A, 'value', threshold, comparator, atol=atol, rtol=rtol) } comp.run(inputs={A: np.ones(A.defaults.variable.shape)}) @@ -938,38 +924,38 @@ class TestWhenFinished: @classmethod def setup_class(self): - self.orig_is_finished_flag = TransferMechanism.is_finished_flag - self.orig_is_finished = TransferMechanism.is_finished - TransferMechanism.is_finished_flag = True - TransferMechanism.is_finished = lambda self, context: self.is_finished_flag + self.orig_is_finished_flag = pnl.TransferMechanism.is_finished_flag + self.orig_is_finished = pnl.TransferMechanism.is_finished + pnl.TransferMechanism.is_finished_flag = True + pnl.TransferMechanism.is_finished = lambda self, context: self.is_finished_flag @classmethod def teardown_class(self): - del TransferMechanism.is_finished_flag - del TransferMechanism.is_finished - TransferMechanism.is_finished_flag = self.orig_is_finished_flag - TransferMechanism.is_finished = self.orig_is_finished + del pnl.TransferMechanism.is_finished_flag + del pnl.TransferMechanism.is_finished + pnl.TransferMechanism.is_finished_flag = self.orig_is_finished_flag + pnl.TransferMechanism.is_finished = self.orig_is_finished def test_WhenFinishedAny_1(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') A.is_finished_flag = True - B = TransferMechanism(function=Linear(intercept=4.0), name='B') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='B') B.is_finished_flag = True - C = TransferMechanism(function=Linear(intercept=1.5), name='C') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='C') for m in [A, B, C]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, C) - comp.add_projection(MappingProjection(), B, C) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + comp.add_projection(pnl.MappingProjection(), A, C) + comp.add_projection(pnl.MappingProjection(), B, C) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, EveryNPasses(1)) - sched.add_condition(C, WhenFinishedAny(A, B)) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.EveryNPasses(1)) + sched.add_condition(C, pnl.WhenFinishedAny(A, B)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(C, 1) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(C, 1) output = list(sched.run(termination_conds=termination_conds)) expected_output = [ set([A, B]), C @@ -977,25 +963,25 @@ def test_WhenFinishedAny_1(self): assert output == pytest.helpers.setify_expected_output(expected_output) def test_WhenFinishedAny_2(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') A.is_finished_flag = False - B = TransferMechanism(function=Linear(intercept=4.0), name='B') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='B') B.is_finished_flag = True - C = TransferMechanism(function=Linear(intercept=1.5), name='C') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='C') for m in [A, B, C]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, C) - comp.add_projection(MappingProjection(), B, C) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + comp.add_projection(pnl.MappingProjection(), A, C) + comp.add_projection(pnl.MappingProjection(), B, C) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, EveryNPasses(1)) - sched.add_condition(C, WhenFinishedAny(A, B)) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.EveryNPasses(1)) + sched.add_condition(C, pnl.WhenFinishedAny(A, B)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(A, 5) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(A, 5) output = list(sched.run(termination_conds=termination_conds)) expected_output = [ set([A, B]), C, set([A, B]), C, set([A, B]), C, set([A, B]), C, set([A, B]) @@ -1003,24 +989,24 @@ def test_WhenFinishedAny_2(self): assert output == pytest.helpers.setify_expected_output(expected_output) def test_WhenFinishedAny_noargs(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') - B = TransferMechanism(function=Linear(intercept=4.0), name='B') - C = TransferMechanism(function=Linear(intercept=1.5), name='C') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='B') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='C') for m in [A, B, C]: m.is_finished_flag = False comp.add_node(m) - comp.add_projection(MappingProjection(), A, C) - comp.add_projection(MappingProjection(), B, C) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + comp.add_projection(pnl.MappingProjection(), A, C) + comp.add_projection(pnl.MappingProjection(), B, C) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, Always()) - sched.add_condition(B, Always()) - sched.add_condition(C, Always()) + sched.add_condition(A, pnl.Always()) + sched.add_condition(B, pnl.Always()) + sched.add_condition(C, pnl.Always()) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = WhenFinishedAny() + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.WhenFinishedAny() output = [] i = 0 for step in sched.run(termination_conds=termination_conds): @@ -1037,25 +1023,25 @@ def test_WhenFinishedAny_noargs(self): assert output == pytest.helpers.setify_expected_output(expected_output) def test_WhenFinishedAll_1(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') A.is_finished_flag = True - B = TransferMechanism(function=Linear(intercept=4.0), name='B') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='B') B.is_finished_flag = True - C = TransferMechanism(function=Linear(intercept=1.5), name='C') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='C') for m in [A, B, C]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, C) - comp.add_projection(MappingProjection(), B, C) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + comp.add_projection(pnl.MappingProjection(), A, C) + comp.add_projection(pnl.MappingProjection(), B, C) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, EveryNPasses(1)) - sched.add_condition(C, WhenFinishedAll(A, B)) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.EveryNPasses(1)) + sched.add_condition(C, pnl.WhenFinishedAll(A, B)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(C, 1) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(C, 1) output = list(sched.run(termination_conds=termination_conds)) expected_output = [ set([A, B]), C @@ -1063,25 +1049,25 @@ def test_WhenFinishedAll_1(self): assert output == pytest.helpers.setify_expected_output(expected_output) def test_WhenFinishedAll_2(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') A.is_finished_flag = False - B = TransferMechanism(function=Linear(intercept=4.0), name='B') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='B') B.is_finished_flag = True - C = TransferMechanism(function=Linear(intercept=1.5), name='C') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='C') for m in [A, B, C]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, C) - comp.add_projection(MappingProjection(), B, C) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + comp.add_projection(pnl.MappingProjection(), A, C) + comp.add_projection(pnl.MappingProjection(), B, C) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, EveryNPasses(1)) - sched.add_condition(C, WhenFinishedAll(A, B)) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.EveryNPasses(1)) + sched.add_condition(C, pnl.WhenFinishedAll(A, B)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(A, 5) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(A, 5) output = list(sched.run(termination_conds=termination_conds)) expected_output = [ set([A, B]), set([A, B]), set([A, B]), set([A, B]), set([A, B]), @@ -1089,24 +1075,24 @@ def test_WhenFinishedAll_2(self): assert output == pytest.helpers.setify_expected_output(expected_output) def test_WhenFinishedAll_noargs(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A') - B = TransferMechanism(function=Linear(intercept=4.0), name='B') - C = TransferMechanism(function=Linear(intercept=1.5), name='C') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='B') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='C') for m in [A, B, C]: comp.add_node(m) m.is_finished_flag = False - comp.add_projection(MappingProjection(), A, C) - comp.add_projection(MappingProjection(), B, C) - sched = Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) + comp.add_projection(pnl.MappingProjection(), A, C) + comp.add_projection(pnl.MappingProjection(), B, C) + sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, Always()) - sched.add_condition(B, Always()) - sched.add_condition(C, Always()) + sched.add_condition(A, pnl.Always()) + sched.add_condition(B, pnl.Always()) + sched.add_condition(C, pnl.Always()) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = WhenFinishedAll() + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.WhenFinishedAll() output = [] i = 0 for step in sched.run(termination_conds=termination_conds): @@ -1125,46 +1111,46 @@ def test_WhenFinishedAll_noargs(self): @pytest.mark.psyneulink class TestAbsolute: - A = TransferMechanism(name='scheduler-pytests-A') - B = TransferMechanism(name='scheduler-pytests-B') - C = TransferMechanism(name='scheduler-pytests-C') + A = pnl.TransferMechanism(name='scheduler-pytests-A') + B = pnl.TransferMechanism(name='scheduler-pytests-B') + C = pnl.TransferMechanism(name='scheduler-pytests-C') @pytest.mark.parametrize( 'conditions, termination_conds', [ ( - {A: TimeInterval(repeat=8), B: TimeInterval(repeat=4), C: TimeInterval(repeat=2)}, - {TimeScale.ENVIRONMENT_STATE_UPDATE: AfterNCalls(A, 2)}, + {A: pnl.TimeInterval(repeat=8), B: pnl.TimeInterval(repeat=4), C: pnl.TimeInterval(repeat=2)}, + {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.AfterNCalls(A, 2)}, ), ( - {A: TimeInterval(repeat=5), B: TimeInterval(repeat=3), C: TimeInterval(repeat=1)}, - {TimeScale.ENVIRONMENT_STATE_UPDATE: AfterNCalls(A, 2)}, + {A: pnl.TimeInterval(repeat=5), B: pnl.TimeInterval(repeat=3), C: pnl.TimeInterval(repeat=1)}, + {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.AfterNCalls(A, 2)}, ), ( - {A: TimeInterval(repeat=3), B: TimeInterval(repeat=2)}, - {TimeScale.ENVIRONMENT_STATE_UPDATE: AfterNCalls(A, 2)}, + {A: pnl.TimeInterval(repeat=3), B: pnl.TimeInterval(repeat=2)}, + {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.AfterNCalls(A, 2)}, ), ( - {A: TimeInterval(repeat=5), B: TimeInterval(repeat=7)}, - {TimeScale.ENVIRONMENT_STATE_UPDATE: AfterNCalls(B, 2)}, + {A: pnl.TimeInterval(repeat=5), B: pnl.TimeInterval(repeat=7)}, + {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.AfterNCalls(B, 2)}, ), ( - {A: TimeInterval(repeat=1200), B: TimeInterval(repeat=1000)}, - {TimeScale.ENVIRONMENT_STATE_UPDATE: AfterNCalls(A, 3)}, + {A: pnl.TimeInterval(repeat=1200), B: pnl.TimeInterval(repeat=1000)}, + {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.AfterNCalls(A, 3)}, ), ( - {A: TimeInterval(repeat=0.33333), B: TimeInterval(repeat=0.66666)}, - {TimeScale.ENVIRONMENT_STATE_UPDATE: AfterNCalls(B, 3)}, + {A: pnl.TimeInterval(repeat=0.33333), B: pnl.TimeInterval(repeat=0.66666)}, + {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.AfterNCalls(B, 3)}, ), # smaller than default units cause floating point issue without mitigation ( - {A: TimeInterval(repeat=2 * _unit_registry.us), B: TimeInterval(repeat=4 * _unit_registry.us)}, - {TimeScale.ENVIRONMENT_STATE_UPDATE: AfterNCalls(B, 3)}, + {A: pnl.TimeInterval(repeat=2 * pnl._unit_registry.us), B: pnl.TimeInterval(repeat=4 * pnl._unit_registry.us)}, + {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.AfterNCalls(B, 3)}, ), ] ) def test_TimeInterval_linear_everynms(self, conditions, termination_conds): - comp = Composition() + comp = pnl.Composition() comp.add_linear_processing_pathway([self.A, self.B, self.C]) comp.scheduler.add_condition_set(conditions) @@ -1186,64 +1172,64 @@ def test_TimeInterval_linear_everynms(self, conditions, termination_conds): [ ( { - A: TimeInterval(repeat=10, start=100), - B: TimeInterval(repeat=10, start=300), - C: TimeInterval(repeat=10, start=400) + A: pnl.TimeInterval(repeat=10, start=100), + B: pnl.TimeInterval(repeat=10, start=300), + C: pnl.TimeInterval(repeat=10, start=400) }, - {TimeScale.ENVIRONMENT_STATE_UPDATE: TimeInterval(start=500)} + {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.TimeInterval(start=500)} ), ( { - A: TimeInterval(start=100), - B: TimeInterval(start=300), - C: TimeInterval(start=400) + A: pnl.TimeInterval(start=100), + B: pnl.TimeInterval(start=300), + C: pnl.TimeInterval(start=400) }, - {TimeScale.ENVIRONMENT_STATE_UPDATE: TimeInterval(start=500)} + {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.TimeInterval(start=500)} ), ( { - A: TimeInterval(repeat=2, start=105), - B: TimeInterval(repeat=7, start=317), - C: TimeInterval(repeat=11, start=431) + A: pnl.TimeInterval(repeat=2, start=105), + B: pnl.TimeInterval(repeat=7, start=317), + C: pnl.TimeInterval(repeat=11, start=431) }, - {TimeScale.ENVIRONMENT_STATE_UPDATE: TimeInterval(start=597)} + {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.TimeInterval(start=597)} ), ( { - A: TimeInterval(repeat=10, start=100, start_inclusive=False), - B: TimeInterval(repeat=10, start=300), - C: TimeInterval(repeat=10, start=400) + A: pnl.TimeInterval(repeat=10, start=100, start_inclusive=False), + B: pnl.TimeInterval(repeat=10, start=300), + C: pnl.TimeInterval(repeat=10, start=400) }, - {TimeScale.ENVIRONMENT_STATE_UPDATE: TimeInterval(start=500)} + {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.TimeInterval(start=500)} ), ( { - A: TimeInterval(repeat=10, start=100, start_inclusive=False), - B: TimeInterval(repeat=10, start=300), - C: TimeInterval(repeat=10, start=400) + A: pnl.TimeInterval(repeat=10, start=100, start_inclusive=False), + B: pnl.TimeInterval(repeat=10, start=300), + C: pnl.TimeInterval(repeat=10, start=400) }, - {TimeScale.ENVIRONMENT_STATE_UPDATE: TimeInterval(start=500, start_inclusive=False)} + {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.TimeInterval(start=500, start_inclusive=False)} ), ( { - A: TimeInterval(repeat=10, start=100), - B: TimeInterval(repeat=10, start=100, end=200), - C: TimeInterval(repeat=10, start=400) + A: pnl.TimeInterval(repeat=10, start=100), + B: pnl.TimeInterval(repeat=10, start=100, end=200), + C: pnl.TimeInterval(repeat=10, start=400) }, - {TimeScale.ENVIRONMENT_STATE_UPDATE: TimeInterval(start=500, start_inclusive=False)} + {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.TimeInterval(start=500, start_inclusive=False)} ), ( { - A: TimeInterval(repeat=10, start=100), - B: TimeInterval(repeat=10, start=100, end=200, end_inclusive=False), - C: TimeInterval(repeat=10, start=400) + A: pnl.TimeInterval(repeat=10, start=100), + B: pnl.TimeInterval(repeat=10, start=100, end=200, end_inclusive=False), + C: pnl.TimeInterval(repeat=10, start=400) }, - {TimeScale.ENVIRONMENT_STATE_UPDATE: TimeInterval(start=500)} + {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.TimeInterval(start=500)} ), ] ) def test_TimeInterval_no_dependencies(self, conditions, termination_conds): - comp = Composition() + comp = pnl.Composition() comp.add_nodes([self.A, self.B, self.C]) comp.scheduler.add_condition_set(conditions) consideration_set_execution_abs_value = comp.scheduler._get_absolute_consideration_set_execution_unit(termination_conds) @@ -1274,7 +1260,7 @@ def test_TimeInterval_no_dependencies(self, conditions, termination_conds): # this test only runs a single ENVIRONMENT_STATE_UPDATE, so this # timestamp corresponds to its last final_timestamp = comp.scheduler.execution_timestamps[comp.default_execution_id][-1].absolute - term_cond = termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] + term_cond = termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] if term_cond.start_inclusive: assert term_cond.start - consideration_set_execution_abs_value == final_timestamp @@ -1284,29 +1270,29 @@ def test_TimeInterval_no_dependencies(self, conditions, termination_conds): @pytest.mark.parametrize( 'repeat, unit, expected_repeat', [ - (1, None, 1 * _unit_registry.ms), - ('1ms', None, 1 * _unit_registry.ms), - (1 * _unit_registry.ms, None, 1 * _unit_registry.ms), - (1, 'ms', 1 * _unit_registry.ms), - (1, _unit_registry.ms, 1 * _unit_registry.ms), - ('1', _unit_registry.ms, 1 * _unit_registry.ms), - (1 * _unit_registry.ms, _unit_registry.ns, 1 * _unit_registry.ms), - (1000 * _unit_registry.ms, None, 1000 * _unit_registry.ms), + (1, None, 1 * pnl._unit_registry.ms), + ('1ms', None, 1 * pnl._unit_registry.ms), + (1 * pnl._unit_registry.ms, None, 1 * pnl._unit_registry.ms), + (1, 'ms', 1 * pnl._unit_registry.ms), + (1, pnl._unit_registry.ms, 1 * pnl._unit_registry.ms), + ('1', pnl._unit_registry.ms, 1 * pnl._unit_registry.ms), + (1 * pnl._unit_registry.ms, pnl._unit_registry.ns, 1 * pnl._unit_registry.ms), + (1000 * pnl._unit_registry.ms, None, 1000 * pnl._unit_registry.ms), ] ) def test_TimeInterval_time_specs(self, repeat, unit, expected_repeat): if unit is None: - c = TimeInterval(repeat=repeat) + c = pnl.TimeInterval(repeat=repeat) else: - c = TimeInterval(repeat=repeat, unit=unit) + c = pnl.TimeInterval(repeat=repeat, unit=unit) assert c.repeat == expected_repeat @pytest.mark.parametrize( 'repeat, inclusive, last_time', [ - (10, True, 10 * _unit_registry.ms), - (10, False, 11 * _unit_registry.ms), + (10, True, 10 * pnl._unit_registry.ms), + (10, False, 11 * pnl._unit_registry.ms), ] ) def test_TimeTermination( @@ -1319,7 +1305,7 @@ def test_TimeTermination( _, comp = three_node_linear_composition comp.scheduler.termination_conds = { - TimeScale.ENVIRONMENT_STATE_UPDATE: TimeTermination(repeat, inclusive) + pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.TimeTermination(repeat, inclusive) } list(comp.scheduler.run()) diff --git a/tests/scheduling/test_scheduler.py b/tests/scheduling/test_scheduler.py index a1990793..c7b92d5e 100644 --- a/tests/scheduling/test_scheduler.py +++ b/tests/scheduling/test_scheduler.py @@ -8,17 +8,6 @@ import graph_scheduler as gs -from psyneulink import _unit_registry -from psyneulink.core.components.functions.stateful.integratorfunctions import DriftDiffusionIntegrator, SimpleIntegrator -from psyneulink.core.components.functions.nonstateful.transferfunctions import Linear -from psyneulink.core.components.mechanisms.processing.integratormechanism import IntegratorMechanism -from psyneulink.core.components.mechanisms.processing.transfermechanism import TransferMechanism -from psyneulink.core.components.projections.pathway.mappingprojection import MappingProjection -from psyneulink.core.compositions.composition import Composition, EdgeType -from psyneulink.core.globals.keywords import VALUE -from psyneulink.core.scheduling.condition import AfterNCalls, AfterNPasses, AfterNEnvironmentStateUpdates, AfterPass, All, AllHaveRun, Always, Any, AtNCalls, AtPass, BeforeNCalls, BeforePass, EveryNCalls, EveryNPasses, JustRan, TimeInterval, WhenFinished -from psyneulink.core.scheduling.time import TimeScale -from psyneulink.library.components.mechanisms.processing.integrator.ddm import DDM logger = logging.getLogger(__name__) @@ -64,7 +53,7 @@ def test_create_multiple_contexts(self): graph = {'A': set()} scheduler = gs.Scheduler(graph) - scheduler.get_clock(scheduler.default_execution_id)._increment_time(TimeScale.ENVIRONMENT_STATE_UPDATE) + scheduler.get_clock(scheduler.default_execution_id)._increment_time(pnl.TimeScale.ENVIRONMENT_STATE_UPDATE) eid = 'eid' eid1 = 'eid1' @@ -72,7 +61,7 @@ def test_create_multiple_contexts(self): assert scheduler.clocks[eid].time.environment_state_update == 0 - scheduler.get_clock(scheduler.default_execution_id)._increment_time(TimeScale.ENVIRONMENT_STATE_UPDATE) + scheduler.get_clock(scheduler.default_execution_id)._increment_time(pnl.TimeScale.ENVIRONMENT_STATE_UPDATE) assert scheduler.clocks[eid].time.environment_state_update == 0 @@ -82,19 +71,19 @@ def test_create_multiple_contexts(self): @pytest.mark.psyneulink def test_two_compositions_one_scheduler(self): - comp1 = Composition() - comp2 = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + comp1 = pnl.Composition() + comp2 = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') comp1.add_node(A) comp2.add_node(A) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp1)) - sched.add_condition(A, BeforeNCalls(A, 5, time_scale=TimeScale.LIFE)) + sched.add_condition(A, pnl.BeforeNCalls(A, 5, time_scale=pnl.TimeScale.LIFE)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(6) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNPasses(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(6) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNPasses(1) comp1.run( inputs={A: [[0], [1], [2], [3], [4], [5]]}, scheduler=sched, @@ -123,17 +112,17 @@ def test_two_compositions_one_scheduler(self): @pytest.mark.psyneulink def test_one_composition_two_contexts(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') comp.add_node(A) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, BeforeNCalls(A, 5, time_scale=TimeScale.LIFE)) + sched.add_condition(A, pnl.BeforeNCalls(A, 5, time_scale=pnl.TimeScale.LIFE)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(6) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNPasses(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(6) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNPasses(1) eid = 'eid' comp.run( inputs={A: [[0], [1], [2], [3], [4], [5]]}, @@ -180,39 +169,39 @@ def test_one_composition_two_contexts(self): @pytest.mark.psyneulink def test_change_termination_condition(self): - D = DDM( - function=DriftDiffusionIntegrator( + D = pnl.DDM( + function=pnl.DriftDiffusionIntegrator( threshold=10, time_step_size=1.0 ), execute_until_finished=False, reset_stateful_function_when=pnl.Never() ) - C = Composition(pathways=[D]) + C = pnl.Composition(pathways=[D]) - D.set_log_conditions(VALUE) + D.set_log_conditions(pnl.VALUE) def change_termination_processing(): if C.termination_processing is None: - C.scheduler.termination_conds = {TimeScale.ENVIRONMENT_STATE_UPDATE: WhenFinished(D)} - C.termination_processing = {TimeScale.ENVIRONMENT_STATE_UPDATE: WhenFinished(D)} - elif isinstance(C.termination_processing[TimeScale.ENVIRONMENT_STATE_UPDATE], AllHaveRun): - C.scheduler.termination_conds = {TimeScale.ENVIRONMENT_STATE_UPDATE: WhenFinished(D)} - C.termination_processing = {TimeScale.ENVIRONMENT_STATE_UPDATE: WhenFinished(D)} + C.scheduler.termination_conds = {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.WhenFinished(D)} + C.termination_processing = {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.WhenFinished(D)} + elif isinstance(C.termination_processing[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE], pnl.AllHaveRun): + C.scheduler.termination_conds = {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.WhenFinished(D)} + C.termination_processing = {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.WhenFinished(D)} else: - C.scheduler.termination_conds = {TimeScale.ENVIRONMENT_STATE_UPDATE: AllHaveRun()} - C.termination_processing = {TimeScale.ENVIRONMENT_STATE_UPDATE: AllHaveRun()} + C.scheduler.termination_conds = {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.AllHaveRun()} + C.termination_processing = {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.AllHaveRun()} change_termination_processing() C.run(inputs={D: [[1.0], [2.0]]}, - # termination_processing={TimeScale.ENVIRONMENT_STATE_UPDATE: WhenFinished(D)}, + # termination_processing={pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.WhenFinished(D)}, call_after_trial=change_termination_processing, reset_stateful_functions_when=pnl.AtConsiderationSetExecution(0), num_trials=4) # EnvironmentStateUpdate 0: - # input = 1.0, termination condition = WhenFinished + # input = 1.0, termination condition = pnl.WhenFinished # 10 passes (value = 1.0, 2.0 ... 9.0, 10.0) # EnvironmentStateUpdate 1: - # input = 2.0, termination condition = AllHaveRun + # input = 2.0, termination condition = pnl.AllHaveRun # 1 pass (value = 2.0) expected_results = [[np.array([10.]), np.array([10.])], [np.array([2.]), np.array([1.])], @@ -227,8 +216,8 @@ def test_default_condition_1(self): C = pnl.TransferMechanism(name='C') comp = pnl.Composition(pathways=[[A, C], [A, B, C]]) - comp.scheduler.add_condition(A, AtPass(1)) - comp.scheduler.add_condition(B, Always()) + comp.scheduler.add_condition(A, pnl.AtPass(1)) + comp.scheduler.add_condition(B, pnl.Always()) output = list(comp.scheduler.run()) expected_output = [B, A, B, C] @@ -241,7 +230,7 @@ def test_default_condition_2(self): C = pnl.TransferMechanism(name='C') comp = pnl.Composition(pathways=[[A, B], [C]]) - comp.scheduler.add_condition(C, AtPass(1)) + comp.scheduler.add_condition(C, pnl.AtPass(1)) output = list(comp.scheduler.run()) expected_output = [A, B, {C, A}] @@ -254,8 +243,8 @@ def test_exact_time_mode(self): ) # these cannot run at same execution set unless in EXACT_TIME - sched.add_condition('A', TimeInterval(start=1)) - sched.add_condition('B', TimeInterval(start=1)) + sched.add_condition('A', pnl.TimeInterval(start=1)) + sched.add_condition('B', pnl.TimeInterval(start=1)) list(sched.run()) @@ -317,33 +306,33 @@ class TestLinear: @classmethod def setup_class(self): - self.orig_is_finished_flag = TransferMechanism.is_finished_flag - self.orig_is_finished = TransferMechanism.is_finished - TransferMechanism.is_finished_flag = True - TransferMechanism.is_finished = lambda self, context: self.is_finished_flag + self.orig_is_finished_flag = pnl.TransferMechanism.is_finished_flag + self.orig_is_finished = pnl.TransferMechanism.is_finished + pnl.TransferMechanism.is_finished_flag = True + pnl.TransferMechanism.is_finished = lambda self, context: self.is_finished_flag @classmethod def teardown_class(self): - del TransferMechanism.is_finished_flag - del TransferMechanism.is_finished - TransferMechanism.is_finished_flag = self.orig_is_finished_flag - TransferMechanism.is_finished = self.orig_is_finished + del pnl.TransferMechanism.is_finished_flag + del pnl.TransferMechanism.is_finished + pnl.TransferMechanism.is_finished_flag = self.orig_is_finished_flag + pnl.TransferMechanism.is_finished = self.orig_is_finished def test_no_termination_conds(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') - C = TransferMechanism(function=Linear(intercept=1.5), name='scheduler-pytests-C') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='scheduler-pytests-C') for m in [A, B, C]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) - comp.add_projection(MappingProjection(), B, C) + comp.add_projection(pnl.MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), B, C) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, EveryNCalls(A, 2)) - sched.add_condition(C, EveryNCalls(B, 3)) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.EveryNCalls(A, 2)) + sched.add_condition(C, pnl.EveryNCalls(B, 3)) output = list(sched.run()) @@ -355,24 +344,24 @@ def test_no_termination_conds(self): # tests below are copied from old scheduler, need renaming def test_1(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') - C = TransferMechanism(function=Linear(intercept=1.5), name='scheduler-pytests-C') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='scheduler-pytests-C') for m in [A, B, C]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) - comp.add_projection(MappingProjection(), B, C) + comp.add_projection(pnl.MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), B, C) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, EveryNCalls(A, 2)) - sched.add_condition(C, EveryNCalls(B, 3)) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.EveryNCalls(A, 2)) + sched.add_condition(C, pnl.EveryNCalls(B, 3)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(C, 4, time_scale=TimeScale.ENVIRONMENT_STATE_UPDATE) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(C, 4, time_scale=pnl.TimeScale.ENVIRONMENT_STATE_UPDATE) output = list(sched.run(termination_conds=termination_conds)) expected_output = [ @@ -385,24 +374,24 @@ def test_1(self): assert output == pytest.helpers.setify_expected_output(expected_output) def test_1b(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') - C = TransferMechanism(function=Linear(intercept=1.5), name='scheduler-pytests-C') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='scheduler-pytests-C') for m in [A, B, C]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) - comp.add_projection(MappingProjection(), B, C) + comp.add_projection(pnl.MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), B, C) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, Any(EveryNCalls(A, 2), AfterPass(1))) - sched.add_condition(C, EveryNCalls(B, 3)) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.Any(pnl.EveryNCalls(A, 2), pnl.AfterPass(1))) + sched.add_condition(C, pnl.EveryNCalls(B, 3)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(C, 4, time_scale=TimeScale.ENVIRONMENT_STATE_UPDATE) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(C, 4, time_scale=pnl.TimeScale.ENVIRONMENT_STATE_UPDATE) output = list(sched.run(termination_conds=termination_conds)) expected_output = [ @@ -415,48 +404,48 @@ def test_1b(self): assert output == pytest.helpers.setify_expected_output(expected_output) def test_2(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') - C = TransferMechanism(function=Linear(intercept=1.5), name='scheduler-pytests-C') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='scheduler-pytests-C') for m in [A, B, C]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) - comp.add_projection(MappingProjection(), B, C) + comp.add_projection(pnl.MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), B, C) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, EveryNCalls(A, 2)) - sched.add_condition(C, EveryNCalls(B, 2)) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.EveryNCalls(A, 2)) + sched.add_condition(C, pnl.EveryNCalls(B, 2)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(C, 1, time_scale=TimeScale.ENVIRONMENT_STATE_UPDATE) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(C, 1, time_scale=pnl.TimeScale.ENVIRONMENT_STATE_UPDATE) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A, A, B, A, A, B, C] assert output == pytest.helpers.setify_expected_output(expected_output) def test_3(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') - C = TransferMechanism(function=Linear(intercept=1.5), name='scheduler-pytests-C') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='scheduler-pytests-C') for m in [A, B, C]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) - comp.add_projection(MappingProjection(), B, C) + comp.add_projection(pnl.MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), B, C) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, EveryNCalls(A, 2)) - sched.add_condition(C, All(AfterNCalls(B, 2), EveryNCalls(B, 1))) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.EveryNCalls(A, 2)) + sched.add_condition(C, pnl.All(pnl.AfterNCalls(B, 2), pnl.EveryNCalls(B, 1))) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(C, 4, time_scale=TimeScale.ENVIRONMENT_STATE_UPDATE) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(C, 4, time_scale=pnl.TimeScale.ENVIRONMENT_STATE_UPDATE) output = list(sched.run(termination_conds=termination_conds)) expected_output = [ @@ -466,24 +455,24 @@ def test_3(self): assert output == pytest.helpers.setify_expected_output(expected_output) def test_6(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') - C = TransferMechanism(function=Linear(intercept=1.5), name='scheduler-pytests-C') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='scheduler-pytests-C') for m in [A, B, C]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) - comp.add_projection(MappingProjection(), B, C) + comp.add_projection(pnl.MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), B, C) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, BeforePass(5)) - sched.add_condition(B, AfterNCalls(A, 5)) - sched.add_condition(C, AfterNCalls(B, 1)) + sched.add_condition(A, pnl.BeforePass(5)) + sched.add_condition(B, pnl.AfterNCalls(A, 5)) + sched.add_condition(C, pnl.AfterNCalls(B, 1)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(C, 3) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(C, 3) output = list(sched.run(termination_conds=termination_conds)) expected_output = [ @@ -493,24 +482,24 @@ def test_6(self): assert output == pytest.helpers.setify_expected_output(expected_output) def test_6_two_environment_state_updates(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') - C = TransferMechanism(function=Linear(intercept=1.5), name='scheduler-pytests-C') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='scheduler-pytests-C') for m in [A, B, C]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) - comp.add_projection(MappingProjection(), B, C) + comp.add_projection(pnl.MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), B, C) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, BeforePass(5)) - sched.add_condition(B, AfterNCalls(A, 5)) - sched.add_condition(C, AfterNCalls(B, 1)) + sched.add_condition(A, pnl.BeforePass(5)) + sched.add_condition(B, pnl.AfterNCalls(A, 5)) + sched.add_condition(C, pnl.AfterNCalls(B, 1)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(2) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(C, 3) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(2) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(C, 3) comp.run( inputs={A: [[0], [1], [2], [3], [4], [5]]}, scheduler=sched, @@ -526,63 +515,63 @@ def test_6_two_environment_state_updates(self): assert output == pytest.helpers.setify_expected_output(expected_output) def test_7(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') for m in [A, B]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), A, B) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, EveryNCalls(A, 2)) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.EveryNCalls(A, 2)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = Any(AfterNCalls(A, 1), AfterNCalls(B, 1)) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.Any(pnl.AfterNCalls(A, 1), pnl.AfterNCalls(B, 1)) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A] assert output == pytest.helpers.setify_expected_output(expected_output) def test_8(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') for m in [A, B]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), A, B) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, EveryNCalls(A, 2)) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.EveryNCalls(A, 2)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = All(AfterNCalls(A, 1), AfterNCalls(B, 1)) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.All(pnl.AfterNCalls(A, 1), pnl.AfterNCalls(B, 1)) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A, A, B] assert output == pytest.helpers.setify_expected_output(expected_output) def test_9(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') for m in [A, B]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), A, B) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, WhenFinished(A)) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.WhenFinished(A)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(B, 2) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(B, 2) output = [] i = 0 @@ -597,114 +586,114 @@ def test_9(self): assert output == pytest.helpers.setify_expected_output(expected_output) def test_9b(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') A.is_finished_flag = False - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') for m in [A, B]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), A, B) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, WhenFinished(A)) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.WhenFinished(A)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(5) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(5) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A, A, A, A, A] assert output == pytest.helpers.setify_expected_output(expected_output) def test_10(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') A.is_finished_flag = True - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') for m in [A, B]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), A, B) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, Any(WhenFinished(A), AfterNCalls(A, 3))) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.Any(pnl.WhenFinished(A), pnl.AfterNCalls(A, 3))) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(B, 5) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(B, 5) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A, B, A, B, A, B, A, B, A, B] assert output == pytest.helpers.setify_expected_output(expected_output) def test_10b(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') A.is_finished_flag = False - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') for m in [A, B]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), A, B) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, Any(WhenFinished(A), AfterNCalls(A, 3))) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.Any(pnl.WhenFinished(A), pnl.AfterNCalls(A, 3))) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(B, 4) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(B, 4) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A, A, A, B, A, B, A, B, A, B] assert output == pytest.helpers.setify_expected_output(expected_output) def test_10c(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') A.is_finished_flag = True - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') for m in [A, B]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), A, B) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, All(WhenFinished(A), AfterNCalls(A, 3))) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.All(pnl.WhenFinished(A), pnl.AfterNCalls(A, 3))) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(B, 4) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(B, 4) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A, A, A, B, A, B, A, B, A, B] assert output == pytest.helpers.setify_expected_output(expected_output) def test_10d(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') A.is_finished_flag = False - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') for m in [A, B]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), A, B) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, All(WhenFinished(A), AfterNCalls(A, 3))) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.All(pnl.WhenFinished(A), pnl.AfterNCalls(A, 3))) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AtPass(10) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AtPass(10) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A, A, A, A, A, A, A, A, A, A] @@ -714,107 +703,107 @@ def test_10d(self): # tests with linear compositions ######################################## def test_linear_AAB(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') for m in [A, B]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), A, B) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, EveryNCalls(A, 2)) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.EveryNCalls(A, 2)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNCalls(B, 2, time_scale=TimeScale.ENVIRONMENT_SEQUENCE) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(B, 2, time_scale=TimeScale.ENVIRONMENT_STATE_UPDATE) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNCalls(B, 2, time_scale=pnl.TimeScale.ENVIRONMENT_SEQUENCE) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(B, 2, time_scale=pnl.TimeScale.ENVIRONMENT_STATE_UPDATE) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A, A, B, A, A, B] assert output == pytest.helpers.setify_expected_output(expected_output) def test_linear_ABB(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') for m in [A, B]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), A, B) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, Any(AtPass(0), EveryNCalls(B, 2))) - sched.add_condition(B, Any(EveryNCalls(A, 1), EveryNCalls(B, 1))) + sched.add_condition(A, pnl.Any(pnl.AtPass(0), pnl.EveryNCalls(B, 2))) + sched.add_condition(B, pnl.Any(pnl.EveryNCalls(A, 1), pnl.EveryNCalls(B, 1))) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(B, 8, time_scale=TimeScale.ENVIRONMENT_STATE_UPDATE) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(B, 8, time_scale=pnl.TimeScale.ENVIRONMENT_STATE_UPDATE) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A, B, B, A, B, B, A, B, B, A, B, B] assert output == pytest.helpers.setify_expected_output(expected_output) def test_linear_ABBCC(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') - C = TransferMechanism(function=Linear(intercept=1.5), name='scheduler-pytests-C') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='scheduler-pytests-C') for m in [A, B, C]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) - comp.add_projection(MappingProjection(), B, C) + comp.add_projection(pnl.MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), B, C) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, Any(AtPass(0), EveryNCalls(C, 2))) - sched.add_condition(B, Any(JustRan(A), JustRan(B))) - sched.add_condition(C, Any(EveryNCalls(B, 2), JustRan(C))) + sched.add_condition(A, pnl.Any(pnl.AtPass(0), pnl.EveryNCalls(C, 2))) + sched.add_condition(B, pnl.Any(pnl.JustRan(A), pnl.JustRan(B))) + sched.add_condition(C, pnl.Any(pnl.EveryNCalls(B, 2), pnl.JustRan(C))) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(C, 4, time_scale=TimeScale.ENVIRONMENT_STATE_UPDATE) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(C, 4, time_scale=pnl.TimeScale.ENVIRONMENT_STATE_UPDATE) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A, B, B, C, C, A, B, B, C, C] assert output == pytest.helpers.setify_expected_output(expected_output) def test_linear_ABCBC(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') - C = TransferMechanism(function=Linear(intercept=1.5), name='scheduler-pytests-C') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='scheduler-pytests-C') for m in [A, B, C]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) - comp.add_projection(MappingProjection(), B, C) + comp.add_projection(pnl.MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), B, C) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, Any(AtPass(0), EveryNCalls(C, 2))) - sched.add_condition(B, Any(EveryNCalls(A, 1), EveryNCalls(C, 1))) - sched.add_condition(C, EveryNCalls(B, 1)) + sched.add_condition(A, pnl.Any(pnl.AtPass(0), pnl.EveryNCalls(C, 2))) + sched.add_condition(B, pnl.Any(pnl.EveryNCalls(A, 1), pnl.EveryNCalls(C, 1))) + sched.add_condition(C, pnl.EveryNCalls(B, 1)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(C, 4, time_scale=TimeScale.ENVIRONMENT_STATE_UPDATE) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(C, 4, time_scale=pnl.TimeScale.ENVIRONMENT_STATE_UPDATE) output = list(sched.run(termination_conds=termination_conds)) expected_output = [A, B, C, B, C, A, B, C, B, C] assert output == pytest.helpers.setify_expected_output(expected_output) def test_one_run_twice(self): - A = IntegratorMechanism( + A = pnl.IntegratorMechanism( name='A', default_variable=[0], - function=SimpleIntegrator( + function=pnl.SimpleIntegrator( rate=.5, ) ) - c = Composition(pathways=[A]) + c = pnl.Composition(pathways=[A]) - term_conds = {TimeScale.ENVIRONMENT_STATE_UPDATE: AfterNCalls(A, 2)} + term_conds = {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.AfterNCalls(A, 2)} stim_list = {A: [[1]]} c.run( @@ -831,27 +820,27 @@ def test_one_run_twice(self): np.testing.assert_allclose(expected_output[i], terminal_mech.get_output_values(c)[i]) def test_two_AAB(self): - A = IntegratorMechanism( + A = pnl.IntegratorMechanism( name='A', default_variable=[0], - function=SimpleIntegrator( + function=pnl.SimpleIntegrator( rate=.5 ) ) - B = TransferMechanism( + B = pnl.TransferMechanism( name='B', default_variable=[0], - function=Linear(slope=2.0), + function=pnl.Linear(slope=2.0), ) - c = Composition(pathways=[A, B]) + c = pnl.Composition(pathways=[A, B]) - term_conds = {TimeScale.ENVIRONMENT_STATE_UPDATE: AfterNCalls(B, 1)} + term_conds = {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.AfterNCalls(B, 1)} stim_list = {A: [[1]]} sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(c)) - sched.add_condition(B, EveryNCalls(A, 2)) + sched.add_condition(B, pnl.EveryNCalls(A, 2)) c.scheduler = sched c.run( @@ -868,28 +857,28 @@ def test_two_AAB(self): np.testing.assert_allclose(expected_output[i], terminal_mech.get_output_values(c)[i]) def test_two_ABB(self): - A = TransferMechanism( + A = pnl.TransferMechanism( name='A', default_variable=[0], - function=Linear(slope=2.0), + function=pnl.Linear(slope=2.0), ) - B = IntegratorMechanism( + B = pnl.IntegratorMechanism( name='B', default_variable=[0], - function=SimpleIntegrator( + function=pnl.SimpleIntegrator( rate=.5 ) ) - c = Composition(pathways=[A, B]) + c = pnl.Composition(pathways=[A, B]) - term_conds = {TimeScale.ENVIRONMENT_STATE_UPDATE: AfterNCalls(B, 2)} + term_conds = {pnl.TimeScale.ENVIRONMENT_STATE_UPDATE: pnl.AfterNCalls(B, 2)} stim_list = {A: [[1]]} sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(c)) - sched.add_condition(A, Any(AtPass(0), AfterNCalls(B, 2))) - sched.add_condition(B, Any(JustRan(A), JustRan(B))) + sched.add_condition(A, pnl.Any(pnl.AtPass(0), pnl.AfterNCalls(B, 2))) + sched.add_condition(B, pnl.Any(pnl.JustRan(A), pnl.JustRan(B))) c.scheduler = sched c.run( @@ -914,41 +903,41 @@ def test_two_ABB(self): class TestBranching: @classmethod def setup_class(self): - self.orig_is_finished_flag = TransferMechanism.is_finished_flag - self.orig_is_finished = TransferMechanism.is_finished - TransferMechanism.is_finished_flag = True - TransferMechanism.is_finished = lambda self, context: self.is_finished_flag + self.orig_is_finished_flag = pnl.TransferMechanism.is_finished_flag + self.orig_is_finished = pnl.TransferMechanism.is_finished + pnl.TransferMechanism.is_finished_flag = True + pnl.TransferMechanism.is_finished = lambda self, context: self.is_finished_flag @classmethod def teardown_class(self): - del TransferMechanism.is_finished_flag - del TransferMechanism.is_finished - TransferMechanism.is_finished_flag = self.orig_is_finished_flag - TransferMechanism.is_finished = self.orig_is_finished + del pnl.TransferMechanism.is_finished_flag + del pnl.TransferMechanism.is_finished + pnl.TransferMechanism.is_finished_flag = self.orig_is_finished_flag + pnl.TransferMechanism.is_finished = self.orig_is_finished # triangle: A # / \ # B C def test_triangle_1(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') - C = TransferMechanism(function=Linear(intercept=1.5), name='scheduler-pytests-C') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='scheduler-pytests-C') for m in [A, B, C]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) - comp.add_projection(MappingProjection(), A, C) + comp.add_projection(pnl.MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), A, C) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, EveryNCalls(A, 1)) - sched.add_condition(C, EveryNCalls(A, 1)) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.EveryNCalls(A, 1)) + sched.add_condition(C, pnl.EveryNCalls(A, 1)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(C, 3, time_scale=TimeScale.ENVIRONMENT_STATE_UPDATE) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(C, 3, time_scale=pnl.TimeScale.ENVIRONMENT_STATE_UPDATE) output = list(sched.run(termination_conds=termination_conds)) expected_output = [ @@ -960,24 +949,24 @@ def test_triangle_1(self): assert output == pytest.helpers.setify_expected_output(expected_output) def test_triangle_2(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') - C = TransferMechanism(function=Linear(intercept=1.5), name='scheduler-pytests-C') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='scheduler-pytests-C') for m in [A, B, C]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) - comp.add_projection(MappingProjection(), A, C) + comp.add_projection(pnl.MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), A, C) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, EveryNCalls(A, 1)) - sched.add_condition(C, EveryNCalls(A, 2)) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.EveryNCalls(A, 1)) + sched.add_condition(C, pnl.EveryNCalls(A, 2)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(C, 3, time_scale=TimeScale.ENVIRONMENT_STATE_UPDATE) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(C, 3, time_scale=pnl.TimeScale.ENVIRONMENT_STATE_UPDATE) output = list(sched.run(termination_conds=termination_conds)) expected_output = [ @@ -992,24 +981,24 @@ def test_triangle_2(self): assert output == pytest.helpers.setify_expected_output(expected_output) def test_triangle_3(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') - C = TransferMechanism(function=Linear(intercept=1.5), name='scheduler-pytests-C') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='scheduler-pytests-C') for m in [A, B, C]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) - comp.add_projection(MappingProjection(), A, C) + comp.add_projection(pnl.MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), A, C) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, EveryNCalls(A, 2)) - sched.add_condition(C, EveryNCalls(A, 3)) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.EveryNCalls(A, 2)) + sched.add_condition(C, pnl.EveryNCalls(A, 3)) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(C, 2, time_scale=TimeScale.ENVIRONMENT_STATE_UPDATE) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(C, 2, time_scale=pnl.TimeScale.ENVIRONMENT_STATE_UPDATE) output = list(sched.run(termination_conds=termination_conds)) expected_output = [ @@ -1020,25 +1009,25 @@ def test_triangle_3(self): # this is test 11 of original constraint_scheduler.py def test_triangle_4(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') - C = TransferMechanism(function=Linear(intercept=1.5), name='scheduler-pytests-C') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='scheduler-pytests-C') for m in [A, B, C]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) - comp.add_projection(MappingProjection(), A, C) + comp.add_projection(pnl.MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), A, C) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, EveryNCalls(A, 2)) - sched.add_condition(C, All(WhenFinished(A), AfterNCalls(B, 3))) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.EveryNCalls(A, 2)) + sched.add_condition(C, pnl.All(pnl.WhenFinished(A), pnl.AfterNCalls(B, 3))) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(C, 1) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(C, 1) output = [] i = 0 A.is_finished_flag = False @@ -1053,25 +1042,25 @@ def test_triangle_4(self): assert output == pytest.helpers.setify_expected_output(expected_output) def test_triangle_4b(self): - comp = Composition() - A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') - B = TransferMechanism(function=Linear(intercept=4.0), name='scheduler-pytests-B') - C = TransferMechanism(function=Linear(intercept=1.5), name='scheduler-pytests-C') + comp = pnl.Composition() + A = pnl.TransferMechanism(function=pnl.Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A') + B = pnl.TransferMechanism(function=pnl.Linear(intercept=4.0), name='scheduler-pytests-B') + C = pnl.TransferMechanism(function=pnl.Linear(intercept=1.5), name='scheduler-pytests-C') for m in [A, B, C]: comp.add_node(m) - comp.add_projection(MappingProjection(), A, B) - comp.add_projection(MappingProjection(), A, C) + comp.add_projection(pnl.MappingProjection(), A, B) + comp.add_projection(pnl.MappingProjection(), A, C) sched = pnl.Scheduler(**pytest.helpers.composition_to_scheduler_args(comp)) - sched.add_condition(A, EveryNPasses(1)) - sched.add_condition(B, EveryNCalls(A, 2)) - sched.add_condition(C, All(WhenFinished(A), AfterNCalls(B, 3))) + sched.add_condition(A, pnl.EveryNPasses(1)) + sched.add_condition(B, pnl.EveryNCalls(A, 2)) + sched.add_condition(C, pnl.All(pnl.WhenFinished(A), pnl.AfterNCalls(B, 3))) termination_conds = {} - termination_conds[TimeScale.ENVIRONMENT_SEQUENCE] = AfterNEnvironmentStateUpdates(1) - termination_conds[TimeScale.ENVIRONMENT_STATE_UPDATE] = AfterNCalls(C, 1) + termination_conds[pnl.TimeScale.ENVIRONMENT_SEQUENCE] = pnl.AfterNEnvironmentStateUpdates(1) + termination_conds[pnl.TimeScale.ENVIRONMENT_STATE_UPDATE] = pnl.AfterNCalls(C, 1) output = [] i = 0 A.is_finished_flag = False @@ -1092,24 +1081,24 @@ def test_triangle_4b(self): # this is test 4 of original constraint_scheduler.py # this test has an implicit priority set of A