online

online

Online inference explanation source.

OnlineInferenceExplanationSource

OnlineInferenceExplanationSource(consumer, sample_stride=1, max_samples=None)

Adapts ready inference windows into explanation batches.

Initialize the online inference adapter.

Parameters:

Name	Type	Description	Default
consumer	ExplanationBatchConsumer	Object that generates explanations for each accepted inference window.	required
sample_stride	int	Explain every Nth inference timestep.	1
max_samples	int | None	Optional cap on the number of ready inference windows sent to the consumer.	None

Raises:

Type	Description
ValueError	If sample_stride or max_samples is invalid.

Source code in src/versatil/explainability/sources/online.py

def __init__(
    self,
    consumer: ExplanationBatchConsumer,
    sample_stride: int = 1,
    max_samples: int | None = None,
) -> None:
    """Initialize the online inference adapter.

    Args:
        consumer: Object that generates explanations for each accepted
            inference window.
        sample_stride: Explain every Nth inference timestep.
        max_samples: Optional cap on the number of ready inference windows
            sent to the consumer.

    Raises:
        ValueError: If ``sample_stride`` or ``max_samples`` is invalid.
    """
    if sample_stride <= 0:
        raise ValueError(f"sample_stride must be positive. Got: {sample_stride}")
    if max_samples is not None and max_samples <= 0:
        raise ValueError(
            f"max_samples must be positive when set. Got: {max_samples}"
        )
    self.consumer = consumer
    self.sample_stride = sample_stride
    self.max_samples = max_samples
    self.explained_sample_count = 0

explain_observation_batch

explain_observation_batch(observation, display_observation, environment_indices, timestep)

Generate explanations for one ready online inference batch.

Parameters:

Name	Type	Description	Default
observation	ObservationBatch	The exact observation batch passed to PolicyRuntime.run_inference.	required
display_observation	dict[str, Tensor]	Camera tensors for overlays.	required
environment_indices	list[int]	Environment indices represented by the batch rows.	required
timestep	int	Inference client timestep.	required

Source code in src/versatil/explainability/sources/online.py

def explain_observation_batch(
    self,
    observation: ObservationBatch,
    display_observation: dict[str, torch.Tensor],
    environment_indices: list[int],
    timestep: int,
) -> None:
    """Generate explanations for one ready online inference batch.

    Args:
        observation: The exact observation batch passed to
            ``PolicyRuntime.run_inference``.
        display_observation: Camera tensors for overlays.
        environment_indices: Environment indices represented by the batch
            rows.
        timestep: Inference client timestep.
    """
    if timestep % self.sample_stride != 0:
        return
    batch_size = len(environment_indices)
    accepted_sample_count = self._resolve_accepted_sample_count(
        batch_size=batch_size
    )
    if accepted_sample_count == 0:
        return
    accepted_environment_indices = environment_indices[:accepted_sample_count]
    accepted_observation = dict(observation)
    accepted_display_observation = display_observation
    if accepted_sample_count < batch_size:
        accepted_observation = self._slice_observation_batch(
            observation=observation,
            sample_count=accepted_sample_count,
            batch_size=batch_size,
        )
        accepted_display_observation = self._slice_display_observation(
            display_observation=display_observation,
            sample_count=accepted_sample_count,
            batch_size=batch_size,
        )
    self.consumer.explain_batch(
        batch=ExplanationBatch(
            observation=accepted_observation,
            actions=None,
            display_observation={
                key: value.detach().cpu()
                for key, value in accepted_display_observation.items()
            },
            metadata={
                "source": ExplanationSourceType.ONLINE_INFERENCE.value,
                "environment_indices": accepted_environment_indices,
                "timestep": timestep,
            },
            preprocess_observation=True,
        )
    )
    self.explained_sample_count += accepted_sample_count