From 003ccf7197a11fdd4d3b107b643af1a4e1810155 Mon Sep 17 00:00:00 2001 From: Claire Xenia Wolf Date: Mon, 31 Oct 2022 20:29:32 +0100 Subject: [PATCH] Add color handling via click.style and click.echo Signed-off-by: Claire Xenia Wolf --- sbysrc/sby_core.py | 46 +++++++++++++++++++++++-------------- sbysrc/sby_engine_abc.py | 4 ++-- sbysrc/sby_engine_aiger.py | 6 ++--- sbysrc/sby_engine_btor.py | 6 ++--- sbysrc/sby_engine_smtbmc.py | 6 ++--- 5 files changed, 40 insertions(+), 28 deletions(-) diff --git a/sbysrc/sby_core.py b/sbysrc/sby_core.py index 48da97b..98ddaf0 100644 --- a/sbysrc/sby_core.py +++ b/sbysrc/sby_core.py @@ -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) @@ -105,8 +105,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 +136,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 +171,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 +202,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 +218,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 +226,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 +640,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, []) @@ -680,15 +680,24 @@ class SbyTask(SbyConfig): self.procs_pending.remove(proc) self.taskloop.procs_pending.remove(proc) + def dress_message(self, logmessage): + tm = localtime() + return " ".join([ + click.style("SBY", fg="blue"), + click.style("{:2d}:{:02d}:{:02d}".format(tm.tm_hour, tm.tm_min, tm.tm_sec), fg="green"), + "[" + click.style(self.workdir, fg="blue") + "]", + logmessage + ]) + 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 = self.dress_message(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 +705,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 +1091,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 +1110,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') diff --git a/sbysrc/sby_engine_abc.py b/sbysrc/sby_engine_abc.py index ace68a0..47224d8 100644 --- a/sbysrc/sby_engine_abc.py +++ b/sbysrc/sby_engine_abc.py @@ -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() diff --git a/sbysrc/sby_engine_aiger.py b/sbysrc/sby_engine_aiger.py index 293b0cb..18cb876 100644 --- a/sbysrc/sby_engine_aiger.py +++ b/sbysrc/sby_engine_aiger.py @@ -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 diff --git a/sbysrc/sby_engine_btor.py b/sbysrc/sby_engine_btor.py index 0bb4c05..284a49e 100644 --- a/sbysrc/sby_engine_btor.py +++ b/sbysrc/sby_engine_btor.py @@ -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: diff --git a/sbysrc/sby_engine_smtbmc.py b/sbysrc/sby_engine_smtbmc.py index 614c8b4..6c19268 100644 --- a/sbysrc/sby_engine_smtbmc.py +++ b/sbysrc/sby_engine_smtbmc.py @@ -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":