diff --git a/src/lava/proc/lif/models.py b/src/lava/proc/lif/models.py index 12aee63ce..104bd5b2c 100644 --- a/src/lava/proc/lif/models.py +++ b/src/lava/proc/lif/models.py @@ -1,7 +1,6 @@ # Copyright (C) 2021 Intel Corporation # SPDX-License-Identifier: BSD-3-Clause # See: https://spdx.org/licenses/ - import numpy as np from lava.magma.core.sync.protocols.loihi_protocol import LoihiProtocol from lava.magma.core.model.py.ports import PyInPort, PyOutPort @@ -9,83 +8,86 @@ 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.lif.process import LIF +from lava.proc.lif.process import LIF, TernaryLIF -@implements(proc=LIF, protocol=LoihiProtocol) -@requires(CPU) -@tag('floating_pt') -class PyLifModelFloat(PyLoihiProcessModel): - """Implementation of Leaky-Integrate-and-Fire neural process in floating - point precision. This short and simple ProcessModel can be used for quick - algorithmic prototyping, without engaging with the nuances of a fixed - point implementation. +class AbstractPyLifModelFloat(PyLoihiProcessModel): + """Abstract implementation of floating point precision + leaky-integrate-and-fire neuron model. + + Specific implementations inherit from here. """ a_in: PyInPort = LavaPyType(PyInPort.VEC_DENSE, float) - s_out: PyOutPort = LavaPyType(PyOutPort.VEC_DENSE, bool, precision=1) + s_out = None # This will be an OutPort of different LavaPyTypes u: np.ndarray = LavaPyType(np.ndarray, float) v: np.ndarray = LavaPyType(np.ndarray, float) bias: np.ndarray = LavaPyType(np.ndarray, float) bias_exp: np.ndarray = LavaPyType(np.ndarray, float) du: float = LavaPyType(float, float) dv: float = LavaPyType(float, float) - vth: float = LavaPyType(float, float) - def run_spk(self): - a_in_data = self.a_in.recv() + def spiking_activation(self): + """Abstract method to define the activation function that determines + how spikes are generated. + """ + raise NotImplementedError("spiking activation() cannot be called from " + "an abstract ProcessModel") + + def subthr_dynamics(self, activation_in: np.ndarray): + """Common sub-threshold dynamics of current and voltage variables for + all LIF models. This is where the 'leaky integration' happens. + """ self.u[:] = self.u * (1 - self.du) - self.u[:] += a_in_data - bias = self.bias * (2**self.bias_exp) - self.v[:] = self.v * (1 - self.dv) + self.u + bias - s_out = self.v >= self.vth - self.v[s_out] = 0 # Reset voltage to 0 - self.s_out.send(s_out) + self.u[:] += activation_in + self.v[:] = self.v * (1 - self.dv) + self.u + self.bias + def reset_voltage(self, spike_vector: np.ndarray): + """Voltage reset behaviour. This can differ for different neuron + models.""" + self.v[spike_vector] = 0 -@implements(proc=LIF, protocol=LoihiProtocol) -@requires(CPU) -@tag('bit_accurate_loihi', 'fixed_pt') -class PyLifModelBitAcc(PyLoihiProcessModel): - """Implementation of Leaky-Integrate-and-Fire neural process bit-accurate - with Loihi's hardware LIF dynamics, which means, it mimics Loihi - behaviour bit-by-bit. + def run_spk(self): + """The run function that performs the actual computation during + execution orchestrated by a PyLoihiProcessModel using the + LoihiProtocol. + """ + a_in_data = self.a_in.recv() + self.subthr_dynamics(activation_in=a_in_data) + s_out = self.spiking_activation() + self.reset_voltage(spike_vector=s_out) + self.s_out.send(s_out) - Currently missing features (compared to Loihi 1 hardware): - - refractory period after spiking - - axonal delays - Precisions of state variables - ----------------------------- - du: unsigned 12-bit integer (0 to 4095) - dv: unsigned 12-bit integer (0 to 4095) - bias: signed 13-bit integer (-4096 to 4095). Mantissa part of neuron bias. - bias_exp: unsigned 3-bit integer (0 to 7). Exponent part of neuron bias. - vth: unsigned 17-bit integer (0 to 131071) +class AbstractPyLifModelFixed(PyLoihiProcessModel): + """Abstract implementation of fixed point precision + leaky-integrate-and-fire neuron model. Implementations like those + bit-accurate with Loihi hardware inherit from here. """ a_in: PyInPort = LavaPyType(PyInPort.VEC_DENSE, np.int16, precision=16) - s_out: PyOutPort = LavaPyType(PyOutPort.VEC_DENSE, bool, precision=1) + s_out: None # This will be an OutPort of different LavaPyTypes u: np.ndarray = LavaPyType(np.ndarray, np.int32, precision=24) v: np.ndarray = LavaPyType(np.ndarray, np.int32, precision=24) du: int = LavaPyType(int, np.uint16, precision=12) dv: int = LavaPyType(int, np.uint16, precision=12) bias: np.ndarray = LavaPyType(np.ndarray, np.int16, precision=13) bias_exp: np.ndarray = LavaPyType(np.ndarray, np.int16, precision=3) - vth: np.ndarray = LavaPyType(np.ndarray, np.int32, precision=17) def __init__(self): - super(PyLifModelBitAcc, self).__init__() + super(AbstractPyLifModelFixed, self).__init__() # ds_offset and dm_offset are 1-bit registers in Loihi 1, which are # added to du and dv variables to compute effective decay constants # for current and voltage, respectively. They enable setting decay # constant values to exact 4096 = 2**12. Without them, the range of # 12-bit unsigned du and dv is 0 to 4095. # ToDo: Currently, these instance variables cannot be set from - # outside, but this will change in the future. + # outside. From experience, there have been hardly any applications + # which have needed to change the defaults. It is straight-forward + # to change in the future. self.ds_offset = 1 self.dm_offset = 0 - self.b_vth_computed = False + self.isbiasscaled = False + self.isthrscaled = False self.effective_bias = 0 - self.effective_vth = 0 # Let's define some bit-widths from Loihi # State variables u and v are 24-bits wide self.uv_bitwidth = 24 @@ -97,17 +99,31 @@ def __init__(self): self.vth_shift = 6 self.act_shift = 6 - def run_spk(self): - # Receive synaptic input - a_in_data = self.a_in.recv() + def scale_bias(self): + """Scale bias with bias exponent by taking into account sign of the + exponent. + """ + self.effective_bias = np.where(self.bias_exp >= 0, np.left_shift( + self.bias, self.bias_exp), np.right_shift(self.bias, + -self.bias_exp)) + self.isbiasscaled = True - # Compute effective bias and threshold only once, not every time-step - if not self.b_vth_computed: - self.effective_bias = np.left_shift(self.bias, self.bias_exp) - # In Loihi, user specified threshold is just the mantissa, with a - # constant exponent of 6 - self.effective_vth = np.left_shift(self.vth, self.vth_shift) - self.b_vth_computed = True + def scale_threshold(self): + """Placeholder method for scaling threshold(s). + """ + raise NotImplementedError("spiking activation() cannot be called from " + "an abstract ProcessModel") + + def spiking_activation(self): + """Placeholder method to specify spiking behaviour of a LIF neuron. + """ + raise NotImplementedError("spiking activation() cannot be called from " + "an abstract ProcessModel") + + def subthr_dynamics(self, activation_in: np.ndarray): + """Common sub-threshold dynamics of current and voltage variables for + all LIF models. This is where the 'leaky integration' happens. + """ # Update current # -------------- @@ -119,15 +135,19 @@ def run_spk(self): decayed_curr = np.sign(decayed_curr) * np.right_shift(np.abs( decayed_curr), self.decay_shift) decayed_curr = np.int32(decayed_curr) - # Hardware left-shifts synpatic input for MSB alignment - a_in_data = np.left_shift(a_in_data, self.act_shift) + # Hardware left-shifts synaptic input for MSB alignment + activation_in = np.left_shift(activation_in, self.act_shift) # Add synptic input to decayed current - decayed_curr += a_in_data + decayed_curr += activation_in # Check if value of current is within bounds of 24-bit. Overflows are # handled by wrapping around modulo 2 ** 23. E.g., (2 ** 23) + k # becomes k and -(2**23 + k) becomes -k + sign_of_curr = np.sign(decayed_curr) + # when decayed_curr is 0, we don't care about its sign. But np.mod + # needs something non-zero to avoid the divide-by-zero warning + sign_of_curr[sign_of_curr == 0] = 1 wrapped_curr = np.mod(decayed_curr, - np.sign(decayed_curr) * self.max_uv_val) + sign_of_curr * self.max_uv_val) self.u[:] = wrapped_curr # Update voltage # -------------- @@ -145,9 +165,159 @@ def run_spk(self): updated_volt = decayed_volt + self.u + self.effective_bias self.v[:] = np.clip(updated_volt, neg_voltage_limit, pos_voltage_limit) - # Spike when exceeds threshold - # ---------------------------- - s_out = self.v >= self.effective_vth + def reset_voltage(self, spike_vector: np.ndarray): + """Voltage reset behaviour. This can differ for different neuron + models. + """ + self.v[spike_vector] = 0 + + def run_spk(self): + """The run function that performs the actual computation during + execution orchestrated by a PyLoihiProcessModel using the + LoihiProtocol. + """ + # Receive synaptic input + a_in_data = self.a_in.recv() + + # ToDo: If bias is set through Var.set() API, the Boolean flag + # isbiasscaled does not get reset. This needs to change through + # Var.set() API. Until that change, we will scale bias at every + # time-step. + self.scale_bias() + # # Compute effective bias and threshold only once, not every time-step + # if not self.isbiasscaled: + # self.scale_bias() + + if not self.isthrscaled: + self.scale_threshold() + + self.subthr_dynamics(activation_in=a_in_data) + + s_out = self.spiking_activation() + # Reset voltage of spiked neurons to 0 - self.v[s_out] = 0 + self.reset_voltage(spike_vector=s_out) self.s_out.send(s_out) + + +@implements(proc=LIF, protocol=LoihiProtocol) +@requires(CPU) +@tag('floating_pt') +class PyLifModelFloat(AbstractPyLifModelFloat): + """Implementation of Leaky-Integrate-and-Fire neural process in floating + point precision. This short and simple ProcessModel can be used for quick + algorithmic prototyping, without engaging with the nuances of a fixed + point implementation. + """ + s_out: PyOutPort = LavaPyType(PyOutPort.VEC_DENSE, bool, precision=1) + vth: float = LavaPyType(float, float) + + def spiking_activation(self): + """Spiking activation function for LIF. + """ + return self.v > self.vth + + +@implements(proc=TernaryLIF, protocol=LoihiProtocol) +@requires(CPU) +@tag('floating_pt') +class PyTernLifModelFloat(AbstractPyLifModelFloat): + """Implementation of Ternary Leaky-Integrate-and-Fire neural process in + floating point precision. This ProcessModel builds upon the floating + point ProcessModel for LIF by adding upper and lower threshold voltages. + """ + # Spikes now become 2-bit signed floating point numbers + s_out: PyOutPort = LavaPyType(PyOutPort.VEC_DENSE, float, precision=2) + vth_hi: float = LavaPyType(float, float) + vth_lo: float = LavaPyType(float, float) + + def spiking_activation(self): + """Spiking activation for T-LIF: -1 spikes below lower threshold, + +1 spikes above upper threshold. + """ + return (-1) * (self.v < self.vth_lo) + (self.v > self.vth_hi) + + def reset_voltage(self, spike_vector: np.ndarray): + """Reset voltage of all spiking neurons to 0. + """ + self.v[spike_vector != 0] = 0 # Reset voltage to 0 wherever we spiked + + +@implements(proc=LIF, protocol=LoihiProtocol) +@requires(CPU) +@tag('bit_accurate_loihi', 'fixed_pt') +class PyLifModelBitAcc(AbstractPyLifModelFixed): + """Implementation of Leaky-Integrate-and-Fire neural process bit-accurate + with Loihi's hardware LIF dynamics, which means, it mimics Loihi + behaviour bit-by-bit. + + Currently missing features (compared to Loihi 1 hardware): + - refractory period after spiking + - axonal delays + + Precisions of state variables + ----------------------------- + du: unsigned 12-bit integer (0 to 4095) + dv: unsigned 12-bit integer (0 to 4095) + bias: signed 13-bit integer (-4096 to 4095). Mantissa part of neuron bias. + bias_exp: unsigned 3-bit integer (0 to 7). Exponent part of neuron bias. + vth: unsigned 17-bit integer (0 to 131071). + """ + s_out: PyOutPort = LavaPyType(PyOutPort.VEC_DENSE, bool, precision=1) + vth: int = LavaPyType(int, np.int32, precision=17) + + def __init__(self): + super(PyLifModelBitAcc, self).__init__() + self.effective_vth = 0 + + def scale_threshold(self): + """Scale threshold according to the way Loihi hardware scales it. In + Loihi hardware, threshold is left-shifted by 6-bits to MSB-align it + with other state variables of higher precision. + """ + self.effective_vth = np.left_shift(self.vth, self.vth_shift) + self.isthrscaled = True + + def spiking_activation(self): + """Spike when voltage exceeds threshold. + """ + return self.v > self.effective_vth + + +@implements(proc=TernaryLIF, protocol=LoihiProtocol) +@requires(CPU) +@tag('fixed_pt') +class PyTernLifModelFixed(AbstractPyLifModelFixed): + """Implementation of Ternary Leaky-Integrate-and-Fire neural process + with fixed point precision. + + See Also + -------- + lava.proc.lif.models.PyLifModelBitAcc: Bit-Accurate LIF neuron model + """ + # Spikes now become 2-bit signed integers + s_out: PyOutPort = LavaPyType(PyOutPort.VEC_DENSE, int, precision=2) + vth_hi: np.ndarray = LavaPyType(np.ndarray, np.int32, precision=24) + vth_lo: np.ndarray = LavaPyType(np.ndarray, np.int32, precision=24) + + def __init__(self): + super(PyTernLifModelFixed, self).__init__() + self.effective_vth_hi = 0 + self.effective_vth_lo = 0 + + def scale_threshold(self): + self.effective_vth_hi = np.left_shift(self.vth_hi, self.vth_shift) + self.effective_vth_lo = np.left_shift(self.vth_lo, self.vth_shift) + self.isthrscaled = True + + def spiking_activation(self): + # Spike when exceeds threshold + # ---------------------------- + neg_spikes = self.v < self.effective_vth_lo + pos_spikes = self.v > self.effective_vth_hi + return (-1) * neg_spikes + pos_spikes + + def reset_voltage(self, spike_vector: np.ndarray): + """Reset voltage of all spiking neurons to 0. + """ + self.v[spike_vector != 0] = 0 # Reset voltage to 0 wherever we spiked diff --git a/src/lava/proc/lif/process.py b/src/lava/proc/lif/process.py index 92ea676c3..990341ad6 100644 --- a/src/lava/proc/lif/process.py +++ b/src/lava/proc/lif/process.py @@ -7,7 +7,29 @@ from lava.magma.core.process.ports.ports import InPort, OutPort -class LIF(AbstractProcess): +class AbstractLIF(AbstractProcess): + """Abstract class for variables common to all neurons with leaky + integrator dynamics.""" + def __init__(self, **kwargs): + super().__init__(**kwargs) + shape = kwargs.get("shape", (1,)) + du = kwargs.pop("du", 0) + dv = kwargs.pop("dv", 0) + bias = kwargs.pop("bias", 0) + bias_exp = kwargs.pop("bias_exp", 0) + + self.shape = shape + self.a_in = InPort(shape=shape) + self.s_out = OutPort(shape=shape) + self.u = Var(shape=shape, init=0) + self.v = Var(shape=shape, init=0) + self.du = Var(shape=(1,), init=du) + self.dv = Var(shape=(1,), init=dv) + self.bias = Var(shape=shape, init=bias) + self.bias_exp = Var(shape=shape, init=bias_exp) + + +class LIF(AbstractLIF): """Leaky-Integrate-and-Fire (LIF) neural Process. LIF dynamics abstracts to: @@ -22,26 +44,46 @@ class LIF(AbstractProcess): dv: Inverse of decay time-constant for voltage decay. bias: Mantissa part of neuron bias. bias_exp: Exponent part of neuron bias, if needed. Mostly for fixed point - implementations. Unnecessary for floating point - implementations. If specified, bias = bias * 2**bias_exp. + implementations. Ignored for floating point + implementations. vth: Neuron threshold voltage, exceeding which, the neuron will spike. """ def __init__(self, **kwargs): super().__init__(**kwargs) - shape = kwargs.get("shape", (1,)) - du = kwargs.pop("du", 0) - dv = kwargs.pop("dv", 0) - bias = kwargs.pop("bias", 0) - bias_exp = kwargs.pop("bias_exp", 0) vth = kwargs.pop("vth", 10) - self.shape = shape - self.a_in = InPort(shape=shape) - self.s_out = OutPort(shape=shape) - self.u = Var(shape=shape, init=0) - self.v = Var(shape=shape, init=0) - self.du = Var(shape=(1,), init=du) - self.dv = Var(shape=(1,), init=dv) - self.bias = Var(shape=shape, init=bias) - self.bias_exp = Var(shape=shape, init=bias_exp) self.vth = Var(shape=(1,), init=vth) + + +class TernaryLIF(AbstractLIF): + """Leaky-Integrate-and-Fire (LIF) neural Process with *ternary* spiking + output, i.e., +1, 0, and -1 spikes. When the voltage of a T-LIF neuron + exceeds its upper threshold (UTh), it issues a positive spike and when + the voltage drops below its lower threshold (LTh), it issues a negative + spike. Between the two thresholds, the neuron follows leaky linear + dynamics. + + This class inherits the state variables and ports from AbstractLIF and + adds two new threshold variables for upper and lower thresholds. + + Parameters + ---------- + vth_hi: Upper threshold voltage, exceeding which the neuron spikes +1 + vth_lo: Lower threshold voltage, below which the neuron spikes -1 + + See Also + -------- + lava.proc.lif.process.LIF: 'Regular' leaky-integrate-and-fire neuron for + documentation on rest of the parameters. + """ + + def __init__(self, **kwargs): + super().__init__(**kwargs) + vth_hi = kwargs.pop("vth_hi", 10) + vth_lo = kwargs.pop("vth_lo", -10) + if vth_lo > vth_hi: + raise AssertionError(f"Lower threshold {vth_lo} is larger than the " + f"upper threshold {vth_hi} for Ternary LIF " + f"neurons. Consider switching the values.") + self.vth_hi = Var(shape=(1,), init=vth_hi) + self.vth_lo = Var(shape=(1,), init=vth_lo) diff --git a/tests/lava/proc/lif/test_models.py b/tests/lava/proc/lif/test_models.py index 0abbcf395..bd9277af1 100644 --- a/tests/lava/proc/lif/test_models.py +++ b/tests/lava/proc/lif/test_models.py @@ -15,7 +15,7 @@ from lava.magma.core.run_configs import RunConfig from lava.magma.core.run_conditions import RunSteps from lava.magma.core.sync.protocols.loihi_protocol import LoihiProtocol -from lava.proc.lif.process import LIF +from lava.proc.lif.process import LIF, TernaryLIF class LifRunConfig(RunConfig): @@ -172,7 +172,7 @@ def test_float_pm_no_decay(self): lif.stop() # Gold standard for the test expected_spk_data = np.zeros((num_steps, shape[0])) - expected_spk_data[1:10:2, :] = 1. + expected_spk_data[4:10:5, :] = 1. self.assertTrue(np.all(expected_spk_data == spk_data_through_run)) def test_float_pm_impulse_du(self): @@ -286,7 +286,7 @@ def test_bitacc_pm_no_decay(self): lif.stop() # Gold standard for the test expected_spk_data = np.zeros((num_steps, shape[0])) - expected_spk_data[3:10:4, :] = 1 + expected_spk_data[4:10:5, :] = 1 self.assertTrue(np.all(expected_spk_data == spk_data_through_run)) def test_bitacc_pm_impulse_du(self): @@ -382,6 +382,327 @@ def test_bitacc_pm_impulse_dv(self): # which would be all Loihi-bit-accurate values right shifted by 6 bits expected_float_v = [128, 192, 224, 240, 248, 252, 254, 255] lif_v_float = np.right_shift(np.array(lif_v), 6) - lif_v_float[1:] += 1 + lif_v_float[1:] += 1 # This compensates the drift caused by dsOffset + self.assertListEqual(expected_v_timeseries, lif_v) + self.assertListEqual(expected_float_v, lif_v_float.tolist()) + + +class TestTLIFProcessModelsFloat(unittest.TestCase): + """Tests for ternary LIF floating point neuron model""" + def test_float_pm_neg_no_decay_1(self): + """Tests floating point ternary LIF model with negative bias + driving a neuron without any decay of current and voltage states.""" + shape = (10,) + num_steps = 30 + # Set up external input to 0 + sps = VecSendProcess(shape=shape, num_steps=num_steps, + vec_to_send=np.zeros(shape, dtype=float), + send_at_times=np.ones((num_steps,), dtype=bool)) + # Set up bias = 1 * 2**1 = 2. and threshold = 4. + # du and dv = 0 => bias driven neurons spike at every 2nd time-step. + tlif = TernaryLIF(shape=shape, du=0., dv=0., + bias=(-1) * np.ones(shape, dtype=float), + bias_exp=np.ones(shape, dtype=float), + vth_lo=-7., vth_hi=5.) + # Receive neuron spikes + spr = VecRecvProcess(shape=(num_steps, shape[0])) + sps.s_out.connect(tlif.a_in) + tlif.s_out.connect(spr.s_in) + # Configure execution and run + rcnd = RunSteps(num_steps=num_steps) + rcfg = LifRunConfig(select_tag='floating_pt') + tlif.run(condition=rcnd, run_cfg=rcfg) + # Gather spike data and stop + spk_data_through_run = spr.spk_data.get() + tlif.stop() + # Gold standard for the test + expected_spk_data = np.zeros((num_steps, shape[0])) + expected_spk_data[7:30:8, :] = -1. + self.assertTrue(np.all(expected_spk_data == spk_data_through_run)) + + def test_float_pm_neg_no_decay_2(self): + """Tests +1 and -1 spike responses of a floating point ternary LIF + model driven by alternating spiking inputs. No current or voltage + decay, no bias.""" + shape = (10,) + num_steps = 11 + pos_idx = np.hstack((np.arange(3), np.arange(9, 11))) + send_steps_pos = np.zeros((num_steps,), dtype=bool) + send_steps_pos[pos_idx] = True + send_steps_neg = (1 - send_steps_pos).astype(bool) + # Set up external input to 0 + sps_pos = VecSendProcess(shape=shape, num_steps=num_steps, + vec_to_send=np.ones(shape, dtype=float), + send_at_times=send_steps_pos) + sps_neg = VecSendProcess(shape=shape, num_steps=num_steps, + vec_to_send=(-1) * np.ones(shape, dtype=float), + send_at_times=send_steps_neg) + # Set up bias = 1 * 2**1 = 2. and threshold = 4. + # du and dv = 0 => bias driven neurons spike at every 2nd time-step. + tlif = TernaryLIF(shape=shape, du=0., dv=0., + bias=np.zeros(shape, dtype=float), + bias_exp=np.ones(shape, dtype=float), + vth_lo=-3., vth_hi=5.) + # Receive neuron spikes + spr = VecRecvProcess(shape=(num_steps, shape[0])) + sps_pos.s_out.connect(tlif.a_in) + sps_neg.s_out.connect(tlif.a_in) + tlif.s_out.connect(spr.s_in) + # Configure execution and run + rcnd = RunSteps(num_steps=num_steps) + rcfg = LifRunConfig(select_tag='floating_pt') + tlif.run(condition=rcnd, run_cfg=rcfg) + # Gather spike data and stop + spk_data_through_run = spr.spk_data.get() + tlif.stop() + # Gold standard for the test + expected_spk_data = np.zeros((num_steps, shape[0])) + expected_spk_data[2, :] = 1. + expected_spk_data[9, :] = -1. + self.assertTrue(np.all(expected_spk_data == spk_data_through_run)) + + def test_float_pm_neg_impulse_du(self): + """Tests the impulse response of the floating point ternary LIF + neuron model with current decay but without voltage decay""" + shape = (1,) # a single neuron + num_steps = 8 + # send activation of -128. at timestep = 1 + sps = VecSendProcess(shape=shape, num_steps=num_steps, + vec_to_send=(-2 ** 7) * np.ones(shape, + dtype=float), + send_at_times=np.array([True, False, False, + False, False, False, + False, False])) + # Set up no bias, no voltage decay. Current decay = 0.5 + # Set up threshold high, such that there are no output spikes + tlif = TernaryLIF(shape=shape, + du=0.5, dv=0, + bias=np.zeros(shape, dtype=float), + bias_exp=np.ones(shape, dtype=float), + vth_lo=-256., vth_hi=2) + spr = VecRecvProcess(shape=(num_steps, shape[0])) + sps.s_out.connect(tlif.a_in) + tlif.s_out.connect(spr.s_in) + # Configure to run 1 step at a time + rcnd = RunSteps(num_steps=1) + rcfg = LifRunConfig(select_tag='floating_pt') + lif_u = [] + # Run 1 timestep at a time and collect state variable u + for j in range(num_steps): + tlif.run(condition=rcnd, run_cfg=rcfg) + lif_u.append(tlif.u.get()[0]) + tlif.stop() + # Gold standard for testing: current decay of 0.5 should halve the + # current every time-step + expected_u_timeseries = [-2. ** (7 - j) for j in range(8)] + self.assertListEqual(expected_u_timeseries, lif_u) + + def test_float_pm_neg_impulse_dv(self): + """Tests the impulse response of the floating point ternary LIF + neuron model with voltage decay but without current decay""" + shape = (1,) # a single neuron + num_steps = 8 + # send activation of -128. at timestep = 1 + sps = VecSendProcess(shape=shape, num_steps=num_steps, + vec_to_send=(-2 ** 7) * np.ones(shape, + dtype=float), + send_at_times=np.array([True, False, False, + False, False, False, + False, False])) + # Set up no bias, no current decay. Voltage decay = 0.5 + # Set up threshold high, such that there are no output spikes + tlif = TernaryLIF(shape=shape, + du=0, dv=0.5, + bias=np.zeros(shape, dtype=float), + bias_exp=np.ones(shape, dtype=float), + vth_lo=-256., vth_hi=2.) + spr = VecRecvProcess(shape=(num_steps, shape[0])) + sps.s_out.connect(tlif.a_in) + tlif.s_out.connect(spr.s_in) + # Configure to run 1 step at a time + rcnd = RunSteps(num_steps=1) + rcfg = LifRunConfig(select_tag='floating_pt') + lif_v = [] + # Run 1 timestep at a time and collect state variable u + for j in range(num_steps): + tlif.run(condition=rcnd, run_cfg=rcfg) + lif_v.append(tlif.v.get()[0]) + tlif.stop() + # Gold standard for testing: voltage decay of 0.5 should integrate + # the voltage from -128. to -255., with steps of -64., -32., -16., etc. + expected_v_timeseries = [-128., -192., -224., -240., -248., -252., + -254., -255.] + self.assertListEqual(expected_v_timeseries, lif_v) + + +class TestTLIFProcessModelsFixed(unittest.TestCase): + """Tests for ternary LIF fixed point neuron model""" + def test_fixed_pm_neg_no_decay_1(self): + """Tests fixed point ProcessModel for ternary LIF neurons without any + current or voltage decay, solely driven by (negative) bias""" + shape = (5,) + num_steps = 10 + # Set up external input to 0 + sps = VecSendProcess(shape=shape, num_steps=num_steps, + vec_to_send=np.zeros(shape, dtype=np.int16), + send_at_times=np.ones((num_steps,), dtype=bool)) + # Set up bias = 2 * 2**6 = 128 and threshold = 8<<6 + # du and dv = 0 => bias driven neurons spike at every 4th time-step. + tlif = TernaryLIF(shape=shape, + du=0, dv=0, + bias=(-2) * np.ones(shape, dtype=np.int32), + bias_exp=6 * np.ones(shape, dtype=np.int32), + vth_lo=(-8), vth_hi=2) + # Receive neuron spikes + spr = VecRecvProcess(shape=(num_steps, shape[0])) + sps.s_out.connect(tlif.a_in) + tlif.s_out.connect(spr.s_in) + # Configure execution and run + rcnd = RunSteps(num_steps=num_steps) + rcfg = LifRunConfig(select_tag='fixed_pt') + tlif.run(condition=rcnd, run_cfg=rcfg) + # Gather spike data and stop + spk_data_through_run = spr.spk_data.get() + tlif.stop() + # Gold standard for the test + expected_spk_data = np.zeros((num_steps, shape[0])) + expected_spk_data[4:10:5, :] = -1 + self.assertTrue(np.all(expected_spk_data == spk_data_through_run)) + + def test_fixed_pm_neg_no_decay_2(self): + """Tests fixed point ProcessModel for ternary LIF neurons without any + current or voltage decay, driven by positive and negative spikes and + no bias.""" + shape = (10,) + num_steps = 11 + pos_idx = np.hstack((np.arange(3), np.arange(9, 11))) + send_steps_pos = np.zeros((num_steps,), dtype=bool) + send_steps_pos[pos_idx] = True + send_steps_neg = (1 - send_steps_pos).astype(bool) + # Set up external input to 0 + sps_pos = VecSendProcess(shape=shape, num_steps=num_steps, + vec_to_send=np.ones(shape, dtype=np.int32), + send_at_times=send_steps_pos) + sps_neg = VecSendProcess(shape=shape, num_steps=num_steps, + vec_to_send=(-1) * np.ones(shape, + dtype=np.int32), + send_at_times=send_steps_neg) + # Set up bias = 1 * 2**1 = 2. and threshold = 4. + # du and dv = 0 => bias driven neurons spike at every 2nd time-step. + tlif = TernaryLIF(shape=shape, du=0, dv=0, + bias=np.zeros(shape, dtype=np.int32), + bias_exp=np.ones(shape, dtype=np.int32), + vth_lo=-3, vth_hi=5) + # Receive neuron spikes + spr = VecRecvProcess(shape=(num_steps, shape[0])) + sps_pos.s_out.connect(tlif.a_in) + sps_neg.s_out.connect(tlif.a_in) + tlif.s_out.connect(spr.s_in) + # Configure execution and run + rcnd = RunSteps(num_steps=num_steps) + rcfg = LifRunConfig(select_tag='fixed_pt') + tlif.run(condition=rcnd, run_cfg=rcfg) + # Gather spike data and stop + spk_data_through_run = spr.spk_data.get() + tlif.stop() + # Gold standard for the test + expected_spk_data = np.zeros((num_steps, shape[0])) + expected_spk_data[2, :] = 1. + expected_spk_data[(8, 10), :] = -1. + self.assertTrue(np.all(expected_spk_data == spk_data_through_run)) + + def test_fixed_pm_neg_impulse_du(self): + """Tests the impulse response of the fixed point ternary LIF neuron + model with no voltage decay""" + shape = (1,) # a single neuron + num_steps = 8 + # send activation of 128. at timestep = 1 + sps = VecSendProcess(shape=shape, num_steps=num_steps, + vec_to_send=(-128) * np.ones(shape, + dtype=np.int32), + send_at_times=np.array([True, False, False, + False, False, False, + False, False])) + # Set up no bias, no voltage decay. Current decay is a 12-bit + # unsigned variable in Loihi hardware. Therefore, du = 2047 is + # equivalent to (1/2) * (2**12) - 1. The subtracted 1 is added by + # default in the hardware, via a setting ds_offset, thereby finally + # giving du = 2048 = 0.5 * 2**12 + # Set up threshold high, such that there are no output spikes. By + # default the threshold value here is left-shifted by 6. + tlif = TernaryLIF(shape=shape, + du=2047, dv=0, + bias=np.zeros(shape, dtype=np.int16), + bias_exp=np.ones(shape, dtype=np.int16), + vth_lo=(-256) * np.ones(shape, dtype=np.int32), + vth_hi=2 * np.ones(shape, dtype=np.int32)) + spr = VecRecvProcess(shape=(num_steps, shape[0])) + sps.s_out.connect(tlif.a_in) + tlif.s_out.connect(spr.s_in) + # Configure to run 1 step at a time + rcnd = RunSteps(num_steps=1) + rcfg = LifRunConfig(select_tag='fixed_pt') + lif_u = [] + # Run 1 timestep at a time and collect state variable u + for j in range(num_steps): + tlif.run(condition=rcnd, run_cfg=rcfg) + lif_u.append(tlif.u.get().astype(np.int32)[0]) + tlif.stop() + # Gold standard for testing: current decay of 0.5 should halve the + # current every time-step. + expected_u_timeseries = [(-1) << (13 - j) for j in range(8)] + # Gold standard for floating point equivalent of the current, + # which would be all Loihi-bit-accurate values right shifted by 6 bits + expected_float_u = [(-1) << (7 - j) for j in range(8)] + self.assertListEqual(expected_u_timeseries, lif_u) + self.assertListEqual(expected_float_u, np.right_shift(np.array( + lif_u), 6).tolist()) + + def test_fixed_pm_neg_impulse_dv(self): + """Tests the impulse response of the fixed point ternary LIF neuron + model with no current decay""" + shape = (1,) # a single neuron + num_steps = 8 + # send activation of 128. at timestep = 1 + sps = VecSendProcess(shape=shape, num_steps=num_steps, + vec_to_send=(-128) * np.ones(shape, + dtype=np.int32), + send_at_times=np.array([True, False, False, + False, False, False, + False, False])) + # Set up no bias, no current decay. Voltage decay is a 12-bit + # unsigned variable in Loihi hardware. Therefore, dv = 2048 is + # equivalent to (1/2) * (2**12). + # Set up threshold high, such that there are no output spikes. + # Threshold provided here is left-shifted by 6-bits. + tlif = TernaryLIF(shape=shape, + du=0, dv=2048, + bias=np.zeros(shape, dtype=np.int16), + bias_exp=np.ones(shape, dtype=np.int16), + vth_lo=(-256) * np.ones(shape, dtype=np.int32), + vth_hi=2 * np.ones(shape, dtype=np.int32)) + spr = VecRecvProcess(shape=(num_steps, shape[0])) + sps.s_out.connect(tlif.a_in) + tlif.s_out.connect(spr.s_in) + # Configure to run 1 step at a time + rcnd = RunSteps(num_steps=1) + rcfg = LifRunConfig(select_tag='fixed_pt') + lif_v = [] + # Run 1 timestep at a time and collect state variable u + for j in range(num_steps): + tlif.run(condition=rcnd, run_cfg=rcfg) + lif_v.append(tlif.v.get().astype(np.int32)[0]) + tlif.stop() + # Gold standard for testing: with a voltage decay of 2048, voltage + # should integrate from 128<<6 to 255<<6. But it is slightly smaller, + # because current decay is not exactly 0. Due to the default + # ds_offset = 1 setting in the hardware, current decay = 1. So + # voltage is slightly smaller than 128<<6 to 255<<6. + expected_v_timeseries = [-8192, -12286, -14331, -15351, -15859, -16111, + -16235, -16295] + # Gold standard for floating point equivalent of the voltage, + # which would be all Loihi-bit-accurate values right shifted by 6 bits + expected_float_v = [-128, -192, -224, -240, -248, -252, -254, -255] + lif_v_float = np.right_shift(np.array(lif_v), 6) self.assertListEqual(expected_v_timeseries, lif_v) self.assertListEqual(expected_float_v, lif_v_float.tolist()) diff --git a/tests/lava/proc/lif/test_process.py b/tests/lava/proc/lif/test_process.py index 6caf67d28..e9c902700 100644 --- a/tests/lava/proc/lif/test_process.py +++ b/tests/lava/proc/lif/test_process.py @@ -3,7 +3,7 @@ # See: https://spdx.org/licenses/ import unittest import numpy as np -from lava.proc.lif.process import LIF +from lava.proc.lif.process import LIF, TernaryLIF class TestLIFProcess(unittest.TestCase): @@ -11,15 +11,45 @@ class TestLIFProcess(unittest.TestCase): def test_init(self): """Tests instantiation of LIF""" lif = LIF(shape=(100,), - du=100 * np.ones((100,), dtype=float), - dv=np.ones((100,), dtype=float), + du=100., dv=1., bias=2 * np.ones((100,), dtype=float), bias_exp=np.ones((100,), dtype=float), - vth=np.ones((100,), dtype=float)) + vth=1.) self.assertEqual(lif.shape, (100,)) - self.assertListEqual(lif.du.init.tolist(), 100 * [100.]) - self.assertListEqual(lif.dv.init.tolist(), 100 * [1.]) + self.assertEqual(lif.du.init, 100.) + self.assertEqual(lif.dv.init, 1.) self.assertListEqual(lif.bias.init.tolist(), 100 * [2.]) self.assertListEqual(lif.bias_exp.init.tolist(), 100 * [1.]) - self.assertListEqual(lif.vth.init.tolist(), 100 * [1.]) + self.assertEqual(lif.vth.init, 1.) + + +class TestTLIFProcess(unittest.TestCase): + """Tests for T-LIF class""" + def test_init(self): + """Tests for instantiation of Ternary LIF""" + tlif = TernaryLIF(shape=(100,), + du=100., dv=1., + bias=2 * np.ones((100,), dtype=float), + bias_exp=np.ones((100,), dtype=float), + vth_lo=-3., vth_hi=5.) + + self.assertEqual(tlif.shape, (100,)) + self.assertEqual(tlif.du.init, 100.) + self.assertEqual(tlif.dv.init, 1.) + self.assertListEqual(tlif.bias.init.tolist(), 100 * [2.]) + self.assertListEqual(tlif.bias_exp.init.tolist(), 100 * [1.]) + self.assertEqual(tlif.vth_lo.init, -3.) + self.assertEqual(tlif.vth_hi.init, 5.) + + def test_vth_hi_lo_order(self): + """Test if the check to assert the order of upper and lower + thresholds works properly, i.e., we should get AssertionError if + lower threshold is greater than upper threshold""" + + with(self.assertRaises(AssertionError)): + _ = TernaryLIF(shape=(100,), + du=100., dv=1., + bias=2 * np.ones((100,), dtype=float), + bias_exp=np.ones((100,), dtype=float), + vth_lo=15., vth_hi=5.) diff --git a/tests/lava/proc/monitor/test_monitors.py b/tests/lava/proc/monitor/test_monitors.py index ae86a6536..f276368e2 100644 --- a/tests/lava/proc/monitor/test_monitors.py +++ b/tests/lava/proc/monitor/test_monitors.py @@ -229,8 +229,8 @@ def test_monitor_collects_voltage_and_spike_data_from_lif_neuron(self): spike_data = data2[neuron.name][neuron.s_out.name] # Check if this data match the expected data - self.assertTrue(np.all(volt_data == np.array([[1, 2, 0, 1, 2, 0]]).T)) - self.assertTrue(np.all(spike_data == np.array([[0, 0, 1, 0, 0, 1]]).T)) + self.assertTrue(np.all(volt_data == np.array([[1, 2, 3, 0, 1, 2]]).T)) + self.assertTrue(np.all(spike_data == np.array([[0, 0, 0, 1, 0, 0]]).T)) def test_monitor_collects_voltage_and_spike_data_from_population_lif(self): """Check if two different Monitor process can monitor voltage (Var) and @@ -272,10 +272,10 @@ def test_monitor_collects_voltage_and_spike_data_from_population_lif(self): spike_data = data2[neuron.name][neuron.s_out.name] # Check if this data match the expected data - self.assertTrue(np.all(volt_data == np.array([[1, 2, 0, 1, 2, 0], - [1, 2, 0, 1, 2, 0]]).T)) - self.assertTrue(np.all(spike_data == np.array([[0, 0, 1, 0, 0, 1], - [0, 0, 1, 0, 0, 1]]).T)) + self.assertTrue(np.all(volt_data == np.array([[1, 2, 3, 0, 1, 2], + [1, 2, 3, 0, 1, 2]]).T)) + self.assertTrue(np.all(spike_data == np.array([[0, 0, 0, 1, 0, 0], + [0, 0, 0, 1, 0, 0]]).T)) def test_monitor_collects_data_from_2D_population_lif(self): """Check if two different Monitor process can monitor voltage (Var) and @@ -318,9 +318,9 @@ def test_monitor_collects_data_from_2D_population_lif(self): # Check if this data match the expected data expected_voltages = np.tile( - np.expand_dims(np.array([1, 2, 0, 1, 2, 0]), (1, 2)), (1,) + shape) + np.expand_dims(np.array([1, 2, 3, 0, 1, 2]), (1, 2)), (1,) + shape) expected_spikes = np.tile( - np.expand_dims(np.array([0, 0, 1, 0, 0, 1]), (1, 2)), (1,) + shape) + np.expand_dims(np.array([0, 0, 0, 1, 0, 0]), (1, 2)), (1,) + shape) self.assertTrue(np.all(volt_data == expected_voltages)) self.assertTrue(np.all(spike_data == expected_spikes))