Nonseq

sinabs/backend/dynapcnn/__init__.py

@@ -1,5 +1,14 @@
-from .dynapcnn_network import (  # second one for compatibility purposes
+from .dynapcnn_network import (
     DynapcnnCompatibleNetwork,
     DynapcnnNetwork,
 )
+
+from .dynapcnn_layer import (
+    DynapcnnLayer,
+)
+
+from .dynapcnn_layer_handler import (
+    DynapcnnLayerHandler,
+)
+
 from .dynapcnn_visualizer import DynapcnnVisualizer

sinabs/backend/dynapcnn/chips/dynapcnn.py

@@ -10,9 +10,10 @@
 from sinabs.backend.dynapcnn.config_builder import ConfigBuilder
 from sinabs.backend.dynapcnn.dvs_layer import DVSLayer, expand_to_pair
 from sinabs.backend.dynapcnn.dynapcnn_layer import DynapcnnLayer
+from sinabs.backend.dynapcnn.dynapcnn_network import DynapcnnNetwork
+from sinabs.backend.dynapcnn.dynapcnn_layer_handler import DynapcnnLayerHandler
 from sinabs.backend.dynapcnn.mapping import LayerConstraints
 
-
 class DynapcnnConfigBuilder(ConfigBuilder):
     @classmethod
     def get_samna_module(cls):
@@ -44,26 +45,26 @@ def set_kill_bits(cls, layer: DynapcnnLayer, config_dict: dict) -> dict:
         """
         config_dict = copy.deepcopy(config_dict)
 
-        if layer.conv_layer.bias is not None:
-            (weights, biases) = layer.conv_layer.parameters()
+        if layer.conv.bias is not None:
+            (weights, biases) = layer.conv.parameters()
         else:
-            (weights,) = layer.conv_layer.parameters()
-            biases = torch.zeros(layer.conv_layer.out_channels)
+            (weights,) = layer.conv.parameters()
+            biases = torch.zeros(layer.conv.out_channels)
 
         config_dict["weights_kill_bit"] = (~weights.bool()).tolist()
         config_dict["biases_kill_bit"] = (~biases.bool()).tolist()
 
         # - Neuron states
-        if not layer.spk_layer.is_state_initialised():
+        if not layer.spk.is_state_initialised():
             # then we assign no initial neuron state to DYNAP-CNN.
             f, h, w = layer.get_neuron_shape()
             neurons_state = torch.zeros(f, w, h)
-        elif layer.spk_layer.v_mem.dim() == 4:
+        elif layer.spk.v_mem.dim() == 4:
             # 4-dimensional states should be the norm when there is a batch dim
-            neurons_state = layer.spk_layer.v_mem.transpose(2, 3)[0]
+            neurons_state = layer.spk.v_mem.transpose(2, 3)[0]
         else:
             raise ValueError(
-                f"Current v_mem (shape: {layer.spk_layer.v_mem.shape}) of spiking layer not understood."
+                f"Current v_mem (shape: {layer.spk.v_mem.shape}) of spiking layer not understood."
             )
 
         config_dict["neurons_value_kill_bit"] = (
@@ -73,12 +74,12 @@ def set_kill_bits(cls, layer: DynapcnnLayer, config_dict: dict) -> dict:
         return config_dict
 
     @classmethod
-    def get_dynapcnn_layer_config_dict(cls, layer: DynapcnnLayer):
+    def get_dynapcnn_layer_config_dict(cls, layer: DynapcnnLayer, layer_handler: DynapcnnLayerHandler, all_handlers: dict) -> dict:
         config_dict = {}
         config_dict["destinations"] = [{}, {}]
 
         # Update the dimensions
-        channel_count, input_size_y, input_size_x = layer.input_shape
+        channel_count, input_size_y, input_size_x = layer.in_shape
         dimensions = {"input_shape": {}, "output_shape": {}}
         dimensions["input_shape"]["size"] = {"x": input_size_x, "y": input_size_y}
         dimensions["input_shape"]["feature_count"] = channel_count
@@ -90,148 +91,164 @@ def get_dynapcnn_layer_config_dict(cls, layer: DynapcnnLayer):
         dimensions["output_shape"]["size"]["x"] = w
         dimensions["output_shape"]["size"]["y"] = h
         dimensions["padding"] = {
-            "x": layer.conv_layer.padding[1],
-            "y": layer.conv_layer.padding[0],
+            "x": layer.conv.padding[1],
+            "y": layer.conv.padding[0],
         }
         dimensions["stride"] = {
-            "x": layer.conv_layer.stride[1],
-            "y": layer.conv_layer.stride[0],
+            "x": layer.conv.stride[1],
+            "y": layer.conv.stride[0],
         }
-        dimensions["kernel_size"] = layer.conv_layer.kernel_size[0]
+        dimensions["kernel_size"] = layer.conv.kernel_size[0]
 
-        if dimensions["kernel_size"] != layer.conv_layer.kernel_size[1]:
+        if dimensions["kernel_size"] != layer.conv.kernel_size[1]:
             raise ValueError("Conv2d: Kernel must have same height and width.")
         config_dict["dimensions"] = dimensions
         # Update parameters from convolution
-        if layer.conv_layer.bias is not None:
-            (weights, biases) = layer.conv_layer.parameters()
+        if layer.conv.bias is not None:
+            (weights, biases) = layer.conv.parameters()
         else:
-            (weights,) = layer.conv_layer.parameters()
-            biases = torch.zeros(layer.conv_layer.out_channels)
+            (weights,) = layer.conv.parameters()
+            biases = torch.zeros(layer.conv.out_channels)
         weights = weights.transpose(2, 3)  # Need this to match samna convention
         config_dict["weights"] = weights.int().tolist()
         config_dict["biases"] = biases.int().tolist()
         config_dict["leak_enable"] = biases.bool().any()
-        # config_dict["weights_kill_bit"] = torch.zeros_like(weights).bool().tolist()
-        # config_dict["biases_kill_bit"] = torch.zeros_like(biases).bool().tolist()
 
         # Update parameters from the spiking layer
 
         # - Neuron states
-        if not layer.spk_layer.is_state_initialised():
+        if not layer.spk.is_state_initialised():
             # then we assign no initial neuron state to DYNAP-CNN.
             f, h, w = layer.get_neuron_shape()
             neurons_state = torch.zeros(f, w, h)
-        elif layer.spk_layer.v_mem.dim() == 4:
+        elif layer.spk.v_mem.dim() == 4:
             # 4-dimensional states should be the norm when there is a batch dim
-            neurons_state = layer.spk_layer.v_mem.transpose(2, 3)[0]
+            neurons_state = layer.spk.v_mem.transpose(2, 3)[0]
         else:
             raise ValueError(
-                f"Current v_mem (shape: {layer.spk_layer.v_mem.shape}) of spiking layer not understood."
+                f"Current v_mem (shape: {layer.spk.v_mem.shape}) of spiking layer not understood."
             )
 
         # - Resetting vs returning to 0
-        if isinstance(layer.spk_layer.reset_fn, sinabs.activation.MembraneReset):
+        if isinstance(layer.spk.reset_fn, sinabs.activation.MembraneReset):
             return_to_zero = True
-        elif isinstance(layer.spk_layer.reset_fn, sinabs.activation.MembraneSubtract):
+        elif isinstance(layer.spk.reset_fn, sinabs.activation.MembraneSubtract):
             return_to_zero = False
         else:
             raise Exception(
                 "Unknown reset mechanism. Only MembraneReset and MembraneSubtract are currently understood."
             )
 
-        # if (not return_to_zero) and self.spk_layer.membrane_subtract != self.spk_layer.threshold:
-        #    warn(
-        #        "SpikingConv2dLayer: Subtraction of membrane potential is always by high threshold."
-        #    )
-        if layer.spk_layer.min_v_mem is None:
+        if layer.spk.min_v_mem is None:
             min_v_mem = -(2**15)
         else:
-            min_v_mem = int(layer.spk_layer.min_v_mem)
+            min_v_mem = int(layer.spk.min_v_mem)
         config_dict.update(
             {
                 "return_to_zero": return_to_zero,
-                "threshold_high": int(layer.spk_layer.spike_threshold),
+                "threshold_high": int(layer.spk.spike_threshold),
                 "threshold_low": min_v_mem,
                 "monitor_enable": False,
                 "neurons_initial_value": neurons_state.int().tolist(),
-                # "neurons_value_kill_bit" : torch.zeros_like(neurons_state).bool().tolist()
             }
         )
-        # Update parameters from pooling
-        if layer.pool_layer is not None:
-            config_dict["destinations"][0]["pooling"] = expand_to_pair(
-                layer.pool_layer.kernel_size
-            )[0]
-            config_dict["destinations"][0]["enable"] = True
-        else:
-            pass
+
+        # setting destinations config. based on destinations destination nodes of the nodes withing this `dcnnl`.
+        destinations = []
+        for node_id, destination_nodes in layer_handler.nodes_destinations.items():
+            for dest_node in destination_nodes:
+                core_id = DynapcnnLayerHandler.find_nodes_core_id(dest_node, all_handlers)
+                kernel_size = layer_handler.get_pool_kernel_size(node_id)
+
+                dest_data = {
+                    'layer': core_id,
+                    'enable': True,
+                    'pooling': expand_to_pair(kernel_size if kernel_size else 1),
+                }
+
+                destinations.append(dest_data)
+        config_dict["destinations"] = destinations
 
         # Set kill bits
         config_dict = cls.set_kill_bits(layer=layer, config_dict=config_dict)
 
         return config_dict
-
+            
     @classmethod
-    def write_dynapcnn_layer_config(
-        cls, layer: DynapcnnLayer, chip_layer: "CNNLayerConfig"
-    ):
-        """Write a single layer configuration to the dynapcnn conf object.
+    def write_dynapcnn_layer_config(cls, layer: DynapcnnLayer, chip_layer: "CNNLayerConfig", layer_handler: DynapcnnLayerHandler, all_handlers: dict) -> None:
+        """ Write a single layer configuration to the dynapcnn conf object. Uses the data in `layer` to configure a `CNNLayerConfig` to be 
+        deployed on chip.
 
         Parameters
         ----------
-            layer:
-                The dynapcnn layer to write the configuration for
-            chip_layer: CNNLayerConfig
-                DYNAPCNN configuration object representing the layer to which
-                configuration is written.
+        - layer (DynapcnnLayer): the layer for which the condiguration will be written.
+        - chip_layer (CNNLayerConfig): configuration object representing the layer to which configuration is written.
+        - layer_handler (DynapcnnLayerHandler): ...
+        - all_handlers (dict): ...
         """
-        config_dict = cls.get_dynapcnn_layer_config_dict(layer=layer)
-        # Update configuration of the DYNAPCNN layer
+
+        # extracting from a DynapcnnLayer the config. variables for its CNNLayerConfig.
+        config_dict = cls.get_dynapcnn_layer_config_dict(layer=layer, layer_handler=layer_handler, all_handlers=all_handlers)
+
+        # update configuration of the DYNAPCNN layer.
         chip_layer.dimensions = config_dict["dimensions"]
         config_dict.pop("dimensions")
-        for i in range(len(config_dict["destinations"])):
-            if "pooling" in config_dict["destinations"][i]:
-                chip_layer.destinations[i].pooling = config_dict["destinations"][i][
-                    "pooling"
-                ]
-        config_dict.pop("destinations")
+
+        # set the destinations configuration.
+        for i in range(len(config_dict['destinations'])):
+            chip_layer.destinations[i].layer = config_dict['destinations'][i]['layer']
+            chip_layer.destinations[i].enable = config_dict['destinations'][i]['enable']
+            chip_layer.destinations[i].pooling = config_dict['destinations'][i]['pooling']
+
+        config_dict.pop('destinations')
+
+        # set remaining configuration.
         for param, value in config_dict.items():
             try:
                 setattr(chip_layer, param, value)
             except TypeError as e:
                 raise TypeError(f"Unexpected parameter {param} or value. {e}")
 
     @classmethod
-    def build_config(cls, model: "DynapcnnNetwork", chip_layers: List[int]):
-        layers = model.sequence
+    def build_config(cls, model: DynapcnnNetwork) -> DynapcnnConfiguration:
+        """ Uses `DynapcnnLayer` objects to configure their equivalent chip core via a `CNNLayerConfig` object that is built
+        using using the `DynapcnnLayer` properties. 
+
+        Parameters
+        ----------
+        - model (DynapcnnNetwork): network instance used to read out `DynapcnnLayer` instances.
+
+        Returns
+        ----------
+        - config (DynapcnnConfiguration): an instance of a `DynapcnnConfiguration`.
+        """
         config = cls.get_default_config()
 
-        has_dvs_layer = False
-        i_cnn_layer = 0  # Instantiate an iterator for the cnn cores
-        for i, chip_equivalent_layer in enumerate(layers):
-            if isinstance(chip_equivalent_layer, DVSLayer):
-                chip_layer = config.dvs_layer
-                cls.write_dvs_layer_config(chip_equivalent_layer, chip_layer)
-                has_dvs_layer = True
-            elif isinstance(chip_equivalent_layer, DynapcnnLayer):
-                chip_layer = config.cnn_layers[chip_layers[i_cnn_layer]]
-                cls.write_dynapcnn_layer_config(chip_equivalent_layer, chip_layer)
-                i_cnn_layer += 1
-            else:
-                # in our generated network there is a spurious layer...
-                # should never happen
-                raise TypeError("Unexpected layer in the model")
+        if not isinstance(model, DynapcnnNetwork):
+            raise ValueError(f"`model` has to be of type DynapcnnNetwork, but is {type(model)}.")
+
+        has_dvs_layer = False   # TODO DVSLayer not supported yet.
 
-            if i == len(layers) - 1:
-                # last layer
-                chip_layer.destinations[0].enable = False
-            else:
-                # Set destination layer
-                chip_layer.destinations[0].layer = chip_layers[i_cnn_layer]
-                chip_layer.destinations[0].enable = True
+        # Loop over layers in network and write corresponding configurations
+        for layer_index, ith_dcnnl in model.layers_mapper.items():
+            if isinstance(ith_dcnnl, DVSLayer):
+                # TODO DVSLayer not supported yet.
+                pass
+
+            elif isinstance(ith_dcnnl, DynapcnnLayer):
+                # retrieve assigned core from the handler of this DynapcnnLayer (`ith_dcnnl`) instance.
+                chip_layer = config.cnn_layers[model.layers_handlers[layer_index].assigned_core]
+                # write core configuration.
+                cls.write_dynapcnn_layer_config(ith_dcnnl, chip_layer, model.layers_handlers[layer_index], model.layers_handlers)
 
+            else:
+                # shouldn't happen since type checks are made previously.
+                raise TypeError(f"Layer (index {layer_index}) is unexpected in the model: \n{ith_dcnnl}")
+
         if not has_dvs_layer:
+            # TODO DVSLayer not supported yet.
+            config.dvs_layer.pass_sensor_events = False
+        else:
             config.dvs_layer.pass_sensor_events = False
 
         return config
@@ -293,12 +310,13 @@ def monitor_layers(cls, config: "DynapcnnConfiguration", layers: List):
             config.dvs_layer.monitor_enable = True
             if config.dvs_layer.pooling.x != 1 or config.dvs_layer.pooling.y != 1:
                 warn(
-                    f"DVS layer has pooling and is being monitored. "
+                    "DVS layer has pooling and is being monitored. "
                     "Note that pooling will not be reflected in the monitored events."
                 )
             monitor_layers.remove("dvs")
         for lyr_indx in monitor_layers:
             config.cnn_layers[lyr_indx].monitor_enable = True
+
             if any(
                 dest.pooling != 1 for dest in config.cnn_layers[lyr_indx].destinations
             ):

sinabs/backend/dynapcnn/chips/speck2cmini.py

@@ -1,4 +1,4 @@
-from typing import List
+from typing import List, Dict
 
 import samna
 from samna.speck2cMini.configuration import SpeckConfiguration
@@ -29,8 +29,8 @@ def get_output_buffer(cls):
         return samna.BasicSinkNode_speck2c_mini_event_output_event()
 
     @classmethod
-    def get_dynapcnn_layer_config_dict(cls, layer: DynapcnnLayer):
-        config_dict = super().get_dynapcnn_layer_config_dict(layer=layer)
+    def get_dynapcnn_layer_config_dict(cls, layer: DynapcnnLayer, layers_mapper: Dict[int, DynapcnnLayer]) -> dict:
+        config_dict = super().get_dynapcnn_layer_config_dict(layer=layer, layers_mapper=layers_mapper)
         config_dict.pop("weights_kill_bit")
         config_dict.pop("biases_kill_bit")
         config_dict.pop("neurons_value_kill_bit")

sinabs/backend/dynapcnn/chips/speck2e.py

@@ -5,6 +5,8 @@
 
 from .dynapcnn import DynapcnnConfigBuilder
 
+from typing import Dict
+
 # Since most of the configuration is identical to DYNAP-CNN, we can simply inherit this class
 
 
@@ -30,6 +32,6 @@ def set_kill_bits(cls, layer: DynapcnnLayer, config_dict: dict) -> dict:
         return config_dict
 
     @classmethod
-    def get_dynapcnn_layer_config_dict(cls, layer: DynapcnnLayer):
-        config_dict = super().get_dynapcnn_layer_config_dict(layer=layer)
+    def get_dynapcnn_layer_config_dict(cls, layer: DynapcnnLayer, layers_mapper: Dict[int, DynapcnnLayer]) -> dict:
+        config_dict = super().get_dynapcnn_layer_config_dict(layer=layer, layers_mapper=layers_mapper)
         return config_dict

sinabs/backend/dynapcnn/chips/speck2f.py

@@ -5,6 +5,8 @@
 
 from .dynapcnn import DynapcnnConfigBuilder
 
+from typing import Dict
+
 # Since most of the configuration is identical to DYNAP-CNN, we can simply inherit this class
 
 
@@ -26,8 +28,8 @@ def get_output_buffer(cls):
         return samna.BasicSinkNode_speck2f_event_output_event()
 
     @classmethod
-    def get_dynapcnn_layer_config_dict(cls, layer: DynapcnnLayer):
-        config_dict = super().get_dynapcnn_layer_config_dict(layer=layer)
+    def get_dynapcnn_layer_config_dict(cls, layer: DynapcnnLayer, layers_mapper: Dict[int, DynapcnnLayer]) -> dict:
+        config_dict = super().get_dynapcnn_layer_config_dict(layer=layer, layers_mapper=layers_mapper)
         config_dict.pop("weights_kill_bit")
         config_dict.pop("biases_kill_bit")
         config_dict.pop("neurons_value_kill_bit")