#!/usr/bin/env python3 """ 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 # Write immediately so each query is persisted 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 # ---- Summary ---------------------------------------------------------------- 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())