Skip to content
Merged
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
4 changes: 4 additions & 0 deletions examples/wuji/pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,10 @@ deploy-vision = ["opencv-contrib-python", "pupil-apriltags", "pyyaml"]
# of `deploy` so the headless core stays binary-free. Pinned to match wuji-mjlab.
deploy-hand = ["wujihandpy==1.5.1"]

[project.scripts]
# Unified deploy/real2sim entry: `uv run wuji <command>` (see deploy/README.md).
wuji = "genelab_wuji.cli:main"

[project.entry-points."genelab.extensions"]
genelab_wuji = "genelab_wuji.tasks:register"

Expand Down
68 changes: 68 additions & 0 deletions examples/wuji/src/genelab_wuji/cli.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
"""Unified ``wuji`` command-line entry for the deploy + real2sim toolchain.

Installed as a console script by the ``examples/wuji[deploy]`` extra, so the
whole pipeline runs through one verb instead of long ``python -m`` module paths::

uv run wuji <command> [options]
uv run wuji play --ckpt policy.onnx --real --goal-mode random
uv run wuji <command> --help # per-command options

Each command lazily imports its target module so that, e.g., ``wuji play`` does
not pull in the Hikvision MVS SDK that ``wuji observer`` needs (and vice-versa) —
the observer module raises at import time when the SDK is absent.
"""

from __future__ import annotations

import importlib
import sys

# command -> (module, prefix-args prepended before user args)
# prefix-args let two verbs share one module (hand_utils' check / home subparser).
_COMMANDS: dict[str, tuple[str, tuple[str, ...]]] = {
"check": ("genelab_wuji.deploy.scripts.hand_utils", ("check",)),
"home": ("genelab_wuji.deploy.scripts.hand_utils", ("home",)),
"observer": ("genelab_wuji.deploy.scripts.cube_world_observer", ()),
"viewer": ("genelab_wuji.deploy.scripts.toreal_viewer", ()),
"calib": ("genelab_wuji.deploy.scripts.calib_check", ()),
"play": ("genelab_wuji.deploy.scripts.play_real", ()),
}

_HELP: dict[str, str] = {
"check": "read-only hand-bridge test (connection + encoder sanity)",
"home": "ease-in-out ramp the hand to the grasp pose",
"observer": "Hikvision camera -> ArUco cube pose, published on ZMQ",
"viewer": "real2sim Genesis mirror of the observed cube",
"calib": "calibration viewer: live hand + observed cube vs. digital twin",
"play": "deploy control loop (real/mock) + goal modes + success monitor",
}


def _usage() -> str:
width = max(len(c) for c in _COMMANDS)
lines = ["usage: wuji <command> [options]", "", "commands:"]
lines += [f" {name:<{width}} {_HELP[name]}" for name in _COMMANDS]
lines += ["", "run `wuji <command> --help` for per-command options."]
return "\n".join(lines)


def main(argv: list[str] | None = None) -> int:
argv = list(sys.argv[1:] if argv is None else argv)
if not argv:
print(_usage(), file=sys.stderr)
return 2
cmd, rest = argv[0], argv[1:]
if cmd in ("-h", "--help", "help"):
print(_usage())
return 0
if cmd not in _COMMANDS:
print(f"wuji: unknown command {cmd!r}\n", file=sys.stderr)
print(_usage(), file=sys.stderr)
return 2
module_name, prefix = _COMMANDS[cmd]
module = importlib.import_module(module_name)
return int(module.main([*prefix, *rest]) or 0)


if __name__ == "__main__":
raise SystemExit(main())
21 changes: 12 additions & 9 deletions examples/wuji/src/genelab_wuji/deploy/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -70,29 +70,32 @@ export LD_LIBRARY_PATH=/opt/MVS/lib/64:/opt/MVS/lib/32:$LD_LIBRARY_PATH

## Run

All deploy steps run through the unified `wuji` entry (installed by the `deploy`
extra); `wuji <command> --help` lists each command's options.

```bash
# 0) export a trained policy to ONNX
genelab export Genelab-Reorient-Wuji-Hand-v0 PATH/model.pt --format onnx --out policy.onnx
# 0) export a trained policy to ONNX (genelab's own CLI)
uv run genelab export Genelab-Reorient-Wuji-Hand-v0 PATH/model.pt --format onnx --out policy.onnx

# 1) smoke-test the control loop, no hardware, no ZMQ, no viewer
python -m genelab_wuji.deploy.scripts.play_real --ckpt policy.onnx --mock --no-zmq --no-viewer --steps 100
uv run wuji play --ckpt policy.onnx --mock --no-zmq --no-viewer --steps 100

# 1.5) bring up the real hand bridge (needs wujihandpy): check first, then home
python -m genelab_wuji.deploy.scripts.hand_utils check # READ-ONLY: connection + encoder sanity
python -m genelab_wuji.deploy.scripts.hand_utils home # 3s ease-in-out ramp to the grasp pose
uv run wuji check # READ-ONLY: connection + encoder sanity
uv run wuji home # 3s ease-in-out ramp to the grasp pose

# 2) vision: detect the cube and publish its tag-frame pose on ZMQ:5555 (needs MVS env)
python -m genelab_wuji.deploy.scripts.cube_world_observer --preview # terminal A
python -m genelab_wuji.deploy.scripts.toreal_viewer # terminal B (real2sim mirror)
uv run wuji observer --preview # terminal A
uv run wuji viewer # terminal B (real2sim mirror)

# 2.5) calibration check: home the hand, render live hand + observed cube in the twin
python -m genelab_wuji.deploy.scripts.calib_check # (needs the observer running)
uv run wuji calib # (needs the observer running)

# 3) drive the real hand from the live observer feed (Genesis mirror viewer on by default,
# showing the live hand + observed cube + goal; pass --no-viewer for headless).
# goal modes: --goal-mode random (uniform-SO3, resampled on success) |
# fixed --goal-quat w,x,y,z | external (goal from toreal_viewer ZMQ)
python -m genelab_wuji.deploy.scripts.play_real --ckpt policy.onnx --real --goal-mode random
uv run wuji play --ckpt policy.onnx --real --goal-mode random
```

`play_real` mirrors the live hand (encoders) + observed cube + goal in a Genesis viewer
Expand Down
10 changes: 5 additions & 5 deletions examples/wuji/src/genelab_wuji/deploy/scripts/calib_check.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,9 +15,9 @@
control beyond the initial homing ramp — the hand stays at home.

Run:
python -m genelab_wuji.deploy.scripts.cube_world_observer & # publisher
python -m genelab_wuji.deploy.scripts.calib_check # this tool
python -m genelab_wuji.deploy.scripts.calib_check --mock # no hardware
uv run wuji observer & # publisher
uv run wuji calib # this tool
uv run wuji calib --mock # no hardware

Press Ctrl+C or close the viewer window to exit. Needs a GPU + display (Genesis
viewer); the transform math itself is covered by the headless deploy tests.
Expand Down Expand Up @@ -61,7 +61,7 @@ def _loop(env: Any, drv: HandDriverBase, cube_recv: CubeReceiver | None, rate_hz
print("\n[calib-check] interrupted by user.")


def main() -> int:
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=(__doc__ or "").splitlines()[0])
parser.add_argument("--cube-port", type=int, default=DEFAULT_CUBE_PORT)
parser.add_argument("--host", default="localhost")
Expand All @@ -77,7 +77,7 @@ def main() -> int:
parser.add_argument(
"--mock", action="store_true", help="use the mock hand (no hardware; renders home pose)"
)
args = parser.parse_args()
args = parser.parse_args(argv)

cube_recv: CubeReceiver | None = None
if not args.no_cube_zmq:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,8 @@
- ZMQ publishing on port 5555

Usage:
python -m genelab_wuji.deploy.scripts.cube_world_observer --preview # With visualization
python -m genelab_wuji.deploy.scripts.cube_world_observer # Headless mode
uv run wuji observer --preview # With visualization
uv run wuji observer # Headless mode

On startup, the world coordinate system is auto-sampled (100 frames by default),
then a fixed world frame is used. Press 'w' to resample the world frame.
Expand Down Expand Up @@ -1442,7 +1442,7 @@ def cleanup(self):
print("Cleanup done.")


def main():
def main(argv=None):
import argparse

# Load config from file first
Expand Down Expand Up @@ -1474,7 +1474,7 @@ def main():
"resolving to config/cube_tags<suffix>.json, or a literal "
"path. Default: config/cube_tags.json (54mm).",
)
args = parser.parse_args()
args = parser.parse_args(argv)
cube_config_path = resolve_cube_config_path(args.cube)

# Use config values, CLI args override
Expand Down
6 changes: 3 additions & 3 deletions examples/wuji/src/genelab_wuji/deploy/scripts/hand_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,8 +11,8 @@
and reports tracking error.

Usage:
python -m genelab_wuji.deploy.scripts.hand_utils check
python -m genelab_wuji.deploy.scripts.hand_utils home
uv run wuji check
uv run wuji home

The home pose is ``REORIENT_JOINT_POS`` (via ``config.default_joint_pos``) — the
same grasp keyframe the policy and ``MockHandDriver`` start from, so after ``home``
Expand Down Expand Up @@ -146,7 +146,7 @@ def cmd_check(_args: argparse.Namespace) -> int:

print("\n" + "=" * 60)
print("✓ All read operations succeeded. Hand bridge healthy.")
print("✓ Next: ramp to home with `python -m genelab_wuji.deploy.scripts.hand_utils home`")
print("✓ Next: ramp to home with `uv run wuji home`")
print("=" * 60)
return 0

Expand Down
10 changes: 5 additions & 5 deletions examples/wuji/src/genelab_wuji/deploy/scripts/play_real.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,13 +18,13 @@

Usage:
# Smoke run without hardware (mock hand, random goals):
python -m genelab_wuji.deploy.scripts.play_real --ckpt policy.onnx --goal-mode random --steps 200
uv run wuji play --ckpt policy.onnx --goal-mode random --steps 200

# Real hand, random goals resampled on success:
python -m genelab_wuji.deploy.scripts.play_real --ckpt policy.onnx --real --goal-mode random
uv run wuji play --ckpt policy.onnx --real --goal-mode random

# Real hand, goal driven by toreal_viewer over ZMQ:
python -m genelab_wuji.deploy.scripts.play_real --ckpt policy.onnx --real --goal-mode external
uv run wuji play --ckpt policy.onnx --real --goal-mode external
"""

from __future__ import annotations
Expand Down Expand Up @@ -164,7 +164,7 @@ def close(self) -> None:
self._env.close()


def main() -> int:
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--ckpt", required=True, help="exported policy.onnx")
parser.add_argument("--metadata", default=None, help="policy metadata.json (auto-detected)")
Expand Down Expand Up @@ -192,7 +192,7 @@ def main() -> int:
help="mirror the live hand + cube + goal in a Genesis viewer "
"(default on; pass --no-viewer for headless / mock smoke runs)",
)
args = parser.parse_args()
args = parser.parse_args(argv)

if args.control_dt <= 0:
raise SystemExit("--control-dt must be > 0 (used for joint velocity + success timing)")
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -8,10 +8,10 @@

Usage:
# Terminal 1: real camera -> ZMQ (needs hardware; see cube_world_observer.py)
python -m genelab_wuji.deploy.scripts.cube_world_observer
uv run wuji observer

# Terminal 2: mirror the cube in the Genesis sim
python -m genelab_wuji.deploy.scripts.toreal_viewer
uv run wuji viewer

Run on a host with a GPU + display (Genesis viewer). The transform math itself is
covered by the headless deploy tests.
Expand All @@ -27,12 +27,12 @@
from genelab_wuji.deploy.zmq_bridge import DEFAULT_CUBE_PORT, CubeReceiver


def main() -> int:
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--cube-port", type=int, default=DEFAULT_CUBE_PORT)
parser.add_argument("--host", default="localhost")
parser.add_argument("--fps", type=float, default=60.0, help="viewer refresh rate")
args = parser.parse_args()
args = parser.parse_args(argv)

env = build_reorient_env()
tag_pos_w, tag_quat_w = tag_world_pose(env) # fixed-base hand -> constant
Expand Down
79 changes: 79 additions & 0 deletions tests/test_examples_wuji_cli.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,79 @@
"""Unit tests for the unified ``wuji`` console-script dispatcher.

Covers verb routing (including the shared hand_utils check/home prefix), the
help/usage/unknown-command exit codes, and that each command lazily imports its
target module (so e.g. `wuji play` does not require the camera SDK).
"""

from __future__ import annotations

import sys
import types

import pytest

from genelab_wuji import cli


def test_commands_map_to_real_modules():
# Every advertised verb maps to a deploy.scripts module + a help line.
for name, (module_name, prefix) in cli._COMMANDS.items():
assert module_name.startswith("genelab_wuji.deploy.scripts.")
assert isinstance(prefix, tuple)
assert name in cli._HELP


@pytest.mark.parametrize(
"argv,code",
[
(["--help"], 0),
(["help"], 0),
([], 2),
(["bogus"], 2),
],
)
def test_top_level_exit_codes(argv, code):
assert cli.main(argv) == code


def test_usage_lists_every_command(capsys):
cli.main(["--help"])
out = capsys.readouterr().out
for name in cli._COMMANDS:
assert name in out


def _install_stub(monkeypatch, target_module: str):
"""Replace ``importlib.import_module`` so the dispatched module is a stub
that records the argv it was handed instead of importing real hardware code."""
seen: dict[str, list[str]] = {}
stub = types.ModuleType(target_module)

def _main(argv):
seen["argv"] = list(argv)
return 0

stub.main = _main # type: ignore[attr-defined]
monkeypatch.setattr(cli.importlib, "import_module", lambda name: stub)
return seen


def test_play_routes_args_through(monkeypatch):
seen = _install_stub(monkeypatch, "genelab_wuji.deploy.scripts.play_real")
rc = cli.main(["play", "--ckpt", "p.onnx", "--mock"])
assert rc == 0
assert seen["argv"] == ["--ckpt", "p.onnx", "--mock"]


def test_check_prepends_subcommand(monkeypatch):
# `wuji check` / `wuji home` share hand_utils.main via a prepended verb.
seen = _install_stub(monkeypatch, "genelab_wuji.deploy.scripts.hand_utils")
cli.main(["check", "--foo"])
assert seen["argv"] == ["check", "--foo"]
cli.main(["home"])
assert seen["argv"] == ["home"]


def test_main_reads_sys_argv_when_argv_none(monkeypatch):
monkeypatch.setattr(sys, "argv", ["wuji", "--help"])
assert cli.main() == 0