synthetic

synthetic

Dataset schema for synthetic multimodal benchmark tasks.

SyntheticSchema

SyntheticSchema(zarr_path, metadata, dataset_type=value, task_name=value, num_episodes=1000, seed=42, image_size=64, num_modes=2, trajectory_length=60, noise_std=0.01, num_styles=1, mode_weights=None, num_rollouts=200)

Bases: DatasetSchema

Schema for procedurally generated synthetic 2D navigation datasets.

Unlike CSV or HDF5 schemas, this schema has no raw data files to read. Episodes are generated on-the-fly via the synthetic generators module and written directly to Zarr.

Initialize and validate the synthetic benchmark schema.

Parameters:

Name	Type	Description	Default
zarr_path	str	Path to save/load the zarr store.	required
metadata	DatasetMetadata	Metadata for zarr array creation.	required
dataset_type	str	Type of dataset. Must be 'synthetic'.	value
task_name	str	SyntheticTaskName.value string identifying the task.	value
num_episodes	int	Total episodes to generate, balanced across modes.	1000
seed	int	Random seed for reproducible generation.	42
image_size	int	Side length in pixels of rendered images (square).	64
num_modes	int	Number of behavioral modes.	2
trajectory_length	int	Number of timesteps per episode.	60
noise_std	float	Standard deviation of Gaussian trajectory noise.	0.01
num_styles	int	Number of sinusoidal style variations per corridor.	1
mode_weights	list[float] | None	Per-mode sampling weights. None for uniform.	None
num_rollouts	int	Rollouts per evaluation in the training callback.	200

Source code in src/versatil/data/raw/schemas/custom/synthetic.py

def __init__(
    self,
    zarr_path: str,
    metadata: DatasetMetadata,
    dataset_type: str = DatasetType.SYNTHETIC.value,
    task_name: str = SyntheticTaskName.CIRCLE.value,
    num_episodes: int = 1000,
    seed: int = 42,
    image_size: int = 64,
    num_modes: int = 2,
    trajectory_length: int = 60,
    noise_std: float = 0.01,
    num_styles: int = 1,
    mode_weights: list[float] | None = None,
    num_rollouts: int = 200,
):
    """Initialize and validate the synthetic benchmark schema.

    Args:
        zarr_path: Path to save/load the zarr store.
        metadata: Metadata for zarr array creation.
        dataset_type: Type of dataset. Must be 'synthetic'.
        task_name: SyntheticTaskName.value string identifying the task.
        num_episodes: Total episodes to generate, balanced across modes.
        seed: Random seed for reproducible generation.
        image_size: Side length in pixels of rendered images (square).
        num_modes: Number of behavioral modes.
        trajectory_length: Number of timesteps per episode.
        noise_std: Standard deviation of Gaussian trajectory noise.
        num_styles: Number of sinusoidal style variations per corridor.
        mode_weights: Per-mode sampling weights. None for uniform.
        num_rollouts: Rollouts per evaluation in the training callback.
    """
    if dataset_type != DatasetType.SYNTHETIC.value:
        raise ValueError(
            f"SyntheticSchema only supports dataset_type='{DatasetType.SYNTHETIC.value}', "
            f"got '{dataset_type}'"
        )
    self.task_name = task_name
    self.num_episodes = num_episodes
    self.seed = seed
    self.image_size = image_size
    self.num_modes = num_modes
    self.trajectory_length = trajectory_length
    self.noise_std = noise_std
    self.num_styles = num_styles
    self.mode_weights = mode_weights
    self.num_rollouts = num_rollouts
    self._validate_metadata(metadata)
    super().__init__(
        zarr_path=zarr_path,
        metadata=metadata,
        dataset_type=dataset_type,
    )

get_callbacks

get_callbacks(experiment_config)

Provide a rollout evaluation callback for synthetic training.

Source code in src/versatil/data/raw/schemas/custom/synthetic.py

def get_callbacks(self, experiment_config: ExperimentConfig) -> list[Callback]:
    """Provide a rollout evaluation callback for synthetic training."""
    return [
        SyntheticRolloutCallback(
            task_name=self.task_name,
            num_modes=self.num_modes,
            num_styles=self.num_styles,
            trajectory_length=self.trajectory_length,
            noise_std=self.noise_std,
            num_rollouts=self.num_rollouts,
            image_size=self.image_size,
            log_every_n_epochs=experiment_config.val_every,
        )
    ]

extract_episode

extract_episode(episode_source)

Not applicable for synthetic data.

Synthetic episodes are generated procedurally, not extracted from raw files. Use create_zarr_from_synthetic instead.

Raises:

Type	Description
NotImplementedError	Always, since synthetic data is generated not extracted.

Source code in src/versatil/data/raw/schemas/custom/synthetic.py

def extract_episode(
    self,
    episode_source: Any,
) -> dict[str, np.ndarray]:
    """Not applicable for synthetic data.

    Synthetic episodes are generated procedurally, not extracted from
    raw files. Use create_zarr_from_synthetic instead.

    Raises:
        NotImplementedError: Always, since synthetic data is generated not extracted.
    """
    raise NotImplementedError(
        "SyntheticSchema does not support extract_episode(). "
        "Use create_zarr_from_synthetic.create_replay_buffer_from_synthetic() instead."
    )