# configuration_my_policy.py
from dataclasses import dataclass, field
import torch
from lerobot.configs import PreTrainedConfig
from lerobot.optim import AdamWConfig
from lerobot.optim import CosineDecayWithWarmupSchedulerConfig

@PreTrainedConfig.register_subclass("my_policy")
@dataclass
class MyPolicyConfig(PreTrainedConfig):
    """Configuration class for MyPolicy.

    Args:
        n_obs_steps: Number of observation steps to use as input
        horizon: Action prediction horizon
        n_action_steps: Number of action steps to execute
        hidden_dim: Hidden dimension for the policy network
        # Add your policy-specific parameters here
    """

    dtype: torch.dtype | None = torch.bfloat16

    horizon: int = 50
    n_action_steps: int = 50
    hidden_dim: int = 256

    optimizer_lr: float = 1e-4
    optimizer_weight_decay: float = 1e-4

    def __post_init__(self):
        super().__post_init__()
        if self.n_action_steps > self.horizon:
            raise ValueError("n_action_steps cannot exceed horizon")

    def validate_features(self) -> None:
        """Validate input/output feature compatibility.

        Call this explicitly from your policy's __init__ — the base class does not.
        """
        if not self.image_features:
            raise ValueError("MyPolicy requires at least one image feature.")
        if self.action_feature is None:
            raise ValueError("MyPolicy requires 'action' in output_features.")

    def get_optimizer_preset(self) -> AdamWConfig:
        return AdamWConfig(lr=self.optimizer_lr, weight_decay=self.optimizer_weight_decay)

    def get_scheduler_preset(self):
        """Return a LRSchedulerConfig from lerobot.optim, or None."""
        return None

    @property
    def observation_delta_indices(self) -> list[int] | None:
        """Relative timestep offsets the dataset loader provides per observation.

        Return `None` for single-frame policies. For temporal policies that consume
        multiple past or future frames, return a list of offsets, e.g. `[-20, -10, 0, 10]` for
        3 past frames at stride 10 and 1 future frame at stride 10.
        """
        return None

    @property
    def action_delta_indices(self) -> list[int]:
        """Relative timestep offsets for the action chunk the dataset loader returns."""
        return list(range(self.horizon))

    @property
    def reward_delta_indices(self) -> None:
        return None

# modeling_my_policy.py
import torch
import torch.nn as nn
from typing import Any

from lerobot.policies import PreTrainedPolicy
from lerobot.utils.constants import ACTION
from .configuration_my_policy import MyPolicyConfig

class MyPolicy(PreTrainedPolicy):
    config_class = MyPolicyConfig  # must match the string in @register_subclass
    name = "my_policy"

    def __init__(self, config: MyPolicyConfig, dataset_stats: dict[str, Any] = None):
        super().__init__(config, dataset_stats)
        config.validate_features()  # not called automatically by the base class
        self.config = config
        self.model = ...  # your nn.Module here

    def reset(self):
        """Reset per-episode state. Called by lerobot-eval at the start of each episode."""
        ...

    def get_optim_params(self) -> dict:
        """Return parameters to pass to the optimizer (e.g. with per-group lr/wd)."""
        return {"params": self.parameters()}

    def predict_action_chunk(self, batch: dict[str, torch.Tensor], **kwargs) -> torch.Tensor:
        """Return the full action chunk (B, chunk_size, action_dim) for the current observation."""
        ...

    def select_action(self, batch: dict[str, torch.Tensor], **kwargs) -> torch.Tensor:
        """Return a single action for the current timestep (called every step at inference)."""
        ...

    def forward(self, batch: dict[str, torch.Tensor]) -> tuple[torch.Tensor, dict | None]:
        """Compute the training loss.

        Returns `(loss, output_dict)`. `output_dict` may be `None`; everything in it must be
        logging-friendly Python natives (no tensors with gradients).

        `batch["action_is_pad"]` is a bool mask of shape (B, horizon) that marks
        timesteps padded because the episode ended before `horizon` steps; you
        can exclude those from your loss.
        """
        actions = batch[ACTION]
        action_is_pad = batch.get("action_is_pad")
        ...
        return loss, {"some_loss_component": some_loss_component.item()}

class MyPolicy(PreTrainedPolicy):
    ...
    _fsdp_wrap_modules = ["MyTransformerBlock"]

# processor_my_policy.py
from typing import Any
import torch

from lerobot.processor import PolicyAction, PolicyProcessorPipeline


def make_my_policy_pre_post_processors(
    config,
    dataset_stats: dict[str, dict[str, torch.Tensor]] | None = None,
) -> tuple[
    PolicyProcessorPipeline[dict[str, Any], dict[str, Any]],
    PolicyProcessorPipeline[PolicyAction, PolicyAction],
]:
    preprocessor = ...   # build your PolicyProcessorPipeline for inputs
    postprocessor = ...  # build your PolicyProcessorPipeline for outputs
    return preprocessor, postprocessor

from lerobot.processor import load_pretrained_policy_processors
from lerobot.utils.constants import (
    POLICY_POSTPROCESSOR_DEFAULT_NAME,
    POLICY_PREPROCESSOR_DEFAULT_NAME,
)


def make_my_policy_pre_post_processors_from_pretrained(
    config,
    pretrained_path: str,
    *,
    revision: str | None = None,
    dataset_stats: dict[str, dict[str, torch.Tensor]] | None = None,
    dataset_meta: Any | None = None,
    preprocessor_overrides: dict[str, Any] | None = None,
    postprocessor_overrides: dict[str, Any] | None = None,
    preprocessor_config_filename: str = f"{POLICY_PREPROCESSOR_DEFAULT_NAME}.json",
    postprocessor_config_filename: str = f"{POLICY_POSTPROCESSOR_DEFAULT_NAME}.json",
) -> tuple[
    PolicyProcessorPipeline[dict[str, Any], dict[str, Any]],
    PolicyProcessorPipeline[PolicyAction, PolicyAction],
]:
    preprocessor, postprocessor = load_pretrained_policy_processors(
        pretrained_path,
        revision=revision,
        preprocessor_overrides=preprocessor_overrides,
        postprocessor_overrides=postprocessor_overrides,
        preprocessor_config_filename=preprocessor_config_filename,
        postprocessor_config_filename=postprocessor_config_filename,
    )
    ...  # reconcile the loaded pipelines with the current `config`
    return preprocessor, postprocessor

from lerobot.utils.constants import ACTION, PREDICTION


class MyPolicy(PreTrainedPolicy):
    def select_action(self, batch, **kwargs) -> Tensor | PolicyOutput:
        if not self._action_queue:
            chunk, extras = self.model.sample(batch)
            self._action_queue.extend(chunk.transpose(0, 1))
            if self.config.return_prediction:  # a config field of your choosing
                # The prediction goes with the step that computed it; the others return a bare action.
                prediction = {
                    "observation": {"observation.images.top": extras.next_frames},  # [B, C, H, W]
                    "language": {"subtask": extras.subtasks},  # list[str], one per env
                    "boxes": {
                        "observation.images.top": [
                            {"detections": [{"label": "cube", "bbox": xyxy}]} for xyxy in extras.cube_xyxy
                        ]
                    },
                }
                return {ACTION: self._action_queue.popleft(), PREDICTION: prediction}
        return self._action_queue.popleft()

input_steps = [
    RenderRuntimeMessagesStep(config.recipe),
    RenderTrainingMessagesStep(config.recipe),
    # observation transforms
    # policy-native roles, image placement, chat template, and tokenization
    # device placement
]

class MyPolicy(PreTrainedPolicy):
    def supports_text_generation(self) -> bool:
        return True

    def generate_text(self, batch) -> str:
        token_ids = self.model.generate(**batch)
        return self.tokenizer.decode(token_ids, skip_special_tokens=True)

lerobot_policy_my_policy/
├── pyproject.toml
└── src/
    └── lerobot_policy_my_policy/
        ├── __init__.py
        ├── configuration_my_policy.py
        ├── modeling_my_policy.py
        └── processor_my_policy.py

[project]
name = "lerobot_policy_my_policy"
version = "0.1.0"
dependencies = [
    # your policy-specific dependencies
]
requires-python = ">= 3.12"

[build-system]
build-backend = # your-build-backend
requires = # your-build-system

# __init__.py
"""Custom policy package for LeRobot."""

try:
    import lerobot  # noqa: F401
except ImportError:
    raise ImportError(
        "lerobot is not installed. Please install lerobot to use this policy package."
    )

from .configuration_my_policy import MyPolicyConfig
from .modeling_my_policy import MyPolicy
from .processor_my_policy import make_my_policy_pre_post_processors

__all__ = [
    "MyPolicyConfig",
    "MyPolicy",
    "make_my_policy_pre_post_processors",
]

cd lerobot_policy_my_policy
pip install -e .

# Or install from PyPI if published
pip install lerobot_policy_my_policy

lerobot-train \
    --policy.type my_policy \
    --dataset.repo_id lerobot/pusht \
    --env.type pusht \
    --steps 200000 \
    --policy.push_to_hub false

src/lerobot/policies/my_policy/
├── __init__.py                    # re-exports config + modeling + processor factory
├── configuration_my_policy.py     # MyPolicyConfig + @register_subclass
├── modeling_my_policy.py          # MyPolicy(PreTrainedPolicy)
├── processor_my_policy.py         # make_my_policy_pre_post_processors
└── README.md                      # symlink → ../../../../docs/source/policy_my_policy_README.md

from typing import TYPE_CHECKING
from lerobot.utils.import_utils import _diffusers_available, require_package

if TYPE_CHECKING or _diffusers_available:
    from diffusers.schedulers.scheduling_ddim import DDIMScheduler
else:
    DDIMScheduler = None  # keeps the symbol bindable at import time

class DiffusionPolicy(PreTrainedPolicy):
    def __init__(self, config):
        require_package("diffusers", extra="diffusion")
        super().__init__(config)
        ...

## Results

Evaluated on LIBERO with `lerobot/<policy>_libero`:

| Suite          | Success rate | n_episodes |
| -------------- | -----------: | ---------: |
| libero_spatial |        87.5% |         50 |
| libero_object  |        93.0% |         50 |
| libero_goal    |        81.5% |         50 |
| libero_10      |        62.0% |         50 |
| **average**    |    **81.0%** |        200 |

Reproduce: `lerobot-eval --policy.path=lerobot/<policy>_libero --env.type=libero --env.task=libero_spatial --eval.n_episodes=50` (1× A100 40 GB).