mirrors/sby
3
0
Fork 0
mirror of https://github.com/YosysHQ/sby.git synced 2025-04-05 22:14:08 +00:00
Code Activity

Merge pull request #221 from YosysHQ/claire/click

Browse source 
Add color handling via click.style and click.echo
This commit is contained in:
Claire Xen 2022-11-24 18:19:43 +01:00 committed by GitHub
parent 54cb030738 c29a5bbe8a
commit 9c75e49418
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 54 additions and 43 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
sbysrc/sby.py
View file

@ -19,9 +19,9 @@
import argparse, json, os, sys, shutil, tempfile, re
##yosys-sys-path##
from sby_core import SbyConfig, SbyTask, SbyAbort, SbyTaskloop, process_filename
from sby_core import SbyConfig, SbyTask, SbyAbort, SbyTaskloop, process_filename, dress_message
from sby_jobserver import SbyJobClient, process_jobserver_environment
import time, platform
import time, platform, click
process_jobserver_environment() # needs to be called early
@ -177,9 +177,8 @@ prep -top top
early_logmsgs = list()
def early_log(workdir, msg):
tm = time.localtime()
early_logmsgs.append("SBY {:2d}:{:02d}:{:02d} [{}] {}".format(tm.tm_hour, tm.tm_min, tm.tm_sec, workdir, msg))
print(early_logmsgs[-1])
early_logmsgs.append(dress_message(workdir, msg))
click.echo(early_logmsgs[-1])
def read_sbyconfig(sbydata, taskname):
cfgdata = list()
@ -567,7 +566,6 @@ else:
failed.append(taskname)
if failed and (len(tasknames) > 1 or tasknames[0] is not None):
tm = time.localtime()
print("SBY {:2d}:{:02d}:{:02d} The following tasks failed: {}".format(tm.tm_hour, tm.tm_min, tm.tm_sec, failed))
click.echo(dress_message(None, click.style(f"The following tasks failed: {failed}", fg="red", bold=True)))
sys.exit(retcode)

47
sbysrc/sby_core.py
View file

@ -16,7 +16,7 @@
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
#
import os, re, sys, signal, platform
import os, re, sys, signal, platform, click
if os.name == "posix":
import resource, fcntl
import subprocess
@ -28,7 +28,7 @@ from sby_design import SbyProperty, SbyModule, design_hierarchy
all_procs_running = []
def force_shutdown(signum, frame):
print("SBY ---- Keyboard interrupt or external termination signal ----", flush=True)
click.echo("SBY ---- Keyboard interrupt or external termination signal ----")
for proc in list(all_procs_running):
proc.terminate()
sys.exit(1)
@ -46,6 +46,16 @@ def process_filename(filename):
return filename
def dress_message(workdir, logmessage):
tm = localtime()
if workdir is not None:
logmessage = "[" + click.style(workdir, fg="blue") + "] " + logmessage
return " ".join([
click.style("SBY", fg="blue"),
click.style("{:2d}:{:02d}:{:02d}".format(tm.tm_hour, tm.tm_min, tm.tm_sec), fg="green"),
logmessage
])
class SbyProc:
def __init__(self, task, info, deps, cmdline, logfile=None, logstderr=True, silent=False):
self.running = False
@ -105,8 +115,8 @@ class SbyProc:
def log(self, line):
if line is not None and (self.noprintregex is None or not self.noprintregex.match(line)):
if self.logfile is not None:
print(line, file=self.logfile)
self.task.log(f"{self.info}: {line}")
click.echo(line, file=self.logfile)
self.task.log(f"{click.style(self.info, fg='magenta')}: {line}")
def handle_output(self, line):
if self.terminated or len(line) == 0:
@ -136,7 +146,7 @@ class SbyProc:
return
if self.running:
if not self.silent:
self.task.log(f"{self.info}: terminating process")
self.task.log(f"{click.style(self.info, fg='magenta')}: terminating process")
if os.name == "posix":
try:
os.killpg(self.p.pid, signal.SIGTERM)
@ -171,7 +181,7 @@ class SbyProc:
return
if not self.silent:
self.task.log(f"{self.info}: starting process \"{self.cmdline}\"")
self.task.log(f"{click.style(self.info, fg='magenta')}: starting process \"{self.cmdline}\"")
if os.name == "posix":
def preexec_fn():
@ -202,7 +212,7 @@ class SbyProc:
self.job_lease.done()
if not self.silent:
self.task.log(f"{self.info}: finished (returncode={self.p.returncode})")
self.task.log(f"{click.style(self.info, fg='magenta')}: finished (returncode={self.p.returncode})")
self.task.update_proc_stopped(self)
self.running = False
@ -218,7 +228,7 @@ class SbyProc:
if returncode == 127:
if not self.silent:
self.task.log(f"{self.info}: COMMAND NOT FOUND. ERROR.")
self.task.log(f"{click.style(self.info, fg='magenta')}: COMMAND NOT FOUND. ERROR.")
self.handle_error(returncode)
self.terminated = True
self.task.proc_failed(self)
@ -226,7 +236,7 @@ class SbyProc:
if self.checkretcode and returncode not in self.retcodes:
if not self.silent:
self.task.log(f"{self.info}: task failed. ERROR.")
self.task.log(f"{click.style(self.info, fg='magenta')}: task failed. ERROR.")
self.handle_error(returncode)
self.terminated = True
self.task.proc_failed(self)
@ -640,12 +650,12 @@ class SbyTask(SbyConfig):
self.log_targets = [sys.stdout, self.logfile]
for line in early_logs:
print(line, file=self.logfile, flush=True)
click.echo(line, file=self.logfile)
if not reusedir:
with open(f"{workdir}/config.sby", "w") as f:
for line in sbyconfig:
print(line, file=f)
click.echo(line, file=f)
def engine_list(self):
engines = self.engines.get(None, []) + self.engines.get(self.opt_mode, [])
@ -682,13 +692,13 @@ class SbyTask(SbyConfig):
def log(self, logmessage):
tm = localtime()
line = "SBY {:2d}:{:02d}:{:02d} [{}] {}".format(tm.tm_hour, tm.tm_min, tm.tm_sec, self.workdir, logmessage)
line = dress_message(self.workdir, logmessage)
for target in self.log_targets:
print(line, file=target, flush=True)
click.echo(line, file=target)
def error(self, logmessage):
tm = localtime()
self.log(f"ERROR: {logmessage}")
self.log(click.style(f"ERROR: {logmessage}", fg="red", bold=True))
self.status = "ERROR"
if "ERROR" not in self.expect:
self.retcode = 16
@ -696,7 +706,7 @@ class SbyTask(SbyConfig):
self.retcode = 0
self.terminate()
with open(f"{self.workdir}/{self.status}", "w") as f:
print(f"ERROR: {logmessage}", file=f)
click.echo(f"ERROR: {logmessage}", file=f)
raise SbyAbort(logmessage)
def makedirs(self, path):
@ -1082,7 +1092,10 @@ class SbyTask(SbyConfig):
] + self.summary
for line in self.summary:
self.log(f"summary: {line}")
if line.startswith("Elapsed"):
self.log(f"summary: {line}")
else:
self.log("summary: " + click.style(line, fg="green" if self.status in self.expect else "red", bold=True))
assert self.status in ["PASS", "FAIL", "UNKNOWN", "ERROR", "TIMEOUT"]
@ -1098,7 +1111,7 @@ class SbyTask(SbyConfig):
def write_summary_file(self):
with open(f"{self.workdir}/{self.status}", "w") as f:
for line in self.summary:
print(line, file=f)
click.echo(line, file=f)
def print_junit_result(self, f, junit_ts_name, junit_tc_name, junit_format_strict=False):
junit_time = strftime('%Y-%m-%dT%H:%M:%S')

4
sbysrc/sby_engine_abc.py
View file

@ -16,7 +16,7 @@
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
#
import re, os, getopt
import re, os, getopt, click
from sby_core import SbyProc
def run(mode, task, engine_idx, engine):
@ -83,7 +83,7 @@ def run(mode, task, engine_idx, engine):
task.error(f"engine_{engine_idx}: Could not determine engine status.")
task.update_status(proc_status)
task.log(f"engine_{engine_idx}: Status returned by engine: {proc_status}")
task.log(f"{click.style(f'engine_{engine_idx}', fg='magenta')}: Status returned by engine: {proc_status}")
task.summary.append(f"""engine_{engine_idx} ({" ".join(engine)}) returned {proc_status}""")
task.terminate()

6
sbysrc/sby_engine_aiger.py
View file

@ -16,7 +16,7 @@
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
#
import re, os, getopt
import re, os, getopt, click
from sby_core import SbyProc
def run(mode, task, engine_idx, engine):
@ -98,7 +98,7 @@ def run(mode, task, engine_idx, engine):
aiw_file.close()
task.update_status(proc_status)
task.log(f"engine_{engine_idx}: Status returned by engine: {proc_status}")
task.log(f"{click.style(f'engine_{engine_idx}', fg='magenta')}: Status returned by engine: {proc_status}")
task.summary.append(f"""engine_{engine_idx} ({" ".join(engine)}) returned {proc_status}""")
task.terminate()
@ -159,7 +159,7 @@ def run(mode, task, engine_idx, engine):
proc2.exit_callback = exit_callback2
else:
task.log(f"engine_{engine_idx}: Engine did not produce a counter example.")
task.log(f"{click.style(f'engine_{engine_idx}', fg='magenta')}: Engine did not produce a counter example.")
proc.output_callback = output_callback
proc.exit_callback = exit_callback

6
sbysrc/sby_engine_btor.py
View file

@ -16,7 +16,7 @@
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
#
import re, os, getopt
import re, os, getopt, click
from types import SimpleNamespace
from sby_core import SbyProc
@ -87,11 +87,11 @@ def run(mode, task, engine_idx, engine):
task.error(f"engine_{engine_idx}: Engine terminated without status.")
task.update_status(proc_status.upper())
task.log(f"engine_{engine_idx}: Status returned by engine: {proc_status}")
task.log(f"{click.style(f'engine_{engine_idx}', fg='magenta')}: Status returned by engine: {proc_status}")
task.summary.append(f"""engine_{engine_idx} ({" ".join(engine)}) returned {proc_status}""")
if len(common_state.produced_traces) == 0:
task.log(f"""engine_{engine_idx}: Engine did not produce a{" counter" if mode != "cover" else "n "}example.""")
task.log(f"""{click.style(f'engine_{engine_idx}', fg='magenta')}: Engine did not produce a{" counter" if mode != "cover" else "n "}example.""")
elif len(common_state.produced_traces) <= common_state.print_traces_max:
task.summary.extend(common_state.produced_traces)
else:

6
sbysrc/sby_engine_smtbmc.py
View file

@ -16,7 +16,7 @@
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
#
import re, os, getopt
import re, os, getopt, click
from sby_core import SbyProc
def run(mode, task, engine_idx, engine):
@ -232,7 +232,7 @@ def run(mode, task, engine_idx, engine):
if mode == "bmc" or mode == "cover":
task.update_status(proc_status)
proc_status_lower = proc_status.lower() if proc_status == "PASS" else proc_status
task.log(f"engine_{engine_idx}: Status returned by engine: {proc_status_lower}")
task.log(f"{click.style(f'engine_{engine_idx}', fg='magenta')}: Status returned by engine: {proc_status_lower}")
task.summary.append(f"""engine_{engine_idx} ({" ".join(engine)}) returned {proc_status_lower}""")
if proc_status == "FAIL" and mode != "cover":
@ -260,7 +260,7 @@ def run(mode, task, engine_idx, engine):
elif mode in ["prove_basecase", "prove_induction"]:
proc_status_lower = proc_status.lower() if proc_status == "PASS" else proc_status
task.log(f"""engine_{engine_idx}: Status returned by engine for {mode.split("_")[1]}: {proc_status_lower}""")
task.log(f"""{click.style(f'engine_{engine_idx}', fg='magenta')}: Status returned by engine for {mode.split("_")[1]}: {proc_status_lower}""")
task.summary.append(f"""engine_{engine_idx} ({" ".join(engine)}) returned {proc_status_lower} for {mode.split("_")[1]}""")
if mode == "prove_basecase":

4
sbysrc/sby_mode_bmc.py
View file

@ -16,7 +16,7 @@
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
#
import re, os, getopt
import re, os, getopt, click
from sby_core import SbyProc
def run(task):
@ -25,7 +25,7 @@ def run(task):
task.handle_str_option("aigsmt", "yices")
for engine_idx, engine in task.engine_list():
task.log(f"""engine_{engine_idx}: {" ".join(engine)}""")
task.log(f"{click.style(f'engine_{engine_idx}', fg='magenta')}: {' '.join(engine)}")
task.makedirs(f"{task.workdir}/engine_{engine_idx}")
if engine[0] == "smtbmc":

4
sbysrc/sby_mode_cover.py
View file

@ -16,7 +16,7 @@
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
#
import re, os, getopt
import re, os, getopt, click
from sby_core import SbyProc
def run(task):
@ -24,7 +24,7 @@ def run(task):
task.handle_int_option("append", 0)
for engine_idx, engine in task.engine_list():
task.log(f"""engine_{engine_idx}: {" ".join(engine)}""")
task.log(f"{click.style(f'engine_{engine_idx}', fg='magenta')}: {' '.join(engine)}")
task.makedirs(f"{task.workdir}/engine_{engine_idx}")
if engine[0] == "smtbmc":

4
sbysrc/sby_mode_live.py
View file

@ -16,7 +16,7 @@
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
#
import re, os, getopt
import re, os, getopt, click
from sby_core import SbyProc
def run(task):
@ -25,7 +25,7 @@ def run(task):
task.status = "UNKNOWN"
for engine_idx, engine in task.engine_list():
task.log(f"""engine_{engine_idx}: {" ".join(engine)}""")
task.log(f"{click.style(f'engine_{engine_idx}', fg='magenta')}: {' '.join(engine)}")
task.makedirs(f"{task.workdir}/engine_{engine_idx}")
if engine[0] == "aiger":

4
sbysrc/sby_mode_prove.py
View file

@ -16,7 +16,7 @@
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
#
import re, os, getopt
import re, os, getopt, click
from sby_core import SbyProc
def run(task):
@ -32,7 +32,7 @@ def run(task):
task.induction_procs = list()
for engine_idx, engine in task.engine_list():
task.log(f"""engine_{engine_idx}: {" ".join(engine)}""")
task.log(f"{click.style(f'engine_{engine_idx}', fg='magenta')}: {' '.join(engine)}")
task.makedirs(f"{task.workdir}/engine_{engine_idx}")
if engine[0] == "smtbmc":