Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion agent_context/MAP.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -299,7 +299,8 @@ topics:
title: Robot Workspace
aliases: [robot workspace, workspace analyzer, reachability cache, 机器人工作空间, 可达空间, 工作空间缓存]
keywords: [WorkspaceAnalyzer, AnalysisMode, ResultsCache, RobotWorkspace, RobotWorkspaceCfg, workspace_cfg,
sample_reachable_pose, analyze-workspace, workspace-cache, results.npz]
sample_reachable_pose, analyze-workspace, workspace-cache, results.npz, manipulability_scores,
ManipulabilityMetric]
paths: [topics/robot-workspace/robot-workspace.md]
source_of_truth:
- embodichain/lab/sim/motion/workspace/analyzer.py
Expand Down
24 changes: 24 additions & 0 deletions agent_context/topics/robot-workspace/analysis-and-cache.md
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,30 @@
| `workspace-cache list/clean` cannot see ResultsCache entries | It targets legacy session caches; use analyzer preview/cache directory semantics for result entries. |
| Concurrent analyzers expose corrupt/incomplete cache | `ResultsCache.save()` writes result then metadata directly with no observed lock or temp-file replacement. Add atomic write/lock coverage before relying on shared concurrent writers. |

### Manipulability scores

- When the metric config enables manipulability (default `ALL`), the analyzer
computes per-configuration Yoshikawa scores `w = sqrt(det(J J^T))` from the
active solver's Jacobian after analysis. `manipulability_scores` is
row-aligned with `joint_configurations` (and therefore with reachable points
in Cartesian/plane modes), stored in `results.npz`, and restored on cache
hits; aggregates land in `metrics["manipulability"]`.
- `ManipulabilityMetric` produces statistics only from Jacobians or
precomputed scores. The former centroid-distance placeholder was measured
to be negatively correlated with true manipulability (corr ≈ −0.37 on
Franka) and now yields a warning plus empty results instead.
- Cost is negligible: batched Jacobian + determinant is ~10 ms per 470
configurations on GPU.

### Seed selection for Cartesian/plane IK

Cartesian and plane analysis verify reachability through the solver's
multi-start `get_ik`, so `PytorchSolverCfg.enable_seed_selection` applies
without analyzer changes. Measured on Franka (4000 identical points, warm):
enabling it at unchanged `num_samples=30` detected slightly more reachable
points at 3.3× lower wall time; `num_samples=8` reached 4.1× with <1%
detection loss. Analytic solvers (OPW/SRS/UR) are unaffected.

### Sampling and allocation controls

- `SamplingConfig` and `analyze-workspace --sampler` default to scrambled Sobol.
Expand Down
127 changes: 125 additions & 2 deletions embodichain/lab/sim/motion/workspace/analyzer.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,10 @@
VisualizationType,
VisualizationConfig,
MetricConfig,
MetricType,
)
from embodichain.lab.sim.motion.workspace.metrics.manipulability_metric import (
ManipulabilityMetric,
)
from embodichain.lab.sim.motion.workspace.samplers import (
SamplerFactory,
Expand Down Expand Up @@ -228,6 +232,7 @@ def __init__(
self.metrics_results: Dict[str, Any] = {}
self.current_mode: AnalysisMode | None = None
self.success_rates: torch.Tensor | None = None
self.manipulability_scores: torch.Tensor | None = None
# Path of the most recently written/read results cache entry (None until
# a disk results cache is used). Exposed for CLI consumers.
self._last_cache_path: Path | None = None
Expand Down Expand Up @@ -1411,6 +1416,11 @@ def analyze(
if cached_results is not None:
logger.log_info("Loaded results from cache")
self._restore_analysis_state(cached_results)
# Metric settings are not part of the cache key: entries
# written under a different metric configuration (or before
# manipulability existed) are repaired here from the cached
# joint configurations.
self._apply_manipulability(cached_results)
self._log_analysis_summary(cached_results)
if visualize:
self._visualize_workspace()
Expand Down Expand Up @@ -1566,6 +1576,7 @@ def analyze(
logger.log_info("[3/3] Computing metrics...")
metrics = self._compute_metrics()
results["metrics"] = metrics
self._apply_manipulability(results)
results["config"] = self.config
results["analysis_time"] = time.time() - start_time

Expand Down Expand Up @@ -2011,6 +2022,115 @@ def _generate_point_colors(self, points: np.ndarray) -> np.ndarray:
)
return colors

def _manipulability_enabled(self) -> bool:
"""Whether manipulability computation is selected in the metric config."""
enabled = self.config.metric.enabled_metrics or []
return MetricType.ALL in enabled or MetricType.MANIPULABILITY in enabled

def _compute_manipulability_values(
self,
) -> tuple[torch.Tensor | None, torch.Tensor | None]:
"""Compute per-configuration Yoshikawa scores and condition numbers.

Uses the active control part's solver Jacobian on the stored
``joint_configurations``, so every row stays aligned with the
configuration (and, in Cartesian/plane modes, with the reachable
point) at the same index: ``w = sqrt(det(J @ J^T))``. Condition
numbers (max/min singular value) are computed only when the metric
config enables isotropy.

Returns:
Tuple of scores ``(N,)`` and condition numbers ``(N,)`` (or
``None``) on the analysis device; ``(None, None)`` when no
configurations or no solver Jacobian are available.
"""
qpos = self.joint_configurations
if qpos is None or len(qpos) == 0:
return None, None
solver = self.robot.get_solver(self.control_part_name)
if solver is None:
logger.log_warning(
"No solver available for manipulability computation; skipping."
)
return None, None

want_isotropy = self.config.metric.manipulability.compute_isotropy
chunk_size = 10000
scores, conditions = [], []
with torch.no_grad():
for start in range(0, len(qpos), chunk_size):
chunk = torch.as_tensor(
qpos[start : start + chunk_size],
dtype=torch.float32,
device=solver.device,
)
jac = solver.get_jacobian(chunk)
jjt = jac @ jac.transpose(1, 2)
scores.append(torch.sqrt(torch.clamp(torch.det(jjt), min=0.0)))
Comment thread
greptile-apps[bot] marked this conversation as resolved.
if want_isotropy:
singulars = torch.linalg.svdvals(jac)
conditions.append(
singulars[:, 0] / torch.clamp(singulars[:, -1], min=1e-15)
)
return (
torch.cat(scores).to(self.device),
torch.cat(conditions).to(self.device) if conditions else None,
)

def _apply_manipulability(self, results: Dict[str, Any]) -> None:
"""Attach manipulability scores and aggregates to a results dict.

Used on both the fresh-analysis path and the cache-hit path. Metric
settings are deliberately not part of the results-cache key, so a
cached entry may have been produced under a different metric
configuration (or before scores existed); recomputing from the cached
``joint_configurations`` costs milliseconds and repairs such entries
transparently.
"""
if not self._manipulability_enabled():
# Strip fields a differently-configured producer may have cached,
# so cached and fresh analyses expose the same result contract.
self.manipulability_scores = None
results.pop("manipulability_scores", None)
metrics = results.get("metrics")
if isinstance(metrics, dict):
metrics.pop("manipulability", None)
return
Comment thread
greptile-apps[bot] marked this conversation as resolved.
scores, conditions = self._compute_manipulability_values()
if scores is None:
# Enabled but not computable here (e.g. no solver): cached scores
# remain valid — they are pure kinematics of the same joint
# configurations. Per-point condition numbers are not cached, so
# condition statistics are unavailable on this path.
cached = results.get("manipulability_scores")
if cached is None:
self.manipulability_scores = None
metrics = results.get("metrics")
if isinstance(metrics, dict):
metrics.pop("manipulability", None)
return
scores = torch.as_tensor(cached).to(self.device)
conditions = None
self.manipulability_scores = scores
results["manipulability_scores"] = scores
# Aggregates are always recomputed under the CURRENT metric
# configuration: cached aggregates may reflect a different
# jacobian_threshold or isotropy setting, since metric settings are
# not part of the cache key.
metric = ManipulabilityMetric(self.config.metric.manipulability)
metrics = results.setdefault("metrics", {})
metrics["manipulability"] = metric.compute(
(
self.workspace_points.cpu().numpy()
if self.workspace_points is not None
else np.empty((0, 3))
),
manipulability_scores=scores.cpu().numpy(),
condition_numbers=(
conditions.cpu().numpy() if conditions is not None else None
),
)

def _compute_metrics(self) -> Dict[str, Any]:
"""Compute workspace metrics based on configuration."""
if self.workspace_points is None or len(self.workspace_points) == 0:
Expand All @@ -2019,8 +2139,7 @@ def _compute_metrics(self) -> Dict[str, Any]:

metrics = {}

# TODO: Implement metric computation using metrics module
# For now, compute basic statistics
# Basic geometric statistics
points_np = self.workspace_points.cpu().numpy()

metrics["bounding_box"] = {
Expand All @@ -2036,6 +2155,9 @@ def _compute_metrics(self) -> Dict[str, Any]:
# Approximate volume (bounding box)
metrics["bounding_box_volume"] = float(np.prod(dimensions))

# Manipulability aggregates are attached by _apply_manipulability,
# which also runs on the cache-hit path.

logger.log_info(f"Computed {len(metrics)} metrics")

return metrics
Expand Down Expand Up @@ -2409,6 +2531,7 @@ def _restore_analysis_state(self, results: Dict[str, Any]) -> None:
)
self.joint_configurations = results.get("joint_configurations")
self.success_rates = results.get("success_rates")
self.manipulability_scores = results.get("manipulability_scores")
if mode_str in ("cartesian_space", "plane_sampling"):
self.reachable_points = results.get("reachable_points")
self.reachability_mask = results.get("reachability_mask")
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,7 @@
"joint_configurations",
"success_rates",
"reachability_mask",
"manipulability_scores",
)

# Scalar/dict fields stored in meta.json (JSON-serializable form).
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

from typing import Dict, Any
import numpy as np
from embodichain.utils import logger
from embodichain.lab.sim.motion.workspace.metrics.base_metric import (
BaseMetric,
)
Expand All @@ -27,8 +28,10 @@
class ManipulabilityMetric(BaseMetric):
"""Manipulability metric for workspace analysis.

Computes dexterity and manipulability measures throughout the workspace.
Note: Full implementation requires robot Jacobian computation.
Computes Yoshikawa manipulability statistics from robot Jacobians or from
precomputed per-point scores. Without either input no statistics are
produced: an earlier centroid-distance placeholder was measured to be
*negatively* correlated with true manipulability and has been removed.
"""

def __init__(self, config: ManipulabilityConfig | None = None):
Expand All @@ -44,6 +47,8 @@ def compute(
workspace_points: np.ndarray,
joint_configurations: np.ndarray | None = None,
jacobians: np.ndarray | None = None,
manipulability_scores: np.ndarray | None = None,
condition_numbers: np.ndarray | None = None,
**kwargs,
) -> Dict[str, Any]:
"""Compute manipulability metrics.
Expand All @@ -52,6 +57,11 @@ def compute(
workspace_points: Workspace points in Cartesian space, shape (N, 3).
joint_configurations: Joint configurations, shape (N, num_joints).
jacobians: Precomputed Jacobian matrices, shape (N, 6, num_joints).
manipulability_scores: Precomputed per-point Yoshikawa scores,
shape (N,). Takes precedence over ``jacobians``.
condition_numbers: Precomputed per-point Jacobian condition
numbers, shape (N,). Used for isotropy statistics when
``jacobians`` is not provided.
**kwargs: Additional arguments.

Returns:
Expand All @@ -60,7 +70,12 @@ def compute(
- std_manipulability: Standard deviation
- min_manipulability: Minimum value
- max_manipulability: Maximum value
- mean_condition: Average condition number (if isotropy enabled)
- mean_condition: Average condition number (if isotropy enabled
and Jacobians were provided)

Without ``jacobians`` or ``manipulability_scores`` an empty dict is
returned: true manipulability cannot be derived from Cartesian
points alone, and fabricated statistics are worse than none.
"""
points = self._to_numpy(workspace_points)

Expand All @@ -72,31 +87,22 @@ def compute(
"max_manipulability": 0.0,
}

# If Jacobians are not provided, we cannot compute true manipulability
# Return placeholder statistics
if jacobians is None:
# Estimate based on distance from centroid (simple heuristic)
centroid = points.mean(axis=0)
distances = np.linalg.norm(points - centroid, axis=1)

# Normalize to [0, 1] range (higher manipulability near center)
max_dist = distances.max() if distances.max() > 0 else 1.0
manipulability_scores = 1.0 - (distances / max_dist)

# Filter by threshold
valid_mask = manipulability_scores >= self.config.jacobian_threshold
valid_scores = manipulability_scores[valid_mask]

if len(valid_scores) == 0:
valid_scores = np.array([0.0])
else:
# Compute true manipulability from Jacobians
if manipulability_scores is not None:
manipulability_scores = self._to_numpy(manipulability_scores)
elif jacobians is not None:
manipulability_scores = self._compute_manipulability_index(jacobians)
valid_mask = manipulability_scores >= self.config.jacobian_threshold
valid_scores = manipulability_scores[valid_mask]

if len(valid_scores) == 0:
valid_scores = np.array([0.0])
else:
logger.log_warning(
"ManipulabilityMetric needs jacobians or precomputed scores; "
"skipping (no placeholder statistics are produced)."
)
self.results = {}
return self.results

valid_mask = manipulability_scores >= self.config.jacobian_threshold
valid_scores = manipulability_scores[valid_mask]
if len(valid_scores) == 0:
valid_scores = np.array([0.0])

self.results = {
"mean_manipulability": float(valid_scores.mean()),
Expand All @@ -107,10 +113,13 @@ def compute(
}

# Compute isotropy if requested
if self.config.compute_isotropy and jacobians is not None:
condition_numbers = self._compute_condition_numbers(jacobians)
self.results["mean_condition"] = float(condition_numbers.mean())
self.results["std_condition"] = float(condition_numbers.std())
if self.config.compute_isotropy:
if condition_numbers is None and jacobians is not None:
condition_numbers = self._compute_condition_numbers(jacobians)
if condition_numbers is not None:
condition_numbers = self._to_numpy(condition_numbers)
self.results["mean_condition"] = float(condition_numbers.mean())
self.results["std_condition"] = float(condition_numbers.std())

return self.results

Expand Down
3 changes: 3 additions & 0 deletions tests/sim/motion/workspace/test_analysis_batching.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,9 @@ def _analyzer(**kwargs) -> WorkspaceAnalyzer:
robot.get_qpos.return_value = torch.zeros(3, 3)
robot.cfg = SimpleNamespace(uid="test", fpath=None, solver_cfg={})
robot._solvers = {}
# Faithful to Robot.get_solver with no solvers attached: manipulability
# computation then skips with a warning instead of touching a Mock.
robot.get_solver.return_value = None

def fk(qpos, **kw):
pose = torch.eye(4).expand(*qpos.shape[:-1], 4, 4).clone()
Expand Down
Loading