base

base

Base runtime wrapper for policy inference.

PolicyRuntime

PolicyRuntime(checkpoint_loader, client_identifier)

Bases: ABC

Base class for policy inference runtimes.

Initialize runtime metadata delegation.

Source code in src/versatil/inference/policy_runtime/base.py

def __init__(
    self,
    checkpoint_loader: BaseCheckpointLoader,
    client_identifier: str,
) -> None:
    """Initialize runtime metadata delegation."""
    self.checkpoint_loader = checkpoint_loader
    self._client_identifier = client_identifier
    self._policy = checkpoint_loader.policy

policy property

policy

Get the runtime policy module.

device property

device

Device where policy inference runs.

checkpoint_path property

checkpoint_path

Path to the loaded checkpoint.

client_identifier property

client_identifier

Stable identifier used when registering with a server.

config property

config

Loaded training configuration.

tokenizer property

tokenizer

Tokenizer saved with the checkpoint, if present.

observation_space property

observation_space

Observation space expected by the policy.

action_space property

action_space

Action space produced by the policy.

prediction_horizon property

prediction_horizon

Number of future action steps predicted per policy call.

observation_horizon property

observation_horizon

Number of observation steps consumed per policy call.

denoising_thresholds property

denoising_thresholds

Per-action denoising thresholds loaded from the checkpoint.

depth_clamp_ranges property

depth_clamp_ranges

Per-camera min/max ranges for clamping depth observations.

run_inference abstractmethod

run_inference(obs_dict)

Run inference on preprocessed observations.

Source code in src/versatil/inference/policy_runtime/base.py

@abstractmethod
def run_inference(
    self,
    obs_dict: dict[str, torch.Tensor],
) -> dict[str, torch.Tensor]:
    """Run inference on preprocessed observations."""