app/preprocessing/steps/s07_augment.py

import numpy as np
from app.preprocessing.base import PreprocessingStep, PreprocessingContext


class AugmentStep(PreprocessingStep):
    name = "augment"
    description = "Optional augmentations (flip, rotate, brightness, contrast) — disabled by default for inference"
    version = "1.0.0"
    order = 7
    enabled = False
    required = False

    async def process(self, ctx: PreprocessingContext, params: dict) -> PreprocessingContext:
        applied = []
        skipped = []

        if ctx.image_array is None:
            ctx.step_outputs["augment"] = {"applied": [], "skipped": ["all"], "reason": "no image_array"}
            return ctx

        arr = ctx.image_array

        if params.get("horizontal_flip", False):
            arr = np.fliplr(arr)
            applied.append("horizontal_flip")
        else:
            skipped.append("horizontal_flip")

        if params.get("vertical_flip", False):
            arr = np.flipud(arr)
            applied.append("vertical_flip")
        else:
            skipped.append("vertical_flip")

        rotation = params.get("rotation_degrees", 0)
        if rotation != 0:
            from PIL import Image
            import io
            # Rotate via PIL for simplicity
            pil_img = ctx.image
            if pil_img is not None:
                pil_img = pil_img.rotate(rotation, expand=False)
                ctx.image = pil_img
                arr = np.array(pil_img, dtype=np.float32)
            applied.append(f"rotation_{rotation}deg")
        else:
            skipped.append("rotation")

        brightness_range = params.get("brightness_range", [1.0, 1.0])
        if brightness_range[0] != 1.0 or brightness_range[1] != 1.0:
            factor = np.random.uniform(brightness_range[0], brightness_range[1])
            arr = np.clip(arr * factor, arr.min(), arr.max())
            applied.append(f"brightness_{factor:.2f}")
        else:
            skipped.append("brightness")

        contrast_range = params.get("contrast_range", [1.0, 1.0])
        if contrast_range[0] != 1.0 or contrast_range[1] != 1.0:
            factor = np.random.uniform(contrast_range[0], contrast_range[1])
            mean = arr.mean()
            arr = np.clip((arr - mean) * factor + mean, arr.min(), arr.max())
            applied.append(f"contrast_{factor:.2f}")
        else:
            skipped.append("contrast")

        ctx.image_array = arr
        ctx.step_outputs["augment"] = {"applied": applied, "skipped": skipped}

        return ctx