Skip to content

Tforms

dataphy.visionpack.tforms

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 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
11
12
13
14
15
16
17
18
19
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
13
14
15
16
17
18
19
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
14
15
16
17
18
19
20
21
22
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
10
11
12
13
14
15
16
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
11
12
13
14
15
16
17
18
19
20
21
22
23
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
10
11
12
13
14
15
16
17
18
19
20
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
11
12
13
14
15
16
17
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
15
16
17
18
19
20
21
22
23
24
25
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
105
106
107
108
109
110
111
112
113
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
87
88
89
90
91
92
93
94
95
96
97
98
99
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
163
164
165
166
167
168
169
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
206
207
208
209
210
211
212
213
214
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
14
15
16
17
18
19
20
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
 94
 95
 96
 97
 98
 99
100
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
133
134
135
136
137
138
139
140
141
142
143
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
209
210
211
212
213
214
215
216
217
218
219
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
232
233
234
235
236
237
238
239
240
241
242
243
244
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
183
184
185
186
187
188
189
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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)