|
| 1 | +# Copyright 2023 DeepMind Technologies Limited |
| 2 | +# |
| 3 | +# Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 | +# you may not use this file except in compliance with the License. |
| 5 | +# You may obtain a copy of the License at |
| 6 | +# |
| 7 | +# http://www.apache.org/licenses/LICENSE-2.0 |
| 8 | +# |
| 9 | +# Unless required by applicable law or agreed to in writing, software |
| 10 | +# distributed under the License is distributed on an "AS IS" BASIS, |
| 11 | +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 12 | +# See the License for the specific language governing permissions and |
| 13 | +# limitations under the License. |
| 14 | +# ============================================================================== |
| 15 | +"""Collision primitives – MLX port.""" |
| 16 | + |
| 17 | +from typing import Tuple |
| 18 | + |
| 19 | +import mlx.core as mx |
| 20 | +from mujoco.mjx_mlx._src import math |
| 21 | +# pylint: disable=g-importing-member |
| 22 | +from mujoco.mjx_mlx._src.collision_types import Collision |
| 23 | +from mujoco.mjx_mlx._src.collision_types import GeomInfo |
| 24 | +from mujoco.mjx_mlx._src.types import Data |
| 25 | +from mujoco.mjx_mlx._src.types import Model |
| 26 | +# pylint: enable=g-importing-member |
| 27 | + |
| 28 | + |
| 29 | +def _concat_tree(*collisions): |
| 30 | + """Concatenate collision tuples (dist, pos, frame) along axis 0.""" |
| 31 | + dists = mx.concatenate([c[0] for c in collisions]) |
| 32 | + poss = mx.concatenate([c[1] for c in collisions]) |
| 33 | + frames = mx.concatenate([c[2] for c in collisions]) |
| 34 | + return dists, poss, frames |
| 35 | + |
| 36 | + |
| 37 | +def collider(ncon: int): |
| 38 | + """Wraps collision functions for use by collision_driver.""" |
| 39 | + |
| 40 | + def wrapper(func): |
| 41 | + def collide(m: Model, d: Data, _, geom: mx.array) -> Collision: |
| 42 | + g1, g2 = geom.T[0], geom.T[1] |
| 43 | + # Build per-pair info and call func in a loop (replaces jax.vmap) |
| 44 | + n_pairs = g1.shape[0] if hasattr(g1, 'shape') and len(g1.shape) > 0 else 1 |
| 45 | + results = [] |
| 46 | + for i in range(n_pairs): |
| 47 | + gi1 = int(g1[i]) if n_pairs > 1 else int(g1) |
| 48 | + gi2 = int(g2[i]) if n_pairs > 1 else int(g2) |
| 49 | + info1 = GeomInfo(d.geom_xpos[gi1], d.geom_xmat[gi1], m.geom_size[gi1]) |
| 50 | + info2 = GeomInfo(d.geom_xpos[gi2], d.geom_xmat[gi2], m.geom_size[gi2]) |
| 51 | + dist_i, pos_i, frame_i = func(info1, info2) |
| 52 | + # ensure batch dims |
| 53 | + if len(dist_i.shape) == 0: |
| 54 | + dist_i = mx.expand_dims(dist_i, axis=0) |
| 55 | + if len(pos_i.shape) == 1: |
| 56 | + pos_i = mx.expand_dims(pos_i, axis=0) |
| 57 | + if len(frame_i.shape) == 2: |
| 58 | + frame_i = mx.expand_dims(frame_i, axis=0) |
| 59 | + results.append((dist_i, pos_i, frame_i)) |
| 60 | + if ncon > 1: |
| 61 | + return _concat_tree(*results) |
| 62 | + # ncon == 1: stack across pairs |
| 63 | + dist = mx.concatenate([r[0] for r in results]) |
| 64 | + pos = mx.concatenate([r[1] for r in results]) |
| 65 | + frame = mx.concatenate([r[2] for r in results]) |
| 66 | + return dist, pos, frame |
| 67 | + |
| 68 | + collide.ncon = ncon |
| 69 | + return collide |
| 70 | + |
| 71 | + return wrapper |
| 72 | + |
| 73 | + |
| 74 | +def _plane_sphere( |
| 75 | + plane_normal: mx.array, |
| 76 | + plane_pos: mx.array, |
| 77 | + sphere_pos: mx.array, |
| 78 | + sphere_radius: mx.array, |
| 79 | +) -> Tuple[mx.array, mx.array]: |
| 80 | + """Returns the distance and contact point between a plane and sphere.""" |
| 81 | + dist = mx.sum((sphere_pos - plane_pos) * plane_normal) - sphere_radius |
| 82 | + pos = sphere_pos - plane_normal * (sphere_radius + 0.5 * dist) |
| 83 | + return dist, pos |
| 84 | + |
| 85 | + |
| 86 | +@collider(ncon=1) |
| 87 | +def plane_sphere(plane: GeomInfo, sphere: GeomInfo) -> Collision: |
| 88 | + """Calculates contact between a plane and a sphere.""" |
| 89 | + n = plane.mat[:, 2] |
| 90 | + dist, pos = _plane_sphere(n, plane.pos, sphere.pos, sphere.size[0]) |
| 91 | + return dist, pos, math.make_frame(n) |
| 92 | + |
| 93 | + |
| 94 | +@collider(ncon=2) |
| 95 | +def plane_capsule(plane: GeomInfo, cap: GeomInfo) -> Collision: |
| 96 | + """Calculates two contacts between a capsule and a plane.""" |
| 97 | + n, axis = plane.mat[:, 2], cap.mat[:, 2] |
| 98 | + # align contact frames with capsule axis |
| 99 | + b, b_norm = math.normalize_with_norm(axis - n * mx.sum(n * axis)) |
| 100 | + y, z = mx.array([0.0, 1.0, 0.0]), mx.array([0.0, 0.0, 1.0]) |
| 101 | + b = mx.where(b_norm < 0.5, mx.where((-0.5 < n[1]) & (n[1] < 0.5), y, z), b) |
| 102 | + frame = mx.array([mx.stack([n, b, math._cross(n, b)])]) |
| 103 | + segment = axis * cap.size[1] |
| 104 | + collisions = [] |
| 105 | + for offset in [segment, -segment]: |
| 106 | + dist, pos = _plane_sphere(n, plane.pos, cap.pos + offset, cap.size[0]) |
| 107 | + dist = mx.expand_dims(dist, axis=0) |
| 108 | + pos = mx.expand_dims(pos, axis=0) |
| 109 | + collisions.append((dist, pos, frame)) |
| 110 | + return _concat_tree(*collisions) |
| 111 | + |
| 112 | + |
| 113 | +@collider(ncon=1) |
| 114 | +def plane_ellipsoid(plane: GeomInfo, ellipsoid: GeomInfo) -> Collision: |
| 115 | + """Calculates one contact between an ellipsoid and a plane.""" |
| 116 | + n = plane.mat[:, 2] |
| 117 | + size = ellipsoid.size |
| 118 | + sphere_support = -math.normalize((ellipsoid.mat.T @ n) * size) |
| 119 | + pos = ellipsoid.pos + ellipsoid.mat @ (sphere_support * size) |
| 120 | + dist = mx.sum(n * (pos - plane.pos)) |
| 121 | + pos = pos - n * dist * 0.5 |
| 122 | + return dist, pos, math.make_frame(n) |
| 123 | + |
| 124 | + |
| 125 | +@collider(ncon=3) |
| 126 | +def plane_cylinder(plane: GeomInfo, cylinder: GeomInfo) -> Collision: |
| 127 | + """Calculates three contacts between a cylinder and a plane.""" |
| 128 | + n = plane.mat[:, 2] |
| 129 | + axis = cylinder.mat[:, 2] |
| 130 | + |
| 131 | + # make sure axis points towards plane |
| 132 | + prjaxis = mx.sum(n * axis) |
| 133 | + sign = -math.sign(prjaxis) |
| 134 | + axis, prjaxis = axis * sign, prjaxis * sign |
| 135 | + |
| 136 | + # compute normal distance to cylinder center |
| 137 | + dist0 = mx.sum((cylinder.pos - plane.pos) * n) |
| 138 | + |
| 139 | + # remove component of -normal along axis, compute length |
| 140 | + vec = axis * prjaxis - n |
| 141 | + len_ = math.norm(vec) |
| 142 | + |
| 143 | + vec = mx.where( |
| 144 | + len_ < 1e-12, |
| 145 | + # disk parallel to plane: pick x-axis of cylinder, scale by radius |
| 146 | + cylinder.mat[:, 0] * cylinder.size[0], |
| 147 | + # general configuration: normalize vector, scale by radius |
| 148 | + math.safe_div(vec, len_) * cylinder.size[0], |
| 149 | + ) |
| 150 | + |
| 151 | + # project vector on normal |
| 152 | + prjvec = mx.sum(vec * n) |
| 153 | + |
| 154 | + # scale axis by half-length |
| 155 | + axis = axis * cylinder.size[1] |
| 156 | + prjaxis = prjaxis * cylinder.size[1] |
| 157 | + |
| 158 | + # compute sideways vector: vec1 |
| 159 | + prjvec1 = -prjvec * 0.5 |
| 160 | + vec1 = math.normalize(math._cross(vec, axis)) * cylinder.size[0] |
| 161 | + vec1 = vec1 * mx.sqrt(mx.array(3.0)) * 0.5 |
| 162 | + |
| 163 | + # disk parallel to plane |
| 164 | + d1 = dist0 + prjaxis + prjvec |
| 165 | + d2 = dist0 + prjaxis + prjvec1 |
| 166 | + dist = mx.array([d1, d2, d2]) |
| 167 | + pos = ( |
| 168 | + cylinder.pos |
| 169 | + + axis |
| 170 | + + mx.stack([ |
| 171 | + vec - n * d1 * 0.5, |
| 172 | + vec1 + vec * -0.5 - n * d2 * 0.5, |
| 173 | + -vec1 + vec * -0.5 - n * d2 * 0.5, |
| 174 | + ]) |
| 175 | + ) |
| 176 | + |
| 177 | + # cylinder parallel to plane |
| 178 | + cond = mx.abs(prjaxis) < 1e-3 |
| 179 | + d3 = dist0 - prjaxis + prjvec |
| 180 | + # dist[1] = d3 if cond |
| 181 | + dist_list = [dist[0], mx.where(cond, d3, dist[1]), dist[2]] |
| 182 | + dist = mx.stack(dist_list) |
| 183 | + # pos[1] if cond |
| 184 | + new_pos1 = cylinder.pos + vec - axis - n * d3 * 0.5 |
| 185 | + pos_rows = [pos[0], mx.where(cond, new_pos1, pos[1]), pos[2]] |
| 186 | + pos = mx.stack(pos_rows) |
| 187 | + |
| 188 | + frame = mx.stack([math.make_frame(n)] * 3, axis=0) |
| 189 | + return dist, pos, frame |
| 190 | + |
| 191 | + |
| 192 | +def _sphere_sphere( |
| 193 | + pos1: mx.array, radius1: mx.array, pos2: mx.array, radius2: mx.array |
| 194 | +) -> Tuple[mx.array, mx.array, mx.array]: |
| 195 | + """Returns the penetration, contact point, and normal between two spheres.""" |
| 196 | + n, dist = math.normalize_with_norm(pos2 - pos1) |
| 197 | + n = mx.where(dist == 0.0, mx.array([1.0, 0.0, 0.0]), n) |
| 198 | + dist = dist - (radius1 + radius2) |
| 199 | + pos = pos1 + n * (radius1 + dist * 0.5) |
| 200 | + return dist, pos, n |
| 201 | + |
| 202 | + |
| 203 | +@collider(ncon=1) |
| 204 | +def sphere_sphere(s1: GeomInfo, s2: GeomInfo) -> Collision: |
| 205 | + """Calculates contact between two spheres.""" |
| 206 | + dist, pos, n = _sphere_sphere(s1.pos, s1.size[0], s2.pos, s2.size[0]) |
| 207 | + return dist, pos, math.make_frame(n) |
| 208 | + |
| 209 | + |
| 210 | +@collider(ncon=1) |
| 211 | +def sphere_capsule(sphere: GeomInfo, cap: GeomInfo) -> Collision: |
| 212 | + """Calculates one contact between a sphere and a capsule.""" |
| 213 | + axis, length = cap.mat[:, 2], cap.size[1] |
| 214 | + segment = axis * length |
| 215 | + pt = math.closest_segment_point( |
| 216 | + cap.pos - segment, cap.pos + segment, sphere.pos |
| 217 | + ) |
| 218 | + dist, pos, n = _sphere_sphere(sphere.pos, sphere.size[0], pt, cap.size[0]) |
| 219 | + return dist, pos, math.make_frame(n) |
| 220 | + |
| 221 | + |
| 222 | +@collider(ncon=1) |
| 223 | +def capsule_capsule(cap1: GeomInfo, cap2: GeomInfo) -> Collision: |
| 224 | + """Calculates one contact between two capsules.""" |
| 225 | + axis1, length1 = cap1.mat[:, 2], cap1.size[1] |
| 226 | + axis2, length2 = cap2.mat[:, 2], cap2.size[1] |
| 227 | + seg1, seg2 = axis1 * length1, axis2 * length2 |
| 228 | + pt1, pt2 = math.closest_segment_to_segment_points( |
| 229 | + cap1.pos - seg1, |
| 230 | + cap1.pos + seg1, |
| 231 | + cap2.pos - seg2, |
| 232 | + cap2.pos + seg2, |
| 233 | + ) |
| 234 | + radius1, radius2 = cap1.size[0], cap2.size[0] |
| 235 | + dist, pos, n = _sphere_sphere(pt1, radius1, pt2, radius2) |
| 236 | + return dist, pos, math.make_frame(n) |
0 commit comments