Move to utils. New function: . Fix handling of pooling in deployment

sinabs/backend/dynapcnn/chips/dynapcnn.py

@@ -12,7 +12,7 @@
 
 import sinabs
 from sinabs.backend.dynapcnn.config_builder import ConfigBuilder
-from sinabs.backend.dynapcnn.dvs_layer import DVSLayer, expand_to_pair
+from sinabs.backend.dynapcnn.dvs_layer import DVSLayer
 from sinabs.backend.dynapcnn.dynapcnn_layer import DynapcnnLayer
 from sinabs.backend.dynapcnn.mapping import LayerConstraints
 
@@ -178,12 +178,24 @@ def get_dynapcnn_layer_config_dict(
         destinations = []
         pooling_sizes = layer.pool
         for dest_layer_id, pool in zip(destination_indices, pooling_sizes):
-            dest_data = {
-                "layer": layer2core_map[dest_layer_id],
-                "enable": True,
-                "pooling": expand_to_pair(pool),
-            }
-            destinations.append(dest_data)
+            # Ignore exit point destinations
+            if dest_layer_id >= 0:
+
+                try:
+                    # Use scalar value for pooling
+                    pool = sinabs.utils.collapse_pair(pool)
+                except ValueError:
+                    raise ValueError(
+                        f"Can only do pooling with quadratic kernels. Received {pool}"
+                    )
+
+                dest_data = {
+                    "layer": layer2core_map[dest_layer_id],
+                    "enable": True,
+                    "pooling": pool,
+                }
+                destinations.append(dest_data)
+
         config_dict["destinations"] = destinations
 
         # Set kill bits
@@ -222,8 +234,7 @@ def write_dynapcnn_layer_config(
         )
 
         # update configuration of the DYNAPCNN layer.
-        chip_layer.dimensions = config_dict["dimensions"]
-        config_dict.pop("dimensions")
+        chip_layer.dimensions = config_dict.pop("dimensions")
 
         # set the destinations configuration.
         for dest_idx, destination in enumerate(config_dict.pop("destinations")):
@@ -274,10 +285,10 @@ def build_config(
                 chip_layer = config.cnn_layers[layer2core_map[layer_index]]
                 # write core configuration.
                 cls.write_dynapcnn_layer_config(
-                    ith_dcnnl,
-                    chip_layer,
-                    destination_indices=destination_map[layer_index],
+                    layer=ith_dcnnl,
                     layer2core_map=layer2core_map,
+                    chip_layer=chip_layer,
+                    destination_indices=destination_map[layer_index],
                 )
 
             else:

sinabs/backend/dynapcnn/dvs_layer.py

@@ -3,27 +3,12 @@
 import torch.nn as nn
 
 from sinabs.layers import SumPool2d
+from sinabs.utils import expand_to_pair
 
 from .crop2d import Crop2d
 from .flipdims import FlipDims
 
 
-def expand_to_pair(value) -> (int, int):
-    """Expand a given value to a pair (tuple) if an int is passed.
-
-    Parameters
-    ----------
-    value:
-        int
-
-    Returns
-    -------
-    pair:
-        (int, int)
-    """
-    return (value, value) if isinstance(value, int) else value
-
-
 class DVSLayer(nn.Module):
     """DVSLayer representing the DVS pixel array on chip and/or the pre-processing. The order of
     processing is as follows MergePolarity -> Pool -> Cut -> Flip.

sinabs/backend/dynapcnn/dynapcnn_layer_utils.py

@@ -4,9 +4,9 @@
 from torch import nn
 
 from sinabs import layers as sl
+from sinabs.utils import expand_to_pair
 
 from .dynapcnn_layer import DynapcnnLayer
-from .utils import expand_to_pair
 
 
 def construct_dynapcnnlayers_from_mapper(

sinabs/backend/dynapcnn/dynapcnn_network.py

@@ -570,7 +570,7 @@ def _make_config(
         # Collect cores (chip layers) that are to be monitored
         monitor_chip_layers = []
         for lyr_idx in monitor_layers:
-            if lyr_idx.lower() == "dvs":
+            if str(lyr_idx).lower() == "dvs":
                 monitor_chip_layers.append("dvs")
             else:
                 # Warn when recording layers with pooling

sinabs/backend/dynapcnn/utils.py

@@ -13,9 +13,10 @@
 import torch.nn as nn
 
 import sinabs.layers as sl
+from sinabs.utils import expand_to_pair
 
 from .crop2d import Crop2d
-from .dvs_layer import DVSLayer, expand_to_pair
+from .dvs_layer import DVSLayer
 from .flipdims import FlipDims
 
 if TYPE_CHECKING:

sinabs/utils.py

@@ -1,4 +1,4 @@
-from typing import List
+from typing import Iterable, List, Tuple, TypeVar, Union
 
 import numpy as np
 import torch
@@ -268,3 +268,45 @@ def get_smallest_compatible_time_dimension(model: nn.Module) -> int:
     num_timesteps = abs(get_num_timesteps(model))
     # Use `abs` to turn `-1` to `1`
     return abs(batch_size * num_timesteps)
+
+
+def expand_to_pair(value) -> Tuple[int, int]:
+    """Expand a given value to a pair (tuple) if an int is passed.
+
+    Parameters
+    ----------
+    value:
+        int
+
+    Returns
+    -------
+    pair:
+        (int, int)
+    """
+    return (value, value) if isinstance(value, int) else value
+
+
+T = TypeVar("T")
+def collapse_pair(pair: Union[Iterable[T], T]) -> T:
+    """ Collapse an iterable of equal elements by returning only the first
+
+    Parameters
+    ----------
+    pair: Iterable. All elements should be the same.
+
+    Returns
+    -------
+    First item of `pair`. If `pair` is not iterable it will return `pair` itself.
+
+    Raises
+    ------
+    ValueError if not all elements in `pair` are equal. 
+    """
+    if isinstance(pair, Iterable):
+        items = [x for x in pair]
+        if any(x != items[0] for x in items):
+            raise ValueError("All elements of `pair` must be the same")
+        return items[0]
+    else:
+        return pair
+