diff --git a/src/lava/proc/cost_integrator/models.py b/src/lava/proc/cost_integrator/models.py deleted file mode 100644 index 67c929e5e..000000000 --- a/src/lava/proc/cost_integrator/models.py +++ /dev/null @@ -1,39 +0,0 @@ -# Copyright (C) 2022 Intel Corporation -# SPDX-License-Identifier: BSD-3-Clause -# See: https://spdx.org/licenses/ - -import numpy as np -from lava.magma.core.decorator import implements, requires -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.sync.protocols.loihi_protocol import LoihiProtocol - -from lava.proc.cost_integrator.process import CostIntegrator - - -@implements(proc=CostIntegrator, protocol=LoihiProtocol) -@requires(CPU) -class CostIntegratorModel(PyLoihiProcessModel): - """CPU model for the CostIntegrator process. - - The process adds up local cost components from downstream units comming as - spike payload. It has a min_cost variable which keeps track of the best - cost seen so far, if the new cost is better, the minimum cost is updated - and send as an output spike to an upstream process. - """ - cost_in: PyInPort = LavaPyType(PyInPort.VEC_DENSE, int) - update_buffer: PyOutPort = LavaPyType(PyOutPort.VEC_DENSE, int) - min_cost: np.ndarray = LavaPyType(np.ndarray, int, 32) - cost: np.ndarray = LavaPyType(np.ndarray, int, 32) - - def run_spk(self): - """Execute spiking phase, integrate input, update dynamics and send - messages out.""" - cost = self.cost_in.recv() - if cost < self.min_cost: - self.min_cost[:] = cost - self.update_buffer.send(cost) - else: - self.update_buffer.send(np.asarray([0])) diff --git a/src/lava/proc/cost_integrator/process.py b/src/lava/proc/cost_integrator/process.py deleted file mode 100644 index 513976477..000000000 --- a/src/lava/proc/cost_integrator/process.py +++ /dev/null @@ -1,52 +0,0 @@ -# Copyright (C) 2022 Intel Corporation -# SPDX-License-Identifier: BSD-3-Clause -# See: https://spdx.org/licenses/ -import typing as ty - -from lava.magma.core.process.ports.ports import InPort, OutPort -from lava.magma.core.process.process import AbstractProcess, LogConfig -from lava.magma.core.process.variable import Var - - -class CostIntegrator(AbstractProcess): - """Node that integrates cost components and produces output when a better - cost is found. - - Parameters - ---------- - shape : tuple(int) - The expected number and topology of the input cost components. - name : str, optional - Name of the Process. Default is 'Process_ID', where ID is an - integer value that is determined automatically. - log_config: Configuration options for logging. - - InPorts - ------- - cost_in - input to be additively integrated. - - OutPorts - -------- - update_buffer - OutPort which notifies the next process about the - detection of a better cost. - - Vars - ---- - cost - Holds current cost as addition of input spikes' payloads - - min_cost - Current minimum cost, i.e., the lowest reported cost so far. - """ - - def __init__(self, *, shape: ty.Tuple[int, ...] = (1,), - min_cost: int = 2**24, - name: ty.Optional[str] = None, - log_config: ty.Optional[LogConfig] = None) -> None: - super().__init__(shape=shape, name=name, log_config=log_config) - self.cost_in = InPort(shape=shape) - self.update_buffer = OutPort(shape=shape) - self.cost = Var(shape=shape, init=2**24) - self.min_cost = Var(shape=shape, init=min_cost) diff --git a/src/lava/proc/read_gate/models.py b/src/lava/proc/read_gate/models.py deleted file mode 100644 index c836d62c1..000000000 --- a/src/lava/proc/read_gate/models.py +++ /dev/null @@ -1,77 +0,0 @@ -# Copyright (C) 2022 Intel Corporation -# SPDX-License-Identifier: BSD-3-Clause -# See: https://spdx.org/licenses/ -import numpy as np -from lava.magma.core.decorator import implements, requires -from lava.magma.core.model.py.model import PyLoihiProcessModel -from lava.magma.core.model.py.ports import PyInPort, PyOutPort, PyRefPort -from lava.magma.core.model.py.type import LavaPyType -from lava.magma.core.resources import CPU -from lava.magma.core.sync.protocols.loihi_protocol import LoihiProtocol - -from lava.proc.read_gate.process import ReadGate - - -@implements(ReadGate, protocol=LoihiProtocol) -@requires(CPU) -class ReadGatePyModel(PyLoihiProcessModel): - """CPU model for the ReadGate process. - - The model verifies if better payload (cost) has been notified by the - downstream processes, if so, it reads those processes state and sends out to - the upstream process the new payload (cost) and the network state. - """ - target_cost: int = LavaPyType(int, np.int32, 32) - cost_in: PyInPort = LavaPyType(PyInPort.VEC_DENSE, np.int32, - precision=32) - acknowledgemet: PyInPort = LavaPyType(PyInPort.VEC_DENSE, np.int32, - precision=32) - cost_out: PyOutPort = LavaPyType( - PyOutPort.VEC_DENSE, np.int32, precision=32 - ) - solution_out: PyOutPort = LavaPyType( - PyOutPort.VEC_DENSE, np.int32, precision=32 - ) - send_pause_request: PyOutPort = LavaPyType( - PyOutPort.VEC_DENSE, np.int32, precision=32 - ) - solution_reader = LavaPyType(PyRefPort.VEC_DENSE, np.int32, - precision=32) - min_cost: int = None - solution: np.ndarray = None - - def post_guard(self): - """Decide whether to run post management phase.""" - if self.min_cost: - return True - return False - - def run_spk(self): - """Execute spiking phase, integrate input, update dynamics and - send messages out.""" - cost = self.cost_in.recv() - if cost[0]: - self.min_cost = cost[0] - print("Found a solution with cost: ", self.min_cost) - self.cost_out.send(np.asarray([0])) - self.acknowledgemet.recv() - self.send_pause_request.send(np.asarray([self.time_step])) - self.acknowledgemet.recv() - elif self.solution is not None: - self.cost_out.send(np.asarray([self.min_cost])) - self.solution_out.send(self.solution) - self.send_pause_request.send(np.asarray([self.time_step])) - self.acknowledgemet.recv() - self.solution = None - self.min_cost = None - else: - self.cost_out.send(np.asarray([0])) - self.acknowledgemet.recv() - self.send_pause_request.send(np.asarray([-1])) - self.acknowledgemet.recv() - - def run_post_mgmt(self): - """Execute post management phase.""" - self._req_pause = True - self.solution = self.solution_reader.read() - self._req_pause = False diff --git a/src/lava/proc/read_gate/process.py b/src/lava/proc/read_gate/process.py deleted file mode 100644 index 902e170e9..000000000 --- a/src/lava/proc/read_gate/process.py +++ /dev/null @@ -1,53 +0,0 @@ -# Copyright (C) 2022 Intel Corporation -# SPDX-License-Identifier: BSD-3-Clause -# See: https://spdx.org/licenses/ -import typing as ty - -from lava.magma.core.process.ports.ports import InPort, OutPort, RefPort -from lava.magma.core.process.process import AbstractProcess, LogConfig -from lava.magma.core.process.variable import Var - - -class ReadGate(AbstractProcess): - """Process that triggers solution readout when problem is solved. - - Parameters - ---------- - shape: The shape of the set of units in the downstream process whose state - will be read by ReadGate. - target_cost: cost value at which, once attained by the network, - this process will stop execution. - name: Name of the Process. Default is 'Process_ID', where ID is an - integer value that is determined automatically. - log_config: Configuration options for logging. - - InPorts - ------- - cost_in: Receives a better cost found by the CostIntegrator at the - previous timestep. - - OutPorts - -------- - cost_out: Forwards to an upstream process the better cost notified by the - CostIntegrator. - solution_out: Forwards to an upstream process the better variable assignment - found by the solver network. - send_pause_request: Notifies upstream process to request execution to pause. - """ - - def __init__(self, - shape: ty.Tuple[int, ...], - target_cost=None, - name: ty.Optional[str] = None, - log_config: ty.Optional[LogConfig] = None) -> None: - super().__init__(shape=shape, - target_cost=target_cost, - name=name, - log_config=log_config) - self.target_cost = Var(shape=(1,), init=target_cost) - self.cost_in = InPort(shape=(1,)) - self.acknowledgemet = InPort(shape=(1,)) - self.cost_out = OutPort(shape=(1,)) - self.send_pause_request = OutPort(shape=(1,)) - self.solution_out = OutPort(shape=shape) - self.solution_reader = RefPort(shape=shape) diff --git a/src/lava/proc/scif/models.py b/src/lava/proc/scif/models.py deleted file mode 100644 index 2903fc5d6..000000000 --- a/src/lava/proc/scif/models.py +++ /dev/null @@ -1,282 +0,0 @@ -# Copyright (C) 2022 Intel Corporation -# SPDX-License-Identifier: BSD-3-Clause -# See: https://spdx.org/licenses/ -from abc import abstractmethod - -import numpy as np - -from lava.magma.core.sync.protocols.loihi_protocol import LoihiProtocol -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 implements, requires, tag -from lava.magma.core.model.py.model import PyLoihiProcessModel -from lava.proc.scif.process import CspScif, QuboScif - - -@implements(proc=CspScif, protocol=LoihiProtocol) -@requires(CPU) -@tag('fixed_pt') -class PyModelCspScifFixed(PyLoihiProcessModel): - """Fixed point implementation of Stochastic Constraint Integrate and - Fire (SCIF) neuron for solving CSP problems. - """ - a_in = LavaPyType(PyInPort.VEC_DENSE, int, precision=8) - s_sig_out = LavaPyType(PyOutPort.VEC_DENSE, int, precision=8) - s_wta_out = LavaPyType(PyOutPort.VEC_DENSE, int, precision=8) - - state: np.ndarray = LavaPyType(np.ndarray, int, precision=24) - spk_hist: np.ndarray = LavaPyType(np.ndarray, int, precision=8) - - step_size: np.ndarray = LavaPyType(np.ndarray, int, precision=24) - theta: np.ndarray = LavaPyType(np.ndarray, int, precision=24) - noise_ampl: np.ndarray = LavaPyType(np.ndarray, int, precision=1) - cnstr_intg: np.ndarray = LavaPyType(np.ndarray, int, precision=24) - neg_tau_ref: np.ndarray = LavaPyType(np.ndarray, int, precision=24) - - def __init__(self, proc_params): - super(PyModelCspScifFixed, self).__init__(proc_params) - self.a_in_data = np.zeros(proc_params['shape']) - - def _prng(self, intg_idx_): - """Pseudo-random number generator - """ - - # ToDo: Choosing a 16-bit signed random integer. For bit-accuracy, - # need to replace it with Loihi-conformant LFSR function - prand = np.zeros(shape=(len(intg_idx_),)) - if prand.size > 0: - rand_nums = \ - np.random.randint(-2 ** 15, 2 ** 15 - 1, size=prand.size) - # Assign random numbers only to neurons, for which noise is enabled - prand = rand_nums * self.noise_ampl[intg_idx_] - return prand - - def _update_buffers(self): - # !! Side effect: Changes self.beta !! - - # Populate the buffer for local computation - spk_hist_buffer = self.spk_hist.copy() - spk_hist_buffer &= 3 - self.spk_hist <<= 2 - # Overflow 8-bit unsigned beta to 0 - self.spk_hist[self.spk_hist >= 256] = 0 - - return spk_hist_buffer - - def _gen_sig_spks(self, spk_hist_buffer): - s_sig = np.zeros_like(self.state) - # Gather spike and unsatisfied indices for summation axons - sig_unsat_idx = np.where(spk_hist_buffer == 2) - sig_spk_idx = np.where(np.logical_and(spk_hist_buffer == 1, - self.cnstr_intg == 0)) - - # Assign sigma spikes (+/- 1) - s_sig[sig_unsat_idx] = -1 - s_sig[sig_spk_idx] = 1 - - return s_sig - - def _gen_wta_spks(self, spk_hist_buffer): - # Indices of WTA neurons signifying unsatisfied constraints, based on - # buffered history from previous timestep - wta_unsat_prev_ts_idx = np.where(np.logical_and(spk_hist_buffer == 1, - self.cnstr_intg < 0)) - - # Reset voltages of unsatisfied WTA - self.state[wta_unsat_prev_ts_idx] = 0 - # indices of neurons to be integrated: - intg_idx = np.where(self.state >= 0) - # indices of neurons in refractory: - rfct_idx = np.where(self.state < 0) - - # Indices of WTA neurons that will spike and enter refractory - wta_spk_idx = self._integration_dynamics(intg_idx) - - # Indices of WTA neurons coming out of refractory or those signifying - # unsatisfied constraints - wta_rfct_end_or_unsat_idx = self._refractory_dynamics(rfct_idx) if \ - self.neg_tau_ref != 0 else (np.array([], dtype=np.int32),) - - s_wta = np.zeros_like(self.state) - s_wta[wta_spk_idx] = 1 - s_wta[wta_unsat_prev_ts_idx] = -1 - s_wta[wta_rfct_end_or_unsat_idx] = -1 - - return s_wta - - def _integration_dynamics(self, intg_idx): - - state_to_intg = self.state[intg_idx] # voltages to be integrated - cnstr_to_intg = self.cnstr_intg[intg_idx] # currents to be integrated - spk_hist_to_intg = self.spk_hist[intg_idx] # beta to be integrated - step_size_to_intg = self.step_size[intg_idx] # bias to be integrated - - lfsr = self._prng(intg_idx_=intg_idx) - - state_to_intg = state_to_intg + lfsr + cnstr_to_intg + step_size_to_intg - np.clip(state_to_intg, a_min=0, a_max=2 ** 23 - 1, out=state_to_intg) - - # WTA spike indices when threshold is exceeded - wta_spk_idx = np.where(state_to_intg >= self.theta) # Exceeds threshold - - # Spiking neuron voltages go in refractory (if neg_tau_ref < 0) - state_to_intg[wta_spk_idx] = self.neg_tau_ref # Post spk refractory - spk_hist_to_intg[wta_spk_idx] |= 1 - - # Assign all temporary states to state Vars - self.state[intg_idx] = state_to_intg - self.cnstr_intg[intg_idx] = cnstr_to_intg - self.spk_hist[intg_idx] = spk_hist_to_intg - - return wta_spk_idx - - def _refractory_dynamics(self, rfct_idx): - - # Split/fork state variables u, v, beta - state_in_rfct = self.state[rfct_idx] # voltages in refractory - cnstr_in_rfct = self.cnstr_intg[rfct_idx] # currents in refractory - spk_hist_in_rfct = self.spk_hist[rfct_idx] # beta in refractory - - # Refractory dynamics - state_in_rfct += 1 # voltage increments by 1 every step - spk_hist_in_rfct |= 3 - # Second set of unsatisfied WTA indices based on v and u in refractory - wta_unsat_idx_2 = \ - np.where(np.logical_or(state_in_rfct == 0, cnstr_in_rfct < 0)) - - # Reset voltage of unsatisfied WTA in refractory - state_in_rfct[wta_unsat_idx_2] = 0 - spk_hist_in_rfct[wta_unsat_idx_2] &= 2 - - # Assign all temporary states to state Vars - self.state[rfct_idx] = state_in_rfct - self.cnstr_intg[rfct_idx] = cnstr_in_rfct - self.spk_hist[rfct_idx] = spk_hist_in_rfct - - return wta_unsat_idx_2 - - def run_spk(self) -> None: - # Receive synaptic input - self.a_in_data = self.a_in.recv() - - # Add the incoming activation and saturate to min-max limits - np.clip(self.cnstr_intg + self.a_in_data, a_min=-2 ** 23, - a_max=2 ** 23 - 1, out=self.cnstr_intg) - - # !! Side effect: Changes self.beta !! - spk_hist_buffer = self._update_buffers() - - # Generate Sigma spikes - s_sig = self._gen_sig_spks(spk_hist_buffer) - - # Generate WTA spikes - s_wta = self._gen_wta_spks(spk_hist_buffer) - - # Send out spikes - self.s_sig_out.send(s_sig) - self.s_wta_out.send(s_wta) - - -@implements(proc=QuboScif, protocol=LoihiProtocol) -@requires(CPU) -@tag('fixed_pt') -class PyModelQuboScifFixed(PyLoihiProcessModel): - """Fixed point implementation of Stochastic Constraint Integrate and - Fire (SCIF) neuron for solving QUBO problems. - """ - a_in = LavaPyType(PyInPort.VEC_DENSE, int, precision=8) - s_sig_out = LavaPyType(PyOutPort.VEC_DENSE, int, precision=8) - s_wta_out = LavaPyType(PyOutPort.VEC_DENSE, int, precision=8) - - state: np.ndarray = LavaPyType(np.ndarray, int, precision=24) - spk_hist: np.ndarray = LavaPyType(np.ndarray, int, precision=8) - - step_size: np.ndarray = LavaPyType(np.ndarray, int, precision=24) - theta: np.ndarray = LavaPyType(np.ndarray, int, precision=24) - noise_ampl: np.ndarray = LavaPyType(np.ndarray, int, precision=1) - - cost_diagonal: np.ndarray = LavaPyType(np.ndarray, int, precision=24) - noise_shift: np.ndarray = LavaPyType(np.ndarray, int, precision=24) - - def __init__(self, proc_params): - super(PyModelQuboScifFixed, self).__init__(proc_params) - self.a_in_data = np.zeros(proc_params['shape']) - - def _prng(self): - """Pseudo-random number generator - """ - - # ToDo: Choosing a 16-bit signed random integer. For bit-accuracy, - # need to replace it with Loihi-conformant LFSR function - prand = np.zeros(shape=self.state.shape) - if prand.size > 0: - rand_nums = \ - np.random.randint(0, (2 ** 16) - 1, size=prand.size) - # Assign random numbers only to neurons, for which noise is enabled - prand = np.right_shift((rand_nums * self.noise_ampl).astype(int), - self.noise_shift) - - return prand - - def _update_buffers(self): - # !! Side effect: Changes self.beta !! - - # Populate the buffer for local computation - spk_hist_buffer = self.spk_hist.copy() - spk_hist_buffer &= 1 - self.spk_hist <<= 1 - # The following ensures that we flush all history beyond 3 timesteps - # The number '3' comes from the fact that it takes 3 timesteps to - # read out solutions after a minimum cost is detected (due to - # downstream connected process-chain) - self.spk_hist &= 7 - - return spk_hist_buffer - - def _gen_sig_spks(self, spk_hist_buffer): - s_sig = np.zeros_like(self.state) - # If we have fired in the previous time-step, we send out the local - # cost now, i.e., when spk_hist_buffer == 1 - sig_spk_idx = np.where(spk_hist_buffer == 1) - # Compute the local cost - s_sig[sig_spk_idx] = self.cost_diagonal[sig_spk_idx] + \ - self.a_in_data[sig_spk_idx] - - return s_sig - - def _gen_wta_spks(self): - lfsr = self._prng() - - self.state += lfsr + self.step_size + self.cost_diagonal + \ - self.a_in_data - np.clip(self.state, a_min=-(2 ** 23), a_max=2 ** 23 - 1, - out=self.state) - - # WTA spike indices when threshold is exceeded - wta_spk_idx = np.where(self.state >= self.theta) # Exceeds threshold - # Spiking neuron voltages go in refractory (if neg_tau_ref < 0) - self.state[wta_spk_idx] = 0 - self.spk_hist[wta_spk_idx] |= 1 - - s_wta = np.zeros_like(self.state) - s_wta[wta_spk_idx] = 1 - - return s_wta - - def run_spk(self) -> None: - # Receive synaptic input - self.a_in_data = self.a_in.recv().astype(int) - - # !! Side effect: Changes self.beta !! - spk_hist_buffer = self._update_buffers() - - # Generate Sigma spikes - s_sig = self._gen_sig_spks(spk_hist_buffer) - - # Generate WTA spikes - s_wta = self._gen_wta_spks() - - # Send out spikes - self.s_sig_out.send(s_sig) - self.s_wta_out.send(s_wta) diff --git a/src/lava/proc/scif/process.py b/src/lava/proc/scif/process.py deleted file mode 100644 index 4ae06c40e..000000000 --- a/src/lava/proc/scif/process.py +++ /dev/null @@ -1,100 +0,0 @@ -# Copyright (C) 2021-22 Intel Corporation -# SPDX-License-Identifier: BSD-3-Clause -# See: https://spdx.org/licenses/ - -import typing as ty -from numpy import typing as npty - -import numpy as np - -from lava.magma.core.process.process import AbstractProcess -from lava.magma.core.process.variable import Var -from lava.magma.core.process.ports.ports import InPort, OutPort - - -class AbstractScif(AbstractProcess): - """Abstract Process for Stochastic Constraint Integrate-and-Fire - (SCIF) neurons. - """ - - def __init__( - self, - *, - shape: ty.Tuple[int, ...], - step_size: ty.Optional[int] = 1, - theta: ty.Optional[int] = 4, - noise_amplitude: ty.Optional[int] = 0) -> None: - """ - Stochastic Constraint Integrate and Fire neuron Process. - - Parameters - ---------- - shape: Tuple - Number of neurons. Default is (1,). - step_size: int - bias current driving the SCIF neuron. Default is 1 (arbitrary). - theta: int - threshold above which a SCIF neuron would fire winner-take-all - spike. Default is 4 (arbitrary). - """ - super().__init__(shape=shape) - - self.a_in = InPort(shape=shape) - self.s_sig_out = OutPort(shape=shape) - self.s_wta_out = OutPort(shape=shape) - - self.state = Var(shape=shape, init=np.zeros(shape=shape).astype(int)) - self.spk_hist = Var(shape=shape, init=np.zeros(shape=shape).astype(int)) - self.noise_ampl = Var(shape=shape, init=noise_amplitude) - - self.step_size = Var(shape=shape, init=int(step_size)) - self.theta = Var(shape=(1,), init=int(theta)) - - @property - def shape(self) -> ty.Tuple[int, ...]: - return self.proc_params['shape'] - - -class CspScif(AbstractScif): - """Stochastic Constraint Integrate-and-Fire neurons to solve CSPs. - """ - - def __init__(self, - *, - shape: ty.Tuple[int, ...], - step_size: ty.Optional[int] = 1, - theta: ty.Optional[int] = 4, - neg_tau_ref: ty.Optional[int] = -5, - noise_amplitude: ty.Optional[int] = 0): - - super(CspScif, self).__init__(shape=shape, - step_size=step_size, - theta=theta, - noise_amplitude=noise_amplitude) - self.neg_tau_ref = Var(shape=(1,), init=int(neg_tau_ref)) - self.cnstr_intg = Var(shape=shape, init=np.zeros(shape=shape).astype( - int)) - - -class QuboScif(AbstractScif): - """Stochastic Constraint Integrate-and-Fire neurons to solve QUBO - problems. - """ - - def __init__(self, - *, - shape: ty.Tuple[int, ...], - cost_diag: npty.NDArray, - step_size: ty.Optional[int] = 1, - theta: ty.Optional[int] = 4, - noise_amplitude: ty.Optional[int] = 0, - noise_shift: ty.Optional[int] = 8): - - super(QuboScif, self).__init__(shape=shape, - step_size=step_size, - theta=theta, - noise_amplitude=noise_amplitude) - self.cost_diagonal = Var(shape=shape, init=cost_diag) - # User provides a desired precision. We convert it to the amount by - # which unsigned 16-bit noise is right-shifted: - self.noise_shift = Var(shape=shape, init=noise_shift) diff --git a/tests/lava/proc/scif/__init__.py b/tests/lava/proc/scif/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/lava/proc/scif/test_models.py b/tests/lava/proc/scif/test_models.py deleted file mode 100644 index a532161ae..000000000 --- a/tests/lava/proc/scif/test_models.py +++ /dev/null @@ -1,424 +0,0 @@ - -# INTEL CORPORATION CONFIDENTIAL AND PROPRIETARY -# -# Copyright © 2021-2022 Intel Corporation. -# -# This software and the related documents are Intel copyrighted -# materials, and your use of them is governed by the express -# license under which they were provided to you (License). Unless -# the License provides otherwise, you may not use, modify, copy, -# publish, distribute, disclose or transmit this software or the -# related documents without Intel's prior written permission. -# -# This software and the related documents are provided as is, with -# no express or implied warranties, other than those that are -# expressly stated in the License. -import sys -import unittest -import numpy as np -from typing import Tuple, Dict - -from lava.magma.core.run_configs import Loihi2SimCfg -from lava.magma.core.run_conditions import RunSteps -from lava.proc.scif.process import CspScif, QuboScif -from lava.proc.lif.process import LIF -from lava.proc.dense.process import Dense -from lava.proc.io.source import RingBuffer as SpikeSource - -verbose = True if (('-v' in sys.argv) or ('--verbose' in sys.argv)) else False - - -class TestCspScifModels(unittest.TestCase): - """Tests for CspScif neuron""" - - def run_test( - self, - num_steps: int, - num_neurons: int, - step_size: int, - theta: int, - neg_tau_ref: int, - wt: int, - t_inj_spk: Dict[int, int], # time_step -> payload dict to inject - tag: str = 'fixed_pt' - ) -> Tuple[np.ndarray, np.ndarray, np.ndarray]: - - spk_src = SpikeSource(data=np.array([[0] * num_neurons]).reshape( - num_neurons, 1).astype(int)) - # TODO (MR): The weight of -1 is now being correctly encoded as -1. - # It was written assuming the weight would be truncated to -2. - dense_in = Dense(weights=(-1) * np.eye(num_neurons), - num_message_bits=16) - csp_scif = CspScif(shape=(num_neurons,), - step_size=step_size, - theta=theta, - neg_tau_ref=neg_tau_ref) - dense_wta = Dense(weights=wt * np.eye(num_neurons), - num_message_bits=16) - dense_sig = Dense(weights=wt * np.eye(num_neurons), - num_message_bits=16) - lif_wta = LIF(shape=(num_neurons,), - du=4095, - dv=4096, - bias_mant=0, - vth=2 ** 17 - 1) - lif_sig = LIF(shape=(num_neurons,), - du=4095, - dv=4096, - bias_mant=0, - vth=2 ** 17 - 1) - spk_src.s_out.connect(dense_in.s_in) - dense_in.a_out.connect(csp_scif.a_in) - csp_scif.s_wta_out.connect(dense_wta.s_in) - csp_scif.s_sig_out.connect(dense_sig.s_in) - dense_wta.a_out.connect(lif_wta.a_in) - dense_sig.a_out.connect(lif_sig.a_in) - - run_condition = RunSteps(num_steps=1) - run_config = Loihi2SimCfg(select_tag=tag) - - volts_scif = [] - volts_lif_wta = [] - volts_lif_sig = [] - for j in range(num_steps): - if j + 1 in t_inj_spk: - spk_src.data.set(np.array( - [[t_inj_spk[j + 1]] * num_neurons]).astype(int)) - csp_scif.run(condition=run_condition, run_cfg=run_config) - spk_src.data.set(np.array([[0] * num_neurons]).astype(int)) - volts_scif.append(csp_scif.state.get()) - # Get the voltage of LIF attached to WTA - v_wta = lif_wta.v.get() - # Transform the voltage into +/- 1 spike - v_wta = (v_wta / wt).astype(int) # De-scale the weight - v_wta = np.right_shift(v_wta, 6) # downshift DendAccum's effect - # Append to list - volts_lif_wta.append(v_wta) - # Get the voltage of LIF attached to Sig - v_sig = lif_sig.v.get() - # Transform the voltage into +/- 1 spike - v_sig = (v_sig / wt).astype(int) # De-scale the weight - v_sig = np.right_shift(v_sig, 6) # downshift DendAccum's effect - # Append to list - volts_lif_sig.append(v_sig) - - csp_scif.stop() - - return np.array(volts_scif).astype(int), \ - np.array(volts_lif_wta).astype(int), \ - np.array(volts_lif_sig).astype(int) - - def test_scif_fixed_pt_no_noise(self) -> None: - """Test a single SCIF neuron without noise, but with a constant bias. - The neuron is expected to spike with a regular period, on WTA as well as - Sigma axons. After excitatory spikes on two consecutive time-steps, the - neuron goes into inhibition and sends 2 inhibitory spikes of payload -1 - at the end of its refractory period. - """ - num_neurons = np.random.randint(1, 11) - step_size = 1 - theta = 4 - neg_tau_ref = -5 - wt = 2 - total_period = theta // step_size - neg_tau_ref - num_epochs = 10 - num_steps = num_epochs * total_period + (theta // step_size) - v_scif, v_lif_wta, v_lif_sig = self.run_test(num_steps=num_steps, - num_neurons=num_neurons, - step_size=step_size, - theta=theta, - neg_tau_ref=neg_tau_ref, - wt=wt, - t_inj_spk={}) - spk_idxs = np.array([theta // step_size - 1 + j * total_period for j in - range(num_epochs)]).astype(int) - wta_pos_spk_idxs = spk_idxs + 1 - sig_pos_spk_idxs = wta_pos_spk_idxs + 1 - wta_neg_spk_idxs = wta_pos_spk_idxs - neg_tau_ref - sig_neg_spk_idxs = sig_pos_spk_idxs - neg_tau_ref - self.assertTrue(np.all(v_scif[spk_idxs] == neg_tau_ref)) - self.assertTrue(np.all(v_lif_wta[wta_pos_spk_idxs] == 1)) - self.assertTrue(np.all(v_lif_sig[sig_pos_spk_idxs] == 1)) - self.assertTrue(np.all(v_lif_wta[wta_neg_spk_idxs] == -1)) - self.assertTrue(np.all(v_lif_sig[sig_neg_spk_idxs] == -1)) - - def test_scif_fp_no_noise_interrupt_rfct_mid(self) -> None: - """ - Test a single SCIF neuron without LFSR noise, but with a constant bias. - - An inhibitory spike is injected in the middle of the refractory - period after the neuron spikes for the first time. The inhibition - interrupts the refractory period. The neuron issues negative spikes - at WTA and Sigma axons on consecutive time-steps. - - An excitatory spike is injected to nullify the inhibition and neuron - starts spiking periodically again. - """ - num_neurons = np.random.randint(1, 11) - step_size = 1 - theta = 4 - neg_tau_ref = -5 - wt = 2 - t_inj_spk = {7: 1, 11: -1} - inj_times = list(t_inj_spk.keys()) - total_period = (theta // step_size) - neg_tau_ref - num_epochs = 5 - num_steps = \ - (theta // step_size) + num_epochs * total_period + inj_times[1] - v_scif, v_lif_wta, v_lif_sig = self.run_test(num_steps=num_steps, - num_neurons=num_neurons, - step_size=step_size, - theta=theta, - neg_tau_ref=neg_tau_ref, - wt=wt, - t_inj_spk=t_inj_spk) - # Test pre-inhibitory-injection SCIF voltage and spiking - spk_idxs_pre_inj = np.array([theta // step_size]).astype(int) - 1 - wta_pos_spk_pre_inj = spk_idxs_pre_inj + 1 - sig_pos_spk_pre_inj = wta_pos_spk_pre_inj + 1 - inh_inj = inj_times[0] - wta_neg_spk_rfct_interrupt = inh_inj + 1 - sig_neg_spk_rfct_interrupt = wta_neg_spk_rfct_interrupt + 1 - self.assertTrue(np.all(v_scif[spk_idxs_pre_inj] == neg_tau_ref)) - self.assertTrue(np.all(v_lif_wta[wta_pos_spk_pre_inj] == 1)) - self.assertTrue(np.all(v_lif_sig[sig_pos_spk_pre_inj] == 1)) - self.assertTrue(np.all(v_scif[inh_inj] == 0)) - self.assertTrue(np.all(v_lif_wta[wta_neg_spk_rfct_interrupt] == -1)) - self.assertTrue(np.all(v_lif_sig[sig_neg_spk_rfct_interrupt] == -1)) - # Test post-inhibitory-injection SCIF voltage and spiking - idx_lst = [inj_times[1] + (theta // step_size) - 1 + j * total_period - for j in range(num_epochs)] - spk_idxs_post_inj = np.array(idx_lst).astype(int) - wta_pos_spk_idxs = spk_idxs_post_inj + 1 - sig_pos_spk_idxs = wta_pos_spk_idxs + 1 - wta_neg_spk_idxs = wta_pos_spk_idxs - neg_tau_ref - sig_neg_spk_idxs = sig_pos_spk_idxs - neg_tau_ref - self.assertTrue(np.all(v_scif[spk_idxs_post_inj] == neg_tau_ref)) - self.assertTrue(np.all(v_lif_wta[wta_pos_spk_idxs] == 1)) - self.assertTrue(np.all(v_lif_sig[sig_pos_spk_idxs] == 1)) - self.assertTrue(np.all(v_lif_wta[wta_neg_spk_idxs] == -1)) - self.assertTrue(np.all(v_lif_sig[sig_neg_spk_idxs] == -1)) - - def test_scif_fp_no_noise_interrupt_rfct_beg(self) -> None: - """ - Test a single SCIF neuron without LFSR noise, but with a constant bias. - - An inhibitory spike is injected at the very start of the refractory - period after the neuron spikes for the first time. The inhibition - interrupts the refractory period. The neuron issues a negative spike - at WTA axons to nullify its positive spike. No spikes are issued on - the Sigma axon. - - An excitatory spike is injected to nullify the inhibition and neuron - starts spiking periodically again. - """ - num_neurons = np.random.randint(1, 11) - step_size = 1 - theta = 4 - neg_tau_ref = -5 - wt = 2 - t_inj_spk = {4: 1, 8: -1} - inj_times = list(t_inj_spk.keys()) - total_period = theta // step_size - neg_tau_ref - num_epochs = 5 - num_steps = \ - num_epochs * total_period + (theta // step_size) + inj_times[1] - v_scif, v_lif_wta, v_lif_sig = self.run_test(num_steps=num_steps, - num_neurons=num_neurons, - step_size=step_size, - theta=theta, - neg_tau_ref=neg_tau_ref, - wt=wt, - t_inj_spk=t_inj_spk) - # Test pre-inhibitory-injection SCIF voltage and spiking - spk_idxs_pre_inj = np.array([theta // step_size]).astype(int) - 1 - wta_pos_spk_pre_inj = spk_idxs_pre_inj + 1 - wta_neg_spk_pre_inj = wta_pos_spk_pre_inj + 1 - inh_inj = inj_times[0] - self.assertTrue(np.all(v_scif[spk_idxs_pre_inj] == neg_tau_ref)) - self.assertTrue(np.all(v_lif_wta[wta_pos_spk_pre_inj] == 1)) - self.assertTrue(np.all(v_lif_wta[wta_neg_spk_pre_inj] == -1)) - self.assertTrue(np.all(v_lif_sig[wta_neg_spk_pre_inj] == 0)) - self.assertTrue(np.all(v_scif[inh_inj] == 0)) - - # Test post-inhibitory-injection SCIF voltage and spiking - idx_lst = [inj_times[1] + (theta // step_size) - 1 + j * total_period - for j in range(num_epochs)] - spk_idxs_post_inj = np.array(idx_lst).astype(int) - wta_pos_spk_idxs = spk_idxs_post_inj + 1 - sig_pos_spk_idxs = wta_pos_spk_idxs + 1 - wta_neg_spk_idxs = wta_pos_spk_idxs - neg_tau_ref - sig_neg_spk_idxs = sig_pos_spk_idxs - neg_tau_ref - self.assertTrue(np.all(v_scif[spk_idxs_post_inj] == neg_tau_ref)) - self.assertTrue(np.all(v_lif_wta[wta_pos_spk_idxs] == 1)) - self.assertTrue(np.all(v_lif_sig[sig_pos_spk_idxs] == 1)) - self.assertTrue(np.all(v_lif_wta[wta_neg_spk_idxs] == -1)) - self.assertTrue(np.all(v_lif_sig[sig_neg_spk_idxs] == -1)) - - -class TestQuboScifModels(unittest.TestCase): - """Tests for sigma delta neuron""" - - def run_test( - self, - num_steps: int, - num_neurons: int, - cost_diag: np.ndarray, - step_size: int, - theta: int, - wt: int, - t_inj_spk: Dict[int, int], # time_step -> payload dict to inject - tag: str = 'fixed_pt' - ) -> Tuple[np.ndarray, np.ndarray, np.ndarray]: - - spk_src = SpikeSource(data=np.array([[0] * num_neurons]).reshape( - num_neurons, 1).astype(int)) - dense_in = Dense(weights=(-1) * np.eye(num_neurons), - num_message_bits=16) - qubo_scif = QuboScif(shape=(num_neurons,), - cost_diag=cost_diag, - step_size=step_size, - theta=theta) - dense_wta = Dense(weights=wt * np.eye(num_neurons), - num_message_bits=16) - dense_sig = Dense(weights=wt * np.eye(num_neurons), - num_message_bits=16) - lif_wta = LIF(shape=(num_neurons,), - du=4095, - dv=4096, - bias_mant=0, - vth=2 ** 17 - 1) - lif_sig = LIF(shape=(num_neurons,), - du=4095, - dv=4096, - bias_mant=0, - vth=2 ** 17 - 1) - spk_src.s_out.connect(dense_in.s_in) - dense_in.a_out.connect(qubo_scif.a_in) - qubo_scif.s_wta_out.connect(dense_wta.s_in) - qubo_scif.s_sig_out.connect(dense_sig.s_in) - dense_wta.a_out.connect(lif_wta.a_in) - dense_sig.a_out.connect(lif_sig.a_in) - - run_condition = RunSteps(num_steps=1) - run_config = Loihi2SimCfg(select_tag=tag) - - volts_scif = [] - volts_lif_wta = [] - volts_lif_sig = [] - for j in range(num_steps): - if j + 1 in t_inj_spk: - spk_src.data.set( - np.array([[t_inj_spk[j + 1]] * num_neurons]).astype(int)) - qubo_scif.run(condition=run_condition, run_cfg=run_config) - spk_src.data.set(np.array([[0] * num_neurons]).astype(int)) - volts_scif.append(qubo_scif.state.get()) - # Get the voltage of LIF attached to WTA - v_wta = lif_wta.v.get() - # Transform the voltage into +/- 1 spike - v_wta = (v_wta / wt).astype(int) # De-scale the weight - v_wta = np.right_shift(v_wta, 6) # downshift DendAccum's effect - # Append to list - volts_lif_wta.append(v_wta) - # Get the voltage of LIF attached to Sig - v_sig = lif_sig.v.get() - # Transform the voltage into +/- 1 spike - v_sig = (v_sig / wt).astype(int) # De-scale the weight - v_sig = np.right_shift(v_sig, 6) # downshift DendAccum's effect - # Append to list - volts_lif_sig.append(v_sig) - - qubo_scif.stop() - - return np.array(volts_scif).astype(int), \ - np.array(volts_lif_wta).astype(int), \ - np.array(volts_lif_sig).astype(int) - - def test_scif_fixed_pt_no_noise(self) -> None: - """Test a single SCIF neuron without noise, but with a constant bias. - The neuron is expected to spike with a regular period, on WTA as well as - Sigma axons. After excitatory spikes on two consecutive time-steps, the - neuron goes into inhibition and sends 2 inhibitory spikes of payload -1 - at the end of its refractory period. - """ - num_neurons = np.random.randint(1, 11) - cost_diag = np.arange(1, num_neurons + 1) - step_size = 1 - theta = 4 - wt = 2 - total_period = theta // step_size - num_epochs = 10 - num_steps = num_epochs * total_period + (theta // step_size) - v_scif, v_lif_wta, v_lif_sig = self.run_test(num_steps=num_steps, - num_neurons=num_neurons, - cost_diag=cost_diag, - step_size=step_size, - theta=theta, - wt=wt, - t_inj_spk={}) - spk_idxs = np.array([theta // step_size - 1 + j * total_period for j in - range(num_epochs)]).astype(int) - wta_pos_spk_idxs = spk_idxs + 1 - sig_pos_spk_idxs = wta_pos_spk_idxs + 1 - self.assertTrue(np.all(v_scif[spk_idxs] == 0)) - self.assertTrue(np.all(v_lif_wta[wta_pos_spk_idxs] == 1)) - self.assertTrue(np.all(v_lif_sig[sig_pos_spk_idxs] == cost_diag)) - - def test_scif_fp_no_noise_interrupt_rfct_mid(self) -> None: - """ - Test a single SCIF neuron without LFSR noise, but with a constant bias. - - An inhibitory spike is injected in the middle of the refractory - period after the neuron spikes for the first time. The inhibition - interrupts the refractory period. The neuron issues negative spikes - at WTA and Sigma axons on consecutive time-steps. - - An excitatory spike is injected to nullify the inhibition and neuron - starts spiking periodically again. - """ - num_neurons = np.random.randint(1, 11) - cost_diag_coeff = -2 - cost_diag = cost_diag_coeff * np.ones(num_neurons,) - step_size = 3 - effective_bias = step_size + cost_diag_coeff - theta = 4 - wt = 2 - t_inj_spk = {5: -1, 7: -1, 9: 2} - inj_times = list(t_inj_spk.keys()) - total_period = (theta // effective_bias) - num_epochs = 5 - num_steps = (num_epochs + 1) * total_period + inj_times[1] - v_scif, v_lif_wta, v_lif_sig = self.run_test(num_steps=num_steps, - num_neurons=num_neurons, - cost_diag=cost_diag, - step_size=step_size, - theta=theta, - wt=wt, - t_inj_spk=t_inj_spk) - - # Test pre-inhibitory-injection SCIF voltage and spiking - spk_idxs_pre_inj = np.array([theta // effective_bias]).astype(int) - 1 - wta_pos_spk_pre_inj = spk_idxs_pre_inj + 1 - sig_pos_spk_pre_inj = wta_pos_spk_pre_inj + 1 - inh_inj = inj_times[0] - wta_spk_rfct_interrupt = inh_inj - 1 - sig_spk_rfct_interrupt = wta_spk_rfct_interrupt + 1 - self.assertTrue(np.all(v_scif[spk_idxs_pre_inj] == 0)) - self.assertTrue(np.all(v_lif_wta[wta_pos_spk_pre_inj] == 1)) - self.assertTrue(np.all( - v_lif_sig[sig_pos_spk_pre_inj] == cost_diag_coeff)) - v_gt_inh_inj = 2 * effective_bias + t_inj_spk[inh_inj] - self.assertTrue(np.all(v_scif[inh_inj - 1] == v_gt_inh_inj)) - self.assertTrue(np.all(v_lif_wta[wta_spk_rfct_interrupt] == 1)) - self.assertTrue(np.all( - v_lif_sig[sig_spk_rfct_interrupt] == cost_diag_coeff)) - # # Test post-inhibitory-injection SCIF voltage and spiking - idx_lst = [inj_times[2] + 2 + j * total_period for j in range( - num_epochs)] - spk_idxs_post_inj = np.array(idx_lst).astype(int) - wta_pos_spk_idxs = spk_idxs_post_inj + 1 - sig_pos_spk_idxs = wta_pos_spk_idxs + 1 - self.assertTrue(np.all(v_scif[spk_idxs_post_inj] == 0)) - self.assertTrue(np.all(v_lif_wta[wta_pos_spk_idxs] == 1)) - self.assertTrue(np.all( - v_lif_sig[sig_pos_spk_idxs] == cost_diag)) diff --git a/tests/lava/proc/scif/test_process.py b/tests/lava/proc/scif/test_process.py deleted file mode 100644 index 0f6c5ef25..000000000 --- a/tests/lava/proc/scif/test_process.py +++ /dev/null @@ -1,48 +0,0 @@ - -# INTEL CORPORATION CONFIDENTIAL AND PROPRIETARY -# -# Copyright © 2021-2022 Intel Corporation. -# -# This software and the related documents are Intel copyrighted -# materials, and your use of them is governed by the express -# license under which they were provided to you (License). Unless -# the License provides otherwise, you may not use, modify, copy, -# publish, distribute, disclose or transmit this software or the -# related documents without Intel's prior written permission. -# -# This software and the related documents are provided as is, with -# no express or implied warranties, other than those that are -# expressly stated in the License. -import unittest -import numpy as np -from lava.proc.scif.process import CspScif, QuboScif - - -class TestCspScifProcess(unittest.TestCase): - """Tests for CspScif class""" - def test_init(self) -> None: - """Tests instantiation of CspScif""" - scif = CspScif(shape=(10,), - step_size=2, - theta=8, - neg_tau_ref=-10) - - self.assertEqual(scif.shape, (10,)) - self.assertEqual(scif.step_size.init, 2) - self.assertEqual(scif.theta.init, 8) - self.assertEqual(scif.neg_tau_ref.init, -10) - - -class TestQuboScifProcess(unittest.TestCase): - """Tests for QuboScif class""" - def test_init(self) -> None: - """Tests instantiation of QuboScif""" - scif = QuboScif(shape=(10,), - step_size=2, - theta=8, - cost_diag=np.arange(1, 11)) - - self.assertEqual(scif.shape, (10,)) - self.assertEqual(scif.step_size.init, 2) - self.assertEqual(scif.theta.init, 8) - self.assertTrue(np.all(scif.cost_diagonal.init == np.arange(1, 11)))