| |
| 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 |
| 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()) |
|
|