More utilities for cover-float integration into act4

.gitignore

@@ -19,3 +19,4 @@ tests/testvectors/B*_tv.txt
 coverage/covergroups/bins_templates/generated
 tests/readable/B*_parsed.txt
 tests/processed/*/B*.csv
+tests/.stamp/*

Makefile

@@ -2,13 +2,23 @@
 
 RM_CMD ?= rm -rf
 AGGRESSIVENESS ?= 1
+PROCESSED_ONLY ?=
+SILENT ?=
 
 COVER_FLOAT_FLAGS =
 
 ifeq ($(AGGRESSIVENESS), 0)
 	COVER_FLOAT_FLAGS += --partial-output
 endif
 
+ifneq ($(PROCESSED_ONLY),)
+	COVER_FLOAT_FLAGS += --only-processed-vectors
+endif
+
+ifneq ($(SILENT),)
+	COVER_FLOAT_FLAGS += -qq
+endif
+
 MODELS := B1 B2 B3 B6 B7 B8 B9 B10 B11 B12 B13 B14 B15 B16 B20 B21 B25 B26 B27 B29
 
 .PHONY: build clean sim all $(MODELS)
@@ -18,6 +28,9 @@ MODELS := B1 B2 B3 B6 B7 B8 B9 B10 B11 B12 B13 B14 B15 B16 B20 B21 B25 B26 B27 B
 all:
 	uv run --managed-python cover-float-testgen $(COVER_FLOAT_FLAGS)
 
+processed-tests-only:
+	uv run --managed-python cover-float-testgen --partial-output --only-processed-vectors --quiet $(COVER_FLOAT_FLAGS)
+
 # Build target to compile the pybind11 module (if necessary)
 build:
 	@echo "Building python module"
@@ -40,6 +53,11 @@ clean:
 	$(RM_CMD) sim/coverfloat_worklib/
 	$(RM_CMD) sim/transcript
 	$(RM_CMD) sim/coverfloat.ucdb
+	$(RM_CMD) tests/testvectors/B*_tv.txt
+	$(RM_CMD) tests/covervectors/B*_cv.txt
+	$(RM_CMD) tests/readable/B*_parsed.txt
+	$(RM_CMD) tests/processed/*/B*.csv
+	$(RM_CMD) tests/.stamp
 
 # --- Include Dependency Files ---
 # Include auto-generated dependency files if they exist

src/cover_float/cli.py

@@ -15,17 +15,17 @@
 
 import argparse
 import logging
-from concurrent.futures import ProcessPoolExecutor
+from concurrent.futures import Future, ProcessPoolExecutor, as_completed
 from pathlib import Path
 
+from rich.progress import BarColumn, MofNCompleteColumn, Progress, TextColumn, TimeElapsedColumn
+
 import cover_float.common.log as log
 import cover_float.testgen as tg
 from cover_float.common.constants import config
 from cover_float.common.util import SingleThreadedExecutor
 from cover_float.reference import run_test_vector
 
-logging.basicConfig(level=logging.INFO)
-
 
 def main() -> None:
     parser = argparse.ArgumentParser()
@@ -63,22 +63,62 @@ def testgen() -> None:
     parser.add_argument(
         "--partial-output", action="store_true", help="Create a Reduced Number of Tests in Test Heavy Models"
     )
+    parser.add_argument(
+        "--quiet",
+        "-q",
+        action="count",
+        default=0,
+        help="Applying Once Condenses Info Logging to a Single Progress Bar, Twice Eliminates all Logging",
+    )
+    parser.add_argument("--only-processed-vectors", action="store_true", help="Generate Only Processed Test Vectors")
     args = parser.parse_args()
 
     output_dir = Path(args.output_dir)
     single_thread = args.single_thread or (args.models is not None and len(args.models) < 2)
     config.FULL_COVERAGE_TESTGEN = not args.partial_output
+    config.QUIET = args.quiet > 0
+    config.RELEASE = args.only_processed_vectors
+    if args.quiet > 0:
+        logging.basicConfig(level=logging.ERROR)
+    else:
+        logging.basicConfig(level=logging.INFO)
 
     if single_thread:
         executor = SingleThreadedExecutor()
     else:
         executor = ProcessPoolExecutor() if args.jobs is None else ProcessPoolExecutor(max_workers=args.jobs)
 
-    with log.StatusReporter() as logger, executor:
+    with log.StatusReporter(disable=(args.quiet > 0)) as logger, executor:
+        futures: list[Future[None]] = []
+
         if args.models is None:
             for model in tg.model.GLOBAL_MODELS:
-                tg.model.GLOBAL_MODELS[model](output_dir, logger, executor)
+                future = tg.model.GLOBAL_MODELS[model](output_dir, logger, executor)
+                if future is not None:
+                    futures.append(future)
         else:
             for model in args.models:
                 if model in tg.model.GLOBAL_MODELS:
-                    tg.model.GLOBAL_MODELS[model](output_dir, logger, executor)
+                    future = tg.model.GLOBAL_MODELS[model](output_dir, logger, executor)
+                    if future is not None:
+                        futures.append(future)
+
+        if len(futures) == 0:
+            if args.quiet < 2:
+                print(f"No work to be done for {'cover-float' if not args.models else ', '.join(args.models)}")
+            return
+
+        if args.quiet == 1:
+            with Progress(
+                TextColumn("{task.description}"),
+                BarColumn(),
+                MofNCompleteColumn(),
+                TimeElapsedColumn(),
+            ) as progress:
+                for future in progress.track(
+                    as_completed(futures), total=len(futures), description="[cyan]Generating Cover-Float Tests"
+                ):
+                    future.result()
+        else:
+            for future in as_completed(futures):
+                future.result()

src/cover_float/common/constants.py

@@ -148,8 +148,10 @@
 
 @dataclass
 class Config:
-    FULL_COVERAGE_TESTGEN: int = 1
+    FULL_COVERAGE_TESTGEN: bool = True
     CACHE_DIR: str = "build/cache"
+    QUIET: bool = False
+    RELEASE: bool = False
 
 
 config = Config()

src/cover_float/common/log.py

@@ -162,14 +162,21 @@ def render(self, task: Task) -> console.RenderableType:
 
 
 class AsyncLoggingHandler(logging.handlers.QueueListener):
-    def __init__(self, reporter: StatusReporter, listen_to: Queue[Any], *handlers: logging.Handler) -> None:
+    def __init__(self, reporter: StatusReporter, listen_to: Queue[Any], level: int, *handlers: logging.Handler) -> None:
         super().__init__(listen_to, *handlers)
 
         self.reporter = reporter
+        self.level = level
 
     def handle_progress_update(self, record: dict[Any, Any]) -> None:
         try:
             action = record["action"]
+            if self.level > STATUS_LEVEL_NUM:
+                if action == "add_task":
+                    task_id_pipe: Connection = record["pipe_end"]
+                    task_id_pipe.send(-1)
+                return
+
             if action == "update":
                 self.reporter.progress.update(*record["args"], **record["kwargs"])
             elif action == "advance":
@@ -186,8 +193,8 @@ def handle_progress_update(self, record: dict[Any, Any]) -> None:
                 self.reporter.progress.refresh()
             else:
                 logging.info(f"Failed to Log {record}")
-        except Exception as e:
-            logging.info(f"Failed to Log {record}", exc_info=e)
+        except Exception:
+            logging.exception(f"Failed to Log {record}")
 
     def handle(self, record: logging.LogRecord | dict[Any, Any]) -> None:
         if isinstance(record, logging.LogRecord):
@@ -217,7 +224,7 @@ def emit(self, record: logging.LogRecord) -> None:
 
 
 class StatusReporter:
-    def __init__(self) -> None:
+    def __init__(self, *, disable: bool = False) -> None:
         self.active_status_bars: dict[str, TaskID] = {}
         self.progress = Progress(
             SpinnerColumn(),
@@ -234,14 +241,17 @@ def __init__(self) -> None:
 
         # This is the actual handler that prints to console
         self.rich_handler = ProgressAwareLogHandler(self.progress)
-        self.queue_listener = AsyncLoggingHandler(self, self.logging_queue, self.rich_handler)
+        self.queue_listener = AsyncLoggingHandler(
+            self, self.logging_queue, logging.getLogger().level, self.rich_handler
+        )
         logging.getLogger().addHandler(logging.handlers.QueueHandler(self.logging_queue))
 
         # This keeps all of the refreshes in one thread, eliminating all race conditions
         self._refresh_thread: threading.Timer = threading.Timer(0.1, self._reset_refresh_timer)
         self._refresh_thread.daemon = True
 
         self.exiting = False
+        self.disable = disable
 
     def _reset_refresh_timer(self) -> None:
         self.logging_queue.put({"action": "refresh", "args": [], "kwargs": {}})
@@ -252,7 +262,9 @@ def _reset_refresh_timer(self) -> None:
             self._refresh_thread.start()
 
     def __enter__(self) -> StatusReporter:
-        self.progress.start()
+        if not self.disable:
+            self.progress.start()
+
         self.queue_listener.start()
         self._refresh_thread.start()
 

src/cover_float/reference/impl.py

@@ -17,7 +17,7 @@
 
 import cover_float._reference
 import cover_float._unmodified_reference
-from cover_float.common.constants import TEST_VECTOR_WIDTH_HEX_WITH_SEPARATORS
+from cover_float.common.constants import TEST_VECTOR_WIDTH_HEX_WITH_SEPARATORS, config
 
 
 def run_and_store_test_vector(test_vector: str, test_file: TextIO, cover_file: TextIO) -> None:
@@ -27,13 +27,15 @@ def run_and_store_test_vector(test_vector: str, test_file: TextIO, cover_file: T
 
     generated_test_vector = cover_vector[:TEST_VECTOR_WIDTH_HEX_WITH_SEPARATORS]
     test_file.write(generated_test_vector + "\n")
-    cover_file.write(cover_vector.strip() + "\n")
+    if not config.RELEASE:
+        cover_file.write(cover_vector.strip() + "\n")
 
 
 def store_cover_vector(cover_vector: str, test_file: TextIO, cover_file: TextIO) -> None:
     generated_test_vector = cover_vector[:TEST_VECTOR_WIDTH_HEX_WITH_SEPARATORS]
     test_file.write(generated_test_vector + "\n")
-    cover_file.write(cover_vector.strip() + "\n")
+    if not config.RELEASE:
+        cover_file.write(cover_vector.strip() + "\n")
 
 
 def verify_test_vector(test_vector: str) -> bool:

src/cover_float/scripts/postprocess.py

@@ -20,6 +20,7 @@
 
 import csv
 import logging
+import os
 import pathlib
 import time
 from dataclasses import dataclass
@@ -111,13 +112,16 @@ def postprocess_testvectors(
     logger: log.ModelLogger = cast(log.ModelLogger, logging.getLogger(model))
 
     test_vector_file = test_vector_location / f"{model}_tv.txt"
-    readable_vectors_file = readable_vectors_dir / f"{model}_parsed.txt"
+    readable_vectors_file = (
+        readable_vectors_dir / f"{model}_parsed.txt" if not constants.config.RELEASE else pathlib.Path(os.devnull)
+    )
     processed_vectors: dict[str, tuple[csv.DictWriter[str], TextIO]] = {}
     total = 0
     non_riscv = 0
 
     file_size = test_vector_file.stat().st_size
-    readable_vectors_file.parent.mkdir(parents=True, exist_ok=True)
+    if not constants.config.RELEASE:
+        readable_vectors_file.parent.mkdir(parents=True, exist_ok=True)
 
     with (
         test_vector_file.open("r") as test_vectors,
@@ -130,7 +134,8 @@ def postprocess_testvectors(
         for line in test_vectors.readlines():
             parsed = parse_test_vector(line)
             if parsed:
-                readable_vectors.write(format_output(parsed) + "\n")
+                if not constants.config.RELEASE:
+                    readable_vectors.write(format_output(parsed) + "\n")
 
                 if not verify_test_vector(line):
                     logger.exception(