Merge branch 'main' into pr312-export-clean

Author: shivv23

tensormap-backend/app/routers/data_process.py

@@ -11,6 +11,7 @@
 )
 from app.services.data_process import (
     add_target_service,
+    augment_image_service,
     delete_one_target_by_id_service,
     get_all_targets_service,
     get_column_stats_service,
@@ -95,3 +96,28 @@ def preprocess(file_id: uuid_pkg.UUID, request: PreprocessRequest, db: Session =
     logger.debug("Preprocessing file_id=%s with %d transformations", file_id, len(request.transformations))
     body, status_code = preprocess_data(db, file_id=file_id, transformations=request.transformations)
     return JSONResponse(status_code=status_code, content=body)
+
+
+@router.post("/data/augment/image/{file_id}")
+def augment_image(
+    file_id: uuid_pkg.UUID,
+    technique: str = Query(
+        "flip_horizontal",
+        pattern="^(flip_horizontal|flip_vertical|rotate_90|brightness|zoom|gaussian_noise|random_crop)$",
+    ),
+    db: Session = Depends(get_db),
+):
+    """Apply image augmentation techniques to generate synthetic variants.
+
+    Supported techniques:
+    - flip_horizontal: Mirror along vertical axis
+    - flip_vertical: Mirror along horizontal axis
+    - rotate_90: Rotate by 90 degrees
+    - brightness: Increase brightness by 20%
+    - zoom: Zoom to 90% then resize
+    - gaussian_noise: Add Gaussian noise
+    - random_crop: Crop to 85% then resize
+    """
+    logger.debug("Applying %s augmentation to file_id=%s", technique, file_id)
+    body, status_code = augment_image_service(db, file_id=file_id, technique=technique)
+    return JSONResponse(status_code=status_code, content=body)

tensormap-backend/app/services/data_process.py

@@ -1,3 +1,4 @@
+import os
 import uuid as uuid_pkg
 from collections.abc import Callable
 from typing import Any
@@ -370,3 +371,86 @@ def preprocess_data(db: Session, file_id: uuid_pkg.UUID, transformations: list)
     except Exception as e:
         logger.exception("Error preprocessing data: %s", str(e))
         return _resp(500, False, f"Error preprocessing data: {e}")
+
+
+def augment_image_service(
+    db: Session,
+    file_id: uuid_pkg.UUID,
+    technique: str = "flip_horizontal",
+) -> tuple:
+    """Apply image augmentation techniques to generate synthetic variants.
+
+    Supported techniques:
+    - flip_horizontal: Mirror image along vertical axis
+    - flip_vertical: Mirror image along horizontal axis
+    - rotate_90: Rotate image by 90 degrees
+    - brightness: Adjust brightness by 20%
+    - zoom: Zoom to 90% then resize
+    - gaussian_noise: Add Gaussian noise
+    - random_crop: Crop to 85% then resize
+    """
+    file_record = db.get(DataFile, file_id)
+    if not file_record:
+        return _resp(404, False, "Image file not found")
+
+    file_path = file_record.file_path
+    if not file_path or not os.path.exists(file_path):
+        return _resp(404, False, "Image file not found on disk")
+
+    settings = get_settings()
+    output_dir = os.path.join(settings.UPLOAD_DIRECTORY, f"augmented_{file_id}")
+    os.makedirs(output_dir, exist_ok=True)
+
+    supported_formats = {".jpg", ".jpeg", ".png", ".bmp", ".gif", ".webp"}
+    _, ext = os.path.splitext(file_path)
+    if ext.lower() not in supported_formats:
+        return _resp(400, False, f"Unsupported image format: {ext}")
+
+    try:
+        from PIL import Image, ImageEnhance
+
+        original = Image.open(file_path)
+
+        if technique == "flip_horizontal":
+            augmented = original.transpose(Image.FLIP_LEFT_RIGHT)
+        elif technique == "flip_vertical":
+            augmented = original.transpose(Image.FLIP_TOP_BOTTOM)
+        elif technique == "rotate_90":
+            augmented = original.rotate(90, expand=True)
+        elif technique == "brightness":
+            enhancer = ImageEnhance.Brightness(original)
+            augmented = enhancer.enhance(1.2)
+        elif technique == "zoom":
+            width, height = original.size
+            new_width = int(width * 0.9)
+            new_height = int(height * 0.9)
+            left = (width - new_width) // 2
+            top = (height - new_height) // 2
+            cropped = original.crop((left, top, left + new_width, top + new_height))
+            augmented = cropped.resize((width, height), Image.LANCZOS)
+        elif technique == "gaussian_noise":
+            np_img = np.array(original.convert("RGB")).astype(np.float32) / 255.0
+            noise = np.random.normal(0, 0.05, np_img.shape)
+            np_img = np.clip(np_img + noise, 0, 1)
+            augmented = Image.fromarray((np_img * 255).astype(np.uint8))
+        elif technique == "random_crop":
+            width, height = original.size
+            crop_size = int(min(width, height) * 0.85)
+            left = np.random.randint(0, width - crop_size + 1)
+            top = np.random.randint(0, height - crop_size + 1)
+            cropped = original.crop((left, top, left + crop_size, top + crop_size))
+            augmented = cropped.resize((width, height), Image.LANCZOS)
+        else:
+            return _resp(400, False, f"Unknown technique: {technique}")
+
+        output_path = os.path.join(output_dir, f"augmented_0{ext}")
+        augmented.save(output_path)
+
+        logger.info("Applied %s augmentation to file %s", technique, file_id)
+        return _resp(200, True, f"Generated augmented image using {technique}", {"output_path": output_path})
+
+    except ImportError:
+        return _resp(500, False, "Pillow not installed")
+    except Exception as e:
+        logger.exception("Error augmenting image: %s", str(e))
+        return _resp(500, False, f"Error augmenting image: {e}")

tensormap-backend/tests/test_image_augment.py

@@ -0,0 +1,20 @@
+"""Unit tests for image augmentation service."""
+
+import sys
+from unittest.mock import MagicMock
+
+sys.modules.setdefault("tensorflow", MagicMock())
+sys.modules.setdefault("flatten_json", MagicMock())
+sys.modules.setdefault("pandas", MagicMock())
+
+
+class TestImageAugmentation:
+    def test_import(self):
+        from app.services.data_process import augment_image_service
+
+        assert callable(augment_image_service)
+
+    def test_router_import(self):
+        from app.routers.data_process import augment_image
+
+        assert callable(augment_image)