(task_dir: Path)
| 161 | |
| 162 | |
| 163 | def load_task_spec(task_dir: Path) -> TaskSpec: |
| 164 | task_yaml = _load_yaml(task_dir / "task.yaml") |
| 165 | object_universe = _load_yaml(task_dir / "object_universe.yaml") |
| 166 | asset_bindings_path = task_dir / "asset_bindings.yaml" |
| 167 | asset_bindings = _load_yaml(asset_bindings_path) if asset_bindings_path.exists() else {} |
| 168 | base_variation_yaml = _load_yaml(task_dir / "base_variation.yaml") |
| 169 | task_instruction = str(task_yaml.get("instruction") or "") |
| 170 | |
| 171 | object_instances: dict[str, ObjectInstanceSpec] = {} |
| 172 | |
| 173 | for item in object_universe.get("container", []) or []: |
| 174 | instance_id = str(item["instance_id"]) |
| 175 | object_instances[instance_id] = ObjectInstanceSpec( |
| 176 | instance_id=instance_id, |
| 177 | semantic_name=str(item.get("semantic_name") or item.get("role") or "container"), |
| 178 | retrieval_query=str(item.get("retrieval_query") or ""), |
| 179 | description=str(item.get("description") or ""), |
| 180 | tags=_as_string_tuple(item.get("tags")), |
| 181 | role=str(item.get("role") or "container"), |
| 182 | ) |
| 183 | |
| 184 | for key in ("goal_objects", "fail_objects", "decor_objects"): |
| 185 | for item in object_universe.get(key, []) or []: |
| 186 | instance_id = str(item["instance_id"]) |
| 187 | object_instances[instance_id] = ObjectInstanceSpec( |
| 188 | instance_id=instance_id, |
| 189 | semantic_name=str(item.get("semantic_name") or instance_id), |
| 190 | retrieval_query=str(item.get("retrieval_query") or ""), |
| 191 | description=str(item.get("description") or ""), |
| 192 | tags=_as_string_tuple(item.get("tags")), |
| 193 | ) |
| 194 | |
| 195 | semantic_groups = { |
| 196 | "container": tuple(str(item["instance_id"]) for item in object_universe.get("container", []) or []), |
| 197 | "target": tuple(str(item["instance_id"]) for item in object_universe.get("goal_objects", []) or []), |
| 198 | "distractor": tuple(str(item["instance_id"]) for item in object_universe.get("fail_objects", []) or []), |
| 199 | "decor": tuple(str(item["instance_id"]) for item in object_universe.get("decor_objects", []) or []), |
| 200 | } |
| 201 | |
| 202 | structured_asset_bindings = _asset_bindings_from_payload(asset_bindings) |
| 203 | candidate_pools = { |
| 204 | instance_id: tuple(candidate.source_uid for candidate in candidates) |
| 205 | for instance_id, candidates in structured_asset_bindings.items() |
| 206 | } |
| 207 | |
| 208 | defaults = task_yaml.get("defaults", {}) or {} |
| 209 | container_instance_id = str(defaults.get("container_instance_id") or semantic_groups["container"][0]) |
| 210 | primary_goal_relation = str(defaults.get("primary_goal_relation") or "in") |
| 211 | fail_relations = _as_string_tuple((defaults.get("fail_policy", {}) or {}).get("fail_on_relations")) or ("in", "on") |
| 212 | |
| 213 | default_policy = DefaultPolicySpec( |
| 214 | success=PolicyChannelSpec( |
| 215 | mode="all_of", |
| 216 | predicates=(PredicateTemplateSpec(subjects_from="target", relation=primary_goal_relation, target_id=container_instance_id),), |
| 217 | ), |
| 218 | fail=PolicyChannelSpec( |
| 219 | mode="any_of", |
| 220 | predicates=(PredicateTemplateSpec(subjects_from="distractor", relations=fail_relations, target_id=container_instance_id),), |
no test coverage detected