vla / scripts /generate_tasks.py
anhtld's picture
Initial commit: DoVLA-CIL codebase (h=16 breakthrough)
adc02fa verified
Raw
History Blame Contribute Delete
2.24 kB
#!/usr/bin/env python
from __future__ import annotations
import argparse
import os
import sys
from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
from dovla_cil.tasks.predicates import SUPPORTED_PREDICATES # noqa: E402
from dovla_cil.utils.io import write_jsonl
from dovla_cil.utils.seeding import seed_everything
from dovla_cil.vlm.client import VLMClient
from dovla_cil.vlm.task_generator import (
TaskGenerationRequest,
TaskGenerator,
default_task_generation_request,
)
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Generate VLM-assisted DoVLA-CIL task specs.")
parser.add_argument("--num-tasks", type=int, default=8)
parser.add_argument(
"--families",
nargs="*",
default=None,
help="Task families as space-separated or comma-separated values.",
)
parser.add_argument("--out", type=Path, default=Path("outputs/tasks.jsonl"))
parser.add_argument("--mock", action="store_true", help="Use deterministic mock generation.")
parser.add_argument("--seed", type=int, default=17)
args = parser.parse_args(argv)
seed_everything(args.seed)
if args.mock:
os.environ["OPENCLAUDE_MOCK"] = "1"
defaults = default_task_generation_request(num_tasks=args.num_tasks)
families = _parse_list(args.families) or defaults.families
request = TaskGenerationRequest(
num_tasks=args.num_tasks,
families=families,
object_categories=defaults.object_categories,
relations=sorted(SUPPORTED_PREDICATES),
constraints=[f"seed={args.seed}"],
)
generator = TaskGenerator(VLMClient())
tasks = generator.generate_tasks(request)
write_jsonl((task.to_dict() for task in tasks), args.out)
print(f"Wrote {len(tasks)} tasks to {args.out}")
return 0
def _parse_list(values: list[str] | None) -> list[str]:
if not values:
return []
parsed: list[str] = []
for value in values:
parsed.extend(item.strip() for item in value.split(",") if item.strip())
return parsed
if __name__ == "__main__":
raise SystemExit(main())