dovla_cil/tasks/validators.py

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}"
            )