| |
| """ |
| Batch-evaluate the simplified NL -> UPPAAL pipeline on specs S11-S20. |
| |
| Runs one spec at a time; appends each row to the output CSV immediately so |
| progress is never lost. |
| |
| python scripts/run_nl_query_simple_batch.py |
| python scripts/run_nl_query_simple_batch.py --specs 11 12 13 |
| python scripts/run_nl_query_simple_batch.py --out artifacts/ours/trans_query.csv |
| python scripts/run_nl_query_simple_batch.py --model gpt-4o --sleep 2 |
| |
| Output CSV columns: |
| spec_id, nl_query, uppaal_query, result |
| |
| result values: ok | repair_ok | compile_fail | validation_fail | no_output |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import csv |
| import sys |
| import time |
| from pathlib import Path |
|
|
| ROOT = Path(__file__).resolve().parents[2] |
| sys.path.insert(0, str(ROOT / "src")) |
|
|
| from frame.pipeline.model_checking_pipeline import ModelCheckingPipeline |
| from frame.pipeline.nl_query_simple import translate_nl_to_uppaal_query |
| from frame.pipeline.nl_uppaal_query import parse_verifyta_text_verdict |
| from frame.rag_component.llm import LLM |
|
|
| EVAL_CSV = ROOT / "datasets" / "query_translation_eval.csv" |
| GOLD_DIR = ROOT / "datasets" / "gold_models" |
| DEFAULT_OUT = ROOT / "artifacts" / "ours" / "trans_query.csv" |
|
|
| SYSTEM_PROMPT = ( |
| "You are an expert in formal verification with UPPAAL timed automata. " |
| "Your only task is to produce valid UPPAAL CTL query formulas. " |
| "Follow the rules in the user message exactly. " |
| "Output exactly one line: the UPPAAL formula, nothing else." |
| ) |
|
|
| FIELDNAMES = ["spec_id", "nl_query", "uppaal_query", "status", "verdict", "expected", "correct"] |
|
|
|
|
| def load_queries(spec_ids: list[int]) -> dict[int, list[dict]]: |
| rows: dict[int, list[dict]] = {sid: [] for sid in spec_ids} |
| with EVAL_CSV.open(encoding="utf-8", newline="") as f: |
| for row in csv.DictReader(f): |
| sid = int(row["spec_id"]) |
| if sid in rows: |
| rows[sid].append(row) |
| return rows |
|
|
|
|
| def already_done(out_path: Path, spec_id: int) -> set[str]: |
| """Return set of nl_query strings already written for this spec_id.""" |
| if not out_path.is_file(): |
| return set() |
| done: set[str] = set() |
| with out_path.open(encoding="utf-8", newline="") as f: |
| for row in csv.DictReader(f): |
| if row.get("spec_id") and int(row["spec_id"]) == spec_id: |
| done.add(row.get("nl_query", "").strip()) |
| return done |
|
|
|
|
| def _verify(formula: str, xml_path: Path) -> str: |
| """Run verifyta and return 'T', 'F', or '?'.""" |
| if not formula: |
| return "?" |
| from frame.pipeline.model_checking_pipeline import ModelCheckingPipeline |
| from frame.pipeline.nl_uppaal_query import parse_verifyta_text_verdict |
| res, _, errs = ModelCheckingPipeline(str(xml_path)).verify(formula) |
| v = parse_verifyta_text_verdict(res, errors=errs) |
| return {"satisfied": "T", "not_satisfied": "F"}.get(v, "?") |
|
|
|
|
| def ensure_header(out_path: Path) -> None: |
| if out_path.is_file() and out_path.stat().st_size > 0: |
| return |
| out_path.parent.mkdir(parents=True, exist_ok=True) |
| with out_path.open("w", encoding="utf-8", newline="") as f: |
| csv.DictWriter(f, fieldnames=FIELDNAMES).writeheader() |
|
|
|
|
| def append_rows(out_path: Path, rows: list[dict]) -> None: |
| with out_path.open("a", encoding="utf-8", newline="") as f: |
| w = csv.DictWriter(f, fieldnames=FIELDNAMES) |
| for row in rows: |
| w.writerow(row) |
|
|
|
|
| def verdict_label(v: str) -> str: |
| return {"satisfied": "T", "not_satisfied": "F"}.get(v, "?") |
|
|
|
|
| def main() -> int: |
| ap = argparse.ArgumentParser(description=__doc__, |
| formatter_class=argparse.RawDescriptionHelpFormatter) |
| ap.add_argument("--specs", type=int, nargs="+", default=list(range(11, 21)), |
| help="Spec IDs to run (default: 11-20)") |
| ap.add_argument("--out", type=Path, default=DEFAULT_OUT, |
| help="Output CSV path") |
| ap.add_argument("--model", type=str, default="gpt-4o-mini", |
| help="LLM model id") |
| ap.add_argument("--sleep", type=float, default=0.5, |
| help="Seconds to pause between queries (rate-limit safety)") |
| ap.add_argument("--verbose", action="store_true") |
| ap.add_argument("--resume", action="store_true", default=True, |
| help="Skip already-completed rows (default: True)") |
| args = ap.parse_args() |
|
|
| ensure_header(args.out) |
|
|
| llm = LLM( |
| system_prompt=SYSTEM_PROMPT, |
| model_name=args.model, |
| max_tokens=300, |
| ) |
|
|
| query_map = load_queries(args.specs) |
|
|
| total_q = sum(len(v) for v in query_map.values()) |
| total_correct = 0 |
| total_done = 0 |
| total_compile_ok = 0 |
|
|
| for sid in sorted(args.specs): |
| rows = query_map.get(sid, []) |
| if not rows: |
| print(f"[S{sid:02d}] No queries found in eval CSV -- skipping") |
| continue |
|
|
| xml_path = GOLD_DIR / f"S{sid:02d}" / "model.xml" |
| if not xml_path.is_file(): |
| print(f"[S{sid:02d}] Gold model not found: {xml_path} -- skipping") |
| continue |
|
|
| done_set = already_done(args.out, sid) if args.resume else set() |
| checker = ModelCheckingPipeline(str(xml_path)) |
|
|
| print(f"\n[S{sid:02d}] {len(rows)} queries model={xml_path.parent.name}") |
| print(f" {'#':>2} {'status':<14} {'got':>3} {'exp':>3} formula") |
| print(" " + "-" * 80) |
|
|
| spec_rows_out: list[dict] = [] |
| spec_correct = 0 |
|
|
| for i, row in enumerate(rows, 1): |
| nl = row["nl_query"].strip() |
| expected = row.get("answer", "").strip().upper() |
|
|
| if nl in done_set: |
| print(f" {i:>2} {'(skipped)':<14} -- -- (already in output)") |
| continue |
|
|
| formula, status = translate_nl_to_uppaal_query( |
| nl, xml_path, llm=llm, verbose=args.verbose |
| ) |
|
|
| if formula: |
| res, _t, errs = checker.verify(formula) |
| v = parse_verifyta_text_verdict(res, errors=errs) |
| got = verdict_label(v) |
| if v in ("satisfied", "not_satisfied"): |
| total_compile_ok += 1 |
| else: |
| got = "?" |
|
|
| is_correct = (got == expected) |
| if is_correct: |
| spec_correct += 1 |
| mark = "+" if is_correct else "x" |
|
|
| print( |
| f" {i:>2} {status:<14} {got:>3} {expected:>3} [{mark}] " |
| f"{(formula or '(none)')[:65]}" |
| ) |
|
|
| spec_rows_out.append({ |
| "spec_id": sid, |
| "nl_query": nl, |
| "uppaal_query": formula or "", |
| "status": status, |
| "verdict": got, |
| "expected": expected, |
| "correct": "Y" if is_correct else "N", |
| }) |
|
|
| total_done += 1 |
|
|
| |
| append_rows(args.out, [spec_rows_out[-1]]) |
|
|
| if args.sleep > 0: |
| time.sleep(args.sleep) |
|
|
| pct = 100 * spec_correct // len(rows) if rows else 0 |
| print(f" {'':>2} S{sid:02d} score: {spec_correct}/{len(rows)} ({pct}%)") |
| total_correct += spec_correct |
|
|
| |
| print("\n" + "=" * 60) |
| grand_pct = 100 * total_correct // total_done if total_done else 0 |
| qcr_pct = 100 * total_compile_ok // total_done if total_done else 0 |
| print(f"Overall {total_correct}/{total_done} correct ({grand_pct}%) " |
| f"QCR={qcr_pct}% written -> {args.out}") |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
