| #!/usr/bin/env python3 |
| |
| import argparse |
| import logging |
| import subprocess |
| import os |
| import time |
| import shutil |
| import glob |
| |
| from time import strftime, gmtime |
| from distutils.dir_util import copy_tree |
| from pathlib import Path |
| from shutil import rmtree, copy, copytree, get_terminal_size |
| |
| |
| logging.basicConfig(level=logging.INFO, |
| format="%(asctime)10s | %(levelname)-7s | %(message)s", |
| datefmt='%d-%b-%Y %H:%M:%S') |
| |
| REPO_DIR = os.path.dirname(os.path.realpath(__file__)) |
| TORTURE_TESTS_DIR = os.path.join(REPO_DIR, "torture_tests") |
| CARAVEL_SIGNOFF_DIR = os.path.join(os.path.dirname(REPO_DIR), "caravel-signoff") |
| |
| |
| def get_paths(inputFile): |
| paths = [] |
| with open(inputFile, 'r') as reader: |
| paths = reader.readlines() |
| paths = [x.rstrip() for x in paths] |
| paths = list(filter(None, paths)) |
| logging.info("read input file: " + inputFile) |
| return paths |
| |
| def remove_extension(path): |
| path = path.rsplit('.')[0] |
| logging.debug("removed file extension: " + path) |
| return path |
| |
| def get_name(path): |
| path = remove_extension(path) |
| file_name = os.path.basename(path) |
| logging.debug("extracted design name: " + file_name) |
| return file_name |
| |
| def mk_torture_dir(design_name): |
| destination_dir = os.path.join(TORTURE_TESTS_DIR, design_name) |
| destination_gds_dir = os.path.join(destination_dir, "gds") |
| destination_verilog_dir = os.path.join(destination_dir, "verilog") |
| Path(destination_dir).mkdir(parents=True, exist_ok=True) |
| Path(destination_gds_dir).mkdir(parents=True, exist_ok=True) |
| Path(destination_verilog_dir).mkdir(parents=True, exist_ok=True) |
| |
| return (destination_dir, destination_gds_dir, destination_verilog_dir) |
| |
| def mk_dir(path): |
| os.makedirs(path, exist_ok=True) |
| # Path(path).mkdir(parents=True, exist_ok=True) |
| return path |
| |
| def git_push(path): |
| logging.debug("Adding %s to git" % path) |
| commit_msg = f"\"Adding {path} to git\"" |
| command = f"git add {path} && git commit -m {commit_msg} && git push" |
| proc = subprocess.run(command, shell=True, |
| stdout=subprocess.PIPE, stderr=subprocess.PIPE) |
| if proc.returncode: |
| logging.warning("did not push: " + proc.stderr.decode("utf-8")) |
| else: |
| logging.info("pushed a new commit to git") |
| |
| return proc.returncode |
| |
| def git_push_caravel_signoff(path, commit): |
| logging.debug("Adding %s to git" % path) |
| commit_msg = f"\"Automatically adding calibre lvs results to caravel-signoff\"" |
| command = f"git checkout -b {commit} && git add {path} && git commit -m {commit_msg} && git push origin {commit}" |
| proc = subprocess.run(command, shell=True, |
| stdout=subprocess.PIPE, stderr=subprocess.PIPE) |
| if proc.returncode: |
| logging.warning("did not push: " + proc.stderr.decode("utf-8")) |
| else: |
| logging.info("pushed a new commit to git") |
| |
| return proc.returncode |
| |
| |
| def find_branch(pattern): |
| branch = None |
| command = "git ls-remote" |
| try: |
| proc = subprocess.check_output( |
| command, shell=True, universal_newlines=True, stderr=subprocess.STDOUT) |
| remotes = [x.split()[1] for x in proc.rstrip().split('\n')] |
| for remote in remotes: |
| if pattern in remote: |
| branch = remote |
| break |
| except subprocess.CalledProcessError: |
| logging.warning("git ls-remote failed") |
| branch = None |
| return branch |
| |
| |
| def get_output_branch(branch): |
| number = branch.split("output")[1].split("_")[0] |
| output_branch = f"latest-{number}" |
| return output_branch |
| |
| |
| def detele_folder(path): |
| rmtree( |
| os.path.join( |
| TORTURE_TESTS_DIR, |
| get_name(path) |
| ) |
| ) |
| command = f"git rm -r {path} && git commit -m \"removing {path}\" && git push" |
| proc = subprocess.run(command, shell=True, |
| stdout=subprocess.PIPE, stderr=subprocess.PIPE) |
| logging.info(f"removed {path} from git") |
| |
| def git_fetch(branch): |
| logging.info(f"git fetch {branch}...") |
| command = f"git fetch origin {branch}:{branch} --depth=1" |
| subprocess.run(command, shell=True, |
| stdout=subprocess.PIPE, stderr=subprocess.PIPE) |
| |
| def git_checkout(branch): |
| logging.info(f"git checkout {branch}...") |
| command = f"git checkout {branch}" |
| subprocess.run(command, shell=True, |
| stdout=subprocess.PIPE, stderr=subprocess.PIPE) |
| |
| if __name__ == "__main__": |
| parser = argparse.ArgumentParser( |
| description="Calibre design list submission" |
| ) |
| parser.add_argument( |
| "--input-file", |
| "-i", |
| action="store", |
| required=True, |
| default=None |
| ) |
| args = parser.parse_args() |
| input_file = args.input_file |
| |
| paths = get_paths(input_file) |
| files = glob.glob(TORTURE_TESTS_DIR + "/*") |
| |
| for path in paths: |
| logging.info(f"processing {path}...") |
| design_name = get_name(path) |
| design_dir = os.path.dirname(path) |
| torture_tests_dir, torture_tests_gds_dir, torture_tests_verilog_dir = mk_torture_dir(design_name) |
| if os.path.splitext(path)[1] == ".gz" or os.path.splitext(path)[1] == ".gds": |
| logging.info(f"copying {path} to {torture_tests_gds_dir}...") |
| copy(path, torture_tests_gds_dir) |
| else: |
| logging.info(f"copying {path} to {torture_tests_verilog_dir}...") |
| copy(path, torture_tests_verilog_dir) |
| logging.info("copied") |
| |
| for path in files: |
| design_name = get_name(path) |
| design_dir = os.path.dirname(path) |
| torture_tests_dir = path |
| push_status = git_push(torture_tests_dir) |
| branch = find_branch("check-in-progress") |
| logging.info(f"searching for *check-in-progress*{design_name} in remote_branches...") |
| if branch is not None: |
| if design_name not in branch: |
| branch = None |
| while branch is None: |
| time.sleep(10) |
| branch = find_branch("check-in-progress") |
| if branch is not None: |
| if design_name not in branch: |
| branch = None |
| |
| logging.info(f"{branch} found") |
| start_time = time.time() |
| output_branch = get_output_branch(branch) |
| logging.info(f"output branch {output_branch}") |
| logging.info(f"searching for {output_branch} in remote_branches...") |
| branch = find_branch(output_branch) |
| while branch is None: |
| time.sleep(10) |
| branch = find_branch(output_branch) |
| elapsed_time = strftime("%H:%M:%S", gmtime(time.time() - start_time)) |
| |
| logging.info(f"{branch} found") |
| branch_clean_name = branch.split("refs/heads/")[1] |
| git_fetch(branch_clean_name) |
| git_checkout(branch_clean_name) |
| |
| lvs_folder = os.path.join(CARAVEL_SIGNOFF_DIR, "lvs") |
| cdl_folder = os.path.join(CARAVEL_SIGNOFF_DIR, "cdl") |
| lvs_destination = os.path.join(lvs_folder, design_name) |
| if os.path.isdir(lvs_destination): |
| shutil.rmtree(lvs_destination) |
| cdl_destination = os.path.join(cdl_folder, design_name) |
| if os.path.isdir(cdl_destination): |
| shutil.rmtree(cdl_destination) |
| # lvs_destination = mk_dir(lvs_destination) |
| lvs_results = f"{torture_tests_dir}/lvs" |
| cdl_results = f"{torture_tests_dir}/cdl" |
| logging.info(f"copying {lvs_results} to {lvs_destination}...") |
| copytree(lvs_results, lvs_destination) |
| copytree(cdl_results, cdl_destination) |
| logging.info("copied") |
| |
| spef_folder = os.path.join(CARAVEL_SIGNOFF_DIR, "spef/openRCX") |
| spef_results = f"{torture_tests_dir}/pex/{design_name}.spef" |
| logging.info(f"copying {spef_results} to {spef_folder}...") |
| try: |
| copy(spef_results, spef_folder) |
| except: |
| logging.warning(f"Can't find {spef_results}") |
| logging.info("copied") |
| |
| |
| # results_folder = "drcmr" |
| # resutls_destination = f"{design_dir}/calibre/{branch_clean_name}/{results_folder}" |
| # results = f"{torture_tests_dir}/{results_folder}" |
| # logging.info(f"copying {results} to {resutls_destination}...") |
| # copytree(results, resutls_destination) |
| # logging.info("copyied") |
| |
| git_checkout("master") |
| logging.info("back to master branch") |
| delete_status = detele_folder(torture_tests_dir) |
| logging.info(f"{path} done!") |
| logging.info(f"time take {elapsed_time}") |
| |
| print('=' * get_terminal_size((80, 20)).columns) |
