環境プロセッサ

Environment processorsは、ポリシー自身のプロセッサよりも前に環境の観測を変換し、環境よりも前にポリシーのアクションを変換します。各環境設定はget_env_processors()からそのペアを返し、make_env_pre_post_processorsは評価、学習時の評価、およびシミュレーションロボットのためにそれらを構築します。

シミュレーション環境では不要です

シミュレーション環境は、ポリシーに正規化済みの形式で到達します。画像はobservation.images.<camera>（またはobservation.image）、状態はフラットな1つのobservation.stateとして渡され、各シミュレータの画像の向きと状態の構成は、そのsrc/lerobot/sim/integrations/<simulator>.py統合によって一度だけ処理されます。これは評価、ロールアウト、記録のいずれにも、ローカルでもサーブでも適用されます（Simulation: local or servedを参照）。例えばLIBEROでは、その観測アダプタが両方のカメラを反転させ、8次元の状態（エンドエフェクタの位置、軸角の向き、グリップ関節）を構築するため、その環境プロセッサは恒等変換になります。

新しいシミュレーターの観測マッピングはその統合に、制御の慣例はその統合のconventions()に配置し、環境プロセッサに配置しないでください。これにより、シミュレータのロボットとサーブされた環境もそれらを取得できるようになります（ベンチマークの追加を参照）。

環境プロセッサがまだツールである場合

• lerobot.sim 以外の環境。 Isaac Lab Arena は、そのリポジトリ独自の make_env を通じて実行され、バッチ化された torch 観測値を使用します。IsaaclabArenaProcessorStep は、その状態キーを選択して連結し、カメラを変換します。
• ポリシー固有の環境適応。 make_env_pre_post_processors を使用すると、XVLA や Cosmos3 が LIBERO で行っているように、ポリシーが環境のペアを置き換えることができます。

Evaluation は、環境プロセッサが生成するものに対してチェックポイントを受け入れます。具体的には、プリプロセッサ後の観測とポストプロセッサ後のアクションであり、これらはそれぞれ transform_features から読み取られます。したがって、ステップは実行時の動作と同様に、その出力を入力に対する関数としてそこに宣言します。例えば、ソースキーが存在する場合のみ状態を構築するステップなどが挙げられます。リモート推論では環境プロセッサは適用されないため、そこで提供されるチェックポイントは、シミュレータの観測を受け取り、そのアクションを自ら命令する必要があります。

評価時の処理順序は以下の通りです：

observation = env_preprocessor(observation)   # environment-specific
observation = preprocessor(observation)       # policy's saved preprocessing (normalization, ...)
action = postprocessor(policy.select_action(observation))
action = env_postprocessor(action)            # environment-specific

1つの作成

環境プロセッサは、設定から返される通常のプロセッサステップです:

from dataclasses import dataclass

from lerobot.envs.configs import EnvConfig
from lerobot.processor import ObservationProcessorStep, PolicyProcessorPipeline, ProcessorStepRegistry


@dataclass
@ProcessorStepRegistry.register(name="myenv_processor")
class MyEnvProcessorStep(ObservationProcessorStep):
    def observation(self, observation):
        processed = observation.copy()
        processed["observation.state"] = processed.pop("observation.myenv_state")
        return processed

    def transform_features(self, features):
        return features  # update if names or shapes change


@EnvConfig.register_subclass("myenv")
@dataclass
class MyEnvConfig(EnvConfig):
    ...

    def get_env_processors(self):
        return PolicyProcessorPipeline(steps=[MyEnvProcessorStep()]), PolicyProcessorPipeline(steps=[])

環境固有の情報（キーレイアウト、座標系）のみを保持し、正規化やモデル固有の変換はポリシーのプロセッサに委ねてください。