Add visualisation callback for image classification

flash/data/base_viz.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, List, Sequence
+from typing import Any, Dict, List, Sequence, Set
 
 from pytorch_lightning.trainer.states import RunningStage
 
@@ -101,7 +101,13 @@ def show(self, batch: Dict[str, Any], running_stage: RunningStage) -> None:
         """
         Override this function when you want to visualize a composition.
         """
-        for func_name in _PREPROCESS_FUNCS:
+        # filter out the functions to visualise
+        func_name: str = self._fcn_white_list[running_stage]
+        func_names_list: Set[str] = list(set([func_name]) & set(_PREPROCESS_FUNCS))
+        if len(func_names_list) == 0:
+            raise ValueError(f"Invalid function name: {func_name}.")
+
+        for func_name in func_names_list:
             hook_name = f"show_{func_name}"
             if _is_overriden(hook_name, self, BaseVisualization):
                 getattr(self, hook_name)(batch[func_name], running_stage)

flash/data/data_module.py

@@ -13,7 +13,7 @@
 # limitations under the License.
 import os
 import platform
-from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, Union
+from typing import Any, Callable, Dict, Iterable, List, Optional, Set, Tuple, Union
 
 import pytorch_lightning as pl
 import torch
@@ -92,6 +92,9 @@ def __init__(
         # this may also trigger data preloading
         self.set_running_stages()
 
+        # buffer to store the functions to visualise
+        self._fcn_white_list: Dict[str, Set[str]] = {}
+
     @property
     def train_dataset(self) -> Optional[Dataset]:
         """This property returns the train dataset"""
@@ -164,25 +167,35 @@ def _show_batch(self, stage: RunningStage, reset: bool = True) -> None:
                 iter_dataloader = self._reset_iterator(stage)
                 _ = next(iter_dataloader)
             data_fetcher: BaseVisualization = self.data_fetcher
+            # list of functions to visualise
+            data_fetcher._fcn_white_list = self._fcn_white_list
             data_fetcher._show(stage)
             if reset:
                 self.viz.batches[stage] = {}
 
-    def show_train_batch(self, reset: bool = True) -> None:
+    def show_train_batch(self, name: str = 'load_sample', reset: bool = True) -> None:
         """This function is used to visualize a batch from the train dataloader."""
-        self._show_batch(_STAGES_PREFIX[RunningStage.TRAINING], reset=reset)
+        stage_name: str = _STAGES_PREFIX[RunningStage.TRAINING]
+        self._fcn_white_list[stage_name] = name
+        self._show_batch(stage_name, reset=reset)
 
-    def show_val_batch(self, reset: bool = True) -> None:
+    def show_val_batch(self, name: str = 'load_sample', reset: bool = True) -> None:
         """This function is used to visualize a batch from the validation dataloader."""
-        self._show_batch(_STAGES_PREFIX[RunningStage.VALIDATING], reset=reset)
+        stage_name: str = _STAGES_PREFIX[RunningStage.VALIDATING]
+        self._fcn_white_list[stage_name] = name
+        self._show_batch(stage_name, reset=reset)
 
-    def show_test_batch(self, reset: bool = True) -> None:
+    def show_test_batch(self, name: str = 'load_sample', reset: bool = True) -> None:
         """This function is used to visualize a batch from the test dataloader."""
-        self._show_batch(_STAGES_PREFIX[RunningStage.TESTING], reset=reset)
+        stage_name: str = _STAGES_PREFIX[RunningStage.TESTING]
+        self._fcn_white_list[stage_name] = name
+        self._show_batch(stage_name, reset=reset)
 
-    def show_predict_batch(self, reset: bool = True) -> None:
+    def show_predict_batch(self, name: str = 'load_sample', reset: bool = True) -> None:
         """This function is used to visualize a batch from the predict dataloader."""
-        self._show_batch(_STAGES_PREFIX[RunningStage.PREDICTING], reset=reset)
+        stage_name: str = _STAGES_PREFIX[RunningStage.PREDICTING]
+        self._fcn_white_list[stage_name] = name
+        self._show_batch(stage_name, reset=reset)
 
     @staticmethod
     def get_dataset_attribute(dataset: torch.utils.data.Dataset, attr_name: str, default: Optional[Any] = None) -> Any:

flash/utils/imports.py

@@ -5,3 +5,4 @@
 _COCO_AVAILABLE = _module_available("pycocotools")
 _TIMM_AVAILABLE = _module_available("timm")
 _TORCHVISION_AVAILABLE = _module_available("torchvision")
+_MATPLOTLIB_AVAILABLE = _module_available("matplotlib")

flash/vision/classification/data.py

@@ -15,27 +15,37 @@
 import pathlib
 from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple, Type, Union
 
+import numpy as np
 import torch
 import torchvision
 from PIL import Image
+from pytorch_lightning.trainer.states import RunningStage
 from pytorch_lightning.utilities.exceptions import MisconfigurationException
 from torch import nn
 from torch.utils.data import Dataset
 from torch.utils.data._utils.collate import default_collate
 from torchvision.datasets.folder import has_file_allowed_extension, IMG_EXTENSIONS, make_dataset
 
+from flash.core.utils import _is_overriden
 from flash.data.auto_dataset import AutoDataset
+from flash.data.base_viz import BaseVisualization  # for viz
+from flash.data.callback import BaseDataFetcher
 from flash.data.data_module import DataModule
 from flash.data.data_pipeline import DataPipeline
 from flash.data.process import Preprocess
-from flash.utils.imports import _KORNIA_AVAILABLE
+from flash.data.utils import _PREPROCESS_FUNCS
+from flash.utils.imports import _KORNIA_AVAILABLE, _MATPLOTLIB_AVAILABLE
 
 if _KORNIA_AVAILABLE:
-    import kornia.augmentation as K
-    import kornia.geometry.transform as T
+    import kornia as K
 else:
     from torchvision import transforms as T
 
+if _MATPLOTLIB_AVAILABLE:
+    import matplotlib.pyplot as plt
+else:
+    plt = None
+
 
 class ImageClassificationPreprocess(Preprocess):
 
@@ -74,20 +84,37 @@ def _get_predicting_files(samples: Union[Sequence, str]) -> List[str]:
         return files
 
     @classmethod
-    def _load_data_dir(cls, data: Any, dataset: Optional[AutoDataset] = None) -> List[str]:
+    def _load_data_dir(cls, data: Any, dataset: Optional[AutoDataset] = None) -> List[Tuple[str, int]]:
+        # case where we pass a list of files
         if isinstance(data, list):
+            # TODO: define num_classes elsewhere. This is a bad assumption since the list of
+            # labels might not contain the complete set of ids so that you can infer the total
+            # number of classes to train in your dataset.
             dataset.num_classes = len(data)
-            out = []
+            out: List[Tuple[str, int]] = []
             for p, label in data:
                 if os.path.isdir(p):
-                    for f in os.listdir(p):
+                    # TODO: there is an issue here when a path is provided along with labels.
+                    # os.listdir cannot assure the same file order as the passed labels list.
+                    files_list: List[str] = os.listdir(p)
+                    if len(files_list) > 1:
+                        raise ValueError(
+                            f"The provided directory contains more than one file."
+                            f"Directory: {p} -> Contains: {files_list}"
+                        )
+                    for f in files_list:
                         if has_file_allowed_extension(f, IMG_EXTENSIONS):
                             out.append([os.path.join(p, f), label])
-                elif os.path.isfile(p) and has_file_allowed_extension(p, IMG_EXTENSIONS):
+                elif os.path.isfile(p) and has_file_allowed_extension(str(p), IMG_EXTENSIONS):
                     out.append([p, label])
+                else:
+                    raise TypeError(f"Unexpected file path type: {p}.")
             return out
         else:
             classes, class_to_idx = cls._find_classes(data)
+            # TODO: define num_classes elsewhere. This is a bad assumption since the list of
+            # labels might not contain the complete set of ids so that you can infer the total
+            # number of classes to train in your dataset.
             dataset.num_classes = len(classes)
             return make_dataset(data, class_to_idx, IMG_EXTENSIONS, None)
 
@@ -231,6 +258,14 @@ def __init__(
         if self._predict_ds:
             self.set_dataset_attribute(self._predict_ds, 'num_classes', self.num_classes)
 
+    @staticmethod
+    def configure_data_fetcher(*args, **kwargs) -> BaseDataFetcher:
+        return _MatplotlibVisualization(*args, **kwargs)
+
+    def show(self) -> None:
+        """Method to block matplotlib windows."""
+        plt.show()
+
     @staticmethod
     def _check_transforms(transform: Dict[str, Union[nn.Module, Callable]]) -> Dict[str, Union[nn.Module, Callable]]:
         if transform and not isinstance(transform, Dict):
@@ -247,14 +282,16 @@ def _check_transforms(transform: Dict[str, Union[nn.Module, Callable]]) -> Dict[
 
     @staticmethod
     def default_train_transforms():
-        image_size = ImageClassificationData.image_size
+        image_size: Tuple[int, int] = ImageClassificationData.image_size
         if _KORNIA_AVAILABLE and not os.getenv("FLASH_TESTING", "0") == "1":
             #  Better approach as all transforms are applied on tensor directly
             return {
                 "to_tensor_transform": torchvision.transforms.ToTensor(),
-                "post_tensor_transform": nn.Sequential(K.RandomResizedCrop(image_size), K.RandomHorizontalFlip()),
+                "post_tensor_transform": nn.Sequential(
+                    K.augmentation.RandomResizedCrop(image_size), K.augmentation.RandomHorizontalFlip()
+                ),
                 "per_batch_transform_on_device": nn.Sequential(
-                    K.Normalize(torch.tensor([0.485, 0.456, 0.406]), torch.tensor([0.229, 0.224, 0.225])),
+                    K.augmentation.Normalize(torch.tensor([0.485, 0.456, 0.406]), torch.tensor([0.229, 0.224, 0.225])),
                 )
             }
         else:
@@ -267,14 +304,14 @@ def default_train_transforms():
 
     @staticmethod
     def default_val_transforms():
-        image_size = ImageClassificationData.image_size
+        image_size: Tuple[int, int] = ImageClassificationData.image_size
         if _KORNIA_AVAILABLE and not os.getenv("FLASH_TESTING", "0") == "1":
             #  Better approach as all transforms are applied on tensor directly
             return {
                 "to_tensor_transform": torchvision.transforms.ToTensor(),
-                "post_tensor_transform": nn.Sequential(K.RandomResizedCrop(image_size)),
+                "post_tensor_transform": nn.Sequential(K.augmentation.RandomResizedCrop(image_size)),
                 "per_batch_transform_on_device": nn.Sequential(
-                    K.Normalize(torch.tensor([0.485, 0.456, 0.406]), torch.tensor([0.229, 0.224, 0.225])),
+                    K.augmentation.Normalize(torch.tensor([0.485, 0.456, 0.406]), torch.tensor([0.229, 0.224, 0.225])),
                 )
             }
         else:
@@ -521,3 +558,68 @@ def from_filepaths(
             seed=seed,
             **kwargs
         )
+
+
+class _MatplotlibVisualization(BaseVisualization):
+    """Process and show the image batch and its associated label using matplotlib.
+    """
+    max_cols: int = 4  # maximum number of columns we accept
+
+    @staticmethod
+    def _to_numpy(img: Union[torch.Tensor, Image.Image]) -> np.ndarray:
+        out: np.ndarray
+        if isinstance(img, Image.Image):
+            out = np.array(img)
+        elif isinstance(img, torch.Tensor):
+            out = img.squeeze(0).permute(1, 2, 0).cpu().numpy()
+        else:
+            raise TypeError(f"Unknown image type. Got: {type(img)}.")
+        return out
+
+    def _show_images_and_labels(self, data: List[Any], num_samples: int, title: str):
+        # define the image grid
+        cols: int = min(num_samples, self.max_cols)
+        rows: int = num_samples // cols
+
+        # create figure and set title
+        fig, axs = plt.subplots(rows, cols)
+        fig.suptitle(title)
+
+        for i, ax in enumerate(axs.ravel()):
+            # unpack images and labels
+            if isinstance(data, list):
+                _img, _label = data[i]
+            elif isinstance(data, tuple):
+                imgs, labels = data
+                _img, _label = imgs[i], labels[i]
+            else:
+                raise TypeError(f"Unknown data type. Got: {type(data)}.")
+            # convert images to numpy
+            _img: np.ndarray = self._to_numpy(_img)
+            if isinstance(_label, torch.Tensor):
+                _label = int(_label.item())
+            # show image and set label as subplot title
+            ax.imshow(_img)
+            ax.set_title(str(_label))
+            ax.axis('off')
+        plt.show(block=False)
+
+    def show_load_sample(self, samples: List[Any], running_stage: RunningStage):
+        win_title: str = f"{running_stage} - show_load_sample"
+        self._show_images_and_labels(samples, len(samples), win_title)
+
+    def show_pre_tensor_transform(self, samples: List[Any], running_stage: RunningStage):
+        win_title: str = f"{running_stage} - show_pre_tensor_transform"
+        self._show_images_and_labels(samples, len(samples), win_title)
+
+    def show_to_tensor_transform(self, samples: List[Any], running_stage: RunningStage):
+        win_title: str = f"{running_stage} - show_to_tensor_transform"
+        self._show_images_and_labels(samples, len(samples), win_title)
+
+    def show_post_tensor_transform(self, samples: List[Any], running_stage: RunningStage):
+        win_title: str = f"{running_stage} - show_post_tensor_transform"
+        self._show_images_and_labels(samples, len(samples), win_title)
+
+    def show_per_batch_transform(self, batch: List[Any], running_stage):
+        win_title: str = f"{running_stage} - show_per_batch_transform"
+        self._show_images_and_labels(batch[0], batch[0][0].shape[0], win_title)

requirements.txt

@@ -17,3 +17,4 @@ sentencepiece>=0.1.95
 filelock  # comes with 3rd-party dependency
 pycocotools>=2.0.2 ; python_version >= "3.7"
 kornia>=0.5.0
+matplotlib  # used by the visualisation callback

tests/data/test_callbacks.py

@@ -25,7 +25,7 @@
 from flash.data.base_viz import BaseVisualization
 from flash.data.callback import BaseDataFetcher
 from flash.data.data_module import DataModule
-from flash.data.utils import _STAGES_PREFIX
+from flash.data.utils import _PREPROCESS_FUNCS, _STAGES_PREFIX
 from flash.vision import ImageClassificationData
 
 
@@ -87,10 +87,7 @@ def test_base_viz(tmpdir):
     (tmpdir / "a").mkdir()
     (tmpdir / "b").mkdir()
     _rand_image().save(tmpdir / "a" / "a_1.png")
-    _rand_image().save(tmpdir / "a" / "a_2.png")
-
     _rand_image().save(tmpdir / "b" / "a_1.png")
-    _rand_image().save(tmpdir / "b" / "a_2.png")
 
     class CustomBaseVisualization(BaseVisualization):
 
@@ -147,8 +144,9 @@ def configure_data_fetcher(*args, **kwargs) -> CustomBaseVisualization:
 
     for stage in _STAGES_PREFIX.values():
 
-        for _ in range(10):
-            getattr(dm, f"show_{stage}_batch")(reset=False)
+        for fcn_name in _PREPROCESS_FUNCS:
+            fcn = getattr(dm, f"show_{stage}_batch")
+            fcn(fcn_name, reset=False)
 
         is_predict = stage == "predict"
 
@@ -193,3 +191,4 @@ def test_data_loaders_num_workers_to_0(tmpdir):
     assert isinstance(iterator, torch.utils.data.dataloader._SingleProcessDataLoaderIter)
     iterator = iter(datamodule.train_dataloader())
     assert isinstance(iterator, torch.utils.data.dataloader._MultiProcessingDataLoaderIter)
+    assert datamodule.num_workers == 3