Add fixed-point thresholding support

src/finn/custom_op/fpgadataflow/rtl/thresholding_rtl.py

@@ -173,7 +173,22 @@ def prepare_codegen_rtl_values(self, model):
                 "using RoundAndClipThresholds transform before code generation."
             )
         if not idt.is_integer() and wdt.is_integer():
-            raise ValueError("Floating-point inputs and integer thresholds are not supported.")
+            raise ValueError("Non-integer inputs and integer thresholds are not supported.")
+        if idt.is_fixed_point() and not wdt.is_fixed_point():
+            raise ValueError("Fixed-point inputs and floating-point thresholds are not supported.")
+        if wdt.is_fixed_point() and not idt.is_fixed_point():
+            raise ValueError("Floating-point inputs and fixed-point thresholds are not supported.")
+        if wdt.is_fixed_point() and idt.is_fixed_point():
+            if wdt.scale_factor() < idt.scale_factor():
+                raise ValueError(
+                    "Fixed-point thresholds have more fractional bits than input. "
+                    "Run RoundAndClipThresholds to reduce threshold fractional bits."
+                )
+            elif wdt.scale_factor() > idt.scale_factor():
+                raise ValueError(
+                    "Fixed-point inputs and with more fractional bits "
+                    "than thresholds are not supported."
+                )
 
         # If a single threshold value is found, set num_channels to PE
         thresholds = model.get_initializer(self.onnx_node.input[1])

src/finn/transformation/fpgadataflow/convert_to_hw_layers.py

@@ -177,14 +177,16 @@ def apply(self, model):
                 idt = model.get_tensor_datatype(thl_input)
                 tdt = model.get_tensor_datatype(thl_threshold)
 
-                # only infer layers where input and thresholds are integers or fp32
+                # only infer layers where input and thresholds are integers, floats, or fixed-point
                 idt_int = idt.is_integer()
                 tdt_int = tdt.is_integer()
                 idt_fp = idt in ["FLOAT32", "FLOAT16"]
                 tdt_fp = tdt in ["FLOAT32", "FLOAT16"]
-                if not (idt_int or idt_fp):
+                idt_fxp = idt.is_fixed_point()
+                tdt_fxp = tdt.is_fixed_point()
+                if not (idt_int or idt_fp or idt_fxp):
                     continue
-                if not (tdt_int or tdt_fp):
+                if not (tdt_int or tdt_fp or tdt_fxp):
                     continue
 
                 # check layout of inputs/outputs, and convert if needed

src/finn/transformation/streamline/round_thresholds.py

@@ -55,27 +55,45 @@ def apply(self, model: ModelWrapper):  # noqa
                 dtype = model.get_tensor_datatype(node.input[0])
                 # This transformation only applies to thresholding operations
                 # operating on integer inputs
-                if not dtype.is_integer():
+                if not (dtype.is_integer() or dtype.is_fixed_point()):
                     continue
-                # Round thresholds up to nearest integer and clip thresholds
-                # outside the input range
-                #   Note: This might promote the thresholds to float64 and
-                #   introduce extra inaccuracies due to large integers not being
-                #   exactly representable in floating-point representation.
-                #   See for example: np.ceil(np.float32(16777217)) == 16777216
-                new_thresholds = np.clip(np.ceil(thresholds), dtype.min(), dtype.max() + 1)
-                # Convert back to the preferred float32 container type
-                new_thresholds = new_thresholds.astype(np.float32)
-                # Insert the rounded and clipped thresholds back into the model
-                model.set_initializer(node.input[1], new_thresholds)
-                # The rounded and clipped thresholds now fit into a data type
-                # that is one bit bigger than the input datatype
-                # Determine new max_value
-                max_val = dtype.max() + 1
-                if not dtype.signed():
-                    tdt = DataType.get_smallest_possible(max_val)
-                else:
-                    tdt = DataType.get_smallest_possible(-(max_val) - 1)
+                if dtype.is_integer():
+                    # Round thresholds up to nearest integer and clip thresholds
+                    # outside the input range
+                    #   Note: This might promote the thresholds to float64 and
+                    #   introduce extra inaccuracies due to large integers not being
+                    #   exactly representable in floating-point representation.
+                    #   See for example: np.ceil(np.float32(16777217)) == 16777216
+                    new_thresholds = np.clip(np.ceil(thresholds), dtype.min(), dtype.max() + 1)
+                    # Convert back to the preferred float32 container type
+                    new_thresholds = new_thresholds.astype(np.float32)
+                    # Insert the rounded and clipped thresholds back into the model
+                    model.set_initializer(node.input[1], new_thresholds)
+                    # The rounded and clipped thresholds now fit into a data type
+                    # that is one bit bigger than the input datatype
+                    # Determine new max_value
+                    max_val = dtype.max() + 1
+                    if not dtype.signed():
+                        tdt = DataType.get_smallest_possible(max_val)
+                    else:
+                        tdt = DataType.get_smallest_possible(-(max_val) - 1)
+                elif dtype.is_fixed_point():
+                    # Round thresholds up to nearest representable value
+                    # of the input datatype
+                    new_thresholds = np.clip(
+                        np.ceil(thresholds / dtype.scale_factor()) * dtype.scale_factor(),
+                        dtype.min(),
+                        dtype.max() + dtype.scale_factor(),
+                    )
+                    new_thresholds = new_thresholds.astype(np.float32)
+                    model.set_initializer(node.input[1], new_thresholds)
+                    # find smallest underlying integer representation for the thresholds
+                    max_val = dtype.max() / dtype.scale_factor() + 1
+                    tdt_int = DataType.get_smallest_possible(-max_val - 1)
+                    tdt = DataType[
+                        f"FIXED<{tdt_int.bitwidth()},{tdt_int.bitwidth() - dtype.frac_bits()}>"
+                    ]
+
                 model.set_tensor_datatype(node.input[1], tdt)
                 # If hw op we need to set the weight data type attribute as well
                 if op_type.startswith("Thresholding"):

tests/fpgadataflow/test_fpgadataflow_thresholding.py

@@ -70,6 +70,15 @@ def generate_random_threshold_values(
             data_type.max() + 1,
             (num_input_channels, num_steps),
         ).astype(np.float32)
+    elif data_type.is_fixed_point():
+        return (
+            np.random.randint(
+                data_type.min() / data_type.scale_factor(),
+                data_type.max() / data_type.scale_factor() + 1,
+                (num_input_channels, num_steps),
+            ).astype(np.float32)
+            * data_type.scale_factor()
+        )
     else:
         return (np.random.randn(num_input_channels, num_steps) * 1000).astype(
             data_type.to_numpy_dt()
@@ -154,6 +163,7 @@ def make_single_multithresholding_modelwrapper(
         (DataType["UINT5"], DataType["UINT8"]),
         (DataType["FLOAT32"], DataType["FLOAT32"]),
         (DataType["FLOAT16"], DataType["FLOAT16"]),
+        (DataType["FIXED<6,2>"], DataType["FIXED<8,4>"]),
     ],
 )
 @pytest.mark.parametrize("fold", [-1, 1, 2])
@@ -196,6 +206,12 @@ def test_fpgadataflow_thresholding(
             "Thresholds will not be rounded when inputs are floating-point. "
             "Test case is identical with floating-point input and round_thresh=False."
         )
+    if (
+        impl_style == "rtl"
+        and input_data_type.is_fixed_point()
+        and not threshold_data_type.is_fixed_point()
+    ):
+        pytest.skip("Fixed-point inputs and non-fixed-point thresholds are not supported in RTL.")
 
     if fold == -1:
         fold = num_input_channels

tests/transformation/streamline/test_round_thresholds.py

@@ -268,3 +268,216 @@ def test_round_and_clip_thresholds_floats(i_dtype, o_dtype, n_elems):
     out_produced = oxe.execute_onnx(model, {"inp": inp})["out"]
 
     assert np.allclose(out_produced, out_expected, atol=1.0e-3)
+
+
+# Tests the RoundAndClipThresholds transformation under various input, output
+# data type combinations with fixed point inputs. Without proper rounding,
+# this tests only the clipping, range and type-casting behavior of the
+# transformation.
+@pytest.mark.parametrize(
+    "i_dtype",
+    [
+        "FIXED<13,3>",
+        "FIXED<14,3>",
+        "FIXED<15,3>",
+        "FIXED<16,3>",
+    ],
+)
+@pytest.mark.parametrize(
+    "o_dtype",
+    [
+        # Explanation for selecting these test configurations:
+        # 1. Outputs of MultiThreshold are typically much smaller bit-width than the
+        #    inputs and thresholds.
+        # 2. However, with randomly samples thresholds from a rather large range due
+        #    to the selected input bit-widths (see above), we risk not adequately
+        #    covering the input range if we sample too few thresholds. The number of
+        #    thresholds sampled depends on the bit-width of the output, thus we use
+        #    rather high bit-width for testing.
+        # 3. For a "real" model, the quantization procedure *should* take care of
+        #    adequately covering the true input range.
+        "INT8",
+        "UINT8",
+    ],
+)
+@pytest.mark.parametrize(
+    "n_elems",
+    [
+        # Explanation for selecting these test configurations:
+        # 1. Small edge cases and quickly running through tests: 1, 2, 3, 4
+        # 2. Large test case 256, hopefully amplifying any rarely occurring errors
+        1,
+        2,
+        3,
+        4,
+        256,
+    ],
+)
+@pytest.mark.streamline
+def test_round_and_clip_thresholds_fxp(i_dtype, o_dtype, n_elems):
+    i_dtype = DataType[i_dtype]
+    t_dtype = DataType["FIXED<15,3>"]
+    o_dtype = DataType[o_dtype]  # noqa: Duplicate model setup code
+    node = helper.make_node(
+        "MultiThreshold",
+        domain="qonnx.custom_op.general",
+        inputs=["inp", "thresholds"],
+        outputs=["out"],
+        out_dtype=str(o_dtype),
+    )
+    n_thresholds = o_dtype.get_num_possible_values() - 1
+    inp = helper.make_tensor_value_info("inp", TensorProto.FLOAT, [1, n_elems])
+    out = helper.make_tensor_value_info("out", TensorProto.FLOAT, [1, n_elems])
+    thresholds = helper.make_tensor_value_info(
+        "thresholds", TensorProto.FLOAT, [n_elems, n_thresholds]
+    )
+    graph = helper.make_graph([node], "thresholds", [inp, thresholds], [out])
+    model = ModelWrapper(helper.make_model(graph))
+
+    inp = gen_finn_dt_tensor(i_dtype, [1, n_elems])
+    # Draw uniformly random prototype thresholds in [0,+1] range
+    thresholds = np.random.rand(n_elems, n_thresholds)
+    # Type alias to 25-bit signed integer type used to set the range of the
+    # thresholds
+    FXP25 = DataType["FIXED<15,3>"]  # noqa: Variable name not lowercase
+    # Map the prototype thresholds into the test integer range and sort
+    thresholds = np.sort((FXP25.max() - FXP25.min()) * thresholds + FXP25.min())
+    # Set data type annotations for the input and thresholds tensor
+    model.set_tensor_datatype("inp", i_dtype)  # noqa: Duplicate model execution
+    model.set_tensor_datatype("thresholds", t_dtype)
+    model.set_tensor_datatype("out", o_dtype)
+    model.set_initializer("thresholds", thresholds)
+
+    # Execute the model before running the RoundAndClipThresholds transformation
+    out_expected = oxe.execute_onnx(model, {"inp": inp})["out"]
+    # Before rounding the threshold data type must be as annotated
+    assert model.get_tensor_datatype("thresholds") == t_dtype
+
+    model = model.transform(RoundAndClipThresholds())
+
+    max_val = i_dtype.max() / i_dtype.scale_factor() + 1
+    new_tdt_int = DataType.get_smallest_possible(-max_val - 1)
+    new_tdt = DataType[
+        f"FIXED<{new_tdt_int.bitwidth()},{new_tdt_int.bitwidth() - i_dtype.frac_bits()}>"
+    ]
+    assert model.get_tensor_datatype("thresholds") == new_tdt
+    assert model.get_tensor_datatype("out") == o_dtype
+
+    # After this transformation, the container type used to store the thresholds
+    # values must be float32. No other type-cast or type promotion may happen.
+    assert model.get_initializer("thresholds").dtype == np.float32
+    # After rounding, all thresholds must be integers represented as float32
+    assert all(
+        x.is_integer()
+        for x in model.get_initializer("thresholds").flatten() / new_tdt.scale_factor()
+    )
+
+    out_produced = oxe.execute_onnx(model, {"inp": inp})["out"]
+
+    assert np.allclose(out_produced, out_expected, atol=1.0e-3)
+
+
+# Tests the RoundAndClipThresholds transformation under various input, output
+# data type combinations with fixed point inputs. This test case tests actual
+# rounding of floating-point thresholds.
+@pytest.mark.parametrize(
+    "i_dtype",
+    [
+        "FIXED<13,3>",
+        "FIXED<14,3>",
+        "FIXED<15,3>",
+        "FIXED<16,3>",
+    ],
+)
+@pytest.mark.parametrize(
+    "o_dtype",
+    [
+        # Explanation for selecting these test configurations:
+        # 1. Outputs of MultiThreshold are typically much smaller bit-width than the
+        #    inputs and thresholds.
+        # 2. However, with randomly samples thresholds from a rather large range due
+        #    to the selected input bit-widths (see above), we risk not adequately
+        #    covering the input range if we sample too few thresholds. The number of
+        #    thresholds sampled depends on the bit-width of the output, thus we use
+        #    rather high bit-width for testing.
+        # 3. For a "real" model, the quantization procedure *should* take care of
+        #    adequately covering the true input range.
+        "INT8",
+        "UINT8",
+    ],
+)
+@pytest.mark.parametrize(
+    "n_elems",
+    [
+        # Explanation for selecting these test configurations:
+        # 1. Small edge cases and quickly running through tests: 1, 2, 3, 4
+        # 2. Large test case 256, hopefully amplifying any rarely occurring errors
+        1,
+        2,
+        3,
+        4,
+        256,
+    ],
+)
+@pytest.mark.streamline
+def test_round_and_clip_thresholds_fxp_float(i_dtype, o_dtype, n_elems):
+    i_dtype = DataType[i_dtype]
+    t_dtype = DataType["FLOAT32"]
+    o_dtype = DataType[o_dtype]  # noqa: Duplicate model setup code
+    node = helper.make_node(
+        "MultiThreshold",
+        domain="qonnx.custom_op.general",
+        inputs=["inp", "thresholds"],
+        outputs=["out"],
+        out_dtype=str(o_dtype),
+    )
+    n_thresholds = o_dtype.get_num_possible_values() - 1
+    inp = helper.make_tensor_value_info("inp", TensorProto.FLOAT, [1, n_elems])
+    out = helper.make_tensor_value_info("out", TensorProto.FLOAT, [1, n_elems])
+    thresholds = helper.make_tensor_value_info(
+        "thresholds", TensorProto.FLOAT, [n_elems, n_thresholds]
+    )
+    graph = helper.make_graph([node], "thresholds", [inp, thresholds], [out])
+    model = ModelWrapper(helper.make_model(graph))
+
+    inp = gen_finn_dt_tensor(i_dtype, [1, n_elems])
+    # Draw uniformly random prototype thresholds in [0,+1] range
+    thresholds = np.random.rand(n_elems, n_thresholds)
+    # Type alias to 25-bit signed integer type used to set the range of the
+    # thresholds
+    FXP25 = DataType["FIXED<15,3>"]  # noqa: Variable name not lowercase
+    # Map the prototype thresholds into the test integer range and sort
+    thresholds = np.sort((FXP25.max() - FXP25.min()) * thresholds + FXP25.min())
+    # Set data type annotations for the input and thresholds tensor
+    model.set_tensor_datatype("inp", i_dtype)  # noqa: Duplicate model execution
+    model.set_tensor_datatype("thresholds", t_dtype)
+    model.set_tensor_datatype("out", o_dtype)
+    model.set_initializer("thresholds", thresholds)
+
+    # Execute the model before running the RoundAndClipThresholds transformation
+    out_expected = oxe.execute_onnx(model, {"inp": inp})["out"]
+    # Before rounding the threshold data type must be as annotated
+    assert model.get_tensor_datatype("thresholds") == t_dtype
+
+    model = model.transform(RoundAndClipThresholds())
+
+    max_val = i_dtype.max() / i_dtype.scale_factor() + 1
+    new_tdt_int = DataType.get_smallest_possible(-max_val - 1)
+    new_tdt = DataType[
+        f"FIXED<{new_tdt_int.bitwidth()},{new_tdt_int.bitwidth() - i_dtype.frac_bits()}>"
+    ]
+    assert model.get_tensor_datatype("thresholds") == new_tdt
+    assert model.get_tensor_datatype("out") == o_dtype
+
+    # After this transformation, the container type used to store the thresholds
+    # values must be float32. No other type-cast or type promotion may happen.
+    assert model.get_initializer("thresholds").dtype == np.float32
+    # After rounding, all thresholds must be integers represented as float32
+    assert all(
+        x.is_integer()
+        for x in model.get_initializer("thresholds").flatten() / new_tdt.scale_factor()
+    )
+
+    out_produced = oxe.execute_onnx(model, {"inp": inp})["out"]
+
+    assert np.allclose(out_produced, out_expected, atol=1.0e-3)