OpenFPGA/openfpga_flow/scripts/check_qor.py

#####################################################################
# Python script to check if heterogeneous blocks, e.g., RAM and multipliers
# have been inferred during openfpga flow
# # This script will
#   - Check the .csv file generated by openfpga task-run to find out
#     the number of each type of heterogeneous blocks
#####################################################################

import os
from os.path import dirname, abspath, isfile
import shutil
import re
import argparse
import logging
import csv

#####################################################################
# Contants
#####################################################################
csv_name_tag = "name"
csv_metric_tag = "metric"

#####################################################################
# Initialize logger
#####################################################################
logging.basicConfig(format='%(levelname)s: %(message)s', level=logging.DEBUG)

#####################################################################
# Parse the options
# - [mandatory option] the file path to .csv file
#####################################################################
parser = argparse.ArgumentParser(
    description='A checker for hetergeneous block mapping in OpenFPGA flow')
parser.add_argument('--check_csv_file', required=True,
                    help='Specify the to-be-checked csv file constaining flow-run information')
parser.add_argument('--reference_csv_file', required=True,
                    help='Specify the reference csv file constaining flow-run information')
parser.add_argument('--metric_checklist_csv_file', required=True,
                    help='Specify the csv file constaining metrics to be checked')
# By default, allow a 50% tolerance when checking metrics
parser.add_argument('--check_tolerance', default="0.5,1.5",
                    help='Specify the tolerance when checking metrics. Format <lower_bound>,<upper_bound>')
args = parser.parse_args()

#####################################################################
# Check options:
# - Input csv files must be valid
#   Otherwise, error out
#####################################################################
if not isfile(args.check_csv_file):
  logging.error("Invalid csv file to check: " + args.check_csv_file + "\nFile does not exist!\n")
  exit(1)

if not isfile(args.reference_csv_file):
  logging.error("Invalid reference csv file: " + args.reference_csv_file + "\nFile does not exist!\n")
  exit(1)

if not isfile(args.metric_checklist_csv_file):
  logging.error("Invalid metric checklist csv file: " + args.metric_checklist_csv_file + "\nFile does not exist!\n")
  exit(1)

#####################################################################
# Parse a checklist for metrics to be checked
#####################################################################
metric_checklist_csv_file = open(args.metric_checklist_csv_file, "r")
metric_checklist_csv_content = csv.DictReader(filter(lambda row : row[0]!='#', metric_checklist_csv_file), delimiter=',')
# Hash the reference results with the name tag
metric_checklist = []
for row in metric_checklist_csv_content:
  metric_checklist.append(row[csv_metric_tag]);

#####################################################################
# Parse the reference csv file
# Skip any line start with '#' which is treated as comments
#####################################################################
ref_csv_file = open(args.reference_csv_file, "r")
ref_csv_content = csv.DictReader(filter(lambda row : row[0]!='#', ref_csv_file), delimiter=',')
# Hash the reference results with the name tag
ref_results = {}
for row in ref_csv_content:
  ref_results[row[csv_name_tag]] = row;

#####################################################################
# Parse the tolerance to be applied when checking metrics
#####################################################################
lower_bound_factor = float(args.check_tolerance.split(",")[0])
upper_bound_factor = float(args.check_tolerance.split(",")[1])

#####################################################################
# Parse the csv file to check
#####################################################################
with open(args.check_csv_file, newline='') as check_csv_file:
  results_to_check = csv.DictReader(check_csv_file, delimiter=',')
  checkpoint_count = 0
  check_error_count = 0
  for row in results_to_check:
    # Start from line 1 and check information
    for metric_to_check in metric_checklist:
      # Check if the metric is in a range
      if (lower_bound_factor * float(ref_results[row[csv_name_tag]][metric_to_check]) > float(row[metric_to_check])) or (upper_bound_factor * float(ref_results[row[csv_name_tag]][metric_to_check]) < float(row[metric_to_check])) :
        # Check QoR failed, error out
        logging.error("Benchmark " + str(row[csv_name_tag]) + " failed in checking '" + str(metric_to_check) +"'\n" + "Found: " + str(row[metric_to_check]) + " but expected: " + str(ref_results[row[csv_name_tag]][metric_to_check]) + " outside range [" + str(lower_bound_factor * 100) + "%, " + str(upper_bound_factor * 100) + "%]")      
        check_error_count += 1
      # Pass this metric check, increase counter
      checkpoint_count += 1
  logging.info("Checked " + str(checkpoint_count) + " metrics")
  logging.info("See " + str(check_error_count) + " QoR failures")

  if (0 < check_error_count):
    exit(1) 

#####################################################################
# Post checked results on stdout:
# reaching here, it means all the checks have passed
#####################################################################
with open(args.check_csv_file, newline='') as check_csv_file:
  results_to_check = csv.DictReader(check_csv_file, delimiter=',')
  # Print out keywords: name + metric checklist
  print(str(csv_name_tag) + " ", end='') 
  for metric_to_check in metric_checklist:
    print(str(metric_to_check) + " ", end='') 
  print("")

  for row in results_to_check:
    # Start from line 1, print checked metrics
    print(row[csv_name_tag] + " ", end='') 
    for metric_to_check in metric_checklist:
      print(row[metric_to_check] + " ", end='') 
    print("")