Custom data processing

While the Incar Skill System already provides some common processing functionality out of the box, it would be impossible to cover every single usecase. Luckily, it is not that difficult to extend the data processing pipeline! You can write an extension containing a ProcessStep subclass, allowing you to process your data at various places within the processing pipeline.

LinuxWindows

# ensure that your python venv is sourced
incar create_pkg --name [package_name] --processing [path]

# ensure that your python venv is sourced
incar create_pkg --name [package_name] --processing [path]

Ensure that the package name is unique to avoid conflicts with other extensions. path is the location where the extension will be created.

Defining a process step

Let's take a look at the commented template that is now generated for you:

[package_name]_processing/steps.py
from incar.extensions.processing_step import ProcessStep, ProcessHook
from dataclasses import dataclass

@ProcessStep.register_subclass("my_step")
@dataclass
class MyStep(ProcessStep):
    # hooks are explained further down! Here, default hooks can be defined, which can still
    # be overwritten in config files.
    hooks: List[ProcessHook] = field(default_factory = lambda: [ProcessHook.DATASET, ProcessHook.OBSERVATION])
    feature: str = ""
    other_settings: ...

    def __post_init__(self):
        # here we can import externals (e.g. SAM, AnyDepth, ...). This way they
        # only get imported if the step is used.

    def process_single_frame(self, frame: dict[str, np.ndarray]) -> None:
        # here, we can modify a single frame in place.
        # the dict contains all features as keys, and their data as values.
        # This function is called for the OBSERVATION, TELEOP_COMMAND and INFERENCE_COMMAND hooks

    def process_sequence(self, frames: dict[str, np.ndarray]) -> None:
        # here, we can modify the sequence of frames in place.
        # the dict contains all features as keys, and their data as values.
        # the array has the shape [sequence_length, *feature_data_shape]
        # This function is called for the GET_ITEM, GET_ITEM_VALIDATION and POST_INFERENCE hooks

    def process_dataset(self, root_path: str, config: DatasetConfig) -> None:
        # Here, we get the path to the copied dataset and the dataset config.
        # Data is changed by changing the actual files. For example:
        demos = [ f.name for f in os.scandir(root_path) if f.is_dir()]
        demos = sorted(demos, key=lambda x: int(x.split('_')[-1]))

        for demo in tqdm(demos, desc="Processing MyStep"):
            # Get meta-data for the demos
            with open(Path(root_path) / demo / DEMO_META_NAME, "r") as f:
                demo_meta = json.load(f)

            # Get the data of a feature
            feature_path = Path(root_path) / demo / self.feature
            with h5py.File(feature_path / "data.h5", "r") as f:
                feature_data = f['data'][:]

            new_data = ... # transform feature_data

            # Set the data of a feature
            with h5py.File(feature_path / "data.h5", "r+") as f:
                f['data'][:] = new_data

            # If the dataset description changed, e.g. feature shape changed or
            # a new feature was created, don't forget to reflect this change
            # within `config`
            config.features[feature].shape = ...

Success

if possible, put the actual data transformation in a separate function called e.g. operation that the other hooks can call. This way you can maintain consistency in your data processing and reduce hard to catch bugs.

Tip

note that we can modify the whole frame or frames dictionary in place. That means we can also add new features based on other features.

Pipeline hooks

The following hooks exist to add a processing step:

DATASET: This hook can be used when a process step has to be applied to the whole dataset, prior to starting training (a copy of the dataset is always made before any processing occurs). Ideal for e.g. sampling the data to some dt, or downsampling video data.
GET_ITEM: This hook is applied during training when retrieving a batch. Ideal for e.g. applying noise transformations to video data.
GET_ITEM_VALIDATION: This hook is applied when retrieving a batch for validation during the training process.
OBSERVATION: This hook is applied to incoming observation data during inference.
POST_INFERENCE: This hook is applied to the sequence of actions that results from each inference.
TELEOP_COMMAND: This hook is applied to every single user teleoperated command that is being sent to the robot. Ideal for e.g. gains or limiting (gains and limiting are configured by default)
INFERENCE_COMMAND: This hook is applied to every single predicted command that is being sent to the robot. Also Ideal for e.g. gains or limiting.

A process step can be added to multiple hooks, so that behaviour stays consistent. For example, a downsample_video step can be added to both the DATASET and OBSERVATION hooks, to ensure the exact same data processing is achieved for training and inference.

Note that the steps for all hooks except for TELEOP_COMMAND and INFERENCE_COMMAND are part of the policy settings, and thus defined in the policy configs. Steps that hook into TELEOP_COMMAND and INFERENCE_COMMAND have to be added to the workspace_config.

Examples

This might be a bit abstract, so for a better understanding take a look at the following examples to get inspiration and a further understanding:

Using a processing step

Create an extension containing the processing step implementation. Then pip install the extension.

Now, the step can be used just as any other processing step already included in the Incar Skill System. For example, the step above can be used by including the following in your training config:

{
    "policy" : {
        "preprocessing": {
            "steps": [
                {
                    "type": "sample_dt",
                    "dt": 0.1
                },
                {
                    "type": "downsample_video",
                    "features": ["wrist_cam"],
                    "new_size": [240, 320]
                },
                {
                    "type": "my_step",
                    "feature": "specific_feature",
                    "other_settings": ...
                }
                ...
            ],
            ...
        },
        ...
    },
    ...
}

Note

Even though we set the default hooks, we can override them in the config:

{
    "type": "my_step",
    "hooks": ["GET_ITEM"],
    "feature": "specific_feature",
    "other_settings": ...
}

The type value is the same as defined in the @ProcessStep.register_subclass() wrapper, and all other available fields are the fields as defined in the ProcessStep subclass you created. In this example, that is feature, hooks and other_settings. If a field is not set, it will use the standard value as defined in the ProcessStep subclass.

Warning

steps that use the hooks TELEOP_COMMAND or INFERENCE_COMMAND have to be added in the workspace_config.json, as these are independent of the policies

workspace_config.json

{
    "command_processing": [
        {
            "type": "inspire_gripper",
            "hooks": ["TELEOP_COMMAND", "INFERENCE_COMMAND"]
        },
        ...
    ],
    ...
}