File size: 2,679 Bytes
adc02fa | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 | from __future__ import annotations
from dovla_cil.tasks.predicates import SUPPORTED_PREDICATES
from dovla_cil.tasks.schema import TaskSpec
def validate_task(task: TaskSpec) -> None:
if not task.instruction_templates:
raise ValueError(f"Task {task.task_id!r} must have at least one instruction")
if any(not template.strip() for template in task.instruction_templates):
raise ValueError(f"Task {task.task_id!r} has an empty instruction template")
object_ids = [obj.object_id for obj in task.objects]
object_id_set = set(object_ids)
if len(object_ids) != len(object_id_set):
raise ValueError(f"Task {task.task_id!r} has duplicate object ids")
_validate_object_refs(task, "target_object_ids", task.target_object_ids, object_id_set)
_validate_object_refs(task, "reference_object_ids", task.reference_object_ids, object_id_set)
_validate_object_refs(task, "distractor_object_ids", task.distractor_object_ids, object_id_set)
for predicate in task.success_predicates:
if predicate.name not in SUPPORTED_PREDICATES:
raise ValueError(f"Task {task.task_id!r} uses unsupported predicate {predicate.name!r}")
for arg in predicate.args:
if arg not in object_id_set:
raise ValueError(
f"Task {task.task_id!r} predicate {predicate.as_expression()} "
f"references unknown object {arg!r}"
)
_validate_minimal_pairs(task, object_id_set)
def _validate_object_refs(
task: TaskSpec, field_name: str, values: list[str], object_id_set: set[str]
) -> None:
for object_id in values:
if object_id not in object_id_set:
raise ValueError(
f"Task {task.task_id!r} field {field_name} references unknown object {object_id!r}"
)
def _validate_minimal_pairs(task: TaskSpec, object_id_set: set[str]) -> None:
valid_relation_values = SUPPORTED_PREDICATES | {
predicate.name for predicate in task.success_predicates
}
for key in ("target_object", "reference_object"):
for value in task.minimal_pair_factors.get(key, []):
if value not in object_id_set:
raise ValueError(
f"Task {task.task_id!r} minimal_pair_factors[{key!r}] "
f"references unknown object {value!r}"
)
for relation in task.minimal_pair_factors.get("relation", []):
if relation not in valid_relation_values:
raise ValueError(
f"Task {task.task_id!r} minimal_pair_factors['relation'] "
f"uses unsupported relation {relation!r}"
)
|