data-augmentation component and his tests

tasks/data-augmentation/Deployment.ipynb

@@ -0,0 +1,154 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# Data Augmentation - Experimento\n",
+    "\n",
+    "Data Augmentation é uma estratégia bastante utilizada para impulsionar treinamento de modelos. Se baseando em diversas transformações nos dados, as abordagens de Data Augmentation conseguem multiplicar os seus dados mantendo os mesmos rótulos. Exemplos dessas transformações em dados de imagens são rotações, translações, mudança de coloração, e etc.\n",
+    "\n",
+    "A implementação desse componente foi feita utilizando a biblioteca [torchvision](https://pytorch.org/vision/stable/index.html)."
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Declaração de Classe para Predições em Tempo Real\n",
+    "\n",
+    "A tarefa de implantação cria um serviço REST para predições em tempo-real.<br>\n",
+    "Para isso você deve criar uma classe `Model` que implementa o método `predict`."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "%%writefile Model.py\n",
+    "from typing import List, Iterable, Dict, Union\n",
+    "\n",
+    "import numpy as np\n",
+    "import cv2\n",
+    "import torchvision.transforms as T\n",
+    "import joblib\n",
+    "from PIL import Image\n",
+    "import io\n",
+    "from io import StringIO\n",
+    "\n",
+    "class Model:\n",
+    "    \n",
+    "    def __init__(self):\n",
+    "        self.loaded = False\n",
+    "        \n",
+    "    def load(self):\n",
+    "        # Carrega artefatos: estimador, etc\n",
+    "        artifacts = joblib.load(\"/tmp/data/data_augmentation.joblib\")\n",
+    "        self.artifacts = artifacts[\"data_augmentation_parameters\"]\n",
+    "        \n",
+    "        self.parameters = [ self.artifacts[\"augmentation_rate\"],\n",
+    "                            self.artifacts[\"horizontal_flip\"],\n",
+    "                            self.artifacts[\"vertical_flip\"],\n",
+    "                            self.artifacts[\"crop\"],\n",
+    "                            self.artifacts[\"color_jitter\"],\n",
+    "                            self.artifacts[\"perspective\"],\n",
+    "                            self.artifacts[\"rotate\"] ]\n",
+    "\n",
+    "        self.augmentation_rate = self.parameters[0]\n",
+    "        \n",
+    "        # Load Model\n",
+    "\n",
+    "        self.jitter = T.ColorJitter(brightness=.5, hue=.3)\n",
+    "        self.perspective_transformer = T.RandomPerspective(distortion_scale=0.6, p=1.0)\n",
+    "        self.rotater = T.RandomRotation(degrees=(0, 180))\n",
+    "        self.hflip = T.RandomHorizontalFlip()\n",
+    "        self.vflip = T.RandomVerticalFlip() \n",
+    "        \n",
+    "        #self.transformations = [ self.jitter, self.perspective_transformer, self.rotater, self.crop, self.hflip, self.vflip ]\n",
+    "        \n",
+    "        self.loaded = True\n",
+    "        \n",
+    "    def predict(self, X, feature_names, meta=None):\n",
+    "\n",
+    "        if not self.loaded:\n",
+    "            self.load()\n",
+    "            \n",
+    "        # Check if data is a bytes\n",
+    "        if isinstance(X, bytes):\n",
+    "            im_bytes = X # Get image bytes\n",
+    "        \n",
+    "        # If not, should be a list or ndarray\n",
+    "        else:\n",
+    "            # Garantee is a ndarray\n",
+    "            X = np.array(X)\n",
+    "            \n",
+    "            # Seek for extra dimension\n",
+    "            if len(X.shape) == 2:\n",
+    "                im_bytes = X[0,0] # Get image bytes\n",
+    "            \n",
+    "            else:\n",
+    "                im_bytes = X[0] # Get image bytes\n",
+    "        \n",
+    "        # Preprocess img bytes to img_arr\n",
+    "        im_arr = np.frombuffer(im_bytes, dtype=np.uint8)\n",
+    "        img = cv2.imdecode(im_arr, flags=cv2.IMREAD_COLOR)\n",
+    "        img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)\n",
+    "        img = Image.fromarray(img) # convert to PIL image\n",
+    "\n",
+    "        width, height = img.size\n",
+    "        crop_size = (int(width * 0.8),int(height * 0.8))  \n",
+    "        self.crop = T.RandomCrop(crop_size)\n",
+    "\n",
+    "        transformed_images = []\n",
+    "        if self.parameters[1]:\n",
+    "            transformed_images += [self.hflip(img) for _ in range(self.augmentation_rate) ]\n",
+    "        if self.parameters[2]:\n",
+    "            transformed_images += [self.vflip(img) for _ in range(self.augmentation_rate) ]\n",
+    "        if self.parameters[3]:\n",
+    "            transformed_images += [self.crop(img) for _ in range(self.augmentation_rate) ]\n",
+    "        if self.parameters[4]:\n",
+    "            transformed_images += [self.jitter(img) for _ in range(self.augmentation_rate) ]\n",
+    "        if self.parameters[5]:\n",
+    "            transformed_images += [self.perspective_transformer(img) for _ in range(self.augmentation_rate) ]\n",
+    "        if self.parameters[6]:\n",
+    "            transformed_images += [self.rotater(img) for _ in range(self.augmentation_rate) ]\n",
+    "                \n",
+    "        # Compile results        \n",
+    "        results = []\n",
+    "        for transf_img in transformed_images:\n",
+    "            buff = io.BytesIO()\n",
+    "            transf_img.save(buff, format=\"JPEG\")\n",
+    "            results.append(buff.getvalue().decode(\"latin1\"))\n",
+    "             \n",
+    "        return results"
+   ]
+  }
+ ],
+ "metadata": {
+  "celltoolbar": "Tags",
+  "experiment_id": "f795343b-5a9d-46a0-8336-41d5079e49a8",
+  "kernelspec": {
+   "display_name": "Python 3",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.8.8"
+  },
+  "operator_id": "db47ff60-929f-4dad-ad91-efbb373367cd",
+  "task_id": "e7e66db7-1bef-4f64-8c41-834dc112d518"
+ },
+ "nbformat": 4,
+ "nbformat_minor": 4
+}

tests/test_data-augmentation.py

@@ -0,0 +1,138 @@
+import os
+import unittest
+import uuid
+
+import papermill
+import pandas as pd
+import numpy as np
+from base64 import b64encode
+import io
+from PIL import Image
+
+from tests import datasets, server
+
+EXPERIMENT_ID = str(uuid.uuid4())
+OPERATOR_ID = str(uuid.uuid4())
+RUN_ID = str(uuid.uuid4())
+
+TEMPORARY_DIR = "tmp"
+LOCAL_TEST_DATA_PATH = f"/{TEMPORARY_DIR}/data/yolo.zip"
+LOCAL_OUTPUT_DATA_PATH = f"/{TEMPORARY_DIR}/data/"
+EXPERIMENT_NOTEBOOK = "Experiment.ipynb"
+DEPLOYMENT_NOTEBOOK = "Deployment.ipynb"
+DEV_DIR = "/dev/null"
+
+
+class TestDataAugmentation(unittest.TestCase):
+
+    def setUp(self):
+        # Set environment variables needed to run notebooks
+        os.environ["EXPERIMENT_ID"] = EXPERIMENT_ID
+        os.environ["OPERATOR_ID"] = OPERATOR_ID
+        os.environ["RUN_ID"] = RUN_ID
+
+        datasets.yolo()
+
+        os.chdir("tasks/data-augmentation")
+
+    def tearDown(self):
+        datasets.clean()
+        os.chdir("../../")
+
+    def test_experiment_default_parameters(self):
+        papermill.execute_notebook(
+            EXPERIMENT_NOTEBOOK,
+            DEV_DIR,
+            parameters=dict(
+                dataset=LOCAL_TEST_DATA_PATH,
+                image_path="image_path",
+                augmentation_rate=5,
+                horizontal_flip=True,
+                vertical_flip=True,
+                crop=True,
+                color_jitter=True,
+                perspective=True,
+                rotate=True
+            ),
+        )
+
+        # Verify output data
+        files = os.listdir(LOCAL_OUTPUT_DATA_PATH + "yolo/test/")
+
+        # 6 transformations * augmentation_rate + original_image
+        self.assertEqual(len(files), 6*5 + 1)
+        out_data = Image.open(LOCAL_OUTPUT_DATA_PATH + "yolo/test/sunflower_transformed_img_3.png")
+        self.assertEqual(out_data.format, "PNG")
+
+        # Deployment pipeline
+        papermill.execute_notebook(
+            DEPLOYMENT_NOTEBOOK,
+            DEV_DIR,
+        )
+
+
+        #data = datasets.landspaces_test_data()
+        for ext in ['png', 'jpg']:
+
+            data = datasets.image_testdata(kind='objects', ext=ext)
+
+            with server.Server() as s:
+                response = s.test(data=data, timeout=10)
+
+            images = []
+            for raw_str in response["ndarray"]:
+                raw_bytes = bytes(raw_str, "latin1")
+                img = Image.open(io.BytesIO(raw_bytes), formats=["JPEG"])
+                images.append(img)
+
+            self.assertEqual(len(images), 5*6)
+            self.assertEqual(images[0].size, (930, 1048))
+
+    def test_experiment_no_crop_parameters(self):
+        papermill.execute_notebook(
+            EXPERIMENT_NOTEBOOK,
+            DEV_DIR,
+            parameters=dict(
+                dataset=LOCAL_TEST_DATA_PATH,
+                image_path="image_path",
+                augmentation_rate=5,
+                horizontal_flip=True,
+                vertical_flip=True,
+                crop=False,
+                color_jitter=True,
+                perspective=True,
+                rotate=True
+            ),
+        )
+
+        # Verify output data
+        files = os.listdir(LOCAL_OUTPUT_DATA_PATH + "yolo/test/")
+
+        # 6 transformations * augmentation_rate + original_image
+        self.assertEqual(len(files), 5*5 + 1)
+        out_data = Image.open(LOCAL_OUTPUT_DATA_PATH + "yolo/test/sunflower_transformed_img_0.png")
+        self.assertEqual(out_data.format, "PNG")
+
+        # Deployment pipeline
+        papermill.execute_notebook(
+            DEPLOYMENT_NOTEBOOK,
+            DEV_DIR,
+        )
+
+
+        #data = datasets.landspaces_test_data()
+        for ext in ['png', 'jpg']:
+
+            data = datasets.image_testdata(kind='objects', ext=ext)
+
+            with server.Server() as s:
+                response = s.test(data=data, timeout=10)
+
+            images = []
+            for raw_str in response["ndarray"]:
+                raw_bytes = bytes(raw_str, "latin1")
+                img = Image.open(io.BytesIO(raw_bytes), formats=["JPEG"])
+                images.append(img)
+
+            self.assertEqual(len(images), 5*5)
+            self.assertEqual(images[0].size, (930, 1048))