2024scrambling (#43)

New command (scramble_benchmarks) that applies the scrambler to a list of benchmarks.

smtcomp/main.py

@@ -25,6 +25,7 @@
 import smtcomp.list_benchmarks
 import smtcomp.selection
 from smtcomp.unpack import write_cin, read_cin
+import smtcomp.scramble_benchmarks
 
 app = typer.Typer()
 
@@ -450,3 +451,18 @@ def convert(x: defs.Result, year: int) -> dict[str, int | str | float] | None:
 #     benchmarks = defs.BenchmarksOld.model_validate_json(read_cin(src))
 #     benchmarks2 = defs.Benchmarks(files=list(map(conv,benchmarks.files)))
 #     write_cin(dst,benchmarks2.model_dump_json(indent=1))
+
+
+@app.command()
+def scramble_benchmarks(
+    competition_track: str, src: Path, dstdir: Path, scrambler: Path, seed: int, max_workers: int = 8
+) -> None:
+    """
+    Use the scrambler to scramble the listed benchmarks and
+    write them to the destination directory.
+    Acceptable competition track names are single-query,
+    incremental, unsat-core, and model-validation.
+    The src file must contain one benchmark path per line.
+    """
+
+    smtcomp.scramble_benchmarks.scramble(competition_track, src, dstdir, scrambler, seed, max_workers)

smtcomp/scramble_benchmarks.py

@@ -0,0 +1,40 @@
+from pathlib import Path
+import rich
+from rich.progress import track
+import subprocess
+import concurrent.futures
+
+
+def scramble_file(files: list, line: int, dstdir: Path, args: list) -> None:
+    dst = Path.joinpath(dstdir, "scrambled" + str(line) + ".smt2")
+    subprocess.run(args, stdin=Path(str(files[line]).strip()).open("r"), stdout=dst.open("w"))
+
+
+def scramble(
+    competition_track: str, benchmark_list: Path, dstdir: Path, scrambler: Path, seed: int, max_workers: int
+) -> None:
+    args = []
+    if competition_track == "single-query":
+        args = [scrambler, "-term_annot", "pattern", "-seed", str(seed)]
+    elif competition_track == "incremental":
+        args = [scrambler, "-term_annot", "pattern", "-incremental", "true", "-seed", str(seed)]
+    elif competition_track == "model-validation":
+        args = [scrambler, "-term_annot", "pattern", "-gen-model-val", "true", "-seed", str(seed)]
+    elif competition_track == "unsat-core":
+        args = [scrambler, "-term_annot", "pattern", "-gen-unsat-core", "true", "-seed", str(seed)]
+    else:
+        rich.print(f"[red]Not a known track name: {track}[/red]")
+        exit(1)
+
+    line = int(0)
+    files = benchmark_list.open().readlines()
+    dstdir.mkdir(parents=True, exist_ok=True)
+    executor = concurrent.futures.ThreadPoolExecutor(max_workers=max_workers)
+    results = list(
+        track(
+            executor.map(lambda line: scramble_file(files, line, dstdir, args), range(len(files))),
+            total=len(files),
+            description="Scrambling selected benchmarks...",
+        )
+    )
+    executor.shutdown()