Pipeline

dataphy.visionpack.pipeline

Classes

BaseTransform(p: float = 1.0, apply_to: List[str] = None, sync_views: Optional[bool] = None, update_intrinsics: bool = False, mask_protect: List[str] = None, min_visible_mask_pct: float = 0.0, resample: Literal['nearest', 'bilinear', 'bicubic'] = 'bilinear', border_mode: Literal['reflect', 'replicate', 'constant'] = 'reflect', pad_mode: Literal['reflect', 'replicate', 'constant'] = 'reflect', pad_value: float = 0.0, seed_policy: Literal['per_frame', 'per_episode', 'global'] = 'per_frame', **kwargs: Any)

Bases: Module

Base class for all vision transforms with cross-cutting parameters.

Source code in src/dataphy/visionpack/tforms/base.py

def __init__(
    self,
    # Cross-cutting parameters
    p: float = 1.0,
    apply_to: List[str] = None,
    sync_views: Optional[bool] = None,
    update_intrinsics: bool = False,
    mask_protect: List[str] = None,
    min_visible_mask_pct: float = 0.0,
    resample: Literal["nearest", "bilinear", "bicubic"] = "bilinear",
    border_mode: Literal["reflect", "replicate", "constant"] = "reflect",
    pad_mode: Literal["reflect", "replicate", "constant"] = "reflect",
    pad_value: float = 0.0,
    seed_policy: Literal["per_frame", "per_episode", "global"] = "per_frame",
    **kwargs: Any
):
    super().__init__()
    self.p = p
    self.apply_to = apply_to if apply_to is not None else ["rgb"]
    self.sync_views = sync_views
    self.update_intrinsics = update_intrinsics
    self.mask_protect = mask_protect if mask_protect is not None else []
    self.min_visible_mask_pct = min_visible_mask_pct
    self.resample = resample
    self.border_mode = border_mode
    self.pad_mode = pad_mode
    self.pad_value = pad_value
    self.seed_policy = seed_policy

    # Store seed state for per-episode consistency
    self._episode_seed = None
    self._global_seed = random.getstate()

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

RandomCropPad(keep_ratio_min: float = 0.84, scale_range: Optional[List[float]] = None, aspect_ratio_range: Optional[List[float]] = None, center_bias: float = 0.0, min_object_iou: float = 0.0, antialias: bool = True, resize_to: Optional[List[int]] = None, **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/crop.py

def __init__(
    self,
    keep_ratio_min: float = 0.84,
    scale_range: Optional[List[float]] = None,
    aspect_ratio_range: Optional[List[float]] = None,
    center_bias: float = 0.0,
    min_object_iou: float = 0.0,
    antialias: bool = True,
    resize_to: Optional[List[int]] = None,
    **kwargs: Any
):
    super().__init__(**kwargs)

    # Use scale_range if provided, otherwise fall back to keep_ratio_min
    if scale_range is not None:
        self.scale_range = scale_range
    else:
        self.scale_range = [keep_ratio_min, 1.0]

    self.aspect_ratio_range = aspect_ratio_range or [0.98, 1.02]
    self.center_bias = center_bias
    self.min_object_iou = min_object_iou
    self.antialias = antialias
    self.resize_to = resize_to

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

scale_range = scale_range instance-attribute

aspect_ratio_range = aspect_ratio_range or [0.98, 1.02] instance-attribute

center_bias = center_bias instance-attribute

min_object_iou = min_object_iou instance-attribute

antialias = antialias instance-attribute

resize_to = resize_to instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

RandomTranslate(px: int = 8, units: Literal['px', 'percent'] = 'px', distribution: Literal['uniform', 'normal'] = 'uniform', std_px: Optional[float] = None, lock_axis: Literal['both', 'x', 'y'] = 'both', wrap: bool = False, same_delta_per_seq: bool = False, max_pad_pct: float = 1.0, **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/translate.py

def __init__(
    self,
    px: int = 8,
    units: Literal["px", "percent"] = "px",
    distribution: Literal["uniform", "normal"] = "uniform",
    std_px: Optional[float] = None,
    lock_axis: Literal["both", "x", "y"] = "both",
    wrap: bool = False,
    same_delta_per_seq: bool = False,
    max_pad_pct: float = 1.0,
    **kwargs: Any
):
    super().__init__(**kwargs)
    self.px = px
    self.units = units
    self.distribution = distribution
    self.std_px = std_px or (px / 3.0)  # Default std is px/3
    self.lock_axis = lock_axis
    self.wrap = wrap
    self.same_delta_per_seq = same_delta_per_seq
    self.max_pad_pct = max_pad_pct

    # Store per-sequence translation for consistency
    self._seq_translation = None
    self._last_episode_id = None

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

px = px instance-attribute

units = units instance-attribute

distribution = distribution instance-attribute

std_px = std_px or px / 3.0 instance-attribute

lock_axis = lock_axis instance-attribute

wrap = wrap instance-attribute

same_delta_per_seq = same_delta_per_seq instance-attribute

max_pad_pct = max_pad_pct instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

ColorJitter(magnitude: Optional[float] = None, brightness: Optional[float] = None, contrast: Optional[float] = None, saturation: Optional[float] = None, hue: Optional[float] = None, order: Optional[List[str]] = None, space: Literal['rgb', 'hsv', 'hsl'] = 'rgb', preserve_robot_color: bool = False, clip_range: Optional[List[float]] = None, gamma: Optional[float] = None, temperature_shift: Optional[float] = None, correlate_views: float = 0.0, **kwargs: Any)

Bases: BaseTransform

Color jittering transform for image augmentation.

This transform applies random color adjustments to images, including brightness, contrast, saturation, and hue changes. It supports multiple color spaces and can preserve robot colors in robotics datasets.

Parameters:

Name	Type	Description	Default
magnitude	Optional[float]	Global magnitude for all color adjustments (overrides individual params)	None
brightness	Optional[float]	Brightness adjustment range	None
contrast	Optional[float]	Contrast adjustment range	None
saturation	Optional[float]	Saturation adjustment range	None
hue	Optional[float]	Hue adjustment range	None
order	Optional[List[str]]	Order of color operations to apply	None
space	Literal['rgb', 'hsv', 'hsl']	Color space to work in ("rgb", "hsv", "hsl")	'rgb'
preserve_robot_color	bool	Whether to preserve robot colors using masks	False
clip_range	Optional[List[float]]	Range to clip output values	None
gamma	Optional[float]	Gamma correction value	None
temperature_shift	Optional[float]	Color temperature shift	None
correlate_views	float	Probability of using correlated parameters across views	0.0
**kwargs	Any	Additional base transform parameters	{}

Source code in src/dataphy/visionpack/tforms/color.py

def __init__(
    self,
    magnitude: Optional[float] = None,
    brightness: Optional[float] = None,
    contrast: Optional[float] = None,
    saturation: Optional[float] = None,
    hue: Optional[float] = None,
    order: Optional[List[str]] = None,
    space: Literal["rgb", "hsv", "hsl"] = "rgb",
    preserve_robot_color: bool = False,
    clip_range: Optional[List[float]] = None,
    gamma: Optional[float] = None,
    temperature_shift: Optional[float] = None,
    correlate_views: float = 0.0,
    **kwargs: Any
):
    """Initialize the ColorJitter transform.

    Args:
        magnitude: Global magnitude for all color adjustments (overrides individual params)
        brightness: Brightness adjustment range
        contrast: Contrast adjustment range
        saturation: Saturation adjustment range
        hue: Hue adjustment range
        order: Order of color operations to apply
        space: Color space to work in ("rgb", "hsv", "hsl")
        preserve_robot_color: Whether to preserve robot colors using masks
        clip_range: Range to clip output values
        gamma: Gamma correction value
        temperature_shift: Color temperature shift
        correlate_views: Probability of using correlated parameters across views
        **kwargs: Additional base transform parameters
    """
    super().__init__(**kwargs)  # type: ignore

    # Use individual parameters if provided, otherwise fall back to magnitude
    if magnitude is not None and all(x is None for x in [brightness, contrast, saturation, hue]):
        self.brightness = magnitude
        self.contrast = magnitude
        self.saturation = magnitude
        self.hue = magnitude
    else:
        self.brightness = brightness or 0.0
        self.contrast = contrast or 0.0
        self.saturation = saturation or 0.0
        self.hue = hue or 0.0

    self.order = order or ["brightness", "contrast", "saturation", "hue"]
    self.space = space
    self.preserve_robot_color = preserve_robot_color
    self.clip_range = clip_range or [0.0, 1.0]
    self.gamma = gamma
    self.temperature_shift = temperature_shift
    self.correlate_views = correlate_views

    # Store correlated parameters for multi-view consistency
    self._correlated_params = {}
    self._last_episode_id = None

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

brightness = magnitude instance-attribute

contrast = magnitude instance-attribute

saturation = magnitude instance-attribute

hue = magnitude instance-attribute

order = order or ['brightness', 'contrast', 'saturation', 'hue'] instance-attribute

space = space instance-attribute

preserve_robot_color = preserve_robot_color instance-attribute

clip_range = clip_range or [0.0, 1.0] instance-attribute

gamma = gamma instance-attribute

temperature_shift = temperature_shift instance-attribute

correlate_views = correlate_views instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

RandomConv(kernel_variance: float = 0.05, kernel_size: int = 3, per_channel: bool = True, normalize_gain: bool = True, alpha_identity: Optional[List[float]] = None, sample_mode: Literal['gaussian', 'bank', 'hybrid'] = 'gaussian', kernel_bank: Optional[List] = None, bank_prob: float = 0.3, passes: int = 1, boundary: Literal['reflect', 'replicate', 'zeros'] = 'reflect', **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/conv.py

def __init__(
    self,
    kernel_variance: float = 0.05,
    kernel_size: int = 3,
    per_channel: bool = True,
    normalize_gain: bool = True,
    alpha_identity: Optional[List[float]] = None,
    sample_mode: Literal["gaussian", "bank", "hybrid"] = "gaussian",
    kernel_bank: Optional[List] = None,
    bank_prob: float = 0.3,
    passes: int = 1,
    boundary: Literal["reflect", "replicate", "zeros"] = "reflect",
    **kwargs: Any
):
    super().__init__(**kwargs)
    self.kernel_variance = kernel_variance
    self.kernel_size = kernel_size
    self.per_channel = per_channel
    self.normalize_gain = normalize_gain
    self.alpha_identity = alpha_identity or [0.0, 1.0]
    self.sample_mode = sample_mode
    self.kernel_bank = kernel_bank or self._create_default_kernel_bank()
    self.bank_prob = bank_prob
    self.passes = passes
    self.boundary = boundary

    assert kernel_size in [3, 5], "kernel_size must be 3 or 5"
    self.padding = kernel_size // 2

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

kernel_variance = kernel_variance instance-attribute

kernel_size = kernel_size instance-attribute

per_channel = per_channel instance-attribute

normalize_gain = normalize_gain instance-attribute

alpha_identity = alpha_identity or [0.0, 1.0] instance-attribute

sample_mode = sample_mode instance-attribute

kernel_bank = kernel_bank or self._create_default_kernel_bank() instance-attribute

bank_prob = bank_prob instance-attribute

passes = passes instance-attribute

boundary = boundary instance-attribute

padding = kernel_size // 2 instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

Cutout(holes: int = 2, size_range: Union[Tuple[int, int], List[int]] = (16, 32), size_units: Literal['px', 'percent'] = 'px', aspect_ratio_range: Optional[List[float]] = None, shape: Literal['rect', 'ellipse', 'stripe', 'grid'] = 'rect', rotation_deg: float = 0.0, fill_mode: Literal['zeros', 'mean', 'random', 'noise', 'blur'] = 'zeros', fill_value: Optional[float] = None, soft_edges: float = 0.0, position_bias: Literal['center', 'edges', 'uniform'] = 'uniform', target_masks_only: bool = False, depth_fill: Literal['nan', 'const', 'interp'] = 'nan', occlusion_budget: float = 1.0, **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/cutout.py

def __init__(
    self,
    holes: int = 2,
    size_range: Union[Tuple[int, int], List[int]] = (16, 32),
    size_units: Literal["px", "percent"] = "px",
    aspect_ratio_range: Optional[List[float]] = None,
    shape: Literal["rect", "ellipse", "stripe", "grid"] = "rect",
    rotation_deg: float = 0.0,
    fill_mode: Literal["zeros", "mean", "random", "noise", "blur"] = "zeros",
    fill_value: Optional[float] = None,
    soft_edges: float = 0.0,
    position_bias: Literal["center", "edges", "uniform"] = "uniform",
    target_masks_only: bool = False,
    depth_fill: Literal["nan", "const", "interp"] = "nan",
    occlusion_budget: float = 1.0,
    **kwargs: Any
):
    super().__init__(**kwargs)
    self.holes = holes
    self.size_range = list(size_range) if isinstance(size_range, tuple) else size_range
    self.size_units = size_units
    self.aspect_ratio_range = aspect_ratio_range or [0.75, 1.33]
    self.shape = shape
    self.rotation_deg = rotation_deg
    self.fill_mode = fill_mode
    self.fill_value = fill_value if fill_value is not None else 0.0
    self.soft_edges = soft_edges
    self.position_bias = position_bias
    self.target_masks_only = target_masks_only
    self.depth_fill = depth_fill
    self.occlusion_budget = occlusion_budget

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

holes = holes instance-attribute

size_range = list(size_range) if isinstance(size_range, tuple) else size_range instance-attribute

size_units = size_units instance-attribute

aspect_ratio_range = aspect_ratio_range or [0.75, 1.33] instance-attribute

shape = shape instance-attribute

rotation_deg = rotation_deg instance-attribute

fill_mode = fill_mode instance-attribute

fill_value = fill_value if fill_value is not None else 0.0 instance-attribute

soft_edges = soft_edges instance-attribute

position_bias = position_bias instance-attribute

target_masks_only = target_masks_only instance-attribute

depth_fill = depth_fill instance-attribute

occlusion_budget = occlusion_budget instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

RandomScale(scale_range: List[float] = [0.9, 1.1], preserve_aspect: bool = True, **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/scale.py

def __init__(
    self,
    scale_range: List[float] = [0.9, 1.1],
    preserve_aspect: bool = True,
    **kwargs: Any
):
    super().__init__(**kwargs)
    self.scale_range = scale_range
    self.preserve_aspect = preserve_aspect

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

scale_range = scale_range instance-attribute

preserve_aspect = preserve_aspect instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

RandomRotate(deg: float = 7.0, **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/rotate.py

def __init__(
    self,
    deg: float = 7.0,
    **kwargs: Any
):
    super().__init__(**kwargs)
    self.deg = deg

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

deg = deg instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

RandomAffine(shear_deg: float = 5.0, perspective: float = 0.05, **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/affine.py

def __init__(
    self,
    shear_deg: float = 5.0,
    perspective: float = 0.05,
    **kwargs: Any
):
    super().__init__(**kwargs)
    self.shear_deg = shear_deg
    self.perspective = perspective

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

shear_deg = shear_deg instance-attribute

perspective = perspective instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

RandomHFlip(allow_when: Optional[List[str]] = None, **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/flip.py

def __init__(
    self,
    allow_when: Optional[List[str]] = None,
    **kwargs: Any
):
    super().__init__(**kwargs)
    self.allow_when = allow_when or []

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

allow_when = allow_when or [] instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

WBShift(kelvin_shift: Optional[float] = None, mired_shift: Optional[float] = None, per_channel_gain: List[float] = [0.95, 1.05], preserve_robot_color: bool = True, **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/wb_shift.py

def __init__(
    self,
    kelvin_shift: Optional[float] = None,
    mired_shift: Optional[float] = None,
    per_channel_gain: List[float] = [0.95, 1.05],
    preserve_robot_color: bool = True,
    **kwargs: Any
):
    super().__init__(**kwargs)
    self.kelvin_shift = kelvin_shift or 1000.0
    self.mired_shift = mired_shift or 10.0
    self.per_channel_gain = per_channel_gain
    self.preserve_robot_color = preserve_robot_color

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

kelvin_shift = kelvin_shift or 1000.0 instance-attribute

mired_shift = mired_shift or 10.0 instance-attribute

per_channel_gain = per_channel_gain instance-attribute

preserve_robot_color = preserve_robot_color instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

Gamma(gamma: List[float] = [0.9, 1.1], per_channel: bool = False, clip_range: List[float] = [0.0, 1.0], **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/gamma.py

def __init__(
    self,
    gamma: List[float] = [0.9, 1.1],
    per_channel: bool = False,
    clip_range: List[float] = [0.0, 1.0],
    **kwargs: Any
):
    super().__init__(**kwargs)
    self.gamma_range = gamma if isinstance(gamma, list) else [gamma, gamma]
    self.per_channel = per_channel
    self.clip_range = clip_range

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

gamma_range = gamma if isinstance(gamma, list) else [gamma, gamma] instance-attribute

per_channel = per_channel instance-attribute

clip_range = clip_range instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

ToGrayscale(keep_channels: int = 3, **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/grayscale.py

def __init__(
    self,
    keep_channels: int = 3,
    **kwargs: Any
):
    super().__init__(**kwargs)
    self.keep_channels = keep_channels

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

keep_channels = keep_channels instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

MotionBlur(kernel: list = [3, 9], angle_deg: list = [0, 180], prob: float = 0.2, **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/blur.py

def __init__(
    self,
    kernel: list = [3, 9],
    angle_deg: list = [0, 180],
    prob: float = 0.2,
    **kwargs: Any
):
    super().__init__(**kwargs)
    self.kernel_range = kernel
    self.angle_range = angle_deg
    self.prob = prob

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

kernel_range = kernel instance-attribute

angle_range = angle_deg instance-attribute

prob = prob instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

GaussianBlur(sigma: list = [0.3, 1.2], kernel: list = [3, 7], **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/blur.py

def __init__(
    self,
    sigma: list = [0.3, 1.2],
    kernel: list = [3, 7],
    **kwargs: Any
):
    super().__init__(**kwargs)
    self.sigma_range = sigma
    self.kernel_range = kernel

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

sigma_range = sigma instance-attribute

kernel_range = kernel instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

GaussianNoise(sigma: list = [0.005, 0.03], per_channel: bool = True, **kwargs: Any)

Bases: BaseTransform

Gaussian noise transform for image augmentation.

This transform adds random Gaussian noise to images, simulating sensor noise or other random perturbations. It can apply noise per-channel or globally.

Parameters:

Name	Type	Description	Default
sigma	list	Range of noise standard deviation [min, max]	[0.005, 0.03]
per_channel	bool	Whether to apply different noise to each channel	True
**kwargs	Any	Additional base transform parameters	{}

Source code in src/dataphy/visionpack/tforms/noise.py

def __init__(
    self,
    sigma: list = [0.005, 0.03],
    per_channel: bool = True,
    **kwargs: Any
):
    """Initialize the GaussianNoise transform.

    Args:
        sigma: Range of noise standard deviation [min, max]
        per_channel: Whether to apply different noise to each channel
        **kwargs: Additional base transform parameters
    """
    super().__init__(**kwargs)
    self.sigma_range = sigma
    self.per_channel = per_channel

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

sigma_range = sigma instance-attribute

per_channel = per_channel instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

PoissonNoise(scale: list = [0.5, 1.5], **kwargs: Any)

Bases: BaseTransform

Poisson noise transform for image augmentation.

This transform adds Poisson (shot) noise to images, simulating photon noise in camera sensors. It includes error handling for edge cases and fallback to Gaussian approximation when needed.

Parameters:

Name	Type	Description	Default
scale	list	Range of scaling factors for noise intensity [min, max]	[0.5, 1.5]
**kwargs	Any	Additional base transform parameters	{}

Source code in src/dataphy/visionpack/tforms/noise.py

def __init__(
    self,
    scale: list = [0.5, 1.5],
    **kwargs: Any
):
    """Initialize the PoissonNoise transform.

    Args:
        scale: Range of scaling factors for noise intensity [min, max]
        **kwargs: Additional base transform parameters
    """
    super().__init__(**kwargs)
    self.scale_range = scale

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

scale_range = scale instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

SpeckleNoise(sigma: list = [0.01, 0.05], **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/noise.py

def __init__(
    self,
    sigma: list = [0.01, 0.05],
    **kwargs: Any
):
    super().__init__(**kwargs)
    self.sigma_range = sigma

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

sigma_range = sigma instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

SaltPepperNoise(amount: list = [0.001, 0.01], salt_vs_pepper: float = 0.5, **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/noise.py

def __init__(
    self,
    amount: list = [0.001, 0.01],
    salt_vs_pepper: float = 0.5,
    **kwargs: Any
):
    super().__init__(**kwargs)
    self.amount_range = amount
    self.salt_vs_pepper = salt_vs_pepper

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

amount_range = amount instance-attribute

salt_vs_pepper = salt_vs_pepper instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

JPEGCompress(quality: List[int] = [35, 90], **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/artifacts.py

def __init__(
    self,
    quality: List[int] = [35, 90],
    **kwargs: Any
):
    super().__init__(**kwargs)
    self.quality_range = quality

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

quality_range = quality instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

BitDepth(bits: List[int] = [8, 10, 12], **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/artifacts.py

def __init__(
    self,
    bits: List[int] = [8, 10, 12],
    **kwargs: Any
):
    super().__init__(**kwargs)
    self.bits_options = bits

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

bits_options = bits instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

Vignette(strength: List[float] = [0.0, 0.3], softness: List[float] = [0.2, 0.7], center_bias_vignette: List[float] = [-0.1, 0.1], **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/artifacts.py

def __init__(
    self,
    strength: List[float] = [0.0, 0.3],
    softness: List[float] = [0.2, 0.7],
    center_bias_vignette: List[float] = [-0.1, 0.1],
    **kwargs: Any
):
    super().__init__(**kwargs)
    self.strength_range = strength
    self.softness_range = softness
    self.center_bias_range = center_bias_vignette

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

strength_range = strength instance-attribute

softness_range = softness instance-attribute

center_bias_range = center_bias_vignette instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

LensDistortion(k1: List[float] = [-0.1, 0.1], k2: List[float] = [-0.05, 0.05], crop_or_pad: Literal['crop', 'pad'] = 'crop', **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/artifacts.py

def __init__(
    self,
    k1: List[float] = [-0.1, 0.1],
    k2: List[float] = [-0.05, 0.05],
    crop_or_pad: Literal["crop", "pad"] = "crop",
    **kwargs: Any
):
    super().__init__(**kwargs)
    self.k1_range = k1
    self.k2_range = k2
    self.crop_or_pad = crop_or_pad

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

k1_range = k1 instance-attribute

k2_range = k2 instance-attribute

crop_or_pad = crop_or_pad instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

OccluderPatch(count: List[int] = [0, 2], size_rel: List[float] = [0.05, 0.2], fill_mode: List[str] = ['mean', 'random', 'noise', 'patch_bank'], rotation_deg: float = 45.0, occlusion_budget: float = 0.15, patch_bank: Optional[List] = None, **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/occlusion.py

def __init__(
    self,
    count: List[int] = [0, 2],
    size_rel: List[float] = [0.05, 0.2],
    fill_mode: List[str] = ["mean", "random", "noise", "patch_bank"],
    rotation_deg: float = 45.0,
    occlusion_budget: float = 0.15,
    patch_bank: Optional[List] = None,
    **kwargs: Any
):
    super().__init__(**kwargs)
    self.count_range = count
    self.size_rel_range = size_rel
    self.fill_modes = fill_mode if isinstance(fill_mode, list) else [fill_mode]
    self.rotation_deg = rotation_deg
    self.occlusion_budget = occlusion_budget
    self.patch_bank = patch_bank or self._create_default_patch_bank()

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

count_range = count instance-attribute

size_rel_range = size_rel instance-attribute

fill_modes = fill_mode if isinstance(fill_mode, list) else [fill_mode] instance-attribute

rotation_deg = rotation_deg instance-attribute

occlusion_budget = occlusion_budget instance-attribute

patch_bank = patch_bank or self._create_default_patch_bank() instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

SyntheticShadow(dir_deg: List[float] = [0, 360], softness: List[float] = [0.0, 1.0], intensity: List[float] = [0.0, 0.6], shape: List[str] = ['ellipse', 'polyline'], **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/occlusion.py

def __init__(
    self,
    dir_deg: List[float] = [0, 360],
    softness: List[float] = [0.0, 1.0],
    intensity: List[float] = [0.0, 0.6],
    shape: List[str] = ["ellipse", "polyline"],
    **kwargs: Any
):
    super().__init__(**kwargs)
    self.dir_deg_range = dir_deg
    self.softness_range = softness
    self.intensity_range = intensity
    self.shapes = shape if isinstance(shape, list) else [shape]

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

dir_deg_range = dir_deg instance-attribute

softness_range = softness instance-attribute

intensity_range = intensity instance-attribute

shapes = shape if isinstance(shape, list) else [shape] instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

DepthNoise(sigma_mm_at_1m: List[float] = [1.0, 5.0], quad_scale: bool = True, missing_prob: List[float] = [0.0, 0.05], hole_fill: Literal['none', 'interp'] = 'none', profile: Literal['tof', 'stereo', 'structured_light', 'generic'] = 'generic', holes_prob: List[float] = None, edge_dropout_prob: List[float] = None, speckle: List[float] = None, guidance: Literal['none', 'rgb_edges'] = 'none', **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/depth.py

def __init__(
    self,
    sigma_mm_at_1m: List[float] = [1.0, 5.0],
    quad_scale: bool = True,
    missing_prob: List[float] = [0.0, 0.05],
    hole_fill: Literal["none", "interp"] = "none",
    # profile extensions
    profile: Literal["tof", "stereo", "structured_light", "generic"] = "generic",
    holes_prob: List[float] = None,
    edge_dropout_prob: List[float] = None,
    speckle: List[float] = None,
    guidance: Literal["none", "rgb_edges"] = "none",
    **kwargs: Any
):
    super().__init__(**kwargs)
    self.sigma_mm_at_1m_range = sigma_mm_at_1m
    self.quad_scale = quad_scale
    self.missing_prob_range = missing_prob
    self.hole_fill = hole_fill
    self.profile = profile
    self.holes_prob = holes_prob or [0.0, 0.0]
    self.edge_dropout_prob = edge_dropout_prob or [0.0, 0.0]
    self.speckle = speckle or [0.0, 0.0]
    self.guidance = guidance

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

sigma_mm_at_1m_range = sigma_mm_at_1m instance-attribute

quad_scale = quad_scale instance-attribute

missing_prob_range = missing_prob instance-attribute

hole_fill = hole_fill instance-attribute

profile = profile instance-attribute

holes_prob = holes_prob or [0.0, 0.0] instance-attribute

edge_dropout_prob = edge_dropout_prob or [0.0, 0.0] instance-attribute

speckle = speckle or [0.0, 0.0] instance-attribute

guidance = guidance instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

DepthQuantize(bits: List[int] = [10, 12, 16], **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/depth.py

def __init__(
    self,
    bits: List[int] = [10, 12, 16],
    **kwargs: Any
):
    super().__init__(**kwargs)
    self.bits_options = bits

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

bits_options = bits instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

LightingRand(ambient_tint: List[float] = None, directional_intensity: List[float] = None, direction_deg: List[float] = None, softness: List[float] = None, spot_count: List[int] = None, spot_radius: List[float] = None, spot_intensity: List[float] = None, shadow_prob: float = 0.0, preserve_robot_color: bool = False, **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/lighting_rand.py

def __init__(
    self,
    ambient_tint: List[float] = None,
    directional_intensity: List[float] = None,
    direction_deg: List[float] = None,
    softness: List[float] = None,
    spot_count: List[int] = None,
    spot_radius: List[float] = None,
    spot_intensity: List[float] = None,
    shadow_prob: float = 0.0,
    preserve_robot_color: bool = False,
    **kwargs: Any,
):
    # default to operating on RGB images key commonly used by DatasetAugmentor
    if "apply_to" not in kwargs or kwargs["apply_to"] is None:
        kwargs["apply_to"] = ["images"]
    super().__init__(**kwargs)
    self.ambient_tint = ambient_tint or [0.95, 1.05]
    self.directional_intensity = directional_intensity or [0.0, 0.5]
    self.direction_deg = direction_deg or [0.0, 360.0]
    self.softness = softness or [0.2, 0.8]
    self.spot_count = spot_count or [0, 2]
    self.spot_radius = spot_radius or [0.1, 0.3]
    self.spot_intensity = spot_intensity or [0.0, 0.5]
    self.shadow_prob = shadow_prob
    self.preserve_robot_color = preserve_robot_color

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

ambient_tint = ambient_tint or [0.95, 1.05] instance-attribute

directional_intensity = directional_intensity or [0.0, 0.5] instance-attribute

direction_deg = direction_deg or [0.0, 360.0] instance-attribute

softness = softness or [0.2, 0.8] instance-attribute

spot_count = spot_count or [0, 2] instance-attribute

spot_radius = spot_radius or [0.1, 0.3] instance-attribute

spot_intensity = spot_intensity or [0.0, 0.5] instance-attribute

shadow_prob = shadow_prob instance-attribute

preserve_robot_color = preserve_robot_color instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

CameraIntrinsicsJitter(fx_jitter: List[float] = None, fy_jitter: List[float] = None, cx_jitter_px: List[int] = None, cy_jitter_px: List[int] = None, antialias: bool = True, keep_size: bool = True, intrinsics_key: str = 'camera_intrinsics', **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/camera_intrinsics.py

def __init__(
    self,
    fx_jitter: List[float] = None,
    fy_jitter: List[float] = None,
    cx_jitter_px: List[int] = None,
    cy_jitter_px: List[int] = None,
    antialias: bool = True,
    keep_size: bool = True,
    intrinsics_key: str = "camera_intrinsics",
    **kwargs: Any,
):
    if "apply_to" not in kwargs or kwargs["apply_to"] is None:
        kwargs["apply_to"] = ["images"]
    super().__init__(**kwargs)
    self.fx_jitter = fx_jitter or [0.98, 1.02]
    self.fy_jitter = fy_jitter or [0.98, 1.02]
    self.cx_jitter_px = cx_jitter_px or [-4, 4]
    self.cy_jitter_px = cy_jitter_px or [-4, 4]
    self.antialias = antialias
    self.keep_size = keep_size
    self.intrinsics_key = intrinsics_key

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

fx_jitter = fx_jitter or [0.98, 1.02] instance-attribute

fy_jitter = fy_jitter or [0.98, 1.02] instance-attribute

cx_jitter_px = cx_jitter_px or [-4, 4] instance-attribute

cy_jitter_px = cy_jitter_px or [-4, 4] instance-attribute

antialias = antialias instance-attribute

keep_size = keep_size instance-attribute

intrinsics_key = intrinsics_key instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

CameraExtrinsicsJitter(rot_deg: List[float] = None, transl_px: List[int] = None, perspective: float = 0.0, extrinsics_key: str = 'camera_extrinsics', same_delta_per_seq: bool = False, **kwargs: Any)

Bases: BaseTransform

Source code in src/dataphy/visionpack/tforms/camera_extrinsics.py

def __init__(
    self,
    rot_deg: List[float] = None,
    transl_px: List[int] = None,
    perspective: float = 0.0,
    extrinsics_key: str = "camera_extrinsics",
    same_delta_per_seq: bool = False,
    **kwargs: Any,
):
    if "apply_to" not in kwargs or kwargs["apply_to"] is None:
        kwargs["apply_to"] = ["images"]
    super().__init__(**kwargs)
    self.rot_deg = rot_deg or [-3.0, 3.0]
    self.transl_px = transl_px or [-6, 6]
    self.perspective = perspective
    self.extrinsics_key = extrinsics_key
    self.same_delta_per_seq = same_delta_per_seq

    self._seq_params: Optional[Dict[str, float]] = None
    self._last_episode_id: Optional[str] = None

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

rot_deg = rot_deg or [-3.0, 3.0] instance-attribute

transl_px = transl_px or [-6, 6] instance-attribute

perspective = perspective instance-attribute

extrinsics_key = extrinsics_key instance-attribute

same_delta_per_seq = same_delta_per_seq instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

RGBSensorNoise(shot_k: List[float] = None, read_sigma: List[float] = None, iso_range: List[int] = None, exposure_shift: List[float] = None, analog_gain: List[float] = None, crosstalk: float = 0.0, black_level: float = 0.0, saturation_level: float = 1.0, clip: bool = True, **kwargs: Any)

Bases: BaseTransform

RGB sensor noise simulation for realistic camera augmentation.

This transform simulates realistic camera sensor noise including shot noise, read noise, ISO effects, exposure variations, and channel crosstalk.

Parameters:

Name	Type	Description	Default
shot_k	List[float]	Range of shot noise scaling factors [min, max]	None
read_sigma	List[float]	Range of read noise standard deviations [min, max]	None
iso_range	List[int]	Range of ISO values to simulate [min, max]	None
exposure_shift	List[float]	Range of exposure adjustments [min, max]	None
analog_gain	List[float]	Range of analog gain factors [min, max]	None
crosstalk	float	Channel crosstalk coefficient	0.0
black_level	float	Black level offset	0.0
saturation_level	float	Saturation level for clipping	1.0
clip	bool	Whether to clip values to valid range	True
**kwargs	Any	Additional base transform parameters	{}

Source code in src/dataphy/visionpack/tforms/sensor_noise.py

def __init__(
    self,
    shot_k: List[float] = None,
    read_sigma: List[float] = None,
    iso_range: List[int] = None,
    exposure_shift: List[float] = None,
    analog_gain: List[float] = None,
    crosstalk: float = 0.0,
    black_level: float = 0.0,
    saturation_level: float = 1.0,
    clip: bool = True,
    **kwargs: Any,
):
    """Initialize the RGBSensorNoise transform.

    Args:
        shot_k: Range of shot noise scaling factors [min, max]
        read_sigma: Range of read noise standard deviations [min, max]
        iso_range: Range of ISO values to simulate [min, max]
        exposure_shift: Range of exposure adjustments [min, max]
        analog_gain: Range of analog gain factors [min, max]
        crosstalk: Channel crosstalk coefficient
        black_level: Black level offset
        saturation_level: Saturation level for clipping
        clip: Whether to clip values to valid range
        **kwargs: Additional base transform parameters
    """
    if "apply_to" not in kwargs or kwargs["apply_to"] is None:
        kwargs["apply_to"] = ["images"]
    super().__init__(**kwargs)
    self.shot_k = shot_k or [0.5, 1.5]
    self.read_sigma = read_sigma or [0.002, 0.01]
    self.iso_range = iso_range or [100, 800]
    self.exposure_shift = exposure_shift or [-0.2, 0.2]
    self.analog_gain = analog_gain or [0.9, 1.1]
    self.crosstalk = crosstalk
    self.black_level = black_level
    self.saturation_level = saturation_level
    self.clip = clip

Attributes

p = p instance-attribute

apply_to = apply_to if apply_to is not None else ['rgb'] instance-attribute

sync_views = sync_views instance-attribute

update_intrinsics = update_intrinsics instance-attribute

mask_protect = mask_protect if mask_protect is not None else [] instance-attribute

min_visible_mask_pct = min_visible_mask_pct instance-attribute

resample = resample instance-attribute

border_mode = border_mode instance-attribute

pad_mode = pad_mode instance-attribute

pad_value = pad_value instance-attribute

seed_policy = seed_policy instance-attribute

shot_k = shot_k or [0.5, 1.5] instance-attribute

read_sigma = read_sigma or [0.002, 0.01] instance-attribute

iso_range = iso_range or [100, 800] instance-attribute

exposure_shift = exposure_shift or [-0.2, 0.2] instance-attribute

analog_gain = analog_gain or [0.9, 1.1] instance-attribute

crosstalk = crosstalk instance-attribute

black_level = black_level instance-attribute

saturation_level = saturation_level instance-attribute

clip = clip instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Base forward method that handles cross-cutting logic.

Source code in src/dataphy/visionpack/tforms/base.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Base forward method that handles cross-cutting logic."""
    if not self._should_apply():
        return batch

    # Setup seed policy
    episode_id = batch.get("episode_id")
    self._setup_seed(episode_id)

    # Apply transform
    return self._apply_transform(batch)

GreenAugAdapter(texture_bank: str = './textures')

Bases: Module

Source code in src/dataphy/visionpack/adapters/greenaug.py

def __init__(self, texture_bank: str = "./textures"):
    super().__init__()
    self.texture_bank = texture_bank

Attributes

texture_bank = texture_bank instance-attribute

Functions

forward(batch)

Source code in src/dataphy/visionpack/adapters/greenaug.py

def forward(self, batch):
    # TODO: integrate real green-screen compositor
    return batch

RoboEngineAdapter(mode: str = 'texture', max_workers: int = 2, cache: str = None)

Bases: Module

Source code in src/dataphy/visionpack/adapters/roboengine.py

def __init__(self, mode: str = "texture", max_workers: int = 2, cache: str = None):
    super().__init__()
    self.mode, self.max_workers, self.cache = mode, max_workers, cache

Functions

forward(batch)

Source code in src/dataphy/visionpack/adapters/roboengine.py

def forward(self, batch):
    # TODO: call RoboEngine APIs / local lib
    return batch

InpaintAdapter(prompt: str = '', generate_once: bool = True)

Bases: Module

Source code in src/dataphy/visionpack/adapters/inpaint.py

def __init__(self, prompt: str = "", generate_once: bool = True):
    super().__init__()
    self.prompt, self.generate_once = prompt, generate_once

Functions

forward(batch)

Source code in src/dataphy/visionpack/adapters/inpaint.py

def forward(self, batch):
    # TODO: Grounding + SAM + Diffusers pipeline
    return batch

Compose(modules: List[torch.nn.Module])

Bases: Module

Compose multiple transforms into a single pipeline.

This class chains multiple vision transforms together, applying them sequentially to the input batch. Each transform in the pipeline is applied to the output of the previous transform.

Parameters:

Name	Type	Description	Default
modules	List[Module]	List of torch.nn.Module transforms to apply sequentially	required

Source code in src/dataphy/visionpack/pipeline.py

def __init__(self, modules: List[torch.nn.Module]):
    """Initialize the Compose pipeline with a list of transform modules.

    Args:
        modules: List of torch.nn.Module transforms to apply sequentially
    """
    super().__init__()
    self.mods = torch.nn.ModuleList(modules)

Attributes

mods = torch.nn.ModuleList(modules) instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Apply all transforms in sequence to the input batch.

Parameters:

Name	Type	Description	Default
batch	Dict[str, Any]	Dictionary containing input data (e.g., images, masks)	required

Returns:

Type	Description
Dict[str, Any]	Dictionary containing the transformed data

Source code in src/dataphy/visionpack/pipeline.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Apply all transforms in sequence to the input batch.

    Args:
        batch: Dictionary containing input data (e.g., images, masks)

    Returns:
        Dictionary containing the transformed data
    """
    for m in self.mods:
        batch = m(batch)
    return batch

SyncViewsPipeline(pipeline: torch.nn.Module)

Bases: Module

Wrapper that ensures consistent augmentation parameters across multiple views/cameras.

This wrapper ensures that when processing multiple camera views or images from the same scene, the same random augmentation parameters are applied consistently across all views. This is important for maintaining spatial and temporal coherence in multi-view datasets.

Parameters:

Name	Type	Description	Default
pipeline	Module	The underlying pipeline to wrap with view synchronization	required

Source code in src/dataphy/visionpack/pipeline.py

def __init__(self, pipeline: torch.nn.Module):
    """Initialize the SyncViewsPipeline wrapper.

    Args:
        pipeline: The underlying pipeline to wrap with view synchronization
    """
    super().__init__()
    self.pipeline = pipeline

Attributes

pipeline = pipeline instance-attribute

Functions

forward(batch: Dict[str, Any]) -> Dict[str, Any]

Apply the pipeline with synchronized parameters across views.

Parameters:

Name	Type	Description	Default
batch	Dict[str, Any]	Dictionary containing input data for multiple views	required

Returns:

Type	Description
Dict[str, Any]	Dictionary containing the transformed data with synchronized augmentations

Source code in src/dataphy/visionpack/pipeline.py

def forward(self, batch: Dict[str, Any]) -> Dict[str, Any]:
    """Apply the pipeline with synchronized parameters across views.

    Args:
        batch: Dictionary containing input data for multiple views

    Returns:
        Dictionary containing the transformed data with synchronized augmentations
    """
    # For sync_views, we ensure the same random seed is used for all images in the batch
    # This is already handled by setting the seed in build_pipeline, but this wrapper
    # can be extended for more sophisticated multi-view synchronization
    return self.pipeline(batch)

Functions

build_pipeline(config_path: Union[str, Path, Dict[str, Any], Config], device: str = 'cuda', seed: int = 1337) -> torch.nn.Module

Build a vision augmentation pipeline from configuration.

This function creates a complete augmentation pipeline by parsing the configuration and instantiating all the required transforms. The pipeline can be built from a YAML file, a dictionary, or a Config object.

Parameters:

Name	Type	Description	Default
config_path	Union[str, Path, Dict[str, Any], Config]	Configuration source. Can be: - Path to a YAML configuration file - Dictionary containing pipeline configuration - Config object instance	required
device	str	Device to run the pipeline on ("cpu" or "cuda")	'cuda'
seed	int	Random seed for reproducible augmentations	1337

Returns:

Type	Description
Module	A torch.nn.Module representing the complete augmentation pipeline

Raises:

Type	Description
TypeError	If config_path is an unsupported type
FileNotFoundError	If config_path is a file path that doesn't exist
ValidationError	If the configuration is invalid

Source code in src/dataphy/visionpack/pipeline.py

def build_pipeline(
    config_path: Union[str, Path, Dict[str, Any], Config],
    device: str = "cuda",
    seed: int = 1337,
) -> torch.nn.Module:
    """Build a vision augmentation pipeline from configuration.

    This function creates a complete augmentation pipeline by parsing the configuration
    and instantiating all the required transforms. The pipeline can be built from a
    YAML file, a dictionary, or a Config object.

    Args:
        config_path: Configuration source. Can be:
            - Path to a YAML configuration file
            - Dictionary containing pipeline configuration
            - Config object instance
        device: Device to run the pipeline on ("cpu" or "cuda")
        seed: Random seed for reproducible augmentations

    Returns:
        A torch.nn.Module representing the complete augmentation pipeline

    Raises:
        TypeError: If config_path is an unsupported type
        FileNotFoundError: If config_path is a file path that doesn't exist
        ValidationError: If the configuration is invalid
    """
    # Accept either a file path, a loaded dict, or a Config object
    if isinstance(config_path, Config):
        cfg = config_path
    elif isinstance(config_path, (str, Path)):
        with open(config_path, "r") as f:
            cfg = Config.model_validate(yaml.safe_load(f))
    elif isinstance(config_path, dict):
        cfg = Config.model_validate(config_path)
    else:
        raise TypeError(
            f"Unsupported config type: {type(config_path)}. Expected str/Path/dict/Config"
        )
    torch.manual_seed(seed)

    mods: List[torch.nn.Module] = []
    for step in cfg.pipeline.steps:
        ctor = AugmenterRegistry.get(step.name)
        params = step.model_dump(exclude={"name"}, exclude_none=True)
        mods.append(ctor(**params))

    # background adapter is optional and appended last
    if cfg.pipeline.background.adapter != "none":
        from .adapters import factory as adapter_factory
        mods.append(adapter_factory(cfg.pipeline.background))

    # Wrap pipeline with sync_views functionality if enabled
    pipe = Compose(mods).to(device)
    if cfg.pipeline.sync_views:
        pipe = SyncViewsPipeline(pipe)

    return pipe