Skip to content

touching

Touching

Bases: TensorizedRelativeState, KinematicsMixin, BooleanStateMixin

Pairwise rigid-body Touching state.

Cloth-rigid and cloth-cloth pairs go through _get_value's cloth fallback, not the tensorized pipeline.

Source code in OmniGibson/omnigibson/object_states/touching.py
class Touching(TensorizedRelativeState, KinematicsMixin, BooleanStateMixin):
    """
    Pairwise rigid-body Touching state.

    Cloth-rigid and cloth-cloth pairs go through `_get_value`'s cloth fallback,
    not the tensorized pipeline.
    """

    # Per-scene contact mask tables (rebuilt in initialize_view, sized to that scene's R_s / C_s).
    # Each entry covers N rows where N = len(OBJ_IDXS); rows for cloth / absent-in-scene objects are zero.
    _obj_row_mask = None  # list[wp.array2d(N, R_s) uint8 | None]
    _obj_col_mask = None  # list[wp.array2d(N, C_s) uint8 | None]

    # Stage 1 scratch: per-scene (N, C_s) int32 atomic_max target.
    _obj_to_col = None  # list[wp.array2d(N, C_s) int32 | None]

    # Stage 2 scratch: single (S, N, N) int32 atomic_max target + per-scene 2D slice views.
    _pair_scratch = None  # wp.array3d (S, N, N) int32
    _pair_scratch_per_scene = None  # list[wp.array2d(int32)] — row slices of _pair_scratch

    @classproperty
    def value_shape(cls):
        return ()

    @classproperty
    def value_type(cls):
        return th.bool

    @classproperty
    def value_name(cls):
        return "touching"

    @staticmethod
    def _check_cloth_contact(cloth_obj, other_obj):
        other_link_paths = set(other_obj.link_prim_paths)
        return any(contact_prim_path in other_link_paths for contact_prim_path, _ in cloth_obj.root_link.get_contacts())

    @classmethod
    def global_initialize(cls):
        super().global_initialize()
        cls._obj_row_mask = None
        cls._obj_col_mask = None
        cls._obj_to_col = None
        cls._pair_scratch = None
        cls._pair_scratch_per_scene = None

    @classmethod
    def initialize_view(cls):
        # Base class rebuilds OBJ_IDXS, IDX_OBJS, VALUES (with value carry-over).
        super().initialize_view()

        S = len(cls.IDX_OBJS)
        N = len(cls.OBJ_IDXS)

        cls._obj_row_mask = []
        cls._obj_col_mask = []
        cls._obj_to_col = []
        cls._pair_scratch_per_scene = []

        if S == 0 or N == 0:
            cls._pair_scratch = None
            return

        # Allocate single (S, N, N) int32 scratch for stage 2 output; per-scene 2D row slices share storage.
        cls._pair_scratch = wp.zeros((S, N, N), dtype=wp.int32, device="cuda")

        for scene_idx, scene_row in enumerate(cls.IDX_OBJS):
            cls._pair_scratch_per_scene.append(cls._pair_scratch[scene_idx])

            shape = RigidContactAPI.get_contact_matrix_shape(scene_idx)
            if shape is None:
                cls._obj_row_mask.append(None)
                cls._obj_col_mask.append(None)
                cls._obj_to_col.append(None)
                continue

            R_s, C_s = shape

            row_mask_cpu = th.zeros((N, R_s), dtype=th.uint8)
            col_mask_cpu = th.zeros((N, C_s), dtype=th.uint8)

            for obj_idx, obj in enumerate(scene_row):
                if obj is None:
                    continue
                if obj.prim_type == PrimType.CLOTH:
                    # Cloth gets handled by the _get_value fallback; rigid kernel sees zero masks.
                    continue
                link_paths = [link.prim_path for link in obj.links.values()]
                row_mask_cpu[obj_idx] = RigidContactAPI.get_contact_row_mask(scene_idx, link_paths).to(th.uint8)
                col_mask_cpu[obj_idx] = RigidContactAPI.get_contact_col_mask(scene_idx, link_paths).to(th.uint8)

            cls._obj_row_mask.append(
                lazy.isaacsim.core.utils.warp.tensor.create_tensor_from_list(row_mask_cpu, "uint8", device="cuda")
            )
            cls._obj_col_mask.append(
                lazy.isaacsim.core.utils.warp.tensor.create_tensor_from_list(col_mask_cpu, "uint8", device="cuda")
            )
            cls._obj_to_col.append(wp.zeros((N, C_s), dtype=wp.int32, device="cuda"))

    @classmethod
    def _update_values(cls, values):
        if cls._pair_scratch is None or cls.VALUES_WP is None:
            return

        S, N, _ = values.shape
        if S == 0 or N == 0:
            return

        # Zero the (S, N, N) pair scratch in one shot.
        cls._pair_scratch.zero_()

        # Per-scene Stage 1 + Stage 2 launches.
        for scene_idx in range(S):
            obj_to_col = cls._obj_to_col[scene_idx]
            if obj_to_col is None:
                continue
            row_mask = cls._obj_row_mask[scene_idx]
            col_mask = cls._obj_col_mask[scene_idx]
            contact_matrix_wp = RigidContactAPI.get_contact_matrix_wp(scene_idx, current_only=True)
            if contact_matrix_wp is None:
                continue
            R_s = contact_matrix_wp.shape[0]
            C_s = contact_matrix_wp.shape[1]

            # Zero stage-1 scratch.
            obj_to_col.zero_()

            # Stage 1: (N, R_s, C_s) — obj_to_col[i, c] |= row_mask[i, r] & contact[r, c]
            wp.launch(
                kernel=_touching_obj_to_col_kernel,
                dim=(N, R_s, C_s),
                inputs=[row_mask, contact_matrix_wp, obj_to_col],
                device="cuda",
            )

            # Stage 2: (N, N, C_s) — pair[i, j] |= obj_to_col[i, c] & col_mask[j, c]
            wp.launch(
                kernel=_touching_pair_kernel,
                dim=(N, N, C_s),
                inputs=[obj_to_col, col_mask, cls._pair_scratch_per_scene[scene_idx]],
                device="cuda",
            )

        # Stage 3: symmetrize, zero diagonal, int32 → uint8 into VALUES.
        wp.launch(
            kernel=_touching_finalize_kernel,
            dim=(S, N, N),
            inputs=[cls._pair_scratch, cls.VALUES_WP],
            device="cuda",
        )

    def _get_value(self, other):
        # Cloth path: tensorized pipeline doesn't see cloth contacts (cloth isn't in the
        # rigid contact matrix), so fall back to the per-call PhysX get_contacts() check.
        self_is_cloth = self.obj.prim_type == PrimType.CLOTH
        other_is_cloth = other.prim_type == PrimType.CLOTH
        if self_is_cloth and other_is_cloth:
            raise ValueError("Cannot detect contact between two cloth objects.")
        if self_is_cloth:
            return self._check_cloth_contact(self.obj, other)
        if other_is_cloth:
            return self._check_cloth_contact(other, self.obj)

        return super()._get_value(other)