action_postprocessor

action_postprocessor

Action postprocessing for the inference pipeline.

ActionPostprocessor

ActionPostprocessor(action_space, denoising_thresholds)

Converts raw policy output tensors into structured action dicts for the server.

Handles gripper postprocessing (sigmoid for binary), denoising thresholds, and action metadata construction from the action space.

Initialize the action postprocessor.

Parameters:

Name	Type	Description	Default
action_space	ActionSpace	Policy's action space with metadata per key.	required
denoising_thresholds	dict[str, float]	Per-VersatIL-key denoising thresholds.	required

Raises:

Type	Description
ValueError	If two predicted action keys share an action type; the wire format is keyed by ActionComponent and cannot express duplicates.

Source code in src/versatil/inference/action_postprocessor.py

def __init__(
    self,
    action_space: ActionSpace,
    denoising_thresholds: dict[str, float],
):
    """Initialize the action postprocessor.

    Args:
        action_space: Policy's action space with metadata per key.
        denoising_thresholds: Per-VersatIL-key denoising thresholds.

    Raises:
        ValueError: If two predicted action keys share an action type;
            the wire format is keyed by ActionComponent and cannot
            express duplicates.
    """
    action_types: dict[str, str] = {}
    for key, metadata in action_space.actions_metadata.items():
        if not metadata.requires_prediction_head:
            continue
        existing_key = action_types.get(metadata.action_type)
        if existing_key is not None:
            raise ValueError(
                f"Predicted action keys '{existing_key}' and '{key}' both "
                f"map to action type '{metadata.action_type}'; the "
                "structured action format is keyed by action type and "
                "would silently drop one of them."
            )
        action_types[metadata.action_type] = key
    self.action_space = action_space
    self.denoising_thresholds = denoising_thresholds

format_action

format_action(action_dict)

Format action tensors into a structured dict keyed by ActionComponent.

Parameters:

Name	Type	Description	Default
action_dict	dict[str, Tensor]	Dict mapping VersatIL action key to tensor.	required

Returns:

Type	Description
dict[str, list[float]]	Dict mapping ActionComponent value to action values list.

Source code in src/versatil/inference/action_postprocessor.py

def format_action(
    self, action_dict: dict[str, torch.Tensor]
) -> dict[str, list[float]]:
    """Format action tensors into a structured dict keyed by ActionComponent.

    Args:
        action_dict: Dict mapping VersatIL action key to tensor.

    Returns:
        Dict mapping ActionComponent value to action values list.
    """
    components: dict[str, list[float]] = {}
    for key, metadata in self.action_space.actions_metadata.items():
        if not metadata.requires_prediction_head:
            continue
        value = action_dict[key].cpu().detach().float().numpy().flatten()
        if metadata.action_type == ActionComponent.GRIPPER.value:
            value = self._postprocess_gripper_action(
                raw_value=value, action_meta=metadata
            )
        threshold = self.denoising_thresholds.get(key)
        if threshold is not None and np.linalg.norm(value) < threshold:
            value = np.zeros_like(value)
        components[metadata.action_type] = value.tolist()
    return components

build_action_metadata

build_action_metadata()

Build action metadata dict keyed by ActionComponent.

Returns:

Type	Description
dict[str, dict[str, str | int]]	Dict mapping ActionComponent value to metadata entry.

Source code in src/versatil/inference/action_postprocessor.py

def build_action_metadata(self) -> dict[str, dict[str, str | int]]:
    """Build action metadata dict keyed by ActionComponent.

    Returns:
        Dict mapping ActionComponent value to metadata entry.
    """
    metadata: dict[str, dict[str, str | int]] = {}
    for _key, action_meta in self.action_space.actions_metadata.items():
        if not action_meta.requires_prediction_head:
            continue
        entry: dict[str, str | int] = {
            ActionMetadataField.DIMENSION.value: action_meta.prediction_dimension,
        }
        self._add_action_type_metadata(action_meta=action_meta, entry=entry)
        self._add_frame_metadata(action_meta=action_meta, entry=entry)
        self._add_orientation_metadata(action_meta=action_meta, entry=entry)
        self._add_gripper_metadata(action_meta=action_meta, entry=entry)
        metadata[action_meta.action_type] = entry
    return metadata