#!/usr/bin/python3
# ##########################################################################
# Copyright 2020
# ASTRON (Netherlands Institute for Radio Astronomy) <http://www.astron.nl/>
# P.O.Box 2, 7990 AA Dwingeloo, The Netherlands
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ##########################################################################
# ##########################################################################
# Author: Pieter Donker
# Purpose: run vhdl_style_fix.py on all files in the hdl dir using multiple
# workers
# Description:
# -p: number of workers to use
# -v: verbose
# usage:
# cd git/hdl/
# . ./init_hdl.sh
# vhdl_style_fix_all_files.py -p 4
#
# ##########################################################################
from os import getenv, path
from sys import exit, version_info
from time import sleep
from argparse import ArgumentParser, RawTextHelpFormatter
from textwrap import dedent
from subprocess import Popen, PIPE
from multiprocessing import Process, Queue
from queue import Empty
PYTHON_VERSION = version_info[0]
def run_cmd(cmd):
""" run_cmd()
run 'cmd' in a terminal and return response
cmd: command to execute in terminal
return: 'stdout' or 'Error, stderr' in case of a error
"""
proc = Popen(cmd, shell=True, stdout=PIPE, stderr=PIPE)
_stdout, _stderr = proc.communicate()
if _stderr:
stderr = _stderr.decode('utf - 8') if PYTHON_VERSION == 3 else _stderr
print(stderr)
return 'Error, {}'.format(stderr)
stdout = _stdout.decode('utf - 8') if PYTHON_VERSION == 3 else _stdout
return stdout
class Worker(Process):
""" Worker, used for file checking before giving it to vsg to fix.
generated files and files with inside message 'DO NOT EDIT' are skipped.
"""
def __init__(self, id, control, in_data, out_data, verbose=None):
Process.__init__(self)
self.id = id
self.control = control
self.in_data = in_data
self.out_data = out_data
self.verbose = False if verbose is None else verbose
self.stop = False
def is_generated(self, filename):
"""
Look if 'generated' is found in the pathname.
Return the skip message and return True if found, else False
"""
if "generated" in filename:
response = f"skip {filename}: (generated file)"
self.out_data.put(response)
return True
return False
def is_do_not_edit(self, filename):
"""
Look if 'DO NOT EDIT' is found inside the file.
Return the skip message and return True if found, else False
"""
with open(filename, 'r') as fd:
data = fd.read()
if "DO NOT EDIT" in data:
response = f"skip {filename}: (DO NOT EDIT)"
self.out_data.put(response)
return True
return False
def run_fix(self, filename):
"""
Use vhdl_style_fix.py to fix the file
"""
basedir = getenv("HDL_WORK")
fixtool = path.join(basedir, "tools", "vhdl_style_fix.py")
cmd = f"{fixtool} {filename}"
if self.verbose:
cmd += " --verbose"
response = run_cmd(cmd)
self.out_data.put(response.strip())
def run(self):
while not self.stop:
try:
if not self.control.empty():
control = self.control.get()
if control == "stop":
self.stop = True
print(f"stop worker {self.id}")
# get next vhd file to process
filename = self.in_data.get(block=False)
# look if generated is part of the full-filename, skip if found
if self.is_generated(filename):
continue
# look if inside the file is a DO NOT EDIT message, skip if found
if self.is_do_not_edit(filename):
continue
# fix the vhd file with vsg
self.run_fix(filename)
except Empty:
sleep(0.001)
def main():
basedir = getenv("HDL_WORK")
print(f"basedir = {basedir}")
# dirs in basedir to check/fix
checkdirs = ["/libraries/base/", "/libraries/dsp/", "/libraries/io/",
"/libraries/technology/", "/boards/", "/applications/"]
# info for the workers, number of workers to use is given by cmdline arg --proc
n_workers = args.proc
workers = []
worker_control = [Queue() for i in range(n_workers)]
process_queue = Queue()
response_queue = Queue()
n_files = 0
n_files_done = 0
# start the workers
for id in range(n_workers):
if args.verbose:
print(f"start vsg worker {id}")
_p = Worker(id, worker_control[id], process_queue, response_queue)
workers.append(_p)
_p.start()
# put all vhd files in the
for checkdir in checkdirs:
dirname = basedir + checkdir
if args.verbose:
print(f"dirname = {dirname}")
cmd = f'find {dirname} -name "*vhd"'
response = run_cmd(cmd)
for filename in response.splitlines():
process_queue.put(filename)
n_files += 1
if args.verbose:
print(f"put {n_files} in the worker queue")
fd = open("vhdl_fix_all_response.txt", 'w')
while True:
if response_queue.empty():
sleep(0.001)
else:
response = response_queue.get()
if response:
fd.write(response + '\n')
fd.flush()
if args.verbose:
print(response)
n_files_done += 1
if n_files == n_files_done:
if args.verbose:
print("All files processed, stop workers now")
for nr in range(n_workers):
worker_control[nr].put("stop")
sleep(0.5)
for worker in workers:
worker.terminate()
break
fd.close()
return 0
if __name__ == "__main__":
# Parse command line arguments
parser = ArgumentParser(
description="".join(dedent("""
run vhdl-style-guide on all found hdl files with settings from vsg_config.yaml:
with --proc the number of vsg workers can be selected, output from this script
is put in 'vsg_fix_all_response.txt' and send to stdout if --verbose is used.
\n""")),
formatter_class=RawTextHelpFormatter)
parser.add_argument('-p', '--proc', type=int, default=4, help="number of processes to use")
parser.add_argument('-v', '--verbose', action='store_true', help="verbose output")
args = parser.parse_args()
exit_code = main()
exit(exit_code)