diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index 7a06f83..2e3b064 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -36,21 +36,67 @@ jobs:
fi
- name: Run pre-commit
uses: pre-commit/action@v3.0.1
+
test:
name: ${{ matrix.os }} / Python ${{ matrix.python-version }}
runs-on: ${{ matrix.os }}
- timeout-minutes: 30
+ timeout-minutes: 45
strategy:
fail-fast: false
matrix:
os: [ubuntu-latest, windows-latest, macos-latest]
python-version: ['3.11', '3.12', '3.13', '3.14']
+ defaults:
+ run:
+ shell: bash -l {0}
+
steps:
- name: Checkout repository (with submodules)
uses: actions/checkout@v4
- - name: Setup Python
+ - name: Detect matching pinokin branch
+ id: pinokin
+ shell: bash
+ run: |
+ BRANCH="${GITHUB_HEAD_REF:-${GITHUB_REF_NAME}}"
+ if git ls-remote --heads https://github.com/Jepson2k/pinokin.git "$BRANCH" 2>/dev/null | grep -q .; then
+ echo "matching=true" >> "$GITHUB_OUTPUT"
+ echo "branch=$BRANCH" >> "$GITHUB_OUTPUT"
+ else
+ echo "matching=false" >> "$GITHUB_OUTPUT"
+ fi
+
+ # ----------------------------------------------------------------------
+ # Path A: matching pinokin branch -> conda env (provides libpinocchio +
+ # libcoal headers/libs) so pinokin can be built from source. We install
+ # PAROL6 first (which pulls in the pinned pinokin v0.1.6 wheel) and
+ # then override pinokin with the source-built wheel from the matching
+ # branch. This is the only way to exercise unreleased pinokin features
+ # before a release lands.
+ # ----------------------------------------------------------------------
+ - name: Setup Miniforge (matching pinokin branch)
+ if: steps.pinokin.outputs.matching == 'true'
+ uses: conda-incubator/setup-miniconda@v3
+ with:
+ miniforge-version: latest
+ python-version: ${{ matrix.python-version }}
+ conda-remove-defaults: true
+ activate-environment: parol6-test
+
+ - name: Clone matching pinokin branch
+ if: steps.pinokin.outputs.matching == 'true'
+ run: |
+ git clone --depth=1 --branch="${{ steps.pinokin.outputs.branch }}" https://github.com/Jepson2k/pinokin.git pinokin-src
+ conda env update -n parol6-test -f pinokin-src/environment.yml
+
+ # ----------------------------------------------------------------------
+ # Path B: no matching pinokin branch -> plain pip with the pinned
+ # release wheel. Anything in PAROL6 that requires a newer pinokin
+ # feature will fail loudly here (intended).
+ # ----------------------------------------------------------------------
+ - name: Setup Python (pinned pinokin)
+ if: steps.pinokin.outputs.matching != 'true'
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
@@ -58,7 +104,6 @@ jobs:
cache-dependency-path: pyproject.toml
- name: Install package
- shell: bash
run: |
python -m pip install --upgrade pip
pip install -e ".[dev]" pytest-timeout
@@ -72,6 +117,16 @@ jobs:
pip install --force-reinstall "waldoctl @ git+https://github.com/Jepson2k/waldoctl.git@${BRANCH}"
fi
+ # Override the pinned pinokin v0.1.6 wheel with the matching-branch
+ # source build. --force-reinstall + --no-deps swaps just pinokin
+ # without disturbing other resolved dependencies.
+ - name: Override pinokin with matching-branch source build
+ if: steps.pinokin.outputs.matching == 'true'
+ run: |
+ cd pinokin-src
+ pip install . --no-build-isolation --force-reinstall --no-deps
+ python -c "from pinokin import CollisionChecker; print('CollisionChecker available')"
+
- name: Show environment
run: |
python -V
diff --git a/parol6/PAROL6_ROBOT.py b/parol6/PAROL6_ROBOT.py
index 28b1ede..1434c6c 100644
--- a/parol6/PAROL6_ROBOT.py
+++ b/parol6/PAROL6_ROBOT.py
@@ -2,13 +2,14 @@
import atexit
import logging
+import os
from dataclasses import dataclass
from pathlib import Path
from typing import Final
import numpy as np
from numpy.typing import NDArray
-from pinokin import Robot
+from pinokin import CollisionChecker, Robot
from parol6.tools import get_tool_transform
@@ -56,10 +57,171 @@
_urdf_path = str(
Path(__file__).resolve().parent / "urdf_model" / "urdf" / "PAROL6.urdf"
)
+_mesh_dir = str(Path(_urdf_path).resolve().parent.parent)
# Current robot instance (tool transform applied in-place)
robot: Robot = Robot(_urdf_path)
+# Self-collision checker bound to the same pinokin Robot. Built eagerly when
+# ``parol6.config`` is imported (config.py calls ``_init_collision_checker``),
+# i.e. on any ``import parol6``; stays None when collision checking is disabled
+# or geometry fails to load. Treat None as "checks disabled" everywhere.
+# TODO: defer construction to a server-side ``ensure_collision_checker()`` so
+# pure RobotClient script subprocesses don't pay the URDF-rewrite + BVH build.
+collision: CollisionChecker | None = None
+
+
+def _resolved_urdf_for_collision() -> str:
+ """Return a path to a URDF with `package://parol6/...` rewritten to
+ absolute `file://` paths so pinokin's mesh loader can resolve them.
+
+ The PAROL6 URDF was authored for a ROS package layout (meshes at
+ `parol6/meshes/`) but the Python package places them at
+ `parol6/urdf_model/meshes/`. Rewriting at runtime keeps the source
+ URDF unchanged and avoids fragile symlink farms.
+
+ Writes a fresh temp file each call and cleans it up at interpreter exit.
+ """
+ import tempfile
+
+ src = Path(_urdf_path)
+ text = src.read_text()
+ mesh_root = Path(_mesh_dir) / "meshes"
+ # `package://parol6/meshes/foo.STL` -> a plain absolute path coal/assimp can
+ # open. Use a POSIX-style path, NOT a `file://` URI: coal strips the scheme
+ # naively, which on Windows leaves an invalid `/D:/...` (leading slash before
+ # the drive letter). `as_posix()` gives `/abs/...` on POSIX and `D:/abs/...`
+ # on Windows — both openable directly.
+ rewritten = text.replace("package://parol6/meshes/", mesh_root.as_posix() + "/")
+ fd, tmp_path = tempfile.mkstemp(prefix="parol6_collision_", suffix=".urdf")
+ with os.fdopen(fd, "w") as f:
+ f.write(rewritten)
+
+ @atexit.register
+ def _cleanup_tmp_urdf() -> None:
+ try:
+ os.unlink(tmp_path)
+ except OSError:
+ pass
+
+ return tmp_path
+
+
+def _init_collision_checker(enabled: bool, srdf_path: str) -> None:
+ """Build the singleton CollisionChecker when *enabled*.
+
+ Config values are passed in (by ``parol6.config`` after its knobs are
+ defined) rather than imported here, keeping the dependency one-directional
+ — ``config`` imports ``PAROL6_ROBOT``, not the other way around.
+ """
+ global collision
+ if not enabled:
+ collision = None
+ return
+
+ try:
+ # All package:// mesh URIs are rewritten to absolute file:// paths in
+ # the temp URDF, so no package_dirs resolution is needed.
+ urdf_for_collision = _resolved_urdf_for_collision()
+ c = CollisionChecker(robot, urdf_for_collision)
+ if srdf_path and os.path.exists(srdf_path):
+ c.load_srdf(srdf_path)
+ collision = c
+ logger.info(
+ "Collision checker loaded: %d pairs, %d geometry objects",
+ c.num_collision_pairs,
+ c.num_geometry_objects,
+ )
+ except Exception as e: # noqa: BLE001
+ # Enabled but failed to build: fail loud. Silently running the arm with
+ # no collision checking is unsafe; require an explicit opt-out.
+ if os.getenv("PAROL6_ALLOW_NO_COLLISION"):
+ logger.warning(
+ "Collision checker init failed; continuing without it because "
+ "PAROL6_ALLOW_NO_COLLISION is set (UNSAFE): %s",
+ e,
+ )
+ collision = None
+ return
+ raise RuntimeError(
+ "Collision checker failed to initialize. Fix the cause, or set "
+ "PAROL6_ALLOW_NO_COLLISION=1 to run without collision checking "
+ f"(UNSAFE). Original error: {e}"
+ ) from e
+
+
+# Geometry-object names for meshes attached to the collision checker on
+# behalf of the currently-active tool, plus the (tool, variant) they were
+# attached for so an unchanged re-apply can skip the disk reload.
+_active_tool_geom_names: list[str] = []
+_active_tool_geom_key: tuple[str, str | None] | None = None
+
+
+def _refresh_collision_tool_geometry(
+ tool_key: str,
+ variant_key: str | None = None,
+) -> None:
+ """Sync the global collision checker's tool geometry with the active
+ tool. No-op if the checker isn't built yet (so this is safe to call
+ during early module init, before the checker is ensured).
+
+ Skips the work entirely when the (tool, variant) is unchanged: collision
+ mesh placement comes only from ``spec.origin``, never the TCP offset, so a
+ TCP-offset-only ``apply_tool`` would otherwise reload STLs and rebuild BVHs
+ on the control-loop thread for no change.
+ """
+ global _active_tool_geom_key
+ if collision is None:
+ return
+ key = (tool_key, variant_key)
+ if key == _active_tool_geom_key:
+ return
+ # Clear the previous tool's geometry. Mark the key inconsistent until the
+ # new attaches finish, so a mid-loop failure self-repairs on the next call
+ # (otherwise the early-return above would skip a partial attach forever).
+ for name in _active_tool_geom_names:
+ collision.remove_geometry_by_name(name)
+ _active_tool_geom_names.clear()
+ _active_tool_geom_key = None
+
+ from parol6.tools import get_registry
+
+ cfg = None if tool_key == "NONE" else get_registry().get(tool_key)
+ if cfg is not None:
+ # A variant with non-empty meshes wholesale replaces cfg.meshes; an
+ # empty variant falls back to cfg.meshes (deliberately — unlike WC's
+ # swap_tool_mesh, which renders nothing for a mesh-less variant).
+ meshes = cfg.meshes
+ if variant_key:
+ for v in cfg.variants:
+ if v.key == variant_key and v.meshes:
+ meshes = v.meshes
+ break
+ mesh_root = Path(_mesh_dir) / "meshes"
+ try:
+ for spec in meshes:
+ path = mesh_root / spec.file
+ # All current MeshSpecs use rpy=(0,0,0); rotation is baked into
+ # the STL geometry (see _MESH_RPY comment in tools.py). Add a
+ # rotation branch here when a non-identity rpy appears.
+ T = np.eye(4, dtype=np.float64)
+ T[:3, 3] = spec.origin
+ collision.attach_mesh_to_frame(
+ spec.file,
+ str(path),
+ parent_frame="L6",
+ placement=T,
+ )
+ _active_tool_geom_names.append(spec.file)
+ except Exception:
+ # Roll back a partial attach so the checker never holds half a tool.
+ for name in _active_tool_geom_names:
+ collision.remove_geometry_by_name(name)
+ _active_tool_geom_names.clear()
+ raise
+
+ _active_tool_geom_key = key
+
def apply_tool(
tool_name: str,
@@ -95,6 +257,8 @@ def apply_tool(
robot.clear_tool_transform()
logger.info(f"Applied tool {label} (identity)")
+ _refresh_collision_tool_geometry(tool_name, variant_key=variant_key or None)
+
# Initialize with no tool
apply_tool("NONE")
diff --git a/parol6/commands/_collision_guard.py b/parol6/commands/_collision_guard.py
new file mode 100644
index 0000000..08b01d0
--- /dev/null
+++ b/parol6/commands/_collision_guard.py
@@ -0,0 +1,110 @@
+"""Shared self-collision pre-flight check used by motion commands.
+
+`guard_joint_path(positions)` raises `TrajectoryPlanningError(SYS_SELF_COLLISION)`
+if any sampled configuration along the interpolated joint path would self-collide
+(or world-collide given runtime obstacles attached to the singleton checker).
+It runs server-side during trajectory build, so it raises the planning-time
+error type (like the other do_setup guards), not the client-side `MotionError`.
+Disabled-by-config or unloaded-checker scenarios are no-ops.
+"""
+
+from __future__ import annotations
+
+import numpy as np
+from numpy.typing import NDArray
+
+import parol6.PAROL6_ROBOT as PAROL6_ROBOT
+from parol6.config import COLLISION_PATH_SAMPLES
+from parol6.utils.error_catalog import make_error
+from parol6.utils.error_codes import ErrorCode
+from parol6.utils.errors import TrajectoryPlanningError
+
+# Penetration-depth tolerance (metres) for the start-in-collision escape check:
+# a move whose min-distance drops by no more than this counts as "not deeper",
+# absorbing numerical jitter in the signed-distance query.
+_ESCAPE_TOL = 1e-4
+
+
+def _format_pairs(pairs: list[tuple[str, str]]) -> str:
+ """Render colliding (name, name) pairs as a human-readable string.
+
+ Caps at 4 to keep error messages tractable when many pairs collide
+ simultaneously (rare in practice; usually the first one is the
+ actionable one anyway).
+ """
+ if not pairs:
+ return "?"
+ head = pairs[:4]
+ rendered = ", ".join(f"{a} vs {b}" for a, b in head)
+ if len(pairs) > 4:
+ rendered += f" (+{len(pairs) - 4} more)"
+ return rendered
+
+
+def guard_joint_path(positions: NDArray[np.float64]) -> None:
+ """Raise if the path drives into collision.
+
+ `positions` is (N, nq) joint positions in radians. Endpoints are always
+ included; up to COLLISION_PATH_SAMPLES interior samples are checked.
+
+ Normally any collision along the path is rejected. The exception is when the
+ path *starts* already in collision (checker enabled at a colliding pose, or a
+ tool attach created an overlap): rejecting outright would trap the arm, so an
+ *escaping* move is permitted — one that adds no new colliding pair and goes no
+ deeper than the start. Driving into a new collision, or deeper into the
+ current one, still raises.
+ """
+ checker = PAROL6_ROBOT.collision
+ if checker is None:
+ return
+
+ # Load-bearing: normalize to a C-contiguous float64 array for the C++ bindings.
+ pos = np.ascontiguousarray(positions, dtype=np.float64)
+ n = pos.shape[0]
+ if n == 0:
+ return
+
+ # Subsample at COLLISION_PATH_SAMPLES interior points. np.linspace includes
+ # both endpoints, and n > target guarantees spacing > 1 so the rounded
+ # indices are strictly increasing; np.unique stays as cheap insurance.
+ target = max(2, COLLISION_PATH_SAMPLES + 2)
+ if n <= target:
+ idx = None
+ sub = pos # already contiguous float64 — no copy needed
+ else:
+ idx = np.unique(np.linspace(0, n - 1, target).round().astype(int))
+ sub = pos[idx] # fancy indexing yields a fresh contiguous float64 array
+
+ def _raise(sample: int, pairs: list[tuple[str, str]]) -> None:
+ raise TrajectoryPlanningError(
+ make_error(
+ ErrorCode.SYS_SELF_COLLISION,
+ sample=str(sample),
+ total=str(n),
+ pairs=_format_pairs(pairs),
+ )
+ )
+
+ hit = checker.check_path(sub)
+ if hit < 0:
+ return # entire path clear — the common case
+ if hit > 0:
+ # Start is clear but the path drives into a collision — reject it.
+ sample = hit if idx is None else int(idx[hit])
+ _raise(sample, checker.colliding_pairs(pos[sample]))
+
+ # hit == 0: already in collision at the start. Permit an escaping move — one
+ # that introduces no new colliding pair and goes no deeper than the start —
+ # so the arm isn't trapped. (A global min-distance trend alone can't tell an
+ # improving start-collision from a new shallower one, so we check pairs too.)
+ d0 = checker.min_distance(pos[0])
+ start_pairs = set(checker.colliding_pairs(pos[0]))
+ for j in range(1, sub.shape[0]):
+ new_pairs = set(checker.colliding_pairs(sub[j])) - start_pairs
+ deeper = checker.min_distance(sub[j]) < d0 - _ESCAPE_TOL
+ if new_pairs or deeper:
+ sample = j if idx is None else int(idx[j])
+ pairs = (
+ sorted(new_pairs) if new_pairs else checker.colliding_pairs(pos[sample])
+ )
+ _raise(sample, pairs)
diff --git a/parol6/commands/basic_commands.py b/parol6/commands/basic_commands.py
index 59d0f52..768d6e9 100644
--- a/parol6/commands/basic_commands.py
+++ b/parol6/commands/basic_commands.py
@@ -29,6 +29,8 @@
from parol6.config import deg_to_steps
from parol6.server.transports.transport_factory import is_simulation_mode
+import parol6.PAROL6_ROBOT as PAROL6_ROBOT # noqa: N811
+
from .base import (
ExecutionStatusCode,
MotionCommand,
@@ -187,6 +189,19 @@ def execute_step(self, state: "ControllerState") -> ExecutionStatusCode:
se.set_jog_velocity(self._jog_vel_rad)
pos_rad, _vel, _finished = se.tick()
+
+ # Don't stream a self-colliding configuration. This only fires in the
+ # collision case (in_collision is False during normal jogging), so it
+ # never affects ordinary motion; an abrupt stop is acceptable when the
+ # alternative is driving the arm into itself. (The Cartesian jog uses a
+ # graceful CSE-based stop; JogJ has no equivalent smoother, so it halts.)
+ checker = PAROL6_ROBOT.collision
+ if checker is not None and checker.in_collision(pos_rad):
+ logger.warning("[JOGJ] self-collision predicted - stopping jog")
+ se.active = False
+ self.finish()
+ return ExecutionStatusCode.COMPLETED
+
self._q_rad_buf[:] = pos_rad
rad_to_steps(self._q_rad_buf, self._steps_buf)
self.set_move_position(state, self._steps_buf)
diff --git a/parol6/commands/cartesian_commands.py b/parol6/commands/cartesian_commands.py
index 54f4169..54fd3b8 100644
--- a/parol6/commands/cartesian_commands.py
+++ b/parol6/commands/cartesian_commands.py
@@ -9,6 +9,7 @@
import numpy as np
import parol6.PAROL6_ROBOT as PAROL6_ROBOT
+from parol6.commands._collision_guard import guard_joint_path
from parol6.config import (
CART_ANG_JOG_MIN,
CART_LIN_JOG_MIN,
@@ -155,7 +156,11 @@ def execute_step(self, state: "ControllerState") -> ExecutionStatusCode:
if not finished and self._dot_buf > 1e-8:
ik_result = solve_ik(PAROL6_ROBOT.robot, smoothed_pose, self._q_ik_seed)
if ik_result.success and ik_result.q is not None:
- self._track_and_send(state, ik_result.q)
+ # Don't stream a self-colliding config during the release
+ # deceleration; skip the send and let the CSE finish stopping.
+ checker = PAROL6_ROBOT.collision
+ if checker is None or not checker.in_collision(ik_result.q):
+ self._track_and_send(state, ik_result.q)
return ExecutionStatusCode.EXECUTING
cse.active = False
@@ -182,11 +187,18 @@ def execute_step(self, state: "ControllerState") -> ExecutionStatusCode:
if self._vel_ratio > 1.0:
velocity /= self._vel_ratio
- # Set target velocity (WRF transforms to body frame, TRF uses body directly)
- if self.p.frame == "WRF":
- cse.set_jog_velocity_1dof_wrf(self._axis_index, velocity, self.is_rotation)
- else:
- cse.set_jog_velocity_1dof(self._axis_index, velocity, self.is_rotation)
+ # Set target velocity (WRF transforms to body frame, TRF uses body
+ # directly). While stopping (IK failure or predicted self-collision)
+ # leave the CSE target at zero so cse.stop()'s deceleration actually
+ # takes effect — re-commanding full velocity every tick would overwrite
+ # it and the arm would never decelerate (nor reach the vel<1e-6 exit).
+ if not self._ik_stopping:
+ if self.p.frame == "WRF":
+ cse.set_jog_velocity_1dof_wrf(
+ self._axis_index, velocity, self.is_rotation
+ )
+ else:
+ cse.set_jog_velocity_1dof(self._axis_index, velocity, self.is_rotation)
smoothed_pose, smoothed_vel, _finished = cse.tick()
@@ -214,9 +226,25 @@ def execute_step(self, state: "ControllerState") -> ExecutionStatusCode:
return ExecutionStatusCode.COMPLETED
return ExecutionStatusCode.EXECUTING
- # IK succeeded - if we were stopping, recover by resuming jogging
+ # Self-collision predicted at next streamed config? Mirror the
+ # IK-failure graceful-stop pathway: decelerate via cse.stop() rather
+ # than raising mid-jog. Operator regains control after smoothing.
+ if PAROL6_ROBOT.collision is not None and PAROL6_ROBOT.collision.in_collision(
+ ik_result.q
+ ):
+ if not self._ik_stopping:
+ _ik_warn(
+ logger,
+ "[CARTJOG] self-collision predicted - initiating stop",
+ )
+ cse.stop()
+ self._ik_stopping = True
+ return ExecutionStatusCode.EXECUTING
+
+ # Reachable + collision-free again — if we were stopping (IK failure or
+ # predicted self-collision), recover by resuming jogging.
if self._ik_stopping:
- logger.info("[CARTJOG] IK recovered - resuming jog")
+ logger.info("[CARTJOG] constraint cleared - resuming jog")
steps_to_rad(state.Position_in, self._q_rad_buf)
cse.sync_pose(get_fkine_se3(state))
self._q_commanded[:] = self._q_rad_buf
@@ -285,6 +313,9 @@ def _precompute_trajectory(self, state: "ControllerState") -> None:
stop_on_failure=stop_on_failure,
)
+ if not joint_path.is_partial:
+ guard_joint_path(joint_path.positions)
+
if joint_path.is_partial:
ik_valid = joint_path.valid
assert ik_valid is not None
@@ -331,7 +362,7 @@ def _precompute_trajectory(self, state: "ControllerState") -> None:
)
def _compute_target_pose(self, state: "ControllerState") -> None:
- """Compute target pose — absolute or relative based on rel flag."""
+ """Compute target pose - absolute or relative based on rel flag."""
pose = self.p.pose
if self.p.rel:
@@ -443,6 +474,8 @@ def do_setup_with_blend(
self.do_setup(state)
return 0
+ guard_joint_path(joint_path.positions)
+
# Use minimum speed/accel across chain, sum durations when all duration-based
min_speed = self.p.resolved_speed
min_accel = self.p.accel
diff --git a/parol6/commands/curved_commands.py b/parol6/commands/curved_commands.py
index fd2daf2..ad65814 100644
--- a/parol6/commands/curved_commands.py
+++ b/parol6/commands/curved_commands.py
@@ -11,6 +11,7 @@
import numpy as np
+from parol6.commands._collision_guard import guard_joint_path
from parol6.commands.base import TrajectoryMoveCommandBase
from parol6.config import INTERVAL_S, LIMITS, steps_to_rad
from parol6.motion import CircularMotion, JointPath, SplineMotion, TrajectoryBuilder
@@ -154,6 +155,8 @@ def do_setup(self, state: "ControllerState") -> None:
)
)
+ guard_joint_path(joint_path.positions)
+
builder = TrajectoryBuilder(
joint_path=joint_path,
profile=state.motion_profile,
diff --git a/parol6/commands/joint_commands.py b/parol6/commands/joint_commands.py
index 2894127..17f2cb6 100644
--- a/parol6/commands/joint_commands.py
+++ b/parol6/commands/joint_commands.py
@@ -16,6 +16,7 @@
import numpy as np
import parol6.PAROL6_ROBOT as PAROL6_ROBOT
+from parol6.commands._collision_guard import guard_joint_path
from parol6.commands.base import TrajectoryMoveCommandBase
from parol6.config import (
INTERVAL_S,
@@ -86,6 +87,7 @@ def do_setup(self, state: ControllerState) -> None:
current_rad = self._q_rad_buf
joint_path = JointPath.interpolate(current_rad, target_rad, n_samples=50)
+ guard_joint_path(joint_path.positions)
builder = TrajectoryBuilder(
joint_path=joint_path,
profile=state.motion_profile,
@@ -194,6 +196,7 @@ def do_setup_with_blend(
return 0
joint_path = JointPath(positions=positions)
+ guard_joint_path(joint_path.positions)
# Use minimum speed/accel across chain, sum durations when all duration-based
min_speed = self.p.resolved_speed
diff --git a/parol6/config.py b/parol6/config.py
index 476b9c6..9d3c61f 100644
--- a/parol6/config.py
+++ b/parol6/config.py
@@ -564,6 +564,34 @@ def _build_cart_kinodynamic(
dtype=np.float64,
)
+# -----------------------------------------------------------------------------
+# Self-collision checking (pinokin CollisionChecker)
+# -----------------------------------------------------------------------------
+# Pre-flight collision checks reject motion commands whose interpolated
+# joint path enters a self-colliding (or world-colliding) configuration.
+# Disable via env: PAROL6_COLLISION_CHECK=0
+COLLISION_CHECK_ENABLED: bool = os.getenv(
+ "PAROL6_COLLISION_CHECK", "1"
+).strip().lower() in ("1", "true", "yes", "on")
+
+# Number of interior joint-space samples checked along an interpolated path.
+# Endpoints are always checked. 0 => endpoints only.
+# At ~38 us p99 per check on the bundled simplified meshes (see the speed
+# diagnostic), 16 samples is ~0.7 ms per command; world geometry attached at
+# runtime may raise the per-check cost.
+COLLISION_PATH_SAMPLES: int = int(os.getenv("PAROL6_COLLISION_PATH_SAMPLES", "16"))
+
+# Optional SRDF file with disabled-pair info. Defaults to the bundled
+# parol6/urdf_model/srdf/PAROL6.srdf when present.
+_default_srdf = Path(__file__).resolve().parent / "urdf_model" / "srdf" / "PAROL6.srdf"
+COLLISION_SRDF_PATH: str = os.getenv(
+ "PAROL6_COLLISION_SRDF",
+ str(_default_srdf) if _default_srdf.exists() else "",
+)
+
+# Populate PAROL6_ROBOT.collision now that the config knobs are defined.
+PAROL6_ROBOT._init_collision_checker(COLLISION_CHECK_ENABLED, COLLISION_SRDF_PATH)
+
# -----------------------------------------------------------------------------
# Utility Functions
diff --git a/parol6/robot.py b/parol6/robot.py
index 6e3b9e3..f30e827 100644
--- a/parol6/robot.py
+++ b/parol6/robot.py
@@ -1,4 +1,4 @@
-"""Unified PAROL6 robot — lifecycle, configuration, kinematics, and factories.
+"""Unified PAROL6 robot - lifecycle, configuration, kinematics, and factories.
Inherits from ``waldoctl.Robot`` ABC.
All parol6-specific details (subprocess management, pinokin, IK solver, etc.)
@@ -489,7 +489,7 @@ def _resolve_mesh_dir() -> str:
@dataclass
class Parol6IKResult:
- """IK result — structurally compatible with the web commander's IKResult Protocol."""
+ """IK result - structurally compatible with the web commander's IKResult Protocol."""
q: NDArray[np.float64] # radians
success: bool
@@ -504,7 +504,7 @@ class Parol6IKResult:
class Robot(_RobotABC):
- """Unified PAROL6 robot — inherits from waldoctl.Robot ABC.
+ """Unified PAROL6 robot - inherits from waldoctl.Robot ABC.
Combines identity, configuration, FK/IK kinematics, controller lifecycle,
and client factories. Supports both sync and async context managers::
@@ -667,6 +667,11 @@ def set_active_tool(
*tcp_offset_m*: optional (x, y, z) user offset in meters, composed
on top of the tool's registered transform.
*variant_key*: optional variant whose TCP overrides the tool default.
+
+ Note: this mutates the per-instance pinokin Robot, not the global
+ `PAROL6_ROBOT.collision` scene. Server-side tool changes route
+ through `PAROL6_ROBOT.apply_tool`, which is where collision-mesh
+ attachment is wired.
"""
from parol6.tools import get_tool_transform
@@ -769,6 +774,63 @@ def fk_batch(self, joint_path_rad: NDArray[np.float64]) -> NDArray[np.float64]:
result[i, 5] = rpy[2]
return result
+ def in_collision(self, q_rad: NDArray[np.float64]) -> bool:
+ """Return True iff `q_rad` is in self/world collision. False if disabled.
+
+ Queries the process-global ``PAROL6_ROBOT.collision`` checker (shared by
+ all Robot methods here); its tool geometry reflects ``PAROL6_ROBOT
+ .apply_tool`` calls in this process (server / dry-run), not
+ :meth:`set_active_tool`, which only mutates this instance's pinokin Robot.
+ """
+ import parol6.PAROL6_ROBOT as PAROL6_ROBOT
+
+ if PAROL6_ROBOT.collision is None:
+ return False
+ self._load_q_buf(q_rad)
+ return PAROL6_ROBOT.collision.in_collision(self._q_buf)
+
+ def check_trajectory(self, q_path_rad: NDArray[np.float64]) -> int:
+ """Returns first colliding row index in `q_path_rad`, or -1 if clear.
+
+ `q_path_rad` is (N, nq) joint positions in radians.
+ """
+ import parol6.PAROL6_ROBOT as PAROL6_ROBOT
+
+ if PAROL6_ROBOT.collision is None:
+ return -1
+ return PAROL6_ROBOT.collision.check_path(
+ np.ascontiguousarray(q_path_rad, dtype=np.float64)
+ )
+
+ def colliding_pairs(self, q_rad: NDArray[np.float64]) -> list[tuple[str, str]]:
+ """Return list of (name, name) geometry pairs in collision at `q_rad`.
+
+ Names are URDF link names for arm geometry (e.g. ``"L4_0"``) and
+ the user-supplied name for runtime-attached geometry (e.g.
+ ``"ssg48_body_simplified.stl"`` for the active tool's body mesh). Tool
+ geometry is present only when ``PAROL6_ROBOT.apply_tool`` attached it in
+ this process (see :meth:`in_collision`).
+ """
+ import parol6.PAROL6_ROBOT as PAROL6_ROBOT
+
+ if PAROL6_ROBOT.collision is None:
+ return []
+ self._load_q_buf(q_rad)
+ return PAROL6_ROBOT.collision.colliding_pairs(self._q_buf)
+
+ def min_distance(self, q_rad: NDArray[np.float64]) -> float:
+ """Return the minimum clearance over all active pairs at `q_rad`.
+
+ Positive => separation; negative => penetration depth.
+ Returns +inf when collision checking is disabled.
+ """
+ import parol6.PAROL6_ROBOT as PAROL6_ROBOT
+
+ if PAROL6_ROBOT.collision is None:
+ return float("inf")
+ self._load_q_buf(q_rad)
+ return PAROL6_ROBOT.collision.min_distance(self._q_buf)
+
def ik_batch(
self,
poses: NDArray[np.float64],
diff --git a/parol6/tools.py b/parol6/tools.py
index 42f0b2f..da015ec 100644
--- a/parol6/tools.py
+++ b/parol6/tools.py
@@ -508,9 +508,13 @@ def _make_tcp_transform(
# ---------------------------------------------------------------------------
_PNEUMATIC_VERTICAL_MESHES = (
- MeshSpec(file="pneumatic_gripper_vertical_body.stl", role=MeshRole.BODY),
- MeshSpec(file="pneumatic_gripper_vertical_right_jaw.stl", role=MeshRole.JAW),
- MeshSpec(file="pneumatic_gripper_vertical_left_jaw.stl", role=MeshRole.JAW),
+ MeshSpec(file="pneumatic_gripper_vertical_body_simplified.stl", role=MeshRole.BODY),
+ MeshSpec(
+ file="pneumatic_gripper_vertical_right_jaw_simplified.stl", role=MeshRole.JAW
+ ),
+ MeshSpec(
+ file="pneumatic_gripper_vertical_left_jaw_simplified.stl", role=MeshRole.JAW
+ ),
)
_PNEUMATIC_VERTICAL_MOTION = (
LinearMotion(
@@ -524,9 +528,15 @@ def _make_tcp_transform(
)
_PNEUMATIC_HORIZONTAL_MESHES = (
- MeshSpec(file="pneumatic_gripper_horizontal_body.stl", role=MeshRole.BODY),
- MeshSpec(file="pneumatic_gripper_horizontal_right_jaw.stl", role=MeshRole.JAW),
- MeshSpec(file="pneumatic_gripper_horizontal_left_jaw.stl", role=MeshRole.JAW),
+ MeshSpec(
+ file="pneumatic_gripper_horizontal_body_simplified.stl", role=MeshRole.BODY
+ ),
+ MeshSpec(
+ file="pneumatic_gripper_horizontal_right_jaw_simplified.stl", role=MeshRole.JAW
+ ),
+ MeshSpec(
+ file="pneumatic_gripper_horizontal_left_jaw_simplified.stl", role=MeshRole.JAW
+ ),
)
_PNEUMATIC_HORIZONTAL_MOTION = (
LinearMotion(
@@ -581,15 +591,15 @@ def _make_tcp_transform(
)
_SSG48_FINGER_MESHES = (
- MeshSpec(file="ssg48_body.stl", role=MeshRole.BODY),
- MeshSpec(file="ssg48_finger_right.stl", role=MeshRole.JAW),
- MeshSpec(file="ssg48_finger_left.stl", role=MeshRole.JAW),
+ MeshSpec(file="ssg48_body_simplified.stl", role=MeshRole.BODY),
+ MeshSpec(file="ssg48_finger_right_simplified.stl", role=MeshRole.JAW),
+ MeshSpec(file="ssg48_finger_left_simplified.stl", role=MeshRole.JAW),
)
_SSG48_PINCH_MESHES = (
- MeshSpec(file="ssg48_body.stl", role=MeshRole.BODY),
- MeshSpec(file="ssg48_pinch_right.stl", role=MeshRole.JAW),
- MeshSpec(file="ssg48_pinch_left.stl", role=MeshRole.JAW),
+ MeshSpec(file="ssg48_body_simplified.stl", role=MeshRole.BODY),
+ MeshSpec(file="ssg48_pinch_right_simplified.stl", role=MeshRole.JAW),
+ MeshSpec(file="ssg48_pinch_left_simplified.stl", role=MeshRole.JAW),
)
register_tool(
@@ -646,21 +656,21 @@ def _make_tcp_transform(
)
_MSG_100_MESHES = (
- MeshSpec(file="msg_ai_100_body.stl", role=MeshRole.BODY),
- MeshSpec(file="msg_ai_100_right_jaw.stl", role=MeshRole.JAW),
- MeshSpec(file="msg_ai_100_left_jaw.stl", role=MeshRole.JAW),
+ MeshSpec(file="msg_ai_100_body_simplified.stl", role=MeshRole.BODY),
+ MeshSpec(file="msg_ai_100_right_jaw_simplified.stl", role=MeshRole.JAW),
+ MeshSpec(file="msg_ai_100_left_jaw_simplified.stl", role=MeshRole.JAW),
)
_MSG_150_MESHES = (
- MeshSpec(file="msg_ai_150_body.stl", role=MeshRole.BODY),
- MeshSpec(file="msg_ai_150_right_jaw.stl", role=MeshRole.JAW),
- MeshSpec(file="msg_ai_150_left_jaw.stl", role=MeshRole.JAW),
+ MeshSpec(file="msg_ai_150_body_simplified.stl", role=MeshRole.BODY),
+ MeshSpec(file="msg_ai_150_right_jaw_simplified.stl", role=MeshRole.JAW),
+ MeshSpec(file="msg_ai_150_left_jaw_simplified.stl", role=MeshRole.JAW),
)
_MSG_200_MESHES = (
- MeshSpec(file="msg_ai_200_body.stl", role=MeshRole.BODY),
- MeshSpec(file="msg_ai_200_right_jaw.stl", role=MeshRole.JAW),
- MeshSpec(file="msg_ai_200_left_jaw.stl", role=MeshRole.JAW),
+ MeshSpec(file="msg_ai_200_body_simplified.stl", role=MeshRole.BODY),
+ MeshSpec(file="msg_ai_200_right_jaw_simplified.stl", role=MeshRole.JAW),
+ MeshSpec(file="msg_ai_200_left_jaw_simplified.stl", role=MeshRole.JAW),
)
register_tool(
@@ -716,7 +726,9 @@ def _make_tcp_transform(
name="Vacuum Gripper",
description="Vacuum gripper (pneumatic valve I/O)",
transform=_make_tcp_transform(z=-0.037),
- meshes=(MeshSpec(file="vacuum_gripper_body.stl", role=MeshRole.BODY),),
+ meshes=(
+ MeshSpec(file="vacuum_gripper_body_simplified.stl", role=MeshRole.BODY),
+ ),
motions=(),
io_port=1,
),
diff --git a/parol6/urdf_model/srdf/PAROL6.srdf b/parol6/urdf_model/srdf/PAROL6.srdf
new file mode 100644
index 0000000..e8026e7
--- /dev/null
+++ b/parol6/urdf_model/srdf/PAROL6.srdf
@@ -0,0 +1,32 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/parol6/urdf_model/urdf/PAROL6.urdf b/parol6/urdf_model/urdf/PAROL6.urdf
index ac81a5f..dce672b 100644
--- a/parol6/urdf_model/urdf/PAROL6.urdf
+++ b/parol6/urdf_model/urdf/PAROL6.urdf
@@ -49,7 +49,7 @@
rpy="0 0 0" />
+ filename="package://parol6/meshes/base_link_simplified.stl" />
@@ -89,7 +89,7 @@
rpy="0 0 0" />
+ filename="package://parol6/meshes/L1_simplified.stl" />
@@ -147,7 +147,7 @@
rpy="0 0 0" />
+ filename="package://parol6/meshes/L2_simplified.stl" />
@@ -205,7 +205,7 @@
rpy="0 0 0" />
+ filename="package://parol6/meshes/L3_simplified.stl" />
@@ -263,7 +263,7 @@
rpy="0 0 0" />
+ filename="package://parol6/meshes/L4_simplified.stl" />
@@ -321,7 +321,7 @@
rpy="0 0 0" />
+ filename="package://parol6/meshes/L5_simplified.stl" />
@@ -379,7 +379,7 @@
rpy="0 0 0" />
+ filename="package://parol6/meshes/L6_simplified.stl" />
diff --git a/parol6/utils/error_catalog.py b/parol6/utils/error_catalog.py
index 94b61dc..7ab38fd 100644
--- a/parol6/utils/error_catalog.py
+++ b/parol6/utils/error_catalog.py
@@ -165,6 +165,12 @@ class _ErrorTemplate:
effect="Profile not changed.",
remedy="Use one of: TOPPRA, RUCKIG, QUINTIC, TRAPEZOID, LINEAR.",
),
+ ErrorCode.SYS_SELF_COLLISION: _ErrorTemplate(
+ title="Self-collision predicted",
+ cause="Planned configuration would self-collide at sample {sample} of {total}: {pairs}",
+ effect="Motion command rejected before dispatch.",
+ remedy="Choose a different target, add intermediate waypoints, or disable via PAROL6_COLLISION_CHECK=0.",
+ ),
}
diff --git a/parol6/utils/error_codes.py b/parol6/utils/error_codes.py
index 5e32b39..6b6eaec 100644
--- a/parol6/utils/error_codes.py
+++ b/parol6/utils/error_codes.py
@@ -38,3 +38,4 @@ class ErrorCode(IntEnum):
SYS_ESTOP_ACTIVE = 51
SYS_PORT_SAVE_FAILED = 52
SYS_PROFILE_INVALID = 53
+ SYS_SELF_COLLISION = 54
diff --git a/pyproject.toml b/pyproject.toml
index e281677..e11c840 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -46,7 +46,7 @@ dependencies = [
"psutil>=5.9",
"msgspec>=0.18",
"ormsgpack>=1.4.0",
- "waldoctl @ git+https://github.com/Jepson2k/waldoctl.git@v0.2.0",
+ "waldoctl @ git+https://github.com/Jepson2k/waldoctl.git@v0.4.0",
]
[tool.setuptools.packages.find]
@@ -112,6 +112,14 @@ include = [
[tool.ty.overrides.rules]
invalid-type-form = "ignore"
+# pinokin is a compiled extension; the lint env's build can predate
+# CollisionChecker, so ty can't resolve that symbol (it resolves fine at runtime
+# and against a matching pinokin build).
+[[tool.ty.overrides]]
+include = ["parol6/PAROL6_ROBOT.py"]
+[tool.ty.overrides.rules]
+unresolved-import = "ignore"
+
[tool.setuptools]
include-package-data = true
diff --git a/tests/unit/test_collision_integration.py b/tests/unit/test_collision_integration.py
new file mode 100644
index 0000000..106d5be
--- /dev/null
+++ b/tests/unit/test_collision_integration.py
@@ -0,0 +1,195 @@
+"""Unit tests for PAROL6's pinokin collision integration.
+
+Covers:
+- Singleton checker initialization in PAROL6_ROBOT
+- SRDF disabled-pair count
+- Public Robot.in_collision / colliding_pairs / min_distance / check_trajectory
+- guard_joint_path raising TrajectoryPlanningError on a colliding sample
+"""
+
+from __future__ import annotations
+
+import numpy as np
+import pytest
+
+import parol6.PAROL6_ROBOT as PAROL6_ROBOT
+import parol6.config # noqa: F401 - imports trigger collision-checker init
+from parol6 import Robot
+from parol6.commands._collision_guard import guard_joint_path
+from parol6.utils.error_codes import ErrorCode
+from parol6.utils.errors import TrajectoryPlanningError
+
+
+def test_singleton_checker_initialized():
+ assert PAROL6_ROBOT.collision is not None
+ assert PAROL6_ROBOT.collision.num_geometry_objects > 0
+ assert PAROL6_ROBOT.collision.num_collision_pairs > 0
+
+
+def test_srdf_disabled_pairs_reduce_pair_count():
+ # Without SRDF: 7 link geometries -> 21 pairs minus 6 parent/child adjacent = 15.
+ # The bundled SRDF disables 7 more pairs (base<->L4/L5/L6, L1<->L4/L5/L6, L2<->L4).
+ assert PAROL6_ROBOT.collision.num_collision_pairs == 8
+
+
+def test_home_is_clear():
+ q = np.zeros(PAROL6_ROBOT.robot.nq)
+ assert PAROL6_ROBOT.collision.in_collision(q) is False
+
+
+def test_robot_collision_methods_clear_at_home():
+ """Public Robot collision methods report clear at/near home on one instance.
+ (Robot defines no __del__; an unstarted instance holds no subprocess, so the
+ old per-test try/finally del was a no-op.)"""
+ r = Robot()
+ q = np.zeros(6)
+ assert r.in_collision(q) is False
+ d = r.min_distance(q)
+ assert d > 0.0 and d != float("inf")
+ # Tiny perturbation around home — definitely clear.
+ q_path = np.linspace(np.zeros(6), 0.01 * np.ones(6), 5)
+ assert r.check_trajectory(q_path) == -1
+
+
+def test_guard_joint_path_clear_returns_none():
+ positions = np.zeros((10, 6))
+ # No exception means no collision detected.
+ guard_joint_path(positions)
+
+
+def test_guard_joint_path_raises_on_explicit_collision(monkeypatch):
+ """Force a fake collision by patching the singleton checker temporarily.
+ monkeypatch auto-restores the real checker at teardown."""
+
+ class FakeChecker:
+ def in_collision(self, q):
+ return True
+
+ def check_path(self, q):
+ return 2
+
+ def colliding_pairs(self, q):
+ return [("ssg48_body_simplified.stl", "L4_0")]
+
+ monkeypatch.setattr(PAROL6_ROBOT, "collision", FakeChecker())
+
+ positions = np.zeros((5, 6))
+ with pytest.raises(TrajectoryPlanningError) as exc_info:
+ guard_joint_path(positions)
+ err = exc_info.value.robot_error
+ assert err.code == int(ErrorCode.SYS_SELF_COLLISION)
+ # Cause string should embed the named pair, not raw int indices.
+ assert "ssg48_body_simplified.stl vs L4_0" in err.cause
+
+
+def test_guard_disabled_when_checker_is_none(monkeypatch):
+ monkeypatch.setattr(PAROL6_ROBOT, "collision", None)
+ positions = np.zeros((5, 6))
+ # No exception, returns None (no-op).
+ assert guard_joint_path(positions) is None
+
+
+def test_no_spurious_self_overlap_at_home_or_joint_limits():
+ """Audit the bundled simplified collision STLs against the SRDF.
+
+ Asserts that the checker reports no colliding pairs at home, at each
+ joint's lower/upper limit (single-axis), at every (low, high) corner
+ of the joint-limit hypercube, and across a handful of seeded random
+ configs. If this fails, the assertion message identifies the named
+ pair — add it to parol6/urdf_model/srdf/PAROL6.srdf and re-run.
+ """
+ import itertools
+
+ lo = PAROL6_ROBOT._joint_limits_radian[:, 0]
+ hi = PAROL6_ROBOT._joint_limits_radian[:, 1]
+ checker = PAROL6_ROBOT.collision
+
+ configs: list[tuple[str, np.ndarray]] = [("home", np.zeros(6))]
+ for j in range(6):
+ q_lo = np.zeros(6)
+ q_lo[j] = lo[j]
+ q_hi = np.zeros(6)
+ q_hi[j] = hi[j]
+ configs.append((f"J{j}=low", q_lo))
+ configs.append((f"J{j}=high", q_hi))
+
+ for bits in itertools.product((0, 1), repeat=6):
+ q = np.where(np.array(bits, dtype=bool), hi, lo)
+ configs.append((f"corner_{''.join(map(str, bits))}", q))
+
+ rng = np.random.default_rng(0xC011)
+ for k in range(20):
+ configs.append((f"rand_{k}", rng.uniform(lo, hi)))
+
+ failures: list[str] = []
+ for label, q in configs:
+ pairs = checker.colliding_pairs(np.ascontiguousarray(q, dtype=np.float64))
+ if pairs:
+ failures.append(f"{label}: {pairs}")
+
+ assert not failures, (
+ "Spurious self-collision in the bundled simplified STLs. Add these "
+ "pairs to parol6/urdf_model/srdf/PAROL6.srdf and re-run:\n"
+ + "\n".join(failures[:10])
+ + (f"\n... ({len(failures) - 10} more)" if len(failures) > 10 else "")
+ )
+
+
+def test_collision_check_speed_diagnostic(capsys):
+ """Time in_collision and colliding_pairs across a sampled workspace.
+
+ Diagnostic only — does not assert a threshold. The per-tick mid-motion
+ collision check shipped (JogL release-decel gate, JogJ stop); this prints
+ percentiles to confirm its cost stays well within the 100 Hz tick budget
+ (10 ms).
+
+ Run via: pytest tests/unit/test_collision_integration.py::test_collision_check_speed_diagnostic -v -s
+ """
+ import time
+
+ rng = np.random.default_rng(0xC0FFEE)
+ lo = PAROL6_ROBOT._joint_limits_radian[:, 0]
+ hi = PAROL6_ROBOT._joint_limits_radian[:, 1]
+ qs = [np.zeros(6)] + [
+ np.ascontiguousarray(rng.uniform(lo, hi), dtype=np.float64) for _ in range(99)
+ ]
+ c = PAROL6_ROBOT.collision
+
+ # warm-up so first-call cache effects don't dominate
+ for q in qs[:10]:
+ c.in_collision(q)
+ c.colliding_pairs(q)
+
+ t_bool: list[int] = []
+ for q in qs:
+ t0 = time.perf_counter_ns()
+ c.in_collision(q)
+ t_bool.append(time.perf_counter_ns() - t0)
+
+ t_pairs: list[int] = []
+ for q in qs:
+ t0 = time.perf_counter_ns()
+ c.colliding_pairs(q)
+ t_pairs.append(time.perf_counter_ns() - t0)
+
+ def pct(a: list[int], p: float) -> float:
+ return float(np.percentile(a, p)) / 1000.0 # ns -> us
+
+ with capsys.disabled():
+ print(
+ f"\nin_collision us:"
+ f" min={pct(t_bool, 0):.1f}"
+ f" med={pct(t_bool, 50):.1f}"
+ f" p95={pct(t_bool, 95):.1f}"
+ f" p99={pct(t_bool, 99):.1f}"
+ f" max={pct(t_bool, 100):.1f}"
+ )
+ print(
+ f"colliding_pairs us:"
+ f" min={pct(t_pairs, 0):.1f}"
+ f" med={pct(t_pairs, 50):.1f}"
+ f" p95={pct(t_pairs, 95):.1f}"
+ f" p99={pct(t_pairs, 99):.1f}"
+ f" max={pct(t_pairs, 100):.1f}"
+ )
+ print("servo tick budget: 10000 us (100 Hz)")