Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_mech_in_feed_dict_for_empty_type(self):
comp = Composition()
A = TransferMechanism(default_variable=[0])
B = TransferMechanism(name='B')
comp.add_mechanism(A)
comp.add_mechanism(B)
comp.add_projection(A, MappingProjection(), B)
comp._analyze_graph()
feed_dict_origin = {A: [[0]]}
feed_dict_monitored = {B: [[0]]}
with pytest.raises(ValueError):
comp._validate_feed_dict(feed_dict_monitored, comp.get_mechanisms_by_role(MechanismRole.MONITORED), "monitored")
def test_add_input_state_with_projection_using_add_states(self):
T1 = TransferMechanism()
I = InputState(projections=[T1])
T2 = TransferMechanism()
T2.add_states([I])
assert T2.input_states[1].path_afferents[0].sender.owner is T1
def test_projection_no_args_projection_spec_with_default(self):
p = MappingProjection()
T = TransferMechanism(default_variable=[[0, 0]], input_states=[p])
np.testing.assert_array_equal(T.instance_defaults.variable, np.array([[0, 0]]))
assert len(T.input_states) == 1
#
# B--A
# / \
# C------D
# \ |
# -----+-> E
#
# C: 4 x 5 = 20
# B: 20 x 1 = 20
# A: f(20)[0] = 0.50838675
# D: 20 x 5 + 0.50838675 = 100.50838675
# E: (20 + 100.50838675) x 5 = 650.83865743
comp = Composition()
C = TransferMechanism(name="C", function=Linear(slope=5.0))
D = TransferMechanism(name="D", function=Linear(slope=5.0))
B = ObjectiveMechanism(function=Linear,
#monitored_output_states=[C], #setup later
name='LC ObjectiveMechanism')
A = LCControlMechanism(name="A", modulation=ModulationParam.ADDITIVE)
# modulated_mechanisms=D,
# objective_mechanism=B # setup later
E = TransferMechanism(name="E", function=Linear(slope=5.0))
comp.add_mechanism(A)
comp.add_mechanism(B)
comp.add_mechanism(C)
comp.add_mechanism(D)
comp.add_mechanism(E)
comp.add_projection(A, ControlProjection(sender=A.control_signals[0], receiver=D.parameter_states[SLOPE]), D)
comp.add_projection(C, MappingProjection(sender=C, receiver=B), B)
comp.add_projection(C, MappingProjection(sender=C, receiver=D), D)
def test_sender_receiver_not_specified(self):
comp = Composition()
A = TransferMechanism(name="A [transfer]", function=Linear(slope=2.0))
B = TransferMechanism(name="B [transfer]", function=Linear(slope=5.0))
comp.add_mechanism(A)
comp.add_mechanism(B)
comp.add_projection(A, MappingProjection(), B)
comp._analyze_graph()
inputs_dict = {A: [1, 2, 3, 4]}
sched = Scheduler(composition=comp)
output = comp.run(
inputs=inputs_dict,
scheduler_processing=sched
)
assert 40.0 == output[0][0]
def test_triangle(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')
mechs = [A, B, C]
for m in mechs:
comp.add_mechanism(m)
comp.add_projection(A, MappingProjection(), B)
comp.add_projection(B, MappingProjection(), C)
assert len(comp.graph_processing.vertices) == 3
assert len(comp.graph_processing.comp_to_vertex) == 3
for m in mechs:
assert m in comp.graph_processing.comp_to_vertex
assert comp.graph_processing.get_parents_from_component(A) == []
assert comp.graph_processing.get_parents_from_component(B) == [comp.graph_processing.comp_to_vertex[A]]
assert comp.graph_processing.get_parents_from_component(C) == [comp.graph_processing.comp_to_vertex[B]]
def test_cycle_x_multiple_incoming(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')
D = TransferMechanism(function=Linear(intercept=1.5), name='D')
E = TransferMechanism(function=Linear(intercept=1.5), name='E')
mechs = [A, B, C, D, E]
for m in mechs:
comp.add_mechanism(m)
comp.add_projection(A, MappingProjection(), C)
comp.add_projection(B, MappingProjection(), C)
comp.add_projection(C, MappingProjection(), D)
comp.add_projection(C, MappingProjection(), E)
comp.add_projection(D, MappingProjection(), A)
comp.add_projection(D, MappingProjection(), B)
comp.add_projection(E, MappingProjection(), A)
comp.add_projection(E, MappingProjection(), B)
assert len(comp.graph_processing.vertices) == 5
assert len(comp.graph_processing.comp_to_vertex) == 5
def test_2_item_tuple_with_state_name_list_and_mechanism(self):
# T1 has OutputStates of with same lengths,
# so T2 should use that length for its InputState (since it is not otherwise specified
T1 = TransferMechanism(input_states=[[0,0],[0,0]])
T2 = TransferMechanism(input_states=[(['RESULT', 'RESULT-1'], T1)])
assert len(T2.input_states[0].value) == 2
assert T2.input_states[0].path_afferents[0].sender.name == 'RESULT'
assert T2.input_states[0].path_afferents[1].sender.name == 'RESULT-1'
# T1 has OutputStates with different lengths,
# so T2 should use its variable default to as format for its InputStates (since it is not otherwise specified
T1 = TransferMechanism(input_states=[[0,0],[0,0,0]])
T2 = TransferMechanism(input_states=[(['RESULT', 'RESULT-1'], T1)])
assert len(T2.input_states[0].value) == 1
assert T2.input_states[0].path_afferents[0].sender.name == 'RESULT'
assert T2.input_states[0].path_afferents[1].sender.name == 'RESULT-1'
def test_5_mechanisms_2_origins_1_terminal(benchmark, mode):
# A ----> C --
# ==> E
# B ----> D --
# 5 x 1 = 5 ----> 5 x 5 = 25 --
# 25 + 25 = 50 ==> 50 * 5 = 250
# 5 x 1 = 5 ----> 5 x 5 = 25 --
comp = Composition()
A = TransferMechanism(name="A", function=Linear(slope=1.0))
B = TransferMechanism(name="B", function=Linear(slope=1.0))
C = TransferMechanism(name="C", function=Linear(slope=5.0))
D = TransferMechanism(name="D", function=Linear(slope=5.0))
E = TransferMechanism(name="E", function=Linear(slope=5.0))
comp.add_mechanism(A)
comp.add_mechanism(B)
comp.add_mechanism(C)
comp.add_mechanism(D)
comp.add_projection(A, MappingProjection(sender=A, receiver=C), C)
comp.add_projection(B, MappingProjection(sender=B, receiver=D), D)
comp.add_mechanism(E)
comp.add_projection(C, MappingProjection(sender=C, receiver=E), E)
comp.add_projection(D, MappingProjection(sender=D, receiver=E), E)
comp._analyze_graph()
inputs_dict = {A: [5.0],
B: [5.0]}
def test_input_state_len_3_brackets_extra_1(self):
comp = Composition()
A = TransferMechanism(default_variable=[0, 1, 2], name='A')
B = TransferMechanism(default_variable=[0, 1, 2], name='B')
comp.add_mechanism(A)
comp.add_mechanism(B)
comp.add_projection(A, MappingProjection(), B)
comp._analyze_graph()
feed_dict_origin = {A: [[[0, 1, 2]]]}
feed_dict_terminal = {B: [[[0, 1, 2]]]}
comp._validate_feed_dict(feed_dict_origin, comp.get_mechanisms_by_role(MechanismRole.ORIGIN), "origin")
comp._validate_feed_dict(feed_dict_terminal, comp.get_mechanisms_by_role(MechanismRole.TERMINAL), "terminal")