From 20ee9a06dcd07de0b53445583580f05e5dc61445 Mon Sep 17 00:00:00 2001 From: szc321 Date: Tue, 20 Jun 2023 13:36:06 +0800 Subject: [PATCH] fix static code analysis Signed-off-by: szc321 --- .../message_infrastructure/pypychannel.py | 2 +- src/lava/proc/dense/models.py | 159 ------------------ 2 files changed, 1 insertion(+), 160 deletions(-) diff --git a/src/lava/magma/runtime/message_infrastructure/pypychannel.py b/src/lava/magma/runtime/message_infrastructure/pypychannel.py index 1d5bd8601..847cd1bd9 100644 --- a/src/lava/magma/runtime/message_infrastructure/pypychannel.py +++ b/src/lava/magma/runtime/message_infrastructure/pypychannel.py @@ -258,7 +258,7 @@ def _req_callback(self, req): not_empty = self.probe() self._queue.put_nowait(0) if self.observer and not not_empty: - self.observer() + self.observer() # pylint: disable=E1102 except EOFError: pass diff --git a/src/lava/proc/dense/models.py b/src/lava/proc/dense/models.py index 01754fc29..84555d21e 100644 --- a/src/lava/proc/dense/models.py +++ b/src/lava/proc/dense/models.py @@ -346,162 +346,3 @@ def run_spk(self): ) self.recv_traces(s_in) - - -class AbstractPyDelayDenseModel(PyLoihiProcessModel): - """Abstract Conn Process with Dense synaptic connections which incorporates - delays into the Conn Process. - """ - weights: np.ndarray = None - delays: np.ndarray = None - a_buff: np.ndarray = None - - def calc_act(self, s_in) -> np.ndarray: - """ - Calculate the activation matrix based on s_in by performing - delay_wgts * s_in. - """ - # First calculating the activations through delay_wgts * s_in - # This matrix is then summed across each row to get the - # activations to the output neurons for different delays. - # This activation vector is reshaped to a matrix of the form - # (n_flat_output_neurons * (max_delay + 1), n_flat_output_neurons) - # which is then transposed to get the activation matrix. - return np.reshape( - np.sum(self.get_delay_wgts_mat(self.weights, - self.delays) * s_in, axis=1), - (np.max(self.delays) + 1, self.weights.shape[0])).T - - @staticmethod - def get_delay_wgts_mat(weights, delays) -> np.ndarray: - """ - Create a matrix where the synaptic weights are separated - by their corresponding delays. The first matrix contains all the - weights, where the delay is equal to zero. The second matrix - contains all the weights, where the delay is equal to one and so on. - These matrices are then stacked together vertically. - - Returns 2D matrix of form - (num_flat_output_neurons * max_delay + 1, num_flat_input_neurons) where - delay_wgts[ - k * num_flat_output_neurons : (k + 1) * num_flat_output_neurons, : - ] - contains the weights for all connections with a delay equal to k. - This allows for the updating of the activation buffer and updating - weights. - """ - return np.vstack([ - np.where(delays == k, weights, 0) - for k in range(np.max(delays) + 1) - ]) - - def update_act(self, s_in): - """ - Updates the activations for the connection. - Clears first column of a_buff and rolls them to the last column. - Finally, calculates the activations for the current time step and adds - them to a_buff. - This order of operations ensures that delays of 0 correspond to - the next time step. - """ - self.a_buff[:, 0] = 0 - self.a_buff = np.roll(self.a_buff, -1) - self.a_buff += self.calc_act(s_in) - - -@implements(proc=DelayDense, protocol=LoihiProtocol) -@requires(CPU) -@tag("floating_pt") -class PyDelayDenseModelFloat(AbstractPyDelayDenseModel): - """Implementation of Conn Process with Dense synaptic connections 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. DelayDense incorporates delays into the Conn - Process. - """ - s_in: PyInPort = LavaPyType(PyInPort.VEC_DENSE, bool, precision=1) - a_out: PyOutPort = LavaPyType(PyOutPort.VEC_DENSE, float) - a_buff: np.ndarray = LavaPyType(np.ndarray, float) - # weights is a 2D matrix of form (num_flat_output_neurons, - # num_flat_input_neurons) in C-order (row major). - weights: np.ndarray = LavaPyType(np.ndarray, float) - # delays is a 2D matrix of form (num_flat_output_neurons, - # num_flat_input_neurons) in C-order (row major). - delays: np.ndarray = LavaPyType(np.ndarray, int) - num_message_bits: np.ndarray = LavaPyType(np.ndarray, int, precision=5) - - def run_spk(self): - # The a_out sent on each timestep is a buffered value from dendritic - # accumulation at timestep t-1. This prevents deadlocking in - # networks with recurrent connectivity structures. - self.a_out.send(self.a_buff[:, 0]) - if self.num_message_bits.item() > 0: - s_in = self.s_in.recv() - else: - s_in = self.s_in.recv().astype(bool) - self.update_act(s_in) - - -@implements(proc=DelayDense, protocol=LoihiProtocol) -@requires(CPU) -@tag("bit_accurate_loihi", "fixed_pt") -class PyDelayDenseModelBitAcc(AbstractPyDelayDenseModel): - """Implementation of Conn Process with Dense synaptic connections that is - bit-accurate with Loihi's hardware implementation of Dense, which means, - it mimics Loihi behaviour bit-by-bit. DelayDense incorporates delays into - the Conn Process. Loihi 2 has a maximum of 6 bits for delays, meaning a - spike can be delayed by 0 to 63 time steps.""" - - s_in: PyInPort = LavaPyType(PyInPort.VEC_DENSE, bool, precision=1) - a_out: PyOutPort = LavaPyType(PyOutPort.VEC_DENSE, np.int32, precision=16) - a_buff: np.ndarray = LavaPyType(np.ndarray, np.int32, precision=16) - # weights is a 2D matrix of form (num_flat_output_neurons, - # num_flat_input_neurons) in C-order (row major). - weights: np.ndarray = LavaPyType(np.ndarray, np.int32, precision=8) - delays: np.ndarray = LavaPyType(np.ndarray, np.int32, precision=6) - num_message_bits: np.ndarray = LavaPyType(np.ndarray, int, precision=5) - - def __init__(self, proc_params): - super().__init__(proc_params) - # Flag to determine whether weights have already been scaled. - self.weights_set = False - - def run_spk(self): - self.weight_exp: int = self.proc_params.get("weight_exp", 0) - - # Since this Process has no learning, weights are assumed to be static - # and only require scaling on the first timestep of run_spk(). - if not self.weights_set: - num_weight_bits: int = self.proc_params.get("num_weight_bits", 8) - sign_mode: SignMode = self.proc_params.get("sign_mode") \ - or determine_sign_mode(self.weights) - - self.weights = clip_weights(self.weights, sign_mode, num_bits=8) - self.weights = truncate_weights(self.weights, - sign_mode, - num_weight_bits) - self.weights_set = True - - # Check if delays are within Loihi 2 constraints - if np.max(self.delays) > 63: - raise ValueError("DelayDense Process 'delays' expects values " - f"between 0 and 63 for Loihi, got " - f"{self.delays}.") - - # The a_out sent at each timestep is a buffered value from dendritic - # accumulation at timestep t-1. This prevents deadlocking in - # networks with recurrent connectivity structures. - self.a_out.send(self.a_buff[:, 0]) - if self.num_message_bits.item() > 0: - s_in = self.s_in.recv() - else: - s_in = self.s_in.recv().astype(bool) - - a_accum = self.calc_act(s_in) - self.a_buff[:, 0] = 0 - self.a_buff = np.roll(self.a_buff, -1) - self.a_buff += ( - np.left_shift(a_accum, self.weight_exp) - if self.weight_exp > 0 - else np.right_shift(a_accum, -self.weight_exp) - )