refactor: unify RLinf python path resolution and to_numpy_tree

docs/source-en/rst_source/development/interfaces.rst

@@ -127,7 +127,9 @@ normal-CLI or Dashboard runtime hook:
 
 Common flags: ``--env-endpoint``, ``--vla-endpoint``. The default ``http`` sends
 JSON over ``POST /call``, encoding NumPy arrays as
-``{"__ndarray__": <base64>, "dtype": ..., "shape": ...}``; switch to ``socket``
+``{"__ndarray__": <base64>, "dtype": ..., "shape": ...}`` and NumPy scalars as
+``{"__npscalar__": <value>, "dtype": ...}`` so dtypes survive the round trip;
+switch to ``socket``
 for large or history-stacked nested-NumPy observations to move length-prefixed
 pickle frames and skip repeated JSON encoding. Pickle is unsafe on untrusted
 input, so only point ``socket`` at trusted endpoints.

docs/source-en/rst_source/usage/advanced_deployment.rst

@@ -84,3 +84,20 @@ that service, and each ``*_PORT`` with the free port you chose at startup.
 Any of the three endpoint flags can be omitted; when one is unset, RPent
 spawns that service locally on a free port. All three default to HTTP when
 the protocol is omitted, and all three accept ``socket://HOST:PORT``.
+
+Custom RLinf checkout
+---------------------
+
+The environment and VLA services import ``rlinf``. When starting them
+manually against a development RLinf checkout instead of the installed
+package, point ``PYTHONPATH`` at the checkout first:
+
+.. code-block:: bash
+
+   export PYTHONPATH=/path/to/rlinf:$PYTHONPATH
+
+Servers spawned by RPent get this automatically: the checkout is resolved
+from ``RPENT_RLINF_ROOT`` (or ``RLINF_REPO_PATH``), falling back to the
+``rlinf`` directory next to the RPent checkout. A resolved path that does
+not exist is harmless — Python ignores invalid ``PYTHONPATH`` entries —
+so the servers import the installed ``rlinf`` package.

docs/source-zh/rst_source/development/interfaces.rst

@@ -120,6 +120,7 @@ runtime 钩子中解析）：
 
 常见：``--env-endpoint``、``--vla-endpoint``。``http`` 为默认，走 ``POST /call``
 传 JSON，其中 NumPy 数组编码为 ``{"__ndarray__": <base64>, "dtype": ..., "shape": ...}``；
+NumPy 标量编码为 ``{"__npscalar__": <value>, "dtype": ...}`` 以保留精确 dtype；
 观测数据很大、或是多帧堆叠的嵌套 NumPy 字典时可改 ``socket``，用带长度前缀的
 pickle 数据帧传输，省掉反复的 JSON 编解码。pickle 不适合不可信输入，socket 只应连接可信端点。
 

docs/source-zh/rst_source/usage/advanced_deployment.rst

@@ -81,3 +81,18 @@ SAM3 服务只加载一次模型，可以由多个 RPent 运行复用。
 某项未指定时，RPent 会在当前机器上启动对应服务并自动选择空闲端口。三个服务省略
 protocol 时都默认使用 HTTP，也都可以通过 ``socket://HOST:PORT`` 改用 socket
 RPC。
+
+自定义 RLinf 源码
+-----------------
+
+环境服务和 VLA 服务需要导入 ``rlinf``。手动启动时如果使用开发中的 RLinf
+源码而非已安装的包，请先把 ``PYTHONPATH`` 指向该源码目录：
+
+.. code-block:: bash
+
+   export PYTHONPATH=/path/to/rlinf:$PYTHONPATH
+
+由 RPent 自动拉起的 server 无需手动设置：RLinf 源码路径从
+``RPENT_RLINF_ROOT`` （或 ``RLINF_REPO_PATH`` ）解析，默认回退到 RPent
+仓库旁边的 ``rlinf`` 目录。解析出的路径不存在也无妨：Python 会忽略无效的
+``PYTHONPATH`` 条目，server 将导入已安装的 ``rlinf`` 包。

robots/dual_franka/env_server.py

@@ -17,22 +17,15 @@
 from __future__ import annotations
 
 import queue
-import sys
 import time
 from typing import Any
 
 import numpy as np
 
 from robots.dual_franka.runtime_config import load_runtime_config
-from robots.franka.env_server import _to_numpy_tree, main
-from rpent.utils.config import get_repo_root, get_rlinf_repo_path
+from robots.franka.env_server import main
 from rpent.utils.logging import get_logger
-
-# Resolve the RLinf checkout before the deferred ``import rlinf`` executes.
-RPENT_ROOT = get_repo_root()
-RLINF_REPO_PATH = get_rlinf_repo_path() or (RPENT_ROOT.parent / "rlinf").resolve()
-if str(RLINF_REPO_PATH) not in sys.path:
-    sys.path.insert(0, str(RLINF_REPO_PATH))
+from rpent.utils.serialization import to_numpy_tree
 
 logger = get_logger("dual_franka_env_server")
 
@@ -145,7 +138,7 @@ def reset(self) -> dict[str, Any]:
             observation, info = self.env.reset()
             return {
                 "ok": True,
-                "info": _to_numpy_tree(info),
+                "info": to_numpy_tree(info),
                 "robot_state": self.get_robot_state(),
                 "states": self._strip_batch(observation.get("states")),
             }
@@ -154,7 +147,7 @@ def reset(self) -> dict[str, Any]:
 
         @staticmethod
         def _strip_batch(value: Any) -> Any:
-            array = _to_numpy_tree(value)
+            array = to_numpy_tree(value)
             if isinstance(array, np.ndarray) and array.ndim > 0 and array.shape[0] == 1:
                 return array[0]
             if isinstance(array, list) and len(array) == 1:
@@ -195,7 +188,7 @@ def _attach_camera_snapshots(self, output: dict[str, Any]) -> dict[str, Any]:
             snapshot_getter = self.env.env.call(
                 "get_wrapper_attr", "get_raw_camera_snapshot"
             )[0]
-            snapshot = _to_numpy_tree(snapshot_getter())
+            snapshot = to_numpy_tree(snapshot_getter())
             output["raw_camera_frames"] = snapshot.get("raw_frames", {})
             output["raw_camera_depths"] = snapshot.get("raw_depths", {})
             perception = self._capture_perception_camera_snapshot()
@@ -268,8 +261,8 @@ def _wrapped_states(self) -> np.ndarray:
         def get_robot_state(self) -> dict[str, Any]:
             left, right = self._arm_states()
             return {
-                "left_arm": _to_numpy_tree(left),
-                "right_arm": _to_numpy_tree(right),
+                "left_arm": to_numpy_tree(left),
+                "right_arm": to_numpy_tree(right),
                 "action_dim": self.action_dim,
                 "per_arm_dim": self.per_arm_dim,
                 "action_scale": self.action_scale.tolist(),
@@ -302,7 +295,7 @@ def get_camera_meta(self) -> dict[str, Any] | None:
             metadata_getter = self.env.env.call(
                 "get_wrapper_attr", "get_raw_camera_metadata"
             )[0]
-            metadata.update(_to_numpy_tree(metadata_getter()))
+            metadata.update(to_numpy_tree(metadata_getter()))
             metadata.update(self._perception_camera_meta)
             return {
                 "cameras": cameras,
@@ -530,9 +523,9 @@ def chunk_step(
             for action in np.asarray(actions, dtype=np.float32):
                 observation, _reward, term, trunc, info = self.env.step(action[None, :])
                 observations.append(self._decorate_step_observation(observation))
-                terminated = terminated or bool(np.asarray(_to_numpy_tree(term)).any())
-                truncated = truncated or bool(np.asarray(_to_numpy_tree(trunc)).any())
-                last_info = _to_numpy_tree(info)
+                terminated = terminated or bool(np.asarray(to_numpy_tree(term)).any())
+                truncated = truncated or bool(np.asarray(to_numpy_tree(trunc)).any())
+                last_info = to_numpy_tree(info)
                 if terminated or truncated:
                     break
             return {

robots/dual_franka/vla_server.py

@@ -18,7 +18,6 @@
 
 import argparse
 import os
-import sys
 import time
 from typing import Any
 
@@ -27,17 +26,10 @@
 from omegaconf import OmegaConf
 
 from rpent.robots.components.vla_facade_base import BaseVLAFacade
-from rpent.utils.config import get_repo_root, get_rlinf_repo_path
 from rpent.utils.logging import get_logger
 
 logger = get_logger("dual_franka_vla_server")
 
-# Resolve the RLinf checkout before the deferred ``import rlinf`` executes.
-RPENT_ROOT = get_repo_root()
-RLINF_REPO_PATH = get_rlinf_repo_path() or (RPENT_ROOT.parent / "rlinf").resolve()
-if str(RLINF_REPO_PATH) not in sys.path:
-    sys.path.insert(0, str(RLINF_REPO_PATH))
-
 
 def build_model_cfg(model_path: str, repo_id: str) -> Any:
     """Build the RLinf OpenPI inference config for dual Franka."""

robots/franka/env_server.py

@@ -17,8 +17,6 @@
 from __future__ import annotations
 
 import argparse
-import dataclasses
-import sys
 import time
 from collections.abc import Callable
 from typing import Any
@@ -27,34 +25,11 @@
 
 from robots.franka.runtime_config import load_runtime_config
 from rpent.robots.components.env_facade_base import BaseEnvFacade
-from rpent.utils.config import get_repo_root, get_rlinf_repo_path
 from rpent.utils.logging import get_logger
+from rpent.utils.serialization import to_numpy_tree
 
 logger = get_logger("franka_env_server")
 
-# Resolve the RLinf checkout before the deferred ``import rlinf`` executes.
-RPENT_ROOT = get_repo_root()
-RLINF_REPO_PATH = get_rlinf_repo_path() or (RPENT_ROOT.parent / "rlinf").resolve()
-if str(RLINF_REPO_PATH) not in sys.path:
-    sys.path.insert(0, str(RLINF_REPO_PATH))
-
-
-def _to_numpy_tree(value: Any) -> Any:
-    """Convert tensors and nested values into pickle-safe CPU/numpy data."""
-    if hasattr(value, "detach") and hasattr(value, "cpu"):
-        return value.detach().cpu().numpy()
-    if dataclasses.is_dataclass(value):
-        return _to_numpy_tree(dataclasses.asdict(value))
-    if isinstance(value, dict):
-        return {key: _to_numpy_tree(item) for key, item in value.items()}
-    if isinstance(value, list):
-        return [_to_numpy_tree(item) for item in value]
-    if isinstance(value, tuple):
-        return tuple(_to_numpy_tree(item) for item in value)
-    if isinstance(value, np.generic):
-        return value.item()
-    return value
-
 
 class FrankaEnvFacade(BaseEnvFacade):
     """Expose the single-Franka ``env.*`` protocol from a Ray-backed worker."""
@@ -151,14 +126,14 @@ def reset(self) -> dict[str, Any]:
             observation, info = self.env.reset()
             return {
                 "ok": True,
-                "info": _to_numpy_tree(info),
+                "info": to_numpy_tree(info),
                 "robot_state": self.get_robot_state(),
                 "states": self._strip_batch(observation.get("states")),
             }
 
         @staticmethod
         def _strip_batch(value: Any) -> Any:
-            array = _to_numpy_tree(value)
+            array = to_numpy_tree(value)
             if isinstance(array, np.ndarray) and array.ndim > 0 and array.shape[0] == 1:
                 return array[0]
             if isinstance(array, list) and len(array) == 1:
@@ -223,7 +198,7 @@ def _raw_tcp_pose(self) -> np.ndarray:
 
         def get_robot_state(self) -> dict[str, Any]:
             return {
-                "raw_base_state": _to_numpy_tree(self._raw_state()),
+                "raw_base_state": to_numpy_tree(self._raw_state()),
                 "action_dim": self.action_dim,
                 "action_scale": self.action_scale.tolist(),
                 "use_relative_frame": self.use_relative_frame,
@@ -234,7 +209,7 @@ def get_camera_meta(self) -> dict[str, Any] | None:
                 metadata = self.env.env.call("get_camera_metadata")[0]
             except Exception as exc:
                 return {"error": str(exc), "error_type": type(exc).__name__}
-            metadata = _to_numpy_tree(metadata)
+            metadata = to_numpy_tree(metadata)
             main_key = self.cfg.env.eval.get("main_image_key")
             names = sorted(metadata.get("cameras", {}))
             extras = [name for name in names if name != main_key]
@@ -371,7 +346,7 @@ def set_gripper(self, *, open: bool) -> dict[str, Any]:
                     gripper=command,
                 )
                 time.sleep(self.controller["gripper_settle_s"])
-                state = _to_numpy_tree(self._raw_state())
+                state = to_numpy_tree(self._raw_state())
                 reached = bool(state.get("gripper_open")) == bool(open)
                 iterations += 1
                 if reached:
@@ -397,9 +372,9 @@ def chunk_step(
             for action in np.asarray(actions, dtype=np.float32):
                 observation, _reward, term, trunc, info = self.env.step(action[None, :])
                 observations.append(self._strip_observation(observation))
-                terminated = terminated or bool(np.asarray(_to_numpy_tree(term)).any())
-                truncated = truncated or bool(np.asarray(_to_numpy_tree(trunc)).any())
-                last_info = _to_numpy_tree(info)
+                terminated = terminated or bool(np.asarray(to_numpy_tree(term)).any())
+                truncated = truncated or bool(np.asarray(to_numpy_tree(trunc)).any())
+                last_info = to_numpy_tree(info)
                 if terminated or truncated:
                     break
             return {

robots/libero/env_server.py

@@ -24,11 +24,8 @@
 import numpy as np
 
 from rpent.robots.components.env_facade_base import BaseEnvFacade
-from rpent.utils.config import (
-    get_repo_root,
-    get_rlinf_repo_path,
-)
 from rpent.utils.logging import get_logger
+from rpent.utils.serialization import to_numpy_tree
 
 # MuJoCo env vars must be set BEFORE importing anything that touches MuJoCo.
 os.environ.setdefault("MUJOCO_GL", "egl")
@@ -39,16 +36,12 @@
 
 logger = get_logger("env_server")
 
-RPENT_ROOT = get_repo_root()
-RLINF_REPO_PATH = get_rlinf_repo_path() or (RPENT_ROOT.parent / "rlinf").resolve()
-if str(RLINF_REPO_PATH) not in sys.path:
-    sys.path.insert(0, str(RLINF_REPO_PATH))
 os.environ.setdefault("ROBOT_PLATFORM", "LIBERO")
 
 # torch and LiberoEnv are only imported at call time (after --cuda-device
 # sets CUDA_VISIBLE_DEVICES in main()); LiberoEnv transitively imports torch.
 if TYPE_CHECKING:
-    import torch  # noqa: F401  (referenced at runtime in _to_numpy_tree)
+    import torch  # noqa: F401  (transitive dep of LiberoEnv; type-check only)
     from rlinf.envs.libero.libero_env import LiberoEnv
 
 
@@ -137,20 +130,6 @@ def make_env(
 # ---------------------------------------------------------------------------
 
 
-def _to_numpy_tree(x):
-    """Recursively convert torch tensors to CPU numpy arrays so the result
-    pickles cleanly across the agent/env_server wire."""
-    if hasattr(x, "detach") and hasattr(x, "cpu") and hasattr(x, "numpy"):
-        return x.detach().cpu().numpy()
-    if isinstance(x, dict):
-        return {k: _to_numpy_tree(v) for k, v in x.items()}
-    if isinstance(x, list):
-        return [_to_numpy_tree(v) for v in x]
-    if isinstance(x, tuple):
-        return tuple(_to_numpy_tree(v) for v in x)
-    return x
-
-
 class LiberoEnvFacade(BaseEnvFacade):
     """Implements :class:`robots.libero.env_client.LiberoEnvClient`
     over :class:`rlinf.envs.libero.libero_env.LiberoEnv`.
@@ -210,20 +189,20 @@ def _expand_chunk(self, actions) -> np.ndarray:
 
     def reset(self):
         obs, info = self._env.reset()
-        obs = self._strip_obs(_to_numpy_tree(obs))
-        return obs, _to_numpy_tree(info)
+        obs = self._strip_obs(to_numpy_tree(obs))
+        return obs, to_numpy_tree(info)
 
     def step(self, action):
         obs, rew, term, trunc, info = self._env.step(self._expand_action(action))
-        obs = self._strip_obs(_to_numpy_tree(obs))
-        term = self._strip(_to_numpy_tree(term))
-        trunc = self._strip(_to_numpy_tree(trunc))
+        obs = self._strip_obs(to_numpy_tree(obs))
+        term = self._strip(to_numpy_tree(term))
+        trunc = self._strip(to_numpy_tree(trunc))
         return (
             obs,
-            self._strip(_to_numpy_tree(rew)),
+            self._strip(to_numpy_tree(rew)),
             term,
             trunc,
-            _to_numpy_tree(info),
+            to_numpy_tree(info),
         )
 
     def chunk_step(self, actions, *, return_all_frames: bool = False):
@@ -241,20 +220,20 @@ def chunk_step(self, actions, *, return_all_frames: bool = False):
         obs_list, rew, term, trunc, info = self._env.chunk_step(
             self._expand_chunk(actions)
         )
-        obs_list = [self._strip_obs(_to_numpy_tree(o)) for o in obs_list]
-        term = self._strip(_to_numpy_tree(term))
-        trunc = self._strip(_to_numpy_tree(trunc))
+        obs_list = [self._strip_obs(to_numpy_tree(o)) for o in obs_list]
+        term = self._strip(to_numpy_tree(term))
+        trunc = self._strip(to_numpy_tree(trunc))
         obs_field = obs_list if return_all_frames else obs_list[-1]
         return (
             obs_field,
-            self._strip(_to_numpy_tree(rew)),
+            self._strip(to_numpy_tree(rew)),
             term,
             trunc,
-            _to_numpy_tree(info),
+            to_numpy_tree(info),
         )
 
     def raw_obs(self) -> dict:
-        return _to_numpy_tree(self._env.current_raw_obs[self._env_idx])
+        return to_numpy_tree(self._env.current_raw_obs[self._env_idx])
 
     def get_env_meta(self) -> dict:
         """Return the meta info this server was launched with."""
@@ -274,7 +253,7 @@ def render_camera(
         width: int = 1024,
         depth: bool = False,
     ):
-        return _to_numpy_tree(
+        return to_numpy_tree(
             self._env.render_camera(
                 camera_name=camera_name,
                 height=height,
@@ -289,7 +268,7 @@ def get_camera_meta(
         height: int = 256,
         width: int = 256,
     ) -> dict | None:
-        return _to_numpy_tree(
+        return to_numpy_tree(
             self._env.get_camera_meta(
                 camera_name=camera_name, height=height, width=width
             )