"""Tier 2: auto-infer a schema from info.json `features[*].names`.""" from __future__ import annotations import logging from pathlib import Path from typing import Any, Optional from .arm_layout import ArmCount, ArmLayoutSpec from .camera_mapping import infer_image_mapping_from_info from .dataset_schema import DatasetSchema logger = logging.getLogger(__name__) # Case-insensitive substring tokens that mark a dimension as "gripper-like" # (absolute target instead of delta). "finger" alone is too permissive — it # matches "fingertip_force_sensor" and "finger_joint_position" which are # arm/sensor dims, not grippers. We only accept "finger" if it's paired with # a closure/position indicator. GRIPPER_TOKENS = ("gripper", "effector") GRIPPER_FINGER_CONTEXTS = ("finger_position", "finger_pose", "finger_angle", "finger_open", "finger_close", "finger_width") # Substrings that *suggest* a dim is gripper/finger-related even if it did not # satisfy the strict ``_is_gripper_name`` whitelist above. Used by the # gripper-inference fallback to tell apart two failure modes when no strict # match is found: # (a) genuinely grippered-but-mislabeled data ("finger" without a recognized # closure/position context, or an exotic "grip"/"claw"/"hand" spelling) # → fail-loud, because silently treating an absolute gripper dim as delta # corrupts the delta transform / stats / supervision. # (b) a robot that genuinely has no gripper channel (e.g. dex-hand robots # whose dims are individual finger joints) → all-delta is correct. # A name containing any of these tokens is treated as "looks grippered" (case a). GRIPPER_HINT_TOKENS = ("gripper", "effector", "finger", "grip", "claw", "hand", "jaw", "pinch") def _match_names_to_list(names: Any) -> list[str]: """Normalize the various shapes `features[k].names` can take. LeRobot v2.1 / v3.0 `names` may be: - None (unlabeled) - list[str] — per-dim labels - dict[str, list] — e.g. {"motors": ["joint_0", ..., "gripper"]} """ if names is None: return [] if isinstance(names, list): return [str(n) for n in names] if isinstance(names, dict): # pick the first inner list (LeRobot convention: {"motors": [...]}) for v in names.values(): if isinstance(v, list): return [str(x) for x in v] return [] def _is_gripper_name(name: str) -> bool: """Return True if the dim name unambiguously denotes a gripper channel. Rejects "fingertip_force_sensor" / "finger_joint_position" — those have "finger" but no gripper-action context. """ low = name.lower() if any(tok in low for tok in GRIPPER_TOKENS): return True if "finger" in low and any(ctx in low for ctx in GRIPPER_FINGER_CONTEXTS): return True return False def _infer_arm_layout( gripper_action_dims: tuple[int, ...], ) -> Optional[ArmLayoutSpec]: """Best-effort ArmLayoutSpec inference from gripper indices. Heuristic (compatible with the canonical single/dual-arm layouts): - One gripper dim at index ``g ∈ {6, 7}`` → SINGLE arm, ``arm_dof = g``, gripper at raw index ``g``. Matches 7-DoF Franka (grip @ 7) and 6-DoF UR5/WidowX (grip @ 6) with the zero-pad at dim 6. - Two gripper dims ``(g0, g1)`` where ``left = g0 ∈ {5, 6}`` AND ``right = g1 - g0 - 1 ∈ {5, 6}`` → DUAL arm, layout ``[arm_l, grip_l, arm_r, grip_r]`` with ``left_gripper_index_in_raw = g0``, ``right_gripper_index_in_raw = g1``. - Anything else (no gripper, or exotic placement) → ``None``. The caller records arm_layout=None, which disables canonical remapping for this dataset (the gripper_action_dims metadata is still used). Returns: An ``ArmLayoutSpec`` if the heuristic succeeds, else ``None``. """ if len(gripper_action_dims) == 1: g = gripper_action_dims[0] if g in (6, 7): return ArmLayoutSpec( arm_count=ArmCount.SINGLE, arm_dof=g, gripper_index_in_raw=g, ) return None if len(gripper_action_dims) == 2: g0, g1 = gripper_action_dims left_arm_dof = g0 right_arm_dof = g1 - g0 - 1 if left_arm_dof in (5, 6) and right_arm_dof in (5, 6): return ArmLayoutSpec( arm_count=ArmCount.DUAL, left_arm_dof=left_arm_dof, right_arm_dof=right_arm_dof, left_gripper_index_in_raw=g0, right_gripper_index_in_raw=g1, ) return None return None def try_infer_from_info( info: dict, robot_type: Optional[str], root: Path, ) -> Optional[DatasetSchema]: """Tier 2 discovery. Returns None if names are too opaque — caller falls back.""" features = info.get("features") or {} # Derive the key FIRST, then read the feature from that same key, so dims # and the emitted key always come from the same alias when both # ("action"/"actions", "observation.state"/"state") are present. action_key = "action" if "action" in features else "actions" state_key = "observation.state" if "observation.state" in features else "state" action_feat = features.get(action_key) or features.get( "actions" if action_key == "action" else "action" ) state_feat = features.get(state_key) or features.get( "state" if state_key == "observation.state" else "observation.state" ) if not (action_feat and state_feat): logger.debug( "try_infer_from_info: missing 'action' / 'state' feature for %s", root ) return None a_names = _match_names_to_list(action_feat.get("names")) s_names = _match_names_to_list(state_feat.get("names")) # Tier B/C detection: opaque single-label vectors like names=["action"]. if len(a_names) <= 1 or len(s_names) <= 1: logger.debug( "try_infer_from_info: names too opaque for %s (|a|=%d |s|=%d)", root, len(a_names), len(s_names), ) return None # Derive delta mask from per-dim names. delta_mask = tuple(not _is_gripper_name(n) for n in a_names) gripper_action_dims = tuple(i for i, d in enumerate(delta_mask) if not d) if not gripper_action_dims: # No dim satisfied the strict gripper whitelist. Two sub-cases: # (a) The robot genuinely has no gripper (e.g. dex-hand robots like # AIRBOT_MMK2 where all 36 dims are individual finger joints). # In that case all-delta is the semantically correct schema. # (b) The data DOES have a gripper dim, but it was named weirdly # enough that the strict matcher missed it (e.g. "finger" with no # recognized closure context, or an exotic "grip"/"claw"/"jaw" # spelling). # # Do NOT blanket fall back to all-delta: that silently mislabels an # absolute gripper dim as delta in case (b) and corrupts the delta # transform / stats / supervision. Action names are detailed here # (|a_names| > 1 established above), so only continue for the # unambiguous case (a) — no action name even hints at a gripper. If any # name looks grippered but failed the strict whitelist, fail loud and # demand an explicit manifest. looks_grippered = any( any(tok in n.lower() for tok in GRIPPER_HINT_TOKENS) for n in a_names ) if looks_grippered: from .dataset_schema import SchemaDiscoveryError raise SchemaDiscoveryError( f"[schema] auto-infer for {root}: action names look like they " f"contain a gripper/finger channel but none matched the strict " f"gripper-name whitelist, so the gripper dim cannot be " f"identified. Treating it as delta would corrupt the delta " f"transform, stats and supervision. Write a " f"meta/labvla_manifest.json declaring gripper_action_dims " f"explicitly (or rename the dim to a recognized form). " f"Action names: {list(a_names)}" ) # Case (a): genuinely no gripper. all-delta is correct. logger.warning( "try_infer_from_info: no gripper channel detected in action names " "for %s (no name hints at a gripper/finger) — using an all-delta " "schema (every dim treated as delta). If this dataset actually has " "an absolute gripper/finger-position target, write a " "meta/labvla_manifest.json to override. First 5 action names: %s", root, list(a_names)[:5], ) delta_mask = tuple([True] * len(a_names)) gripper_action_dims = tuple() action_shape = action_feat.get("shape") or [len(a_names)] state_shape = state_feat.get("shape") or [len(s_names)] schema_id = f"{robot_type or 'unknown'}:{root.name}" # Best-effort canonical arm layout — lets Tier-2 datasets reuse the # deploy-time forward/reverse mapper without needing a manifest. Falls # back to None when the gripper placement is ambiguous (caller then # trains/deploys using gripper_action_dims metadata directly). arm_layout = _infer_arm_layout(gripper_action_dims) return DatasetSchema( schema_id=schema_id, robot_type=robot_type or "unknown", state_keys=(state_key,), action_keys=(action_key,), state_dims=(int(state_shape[0]),), action_dims=(int(action_shape[0]),), delta_mask=delta_mask, gripper_action_dims=gripper_action_dims, image_mapping=infer_image_mapping_from_info(features), source="info_names", source_path=str(root / "meta" / "info.json"), arm_layout=arm_layout, )