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_create_scheduler_from_system_StroopDemo(self):
Color_Input = TransferMechanism(name='Color Input', function=Linear(slope=0.2995))
Word_Input = TransferMechanism(name='Word Input', function=Linear(slope=0.2995))
# Processing Mechanisms (Control)
Color_Hidden = TransferMechanism(
name='Colors Hidden',
function=Logistic(gain=(1.0, ControlProjection)),
)
Word_Hidden = TransferMechanism(
name='Words Hidden',
function=Logistic(gain=(1.0, ControlProjection)),
)
Output = TransferMechanism(
name='Output',
function=Logistic(gain=(1.0, ControlProjection)),
)
# Decision Mechanisms
Target_Rep = TransferMechanism(
name='Target Representation',
function=Linear(
slope=(
1.0,
ControlProjection(
function=Linear,
control_signal_params={ALLOCATION_SAMPLES: signalSearchRange}
)
)
),
prefs=mechanism_prefs
)
Flanker_Rep = TransferMechanism(
name='Flanker Representation',
function=Linear(
slope=(
1.0,
ControlProjection(
function=Linear,
control_signal_params={ALLOCATION_SAMPLES: signalSearchRange}
)
)
),
prefs=mechanism_prefs
)
# Processing Mechanism (Automatic)
Automatic_Component = TransferMechanism(
name='Automatic Component',
function=Linear(slope=(1.0)),
prefs=mechanism_prefs
def test_AfterNPasses(self):
comp = Composition()
A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A')
comp.add_node(A)
sched = Scheduler(composition=comp)
sched.add_condition(A, AfterNPasses(1))
termination_conds = {}
termination_conds[TimeScale.RUN] = AfterNTrials(1)
termination_conds[TimeScale.TRIAL] = 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_multisource_1(self):
comp = Composition()
A1 = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A1')
A2 = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A2')
B1 = TransferMechanism(function=Linear(intercept=4.0), name='B1')
B2 = TransferMechanism(function=Linear(intercept=4.0), name='B2')
B3 = TransferMechanism(function=Linear(intercept=4.0), name='B3')
C1 = TransferMechanism(function=Linear(intercept=1.5), name='C1')
C2 = TransferMechanism(function=Linear(intercept=.5), name='C2')
for m in [A1, A2, B1, B2, B3, C1, C2]:
comp.add_node(m)
comp.add_projection(MappingProjection(), A1, B1)
comp.add_projection(MappingProjection(), A1, B2)
comp.add_projection(MappingProjection(), A2, B1)
comp.add_projection(MappingProjection(), A2, B2)
comp.add_projection(MappingProjection(), A2, B3)
comp.add_projection(MappingProjection(), B1, C1)
comp.add_projection(MappingProjection(), B2, C1)
comp.add_projection(MappingProjection(), B1, C2)
def test_BeforeNCalls(self):
comp = Composition()
A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='A')
comp.add_node(A)
sched = Scheduler(composition=comp)
sched.add_condition(A, BeforeNCalls(A, 3))
termination_conds = {}
termination_conds[TimeScale.RUN] = AfterNTrials(1)
termination_conds[TimeScale.TRIAL] = AtPass(5)
output = list(sched.run(termination_conds=termination_conds))
expected_output = [A, A, A, set(), set()]
assert output == pytest.helpers.setify_expected_output(expected_output)
def test_one_composition_two_contexts(self):
comp = Composition()
A = TransferMechanism(function=Linear(slope=5.0, intercept=2.0), name='scheduler-pytests-A')
comp.add_node(A)
sched = Scheduler(composition=comp)
sched.add_condition(A, BeforeNCalls(A, 5, time_scale=TimeScale.LIFE))
termination_conds = {}
termination_conds[TimeScale.RUN] = AfterNTrials(6)
termination_conds[TimeScale.TRIAL] = AfterNPasses(1)
eid = 'eid'
comp.run(
inputs={A: [[0], [1], [2], [3], [4], [5]]},
scheduler=sched,
termination_processing=termination_conds,
context=eid,
)
def test_LCAMechanism_length_1(self, benchmark, mode):
T = TransferMechanism(function=Linear(slope=1.0))
L = LCAMechanism(function=Linear(slope=2.0),
self_excitation=3.0,
leak=0.5,
competition=1.0, # competition does not matter because we only have one unit
time_step_size=0.1)
C = Composition()
C.add_linear_processing_pathway([T,L])
L.reinitialize_when = Never()
# - - - - - - - Equations to be executed - - - - - - -
# new_transfer_input =
# previous_transfer_input
# + (leak * previous_transfer_input_1 + self_excitation * result1 + competition * result2 + outside_input1) * dt
# + noise
# result = new_transfer_input*2.0
rate=.5
)
)
B = IntegratorMechanism(
name='B',
default_variable=[0],
function=SimpleIntegrator(
rate=1
)
)
C = TransferMechanism(
name='C',
default_variable=[0],
function=Linear(slope=2.0),
)
p = Process(
default_variable=[0],
pathway=[A, C],
name='p'
)
q = Process(
default_variable=[0],
pathway=[B, C],
name='q'
)
s = System(
processes=[p, q],
a_v_FitzHughNagumo=-1.0,
b_v_FitzHughNagumo=1.0,
c_v_FitzHughNagumo=1.0,
d_v_FitzHughNagumo=0.0,
e_v_FitzHughNagumo=-1.0,
f_v_FitzHughNagumo=1.0,
a_w_FitzHughNagumo=1.0,
b_w_FitzHughNagumo=-1.0,
c_w_FitzHughNagumo=0.0,
t_0_FitzHughNagumo=0.0,
base_level_gain=G, # Additionally, we set the parameters k and G to compute the gain equation
scaling_factor_gain=k,
initial_v_FitzHughNagumo=initial_v, # Initialize v
initial_w_FitzHughNagumo=initial_w, # Initialize w
objective_mechanism=pnl.ObjectiveMechanism(
function=psyneulink.core.components.functions.transferfunctions.Linear,
monitored_output_ports=[(
decision_layer, # Project output of T1 and T2 but not distractor from decision layer to LC
np.array([[lcwt], [lcwt], [0.0]])
)],
name='Combine values'
),
modulated_mechanisms=[decision_layer, response_layer], # Modulate gain of decision & response layers
name='LC'
)
# Log value of LC
LC.set_log_conditions('value')
# Set initial gain to G + k*initial_w, when the System runs the very first time,
# since the decison layer executes before the LC and hence needs one initial gain value to start with.
for output_port in LC.output_ports:
----------
control_signal
see `control_signal `
:default value: None
:type:
:read only: True
function
see `function `
:default value: `Linear`
:type: `Function`
"""
function = Parameter(Linear, stateful=False, loggable=False)
control_signal = Parameter(None, read_only=True, getter=_control_signal_getter, setter=_control_signal_setter, pnl_internal=True)
control_signal_params = Parameter(
None,
stateful=False,
loggable=False,
read_only=True,
user=False,
pnl_internal=True
)
projection_sender = ControlMechanism
@tc.typecheck
def __init__(self,