neural_de\transformations\transformation_pipeline.py source
1 """
2 Transformation pipeline for automation of multiple transformations methods
3 """
4 import logging
5 from pathlib import Path
6 from typing import Union
7
8 import numpy as np
9 import yaml
10
11 from neural_de.transformations.transformation import BaseTransformation
12 # from neural_de import transformations
13 from neural_de.utils.twe_logger import log_and_raise
14 import importlib
15
16
17 def camel_to_snake(s):
18 """ This function convert input strings s from Camelcase format to snake case format"""
19 return ''.join(['_' + c.lower() if c.isupper() else c for c in s]).lstrip('_')
20
21
22 class TransformationPipeline(BaseTransformation):
23 """
24 Provides a pipeline object, to facilitate the automation of multiple transformations methods,
25 and/or offer loading from a yaml file.
26
27 You can check the example notebook **examples/Pipeline_example.ipynb** for details on the syntax
28 and usage.
29 An example of valid config file can be found in **examples/config/conf_user.yaml**
30
31 Args:
32 config: either a path toward a yaml configuration file, or a list of dict.
33 logger: It is recommended to use the confiance.ai logger, obtainable with
34 neural_de.utils.get_logger(...). If None, one logging with stdout will be provided.
35 """
36 def __init__(self, config: Union[str, list, Path], logger: logging.Logger = None):
37 super().__init__(logger)
38 if isinstance(config, list):
39 self._pipeline_conf = config
40 else:
41 # Avoid lazy init for the configuration in order to check consistency asap.
42 self._pipeline_conf = self._read_config(config)
43 self._pipeline = None
44 self._logger.info("Config file loaded")
45
46 def _read_config(self, config_path: str) -> dict:
47 """
48 Read user configuration of pipeline that contains different transformations parameters
49
50 Args:
51 config_path: path to a yaml configuration file. See example/Pipeline_example for syntax
52 specification
53 Returns:
54 The loaded configuration
55 """
56 try:
57 with open(config_path, 'r', encoding='utf-8') as yaml_file:
58 config = yaml.safe_load(yaml_file)
59 except FileNotFoundError:
60 log_and_raise(self._logger, FileNotFoundError,
61 f"Config file '{config_path}' not found.")
62 except yaml.YAMLError as e:
63 log_and_raise(self._logger, yaml.YAMLError, f"Error parsing YAML file: {e}")
64 self._logger.info("Config loaded from %s loaded", config_path)
65 return config
66
67 def _init_pipeline(self) -> None:
68 """
69 Initialize every transformation method in the pipeline (once on first transform call).
70 """
71 self._logger.info("Loading all the pipeline methods and models")
72 self._pipeline = []
73 try:
74 # for every transformation, initialize it and store the resulting instance
75 # initialization can be costly, so it is done only once, during first transform call.
76 for transformation_conf in self._pipeline_conf:
77 transformation_name: str = transformation_conf['name']
78 parameters = transformation_conf.get('init_param', {})
79
80 # Get module name corresponding to tranformation
81
82 module_transformation = importlib.import_module("neural_de.transformations." +
83 camel_to_snake(transformation_name))
84
85 transformation = getattr(module_transformation, transformation_name)
86 self._pipeline.append(transformation(**parameters))
87 except KeyError:
- E222 Multiple spaces after operator
89 except AttributeError:
90 log_and_raise(self._logger, AttributeError, "Transformation " + transformation_name +
91 "not found in neural.transformations")
92 except TypeError:
93 log_and_raise(self._logger, TypeError, "Invalid call during initialization of " +
94 transformation_conf["name"])
95 self._logger.info("All pipeline models successfully loaded")
96
97 def transform(self, images: Union[list, np.ndarray]) -> Union[list, np.ndarray]:
98 """
99 Sequentially apply every method of the pipeline on a batch of image, and returns
100 the resulting images.
101
102 Args:
103 images: Batch of images. Each image should be of a ``np.ndarray`` of target_shape *(h,w,
104 channels)*
105 Returns:
106 Resulting batch of images, one per image provided.
107 """
108 # verify if image is a valid batch
109 self._check_batch_validity(images)
110 # Lazy method init, as it can be costly both in term of computing power and ram.
111 if self._pipeline is None:
112 self._init_pipeline()
113 try:
114 # For each method in the pipeline
115 for i, transformation in enumerate(self._pipeline):
116 self._logger.info("Applying method %s to images", transformation)
117 # Retrieves optional transform() parameters if any
118 transformation_parameters = self._pipeline_conf[i]\
119 .get('transform', {})
120
121 # apply the transformation to the images
122 images = transformation.transform(images=images, **transformation_parameters)
123 except TypeError as e:
124 raise log_and_raise(self._logger, TypeError,
125 f"Invalid call during function transform of "
126 f"'{transformation}': {e}")
127 return images