[linting] Pre-commit with python 3.10 on all files

.pre-commit-config.yaml

@@ -60,11 +60,12 @@ repos:
   hooks:
   - id: black
     language_version: python3
+    args: [--line-length=100]
 
 - repo: https://github.com/PyCQA/flake8
   rev: 6.0.0
   hooks:
   - id: flake8
     # black-compatible flake-8 config
-    args: ['--max-line-length=88',  # black default
+    args: ['--max-line-length=100',  # black default
            '--extend-ignore=E203']  # E203 is not PEP8 compliant

notebooks/end2end_example/cybersecurity/dataloader_quantized.py

@@ -48,7 +48,6 @@ def __init__(
         onehot=False,
         train=True,
     ):
-
         self.dataframe = (
             pd.concat([pd.read_csv(file_path_train), pd.read_csv(file_path_test)])
             .reset_index()
@@ -77,9 +76,7 @@ def __getitem__(self, index):
         data_val = self.data[index][:-1]
         return data_val, target
 
-    def dec2bin(
-        self, column: pd.Series, number_of_bits: int, left_msb: bool = True
-    ) -> pd.Series:
+    def dec2bin(self, column: pd.Series, number_of_bits: int, left_msb: bool = True) -> pd.Series:
         """Convert a decimal pd.Series to binary pd.Series with numbers in their
         # base-2 equivalents.
         The output is a numpy nd array.
@@ -133,6 +130,7 @@ def integer_encoding(self, df):
     def quantize_df(self, df):
         """Quantized the input dataframe. The scaling is done by multiplying
         every column by the inverse of the minimum of that column"""
+
         # gets the smallest positive number of a vector
         def get_min_positive_number(vector):
             return vector[vector > 0].min()
@@ -178,24 +176,18 @@ def char_split(s):
             column_data = np.clip(
                 column_data, 0, 4294967295
             )  # clip due to overflow of uint32 of matlab code
-            column_data = self.round_like_matlab_series(
-                column_data
-            )  # round like matlab
+            column_data = self.round_like_matlab_series(column_data)  # round like matlab
             column_data = column_data.astype(np.uint32)  # cast like matlab
 
             if column == "rate":
                 column_data.update(pd.Series(dict_correct_rate_values))
 
             python_quantized_df[column] = (
-                self.dec2bin(column_data, maxbits, left_msb=False)
-                .reshape((-1, 1))
-                .flatten()
+                self.dec2bin(column_data, maxbits, left_msb=False).reshape((-1, 1)).flatten()
             )
 
         for column in python_quantized_df.columns:
-            python_quantized_df[column] = (
-                python_quantized_df[column].apply(char_split).values
-            )
+            python_quantized_df[column] = python_quantized_df[column].apply(char_split).values
 
         python_quantized_df_separated = pd.DataFrame(
             np.column_stack(python_quantized_df.values.T.tolist())

notebooks/end2end_example/cybersecurity/validate-unsw-nb15.py

@@ -57,9 +57,7 @@ def make_unsw_nb15_test_batches(bsize, dataset_root):
         help='name of bitfile (i.e. "resizer.bit")',
         default="../bitfile/finn-accel.bit",
     )
-    parser.add_argument(
-        "--dataset_root", help="dataset root dir for download/reuse", default="."
-    )
+    parser.add_argument("--dataset_root", help="dataset root dir for download/reuse", default=".")
     # parse arguments
     args = parser.parse_args()
     bsize = args.batchsize

src/finn/analysis/fpgadataflow/dataflow_performance.py

@@ -66,9 +66,7 @@ def dataflow_performance(model):
                     max_pred_latency = 0
                 else:
                     # find max of any of predecessors
-                    pred_latencies = map(
-                        lambda x: latency_at_node_output[x.name], predecessors
-                    )
+                    pred_latencies = map(lambda x: latency_at_node_output[x.name], predecessors)
                     max_pred_latency = max(pred_latencies)
                 latency_at_node_output[node.name] = node_cycles + max_pred_latency
     critical_path_cycles = max(latency_at_node_output.values())

src/finn/analysis/fpgadataflow/post_synth_res.py

@@ -86,7 +86,7 @@ def get_instance_stats(inst_name):
         if row != []:
             node_dict = {}
             row = list(row[0])
-            for (restype, ind) in restype_to_ind.items():
+            for restype, ind in restype_to_ind.items():
                 node_dict[restype] = int(row[ind].attrib["contents"])
             return node_dict
         else:

src/finn/analysis/fpgadataflow/res_estimation.py

@@ -62,10 +62,7 @@ def res_estimation_complete(model):
         if is_fpgadataflow_node(node) is True:
             op_type = node.op_type
             inst = registry.getCustomOp(node)
-            if (
-                op_type == "MatrixVectorActivation"
-                or op_type == "VectorVectorActivation"
-            ):
+            if op_type == "MatrixVectorActivation" or op_type == "VectorVectorActivation":
                 orig_restype = inst.get_nodeattr("resType")
                 res_dict[node.name] = []
                 inst.set_nodeattr("resType", "dsp")

src/finn/builder/build_dataflow.py

@@ -91,12 +91,8 @@ def resolve_build_steps(cfg: DataflowBuildConfig, partial: bool = True):
     return steps_as_fxns
 
 
-def resolve_step_filename(
-    step_name: str, cfg: DataflowBuildConfig, step_delta: int = 0
-):
-    step_names = list(
-        map(lambda x: x.__name__, resolve_build_steps(cfg, partial=False))
-    )
+def resolve_step_filename(step_name: str, cfg: DataflowBuildConfig, step_delta: int = 0):
+    step_names = list(map(lambda x: x.__name__, resolve_build_steps(cfg, partial=False)))
     assert step_name in step_names, "start_step %s not found" + step_name
     step_no = step_names.index(step_name) + step_delta
     assert step_no >= 0, "Invalid step+delta combination"
@@ -150,19 +146,13 @@ def build_dataflow_cfg(model_filename, cfg: DataflowBuildConfig):
     for transform_step in build_dataflow_steps:
         try:
             step_name = transform_step.__name__
-            print(
-                "Running step: %s [%d/%d]"
-                % (step_name, step_num, len(build_dataflow_steps))
-            )
+            print("Running step: %s [%d/%d]" % (step_name, step_num, len(build_dataflow_steps)))
             # redirect output to logfile
             if not cfg.verbose:
                 sys.stdout = stdout_logger
                 sys.stderr = stderr_logger
                 # also log current step name to logfile
-                print(
-                    "Running step: %s [%d/%d]"
-                    % (step_name, step_num, len(build_dataflow_steps))
-                )
+                print("Running step: %s [%d/%d]" % (step_name, step_num, len(build_dataflow_steps)))
             # run the step
             step_start = time.time()
             model = transform_step(model, cfg)

src/finn/builder/build_dataflow_config.py

@@ -267,9 +267,7 @@ class DataflowBuildConfig:
 
     #: When `auto_fifo_depths = True`, select which method will be used for
     #: setting the FIFO sizes.
-    auto_fifo_strategy: Optional[
-        AutoFIFOSizingMethod
-    ] = AutoFIFOSizingMethod.LARGEFIFO_RTLSIM
+    auto_fifo_strategy: Optional[AutoFIFOSizingMethod] = AutoFIFOSizingMethod.LARGEFIFO_RTLSIM
 
     #: Avoid using C++ rtlsim for auto FIFO sizing and rtlsim throughput test
     #: if set to True, always using Python instead
@@ -366,9 +364,7 @@ def _resolve_driver_platform(self):
         elif self.shell_flow_type == ShellFlowType.VITIS_ALVEO:
             return "alveo"
         else:
-            raise Exception(
-                "Couldn't resolve driver platform for " + str(self.shell_flow_type)
-            )
+            raise Exception("Couldn't resolve driver platform for " + str(self.shell_flow_type))
 
     def _resolve_fpga_part(self):
         if self.fpga_part is None:
@@ -410,8 +406,7 @@ def _resolve_vitis_platform(self):
             return alveo_default_platform[self.board]
         else:
             raise Exception(
-                "Could not resolve Vitis platform:"
-                " need either board or vitis_platform specified"
+                "Could not resolve Vitis platform:" " need either board or vitis_platform specified"
             )
 
     def _resolve_verification_steps(self):
@@ -429,8 +424,7 @@ def _resolve_verification_io_pair(self):
             )
             verify_input_npy = np.load(self.verify_input_npy)
             assert os.path.isfile(self.verify_expected_output_npy), (
-                "verify_expected_output_npy not found: "
-                + self.verify_expected_output_npy
+                "verify_expected_output_npy not found: " + self.verify_expected_output_npy
             )
             verify_expected_output_npy = np.load(self.verify_expected_output_npy)
             return (

src/finn/builder/build_dataflow_steps.py

@@ -145,9 +145,7 @@ def verify_step(
         in_npy = np.expand_dims(in_npy_all[b], axis=0)
         exp_out_npy = np.expand_dims(exp_out_npy_all[b], axis=0)
         if need_parent:
-            assert (
-                cfg.save_intermediate_models
-            ), "Enable save_intermediate_models for verification"
+            assert cfg.save_intermediate_models, "Enable save_intermediate_models for verification"
             parent_model_fn = intermediate_models_dir + "/dataflow_parent.onnx"
             child_model_fn = intermediate_models_dir + "/verify_%s.onnx" % step_name
             model.save(child_model_fn)
@@ -161,9 +159,7 @@ def verify_step(
                 )
                 print("Attempting to force model shape on verification input")
                 in_npy = in_npy.reshape(exp_ishape)
-            out_dict = execute_parent(
-                parent_model_fn, child_model_fn, in_npy, return_full_ctx=True
-            )
+            out_dict = execute_parent(parent_model_fn, child_model_fn, in_npy, return_full_ctx=True)
             out_npy = out_dict[out_tensor_name]
         else:
             inp_tensor_name = model.graph.input[0].name
@@ -230,9 +226,7 @@ def prepare_for_stitched_ip_rtlsim(verify_model, cfg):
                 inst.set_nodeattr("ipgen_path", "")
                 need_restitch = True
         # StreamingDataWidthConverter must have impl_style=hls
-        for dwc_layer in verify_model.get_nodes_by_op_type(
-            "StreamingDataWidthConverter_Batch"
-        ):
+        for dwc_layer in verify_model.get_nodes_by_op_type("StreamingDataWidthConverter_Batch"):
             inst = getCustomOp(dwc_layer)
             if inst.get_nodeattr("impl_style") != "hls":
                 inst.set_nodeattr("impl_style", "hls")
@@ -382,8 +376,7 @@ def step_create_dataflow_partition(model: ModelWrapper, cfg: DataflowBuildConfig
 
     parent_model = model.transform(
         CreateDataflowPartition(
-            partition_model_dir=cfg.output_dir
-            + "/intermediate_models/supported_op_partitions"
+            partition_model_dir=cfg.output_dir + "/intermediate_models/supported_op_partitions"
         )
     )
     sdp_nodes = parent_model.get_nodes_by_op_type("StreamingDataflowPartition")
@@ -422,9 +415,7 @@ def step_target_fps_parallelization(model: ModelWrapper, cfg: DataflowBuildConfi
             "mem_mode",
             "runtime_writeable_weights",
         ]
-        extract_model_config_to_json(
-            model, cfg.output_dir + "/auto_folding_config.json", hw_attrs
-        )
+        extract_model_config_to_json(model, cfg.output_dir + "/auto_folding_config.json", hw_attrs)
 
     return model
 
@@ -459,9 +450,7 @@ def step_generate_estimate_reports(model: ModelWrapper, cfg: DataflowBuildConfig
         with open(report_dir + "/estimate_layer_cycles.json", "w") as f:
             json.dump(estimate_layer_cycles, f, indent=2)
         estimate_layer_resources = model.analysis(res_estimation)
-        estimate_layer_resources["total"] = aggregate_dict_keys(
-            estimate_layer_resources
-        )
+        estimate_layer_resources["total"] = aggregate_dict_keys(estimate_layer_resources)
         with open(report_dir + "/estimate_layer_resources.json", "w") as f:
             json.dump(estimate_layer_resources, f, indent=2)
         estimate_layer_resources_complete = model.analysis(res_estimation_complete)
@@ -475,8 +464,7 @@ def step_generate_estimate_reports(model: ModelWrapper, cfg: DataflowBuildConfig
         est_fps = n_clock_cycles_per_sec / estimate_network_performance["max_cycles"]
         estimate_network_performance["estimated_throughput_fps"] = est_fps
         est_latency_ns = (
-            estimate_network_performance["critical_path_cycles"]
-            * cfg.synth_clk_period_ns
+            estimate_network_performance["critical_path_cycles"] * cfg.synth_clk_period_ns
         )
         estimate_network_performance["estimated_latency_ns"] = est_latency_ns
         with open(report_dir + "/estimate_network_performance.json", "w") as f:
@@ -497,9 +485,7 @@ def step_minimize_bit_width(model: ModelWrapper, cfg: DataflowBuildConfig):
 def step_hls_codegen(model: ModelWrapper, cfg: DataflowBuildConfig):
     "Generate Vivado HLS code to prepare HLSCustomOp nodes for IP generation."
 
-    model = model.transform(
-        PrepareIP(cfg._resolve_fpga_part(), cfg._resolve_hls_clk_period())
-    )
+    model = model.transform(PrepareIP(cfg._resolve_fpga_part(), cfg._resolve_hls_clk_period()))
     return model
 
 
@@ -599,9 +585,7 @@ def step_set_fifo_depths(model: ModelWrapper, cfg: DataflowBuildConfig):
         "inFIFODepths",
         "outFIFODepths",
     ]
-    extract_model_config_to_json(
-        model, cfg.output_dir + "/final_hw_config.json", hw_attrs
-    )
+    extract_model_config_to_json(model, cfg.output_dir + "/final_hw_config.json", hw_attrs)
 
     # perform FIFO splitting and shallow FIFO removal only after the final config
     # json file has been written. otherwise, since these transforms may add/remove
@@ -612,9 +596,7 @@ def step_set_fifo_depths(model: ModelWrapper, cfg: DataflowBuildConfig):
 
     # after FIFOs are ready to go, call PrepareIP and HLSSynthIP again
     # this will only run for the new nodes (e.g. FIFOs and DWCs)
-    model = model.transform(
-        PrepareIP(cfg._resolve_fpga_part(), cfg._resolve_hls_clk_period())
-    )
+    model = model.transform(PrepareIP(cfg._resolve_fpga_part(), cfg._resolve_hls_clk_period()))
     model = model.transform(HLSSynthIP())
     return model
 
@@ -651,9 +633,7 @@ def step_create_stitched_ip(model: ModelWrapper, cfg: DataflowBuildConfig):
         if cfg.verify_save_rtlsim_waveforms:
             report_dir = cfg.output_dir + "/report"
             os.makedirs(report_dir, exist_ok=True)
-            verify_model.set_metadata_prop(
-                "rtlsim_trace", "%s/verify_rtlsim.vcd" % (report_dir)
-            )
+            verify_model.set_metadata_prop("rtlsim_trace", "%s/verify_rtlsim.vcd" % (report_dir))
         verify_step(verify_model, cfg, "stitched_ip_rtlsim", need_parent=True)
         os.environ["LIVENESS_THRESHOLD"] = str(prev_liveness)
     return model
@@ -674,9 +654,7 @@ def step_measure_rtlsim_performance(model: ModelWrapper, cfg: DataflowBuildConfi
         rtlsim_model = deepcopy(model)
         rtlsim_model = prepare_for_stitched_ip_rtlsim(rtlsim_model, cfg)
         # multi-in/out streams currently not supported in our C++ verilator driver
-        model_multi_io = (
-            len(rtlsim_model.graph.input) > 1 or len(rtlsim_model.graph.output) > 1
-        )
+        model_multi_io = len(rtlsim_model.graph.input) > 1 or len(rtlsim_model.graph.output) > 1
         force_python_rtlsim = cfg.force_python_rtlsim or model_multi_io
         if model_multi_io:
             warnings.warn(
@@ -694,9 +672,7 @@ def step_measure_rtlsim_performance(model: ModelWrapper, cfg: DataflowBuildConfi
                     "rtlsim_trace",
                     "%s/rtlsim_perf_batch_%d.vcd" % (report_dir, rtlsim_bs),
                 )
-            rtlsim_model.set_metadata_prop(
-                "extra_verilator_args", str(["-CFLAGS", "-O3"])
-            )
+            rtlsim_model.set_metadata_prop("extra_verilator_args", str(["-CFLAGS", "-O3"]))
             # run with single input to get latency
             rtlsim_latency_dict = throughput_test_rtlsim(rtlsim_model, 1)
             # run with batch to get stable-state throughput
@@ -712,7 +688,7 @@ def step_measure_rtlsim_performance(model: ModelWrapper, cfg: DataflowBuildConfi
             rtlsim_perf_dict["runtime[ms]"] = runtime_s * 1000
             rtlsim_perf_dict["throughput[images/s]"] = rtlsim_bs / runtime_s
             rtlsim_perf_dict["fclk[mhz]"] = fclk_mhz
-            for (key, val) in rtlsim_perf_dict.items():
+            for key, val in rtlsim_perf_dict.items():
                 if "max_count" in key:
                     del rtlsim_perf_dict[key]
         # estimate stable-state throughput based on latency+throughput
@@ -754,13 +730,9 @@ def step_out_of_context_synthesis(model: ModelWrapper, cfg: DataflowBuildConfig)
     """Run out-of-context synthesis and generate reports.
     Depends on the DataflowOutputType.STITCHED_IP output product."""
     if DataflowOutputType.OOC_SYNTH in cfg.generate_outputs:
-        assert (
-            DataflowOutputType.STITCHED_IP in cfg.generate_outputs
-        ), "OOC needs stitched IP"
+        assert DataflowOutputType.STITCHED_IP in cfg.generate_outputs, "OOC needs stitched IP"
         model = model.transform(
-            SynthOutOfContext(
-                part=cfg._resolve_fpga_part(), clk_period_ns=cfg.synth_clk_period_ns
-            )
+            SynthOutOfContext(part=cfg._resolve_fpga_part(), clk_period_ns=cfg.synth_clk_period_ns)
         )
         report_dir = cfg.output_dir + "/report"
         os.makedirs(report_dir, exist_ok=True)

src/finn/core/onnx_exec.py

@@ -34,9 +34,7 @@
 from finn.core.rtlsim_exec import rtlsim_exec
 
 
-def execute_onnx(
-    model, input_dict, return_full_exec_context=False, start_node=None, end_node=None
-):
+def execute_onnx(model, input_dict, return_full_exec_context=False, start_node=None, end_node=None):
     """Executes given ONNX ModelWrapper with given named inputs.
     If return_full_exec_context is False, a dict of named outputs is returned
     as indicated by the model.graph.output.
@@ -53,9 +51,7 @@ def execute_onnx(
     # if set to "rtlsim" execute model using pyverilator
     model_exec_mode = model.get_metadata_prop("exec_mode")
     if (model_exec_mode is None) or (model_exec_mode == ""):
-        return execute_onnx_base(
-            model, input_dict, return_full_exec_context, start_node, end_node
-        )
+        return execute_onnx_base(model, input_dict, return_full_exec_context, start_node, end_node)
 
     if not model.check_all_tensor_shapes_specified():
         raise Exception("Found unspecified tensor shapes, try infer_shapes")