Skip to content

llvm: Add support for standard DDM output ports #3426

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jvesely merged 3 commits into from
Jan 5, 2026
Merged

llvm: Add support for standard DDM output ports #3426

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f50cb8f
tests/DDM: Split selected_input_array based on expected result
jvesely Oct 6, 2025
3c32e9c
tests/DDM: Add compiled variants of DECISION_OUTCOME tests
jvesely Dec 17, 2025
63af4f4
llvm, Mechanism: Add workaround for 'input_port_variables' Parameter
jvesely Nov 12, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion psyneulink/core/components/mechanisms/mechanism.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3115,7 +3115,7 @@ def _get_param_base_ptr(b, i):
return params_out, builder


def _gen_llvm_output_port_parse_variable(self, ctx, builder, mech_params, mech_state, value, port):
def _gen_llvm_output_port_parse_variable(self, ctx, builder, mech_in, mech_params, mech_state, value, port):
"""Create output port variable based on the variable specification."""
port_spec = port._variable_spec

Expand All @@ -3133,9 +3133,16 @@ def _gen_llvm_output_port_parse_variable(self, ctx, builder, mech_params, mech_s
for spec in canonical_port_spec:
param_name, indices = spec

# "value" is not always stored in mechanism parameters,
# use the location of the freshly calculated result instead
if param_name == VALUE:
base = value

# "input_port_variables" is not used in compilation,
# but it should match the mechanism input
elif param_name == "input_port_variables":
base = mech_in

elif param_name in self.llvm_state_ids:
base = ctx.get_param_or_state_ptr(builder, self, param_name, state_struct_ptr=mech_state)

Expand Down Expand Up @@ -3212,6 +3219,7 @@ def _get_output_port_value_ptr(b, i):
def _get_output_port_variable_ptr(b, i):
ptr = self._gen_llvm_output_port_parse_variable(ctx,
b,
mech_in,
mech_params,
mech_state,
value,
Expand Down
152 changes: 97 additions & 55 deletions tests/mechanisms/test_ddm_mechanism.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
import contextlib
import numpy as np
import pytest
import re

import psyneulink as pnl

Expand Down Expand Up @@ -172,7 +174,7 @@ def test_is_finished_stops_composition(self):
@pytest.mark.composition
class TestInputPorts:

def test_regular_input_mode(self):
def test_regular_input_mode(self, comp_mode):
input_mech = ProcessingMechanism(input_shapes=2)
ddm = DDM(
function=DriftDiffusionAnalytical(),
Expand All @@ -181,16 +183,21 @@ def test_regular_input_mode(self):
)
comp = Composition()
comp.add_linear_processing_pathway(pathway=[input_mech, [[1],[-1]], ddm])
result = comp.run(inputs={input_mech:[1,0]})
np.testing.assert_allclose(ddm.output_ports[0].value, [1])
np.testing.assert_allclose(ddm.output_ports[1].value, [1])

result = comp.run(inputs={input_mech:[1,0]}, execution_mode=comp_mode)

if comp_mode is pnl.ExecutionMode.Python:
np.testing.assert_allclose(ddm.output_ports[0].value, [1])
np.testing.assert_allclose(ddm.output_ports[1].value, [1])

np.testing.assert_allclose(ddm.value,
[[1.00000000e+00], [1.19932930e+00], [9.99664650e-01], [3.35350130e-04],
[1.19932930e+00], [2.48491374e-01], [1.48291009e+00], [1.19932930e+00],
[2.48491374e-01], [1.48291009e+00]])
[[1.00000000e+00], [1.19932930e+00], [9.99664650e-01], [3.35350130e-04],
[1.19932930e+00], [2.48491374e-01], [1.48291009e+00], [1.19932930e+00],
[2.48491374e-01], [1.48291009e+00]])
np.testing.assert_allclose(result, [[1.0], [1.0]])

def test_array_mode(self):
@pytest.mark.parametrize("with_modulation", [True, False], ids=["with modulation", "without modulation"])
def test_array_mode(self, with_modulation, comp_mode):
input_mech = ProcessingMechanism(input_shapes=2)
ddm = DDM(
input_format=ARRAY,
Expand All @@ -200,50 +207,83 @@ def test_array_mode(self):
)
comp = Composition()
comp.add_linear_processing_pathway(pathway=[input_mech, ddm])
result = comp.run(inputs={input_mech:[1,0]})
np.testing.assert_allclose(ddm.output_ports[0].value, [1,0])
np.testing.assert_allclose(ddm.output_ports[1].value, [1,0])

inputs = {input_mech:[1, 0]}
if with_modulation:
control = pnl.ControlMechanism(control_signals=[(pnl.THRESHOLD, ddm)])
comp.add_node(control)
inputs[control] = [1]

result = comp.run(inputs={input_mech:[1, 0]}, execution_mode=comp_mode)

if comp_mode is pnl.ExecutionMode.Python:
np.testing.assert_allclose(ddm.output_ports[0].value, [1, 0])
np.testing.assert_allclose(ddm.output_ports[1].value, [1, 0])

np.testing.assert_allclose(ddm.value,
[[1.00000000e+00], [1.19932930e+00], [9.99664650e-01], [3.35350130e-04],
[1.19932930e+00], [2.48491374e-01], [1.48291009e+00], [1.19932930e+00],
[2.48491374e-01], [1.48291009e+00]])
[[1.00000000e+00], [1.19932930e+00], [9.99664650e-01], [3.35350130e-04],
[1.19932930e+00], [2.48491374e-01], [1.48291009e+00], [1.19932930e+00],
[2.48491374e-01], [1.48291009e+00]])
np.testing.assert_allclose(result, [[1., 0.], [1.0, 0.0]])

class TestOutputPorts:

def test_selected_input_array(self):
action_selection = DDM(
input_format=ARRAY,
function=DriftDiffusionAnalytical(
),
output_ports=[SELECTED_INPUT_ARRAY],
name='DDM'
)
with pytest.raises(MechanismError) as error:
action_selection.execute([1.0])
assert ("Shape ((1,)) of input ([1.]) does not match required shape ((1, 2)) "
"for input to InputPort 'ARRAY' of DDM.") in str(error.value)
action_selection.execute([1.0, 0.0])
@pytest.mark.parametrize('variable, expected_value, expected_result, context',
[pytest.param([1.0],
None,
None,
pytest.raises(MechanismError,
match=re.escape("Shape ((1,)) of input ([1.]) does not match required shape ((1, 2)) for input to InputPort 'ARRAY' of DDM.")),
marks=pytest.mark.llvm_not_implemented,
id="negative"),
pytest.param([1.0, 0.0],
[[1.00000000e+00], [1.19932930e+00], [9.99664650e-01], [3.35350130e-04], [1.19932930e+00],
[2.48491374e-01], [1.48291009e+00], [1.19932930e+00], [2.48491374e-01], [1.48291009e+00]],
[[1., 0.]],
contextlib.nullcontext(),
id="postive")])
def test_selected_input_array(self, variable, expected_value, expected_result, context, mech_mode):
action_selection = DDM(input_format=ARRAY,
function=DriftDiffusionAnalytical(),
output_ports=[SELECTED_INPUT_ARRAY],
name='DDM')

def test_decision_outcome_integrator(self):
ddm = DDM(
function=DriftDiffusionIntegrator(rate=0.5, threshold=0.5, non_decision_time=0.0, noise=0.0),
output_ports=[DECISION_OUTCOME],
name='DDM'
)
assert np.allclose(ddm.execute([10.0]), [[0.5], [1]]) and ddm.output_ports[0].value == [1.0]
assert np.allclose(ddm.execute([-10.0]), [[-0.5], [2]]) and ddm.output_ports[0].value == [0.0]
with context:
EX = pytest.helpers.get_mech_execution(action_selection, mech_mode)
result = EX(variable)

def test_decision_outcome_analytical(self):
ddm = DDM(
function=DriftDiffusionAnalytical(drift_rate=0.5, threshold=0.5, non_decision_time=0.0, noise=0.0001),
output_ports=[DECISION_OUTCOME],
name='DDM'
)
ddm.execute([10.0])
assert ddm.output_ports[0].value == [1.0]
ddm.execute([-10.0])
assert ddm.output_ports[0].value == [0.0]
np.testing.assert_allclose(result, expected_result)
np.testing.assert_allclose(action_selection.value, expected_value)

def test_decision_outcome_integrator(self, mech_mode):
ddm = pnl.DDM(function=DriftDiffusionIntegrator(rate=0.5, threshold=0.5, non_decision_time=0.0, noise=0.0),
output_ports=[DECISION_OUTCOME],
name='DDM')

EX = pytest.helpers.get_mech_execution(ddm, mech_mode)

res = EX([10])

np.testing.assert_allclose(res, [[1.0]])
np.testing.assert_allclose(ddm.value, [[0.5], [1]])

res = EX([-10])

np.testing.assert_allclose(res, [[0.0]])
np.testing.assert_allclose(ddm.value, [[-0.5], [2]])

def test_decision_outcome_analytical(self, mech_mode):
ddm = pnl.DDM(function=DriftDiffusionAnalytical(drift_rate=0.5, threshold=0.5, non_decision_time=0.0, noise=0.0001),
output_ports=[DECISION_OUTCOME],
name='DDM')

EX = pytest.helpers.get_mech_execution(ddm, mech_mode)

res = EX([10.0])
np.testing.assert_allclose(res, [[1.0]])

res = EX([-10.0])
np.testing.assert_allclose(res, [[0.0]])


# ------------------------------------------------------------------------------------------------
Expand Down Expand Up @@ -689,22 +729,24 @@ def test_DDM_threshold_modulation_analytical(comp_mode):
@pytest.mark.composition
@pytest.mark.ddm_mechanism
def test_DDM_threshold_modulation_integrator(comp_mode):
M = pnl.DDM(name='DDM',
execute_until_finished=True,
function=pnl.DriftDiffusionIntegrator(threshold=20),
)
ddm = pnl.DDM(name='DDM',
execute_until_finished=True,
function=pnl.DriftDiffusionIntegrator(threshold=20),
)

control = pnl.ControlMechanism(
control_signals=[(pnl.THRESHOLD, M)])
control = pnl.ControlMechanism(control_signals=[(pnl.THRESHOLD, ddm)])

C = pnl.Composition()
C.add_node(M, required_roles=[pnl.NodeRole.INPUT, pnl.NodeRole.OUTPUT])
C.add_node(ddm, required_roles=[pnl.NodeRole.INPUT, pnl.NodeRole.OUTPUT])
C.add_node(control)
inputs = {M:[1], control:[3]}

# Modulation value 3 changes the threshold from 20 -> 60.
# Input value 1.11 needs ~55 iterations to cross the new
# threshold.
inputs = {ddm:[1.11], control:[3]}
val = C.run(inputs, num_trials=1, execution_mode=comp_mode)

np.testing.assert_allclose(val[0], [60.0])
np.testing.assert_allclose(val[1], [60.0])
np.testing.assert_allclose(val, [[60.0], [55.0]])


@pytest.mark.composition
Expand Down
Loading