Question: How to Implement an InPort That Concatenates Signals from Multiple OutPorts?

[jcrvz]

Greetings!

I hope this message finds you well.

I'm implementing an InPort that receives signals from multiple OutPorts and concatenates these inputs instead of the default summation behaviour. However, I couldn't find detailed information or examples in the documentation or on the web.

I came across the tutorial Connect Processes and Ports, but the section title ("Connect multiple InPorts from a single OutPort") seems misleading or oversimplifying this topic. The tutorial doesn't provide information on how to override or customise the behaviour of ports for this use case.

Could you provide guidance or examples on configuring an InPort to concatenate incoming signals along a specified axis (e.g., using ConcatPort or another approach)? Or does this require overriding methods in the PyProcessModel (e.g., recv)? (I've tried something in this direction, and it does not work as expected) What are the best practices for handling multiple incoming signals?

I'd greatly appreciate any insights or references to additional resources to help with this!

Kind regards,
Jorge Cruz

[kds300]

Hello @jcrvz,

I've recently been trying to accomplish the same thing, but it looks like it might not be implemented yet. When I tried something similar to the following:

s_out: list[OutPort]
s_in: InPort

concat_port = s_out[0].concat_with(s_out[1:], axis=0)
concat_port.connect(s_in)

I get the following error when running the network:

NotImplementedError: ConcatPorts are not yet supported.

What I've been doing to get around this, is splitting my synapse layer in a list of separate processes, one for each of the separate inputs. The code is a lot uglier, but I am able to compile and run networks this way.

Best,
Kevin

[jcrvz]

Hello @kds300

First of all, thank you for sharing your approach. Initially, one only wants to get the code working; then embellishments can come.

I've tried something slightly different, but the result was the same error. In your approach, are you masking those inputs to get them concatenated? Something like, for example, for the second element of concatenating (in this case, stacking), three inputs:

# Initialised mask `self.mask_id = 1`
# ...
# Some operations to get `x_value` with shape = (2,)
pre_output = np.zeros(self.shape)  # self.shape = (3, 2)
pre_output[self.mask_id, :] =  self.x_value[:]
self.x_out.send(pre_output)

I have in mind that form above, and I'm also implementing two different ways.

The first is implementing a process with many inputs, as you mentioned, which are created dynamically. However, I haven't managed to get that number modified dynamically. Now, as in the example above, it is set to 3. I've been looking (unsuccessfully) to get the self.proc_params["num_inputs"] value.

class Concatenator(AbstractProcess):
    def __init__(self, shape, num_inputs):
        super().__init__()
        for i in range(num_inputs):
            setattr(self, f"x_in_{i}", InPort(shape=shape))

        self.X_out      = OutPort(shape=(num_inputs, *shape))
        self.proc_params["num_inputs"] = num_inputs

@implements(proc=Concatenator, protocol=LoihiProtocol)
@requires(CPU)
class PyConcatenatorModel(PyLoihiProcessModel):
    X_out: PyOutPort       = LavaPyType(PyOutPort.VEC_DENSE, float)

    for i in range(3):  # <---------- I don't really like this
        exec(f"x_in_{i} = LavaPyType(PyInPort.VEC_DENSE, float)")

    def __init__(self, proc_params):
        super().__init__(proc_params)
        self.num_inputs = self.proc_params.get("num_inputs", 1)

    def run_spk(self):
        concatenated_data = []
        for i in range(self.num_inputs):
            in_port = getattr(self, f"x_in_{i}")
            data = in_port.recv()  
            concatenated_data.append(data)

        concatenated_data = np.stack(concatenated_data, axis=0)
        self.X_out.send(concatenated_data)

The second approach is to get a server, like an archive, that allows processes to store or access that information. I have a preliminary code done, but I'm afraid it could be a bottleneck in my overall process.

It's too soon for me to know which one would work best. As I've seen in previous experiments, I'm trying to avoid creating too many processes because my machine runs out of memory, even setting ulimit -n 102400. That's why I am preparing different alternatives and testing them.

Kind regards,

Jorge

[kds300]

I haven't really thought about it in terms of masking, but I think that is essentially what I'm doing. In my case, I have a relatively simple network:

{Input} (M,) -> Sparse (N, M) -> LIF (N,) -> Sparse (N, N) -> {Output} (N,)

However, I am using different vth values for each LIF neuron, and therefore need to have a separate process for each neuron. What I've been trying to do is find a way to concatenate the inputs and outputs of the LIF neurons into a single layer. Since the concatenation throws an error, I've decided instead to split the (N, M) sparse process into N (1, M) sparse processes which each connect the input to a single LIF neuron. I then repeat a similar process for the second sparse layer, creating N (N, 1) connections.

Essentially, instead of truly concatenating the LIF outputs, I avoid the issue by splitting up the output connection. However, I haven't tried this at large numbers of processes yet, so I'm not sure if it is scalable.

I like your idea for a Concatenator! I'm going to try some things out to see if there's a way to dynamically set the Process and ProcessModel attributes. I'll let you know if I figure something out!

[kds300]

After some playing around with Python's __new__ method, I created a prototype class for dynamically creating a concatenation process.

The classes AbstractConcatenator and PyAbstractConcatenatorModel are the actual Lava process and model. The classes Concatenator and PyConcatenatorModel use __new__ to dynamically create subclasses of the process and model, then add the required ports before creating the process and model. The prototype only supports concatenating N imports of shape (1,) to a single output of shape (N,). It's only a proof of concept, but I think it should be straightforward to extend it to other inputs.

Here's the code for the prototype:

"""Script demonstrating dynamically created Concatenator process.

"""

import numpy as np

from lava.magma.core.process.process import AbstractProcess
from lava.magma.core.process.ports.ports import InPort, OutPort
from lava.magma.core.sync.protocols.loihi_protocol import LoihiProtocol
from lava.magma.core.model.py.model import PyLoihiProcessModel
from lava.magma.core.model.py.ports import PyInPort, PyOutPort
from lava.magma.core.model.py.type import LavaPyType
from lava.magma.core.resources import CPU
from lava.magma.core.decorator import requires


class AbstractConcatenator(AbstractProcess):
    """Abstract class for a concatenator"""
    x_in: dict[int, InPort]
    x_out: OutPort
    n_inputs: int
    shape: tuple[int]

    def __init__(self, shape):
        super().__init__(shape=shape)

        for i in range(self.n_inputs):
            setattr(self, f"_x_in_{i}", InPort(shape=shape))

        self.x_in = {
            i: getattr(self, f"_x_in_{i}") for i in range(self.n_inputs)
        }
        self.x_out = OutPort(shape=(self.n_inputs,))


@requires(CPU)
class PyAbstractConcatenatorModel(PyLoihiProcessModel):
    x_in: dict[int, PyInPort]
    x_out: PyOutPort = LavaPyType(PyOutPort.VEC_DENSE, float)
    n_inputs: int

    def __init__(self, proc_params):
        super().__init__(proc_params)
        self.shape = proc_params.get("shape", (1,))

    def run_spk(self):
        _inp_data = np.zeros(self.x_out.shape)
        _inp_data = np.concatenate(
            [getattr(self, f"_x_in_{i}").recv() for i in range(self.n_inputs)],
            axis=0
        )
        self.x_out.send(_inp_data)


class Concatenator:
    def __new__(cls, n_inputs, shape):
        model_attrs = {'n_inputs': n_inputs}

        new_model = type(
            f'PyConcatenatorModel_{n_inputs}',
            (PyAbstractConcatenatorModel,),
            model_attrs
        )

        new_proc = type(
            f'Concatenator_{n_inputs}',
            (AbstractConcatenator,),
            {'n_inputs': n_inputs, '_model': new_model}
        )

        return new_proc(shape=shape)


class PyConcatenatorModel:
    def __new__(cls, proc):

        model_attrs = {'n_inputs': proc.n_inputs}
        for i in range(proc.n_inputs):
            model_attrs[f"_x_in_{i}"] = LavaPyType(PyInPort.VEC_DENSE, float)

        new_model = type(
            f'PyConcatenatorModel_{proc.n_inputs}',
            (PyAbstractConcatenatorModel,),
            model_attrs
        )

        # set up implementation details
        new_model.implements_process = type(proc)
        new_model.implements_protocol = LoihiProtocol

        return new_model


if __name__ == "__main__":
    import numpy as np
    rng = np.random.default_rng()

    from lava.proc.io.source import RingBuffer as Source
    from lava.proc.io.sink import RingBuffer as Sink
    from lava.magma.core.run_conditions import RunSteps
    from lava.magma.core.run_configs import Loihi2SimCfg


    num_steps = 10
    n_inputs = 3

    inp_data = [rng.integers(0, 2, size=(1, num_steps)) for _ in range(n_inputs)]

    sources = [Source(data=inp_data[i]) for i in range(n_inputs)]

    # dynamically creates a Concatenator process
    concat = Concatenator(n_inputs=n_inputs, shape=(1,))
    # creates the accompanying ProcessModel
    concat_model = PyConcatenatorModel(concat)

    sink = Sink(shape=concat.x_out.shape, buffer=num_steps)

    for i in range(n_inputs):
        sources[i].s_out.connect(concat.x_in[i])
    concat.x_out.connect(sink.a_in)

    concat.run(
        condition=RunSteps(num_steps=num_steps),
        run_cfg=Loihi2SimCfg(
            select_tag='fixed_pt',
            # required for compiler to know which model to use
            exception_proc_model_map={concat: concat_model}
        )
    )
    out_data = sink.data.get().astype(int)
    concat.stop()

    print("Input Spikes:")
    print(np.concatenate(inp_data, axis=0))
    print("Output Spikes:")
    print(out_data)

[jcrvz]

Excellent implementation, @kds300 !

As you mentioned, I extended it quite quickly and modified it a little to work with floating-point values. I think this is a net implementation worth having in our Lava toolbox.

Best,

Jorge

---

Modified code for floating-point vectors:

"""
    Script demonstrating dynamically created Concatenator process.
"""

import numpy as np

from lava.magma.core.process.process import AbstractProcess
from lava.magma.core.process.ports.ports import InPort, OutPort
from lava.magma.core.sync.protocols.loihi_protocol import LoihiProtocol
from lava.magma.core.model.py.model import PyLoihiProcessModel
from lava.magma.core.model.py.ports import PyInPort, PyOutPort
from lava.magma.core.model.py.type import LavaPyType
from lava.magma.core.resources import CPU
from lava.magma.core.decorator import requires


class AbstractConcatenator(AbstractProcess):
    """Abstract class for a concatenator"""
    x_in:       dict[int, InPort]
    x_out:      OutPort
    n_inputs:   int
    shape:      tuple[int]

    def __init__(self, shape):
        super().__init__(shape=shape)

        for i in range(self.n_inputs):
            setattr(self, f"_x_in_{i}", InPort(shape=shape))

        self.x_in = {
            i: getattr(self, f"_x_in_{i}") for i in range(self.n_inputs)
        }
        self.x_out = OutPort(shape=(self.n_inputs, shape[0]))


@requires(CPU)
class PyAbstractConcatenatorModel(PyLoihiProcessModel):
    x_in: dict[int, PyInPort]
    x_out: PyOutPort = LavaPyType(PyOutPort.VEC_DENSE, float)
    n_inputs: int

    def __init__(self, proc_params):
        super().__init__(proc_params)
        self.shape = proc_params.get("shape", (1,))

    def run_spk(self):
        _inp_data = np.zeros(self.x_out.shape)
        _inp_data = np.stack(
            [getattr(self, f"_x_in_{i}").recv() for i in range(self.n_inputs)],
            axis=0
        )
        # print("Concatenated Input:")
        # print(_inp_data)
        self.x_out.send(_inp_data)


class Concatenator:
    def __new__(cls, n_inputs, shape):
        model_attrs = {'n_inputs': n_inputs}

        new_model = type(
            f'PyConcatenatorModel_{n_inputs}',
            (PyAbstractConcatenatorModel,),
            model_attrs
        )

        new_proc = type(
            f'Concatenator_{n_inputs}',
            (AbstractConcatenator,),
            {'n_inputs': n_inputs, '_model': new_model}
        )

        return new_proc(shape=shape)


class PyConcatenatorModel:
    def __new__(cls, proc):

        model_attrs = {'n_inputs': proc.n_inputs}
        for i in range(proc.n_inputs):
            model_attrs[f"_x_in_{i}"] = LavaPyType(PyInPort.VEC_DENSE, float)

        new_model = type(
            f'PyConcatenatorModel_{proc.n_inputs}',
            (PyAbstractConcatenatorModel,),
            model_attrs
        )

        # set up implementation details
        new_model.implements_process = type(proc)
        new_model.implements_protocol = LoihiProtocol

        return new_model


if __name__ == "__main__":
    import numpy as np
    rng = np.random.default_rng()

    from lava.proc.io.source import RingBuffer as Source
    from lava.proc.io.sink import RingBuffer as Sink
    from lava.magma.core.run_conditions import RunSteps
    from lava.magma.core.run_configs import Loihi2SimCfg

    num_steps = 5
    num_dims = 2
    n_inputs = 3

    inp_data = [rng.uniform(-1, 1, size=(num_dims, num_steps)) for _ in range(n_inputs)]

    sources = [Source(data=inp_data[i]) for i in range(n_inputs)]

    # dynamically creates a Concatenator process
    concat = Concatenator(n_inputs=n_inputs, shape=(num_dims,))
    # creates the accompanying ProcessModel
    concat_model = PyConcatenatorModel(concat)

    sink = Sink(shape=concat.x_out.shape, buffer=num_steps)

    for i in range(n_inputs):
        sources[i].s_out.connect(concat.x_in[i])
    concat.x_out.connect(sink.a_in)

    concat.run(
        condition=RunSteps(num_steps=num_steps),
        run_cfg=Loihi2SimCfg(
            select_tag='floating_pt',
            # required for compiler to know which model to use
            exception_proc_model_map={concat: concat_model}
        )
    )
    out_data = sink.data.get().astype(float)
    concat.stop()

    for step in range(num_steps):
        print(f"Step {step}:")
        print("\tInput Spikes:")
        for i in range(n_inputs):
            print(f"\t\tFrom Source {i}: {inp_data[i][:, step]}")
        print(f"\tExpected Output:")
        print(np.stack([inp_data[i][:, step] for i in range(n_inputs)], axis=0))
        print("\tOutput Spikes:")
        print(out_data[:,:,step])
        print("")

And this was the output:

Step 0:
	Input Spikes:
		From Source 0: [-0.70673381  0.99627328]
		From Source 1: [-0.05978417 -0.66855084]
		From Source 2: [0.65714403 0.65520099]
	Expected Output:
[[-0.70673381  0.99627328]
 [-0.05978417 -0.66855084]
 [ 0.65714403  0.65520099]]
	Output Spikes:
[[-0.70673381  0.99627328]
 [-0.05978417 -0.66855084]
 [ 0.65714403  0.65520099]]

Step 1:
	Input Spikes:
		From Source 0: [ 0.05863927 -0.73365517]
		From Source 1: [0.87141324 0.25726538]
		From Source 2: [ 0.20106895 -0.93913351]
	Expected Output:
[[ 0.05863927 -0.73365517]
 [ 0.87141324  0.25726538]
 [ 0.20106895 -0.93913351]]
	Output Spikes:
[[ 0.05863927 -0.73365517]
 [ 0.87141324  0.25726538]
 [ 0.20106895 -0.93913351]]

Step 2:
	Input Spikes:
		From Source 0: [ 0.05460829 -0.41060273]
		From Source 1: [0.93853204 0.22435938]
		From Source 2: [ 0.33399053 -0.22139826]
	Expected Output:
[[ 0.05460829 -0.41060273]
 [ 0.93853204  0.22435938]
 [ 0.33399053 -0.22139826]]
	Output Spikes:
[[ 0.05460829 -0.41060273]
 [ 0.93853204  0.22435938]
 [ 0.33399053 -0.22139826]]

Step 3:
	Input Spikes:
		From Source 0: [-0.22881415 -0.08005023]
		From Source 1: [-0.21996218  0.11941307]
		From Source 2: [-0.39889051  0.99987259]
	Expected Output:
[[-0.22881415 -0.08005023]
 [-0.21996218  0.11941307]
 [-0.39889051  0.99987259]]
	Output Spikes:
[[-0.22881415 -0.08005023]
 [-0.21996218  0.11941307]
 [-0.39889051  0.99987259]]

Step 4:
	Input Spikes:
		From Source 0: [0.39549258 0.37327979]
		From Source 1: [-0.66319257  0.7610191 ]
		From Source 2: [-0.62931573  0.5373228 ]
	Expected Output:
[[ 0.39549258  0.37327979]
 [-0.66319257  0.7610191 ]
 [-0.62931573  0.5373228 ]]
	Output Spikes:
[[ 0.39549258  0.37327979]
 [-0.66319257  0.7610191 ]
 [-0.62931573  0.5373228 ]]

[kds300]

I'm glad it worked!

Also, I agree that it could be a useful process to have for Lava, at least until the ConcatPort gets implemented. I can create an issue and a branch with this code.

Edit: I just saw that you already created an issue for this so I won't make another one. I'll still create a branch on my lava fork that properly implements the code

[kds300]

I've created a branch on my fork of lava kds300/lava containing the Concatenator process and models under lava/proc/concatenator/.

I uploaded a modified version of my original solution. @jcrvz, I can add you as a collaborator to the repo if you'd like to add your changes. I can also add them myself if you don't care who they're added under.

[jcrvz]

Sure thing, @kds300, include me there, and I'll upload my changes.
Kind regards! Jorge