AC: Refactoring mtcnn models and init in cascade models (#2952)

* Minor refactoring

* Refactoring mtcnn

* Aligned init in cascade models

Author: Anna Grebneva

tools/accuracy_checker/openvino/tools/accuracy_checker/evaluators/custom_evaluators/asr_custom_encoder_decoder_joint.py

@@ -19,7 +19,7 @@
 from .asr_encoder_prediction_joint_evaluator import ASREvaluator
 from .base_models import create_model, BaseCascadeModel, BaseDLSDKModel, BaseONNXModel, BaseOpenVINOModel
 from ...adapters import create_adapter
-from ...utils import generate_layer_name, contains_all, contains_any
+from ...utils import generate_layer_name, contains_all
 from ...config import ConfigError
 
 
@@ -382,22 +382,10 @@ class ASRModel(BaseCascadeModel):
 
     def __init__(self, network_info, adapter_config, launcher, models_args, is_blob, delayed_model_loading=False):
         super().__init__(network_info, launcher)
-        if models_args and not delayed_model_loading:
-            encoder = network_info.get('encoder', {})
-            decoder = network_info.get('decoder', {})
-            joint = network_info.get('joint', {})
-            if not contains_any(encoder, ['model', 'onnx_model']) and models_args:
-                encoder['model'] = models_args[0]
-                encoder['_model_is_blob'] = is_blob
-            if not contains_any(decoder, ['model', 'onnx_model']) and models_args:
-                decoder['model'] = models_args[1 if len(models_args) > 1 else 0]
-                decoder['_model_is_blob'] = is_blob
-            if not contains_any(joint, ['model', 'onnx_model']) and models_args:
-                joint['model'] = models_args[2 if len(models_args) > 2 else 0]
-                joint['_model_is_blob'] = is_blob
-            network_info.update({'encoder': encoder, 'decoder': decoder, 'joint': joint})
-        if not contains_all(network_info, ['encoder', 'decoder', 'joint']) and not delayed_model_loading:
-            raise ConfigError('network_info should contain encoder, prediction and joint fields')
+        parts = ['encoder', 'decoder', 'joint']
+        network_info = self.fill_part_with_model(network_info, parts, models_args, is_blob, delayed_model_loading)
+        if not contains_all(network_info, parts) and not delayed_model_loading:
+            raise ConfigError('network_info should contain encoder, decoder and joint fields')
         self._decoder_mapping = {
             'dlsdk': DLSDKDecoder,
             'openvino': OVDecoder,

tools/accuracy_checker/openvino/tools/accuracy_checker/evaluators/custom_evaluators/asr_encoder_decoder_evaluator.py

@@ -26,7 +26,7 @@
     create_model, create_encoder)
 from ...adapters import create_adapter
 from ...config import ConfigError
-from ...utils import contains_all, contains_any, extract_image_representations, read_pickle
+from ...utils import contains_all, extract_image_representations, read_pickle
 
 
 class AutomaticSpeechRecognitionEvaluator(BaseCustomEvaluator):
@@ -67,17 +67,9 @@ def _process(self, output_callback, calculate_metrics, progress_reporter, metric
 class ASRModel(BaseCascadeModel):
     def __init__(self, network_info, launcher, models_args, is_blob, delayed_model_loading=False):
         super().__init__(network_info, launcher)
-        if models_args and not delayed_model_loading:
-            encoder = network_info.get('encoder', {})
-            decoder = network_info.get('decoder', {})
-            if not contains_any(encoder, ['model', 'onnx_model']) and models_args:
-                encoder['model'] = models_args[0]
-                encoder['_model_is_blob'] = is_blob
-            if not contains_any(decoder, ['model', 'onnx_model']) and models_args:
-                decoder['model'] = models_args[1 if len(models_args) > 1 else 0]
-                decoder['_model_is_blob'] = is_blob
-            network_info.update({'encoder': encoder, 'decoder': decoder})
-        if not contains_all(network_info, ['encoder', 'decoder']) and not delayed_model_loading:
+        parts = ['encoder', 'decoder']
+        network_info = self.fill_part_with_model(network_info, parts, models_args, is_blob, delayed_model_loading)
+        if not contains_all(network_info, parts) and not delayed_model_loading:
             raise ConfigError('network_info should contain encoder and decoder fields')
         self._decoder_mapping = {
             'dlsdk': DecoderDLSDKModel,

tools/accuracy_checker/openvino/tools/accuracy_checker/evaluators/custom_evaluators/asr_encoder_prediction_joint_evaluator.py

@@ -21,7 +21,7 @@
 
 from ...adapters import create_adapter
 from ...config import ConfigError
-from ...utils import contains_all, contains_any, read_pickle, parse_partial_shape
+from ...utils import contains_all, read_pickle, parse_partial_shape
 from .asr_encoder_decoder_evaluator import AutomaticSpeechRecognitionEvaluator
 from .base_models import (
     BaseCascadeModel, BaseDLSDKModel, BaseOpenVINOModel, BaseONNXModel, create_model, create_encoder
@@ -43,21 +43,9 @@ def from_configs(cls, config, delayed_model_loading=False, orig_config=None):
 class ASRModel(BaseCascadeModel):
     def __init__(self, network_info, launcher, models_args, is_blob, adapter_info, delayed_model_loading=False):
         super().__init__(network_info, launcher)
-        if models_args and not delayed_model_loading:
-            encoder = network_info.get('encoder', {})
-            prediction = network_info.get('prediction', {})
-            joint = network_info.get('joint', {})
-            if not contains_any(encoder, ['model', 'onnx_model']) and models_args:
-                encoder['model'] = models_args[0]
-                encoder['_model_is_blob'] = is_blob
-            if not contains_any(prediction, ['model', 'onnx_model']) and models_args:
-                prediction['model'] = models_args[1 if len(models_args) > 1 else 0]
-                prediction['_model_is_blob'] = is_blob
-            if not contains_any(joint, ['model', 'onnx_model']) and models_args:
-                joint['model'] = models_args[2 if len(models_args) > 2 else 0]
-                joint['_model_is_blob'] = is_blob
-            network_info.update({'encoder': encoder, 'prediction': prediction, 'joint': joint})
-        if not contains_all(network_info, ['encoder', 'prediction', 'joint']) and not delayed_model_loading:
+        parts = ['encoder', 'prediction', 'joint']
+        network_info = self.fill_part_with_model(network_info, parts, models_args, is_blob, delayed_model_loading)
+        if not contains_all(network_info, parts) and not delayed_model_loading:
             raise ConfigError('network_info should contain encoder, prediction and joint fields')
         self._encoder_mapping = {
             'dlsdk': EncoderDLSDKModel,

tools/accuracy_checker/openvino/tools/accuracy_checker/evaluators/custom_evaluators/base_models.py

@@ -18,7 +18,7 @@
 import numpy as np
 
 from ...config import ConfigError
-from ...utils import get_path, parse_partial_shape
+from ...utils import get_path, parse_partial_shape, contains_any
 from ...logging import print_info
 
 
@@ -70,6 +70,17 @@ def get_network(self):
     def reset(self):
         pass
 
+    @staticmethod
+    def fill_part_with_model(network_info, parts, models_args, is_blob, delayed_model_loading):
+        if models_args and not delayed_model_loading:
+            for idx, part in enumerate(parts):
+                part_info = network_info.get(part, {})
+                if not contains_any(part_info, ['model', 'onnx_model']) and models_args:
+                    part_info['model'] = models_args[idx if len(models_args) > idx else 0]
+                    part_info['_model_is_blob'] = is_blob
+                network_info.update({part: part_info})
+        return network_info
+
 
 class BaseDLSDKModel:
     def __init__(self, network_info, launcher, suffix=None, delayed_model_loading=False):
@@ -415,3 +426,52 @@ def release(self):
     def automatic_model_search(network_info):
         model = Path(network_info['model'])
         return model
+
+
+class BaseCaffeModel:
+    def __init__(self, network_info, launcher, suffix=None, delayed_model_loading=False):
+        self.network_info = network_info
+        self.launcher = launcher
+        self.default_model_suffix = suffix
+
+    def fit_to_input(self, data, layer_name, layout, precision, tmpl=None):
+        return self.launcher.fit_to_input(data, layer_name, layout, precision, template=tmpl)
+
+    def predict(self, identifiers, input_data):
+        raise NotImplementedError
+
+    def release(self):
+        del self.net
+
+    def automatic_model_search(self, network_info):
+        model = Path(network_info.get('model', ''))
+        weights = network_info.get('weights')
+        if model.is_dir():
+            models_list = list(Path(model).glob('{}.prototxt'.format(self.default_model_name)))
+            if not models_list:
+                models_list = list(Path(model).glob('*.prototxt'))
+            if not models_list:
+                raise ConfigError('Suitable model description is not detected')
+            if len(models_list) != 1:
+                raise ConfigError('Several suitable models found, please specify required model')
+            model = models_list[0]
+        if weights is None or Path(weights).is_dir():
+            weights_dir = weights or model.parent
+            weights = Path(weights_dir) / model.name.replace('prototxt', 'caffemodel')
+            if not weights.exists():
+                weights_list = list(weights_dir.glob('*.caffemodel'))
+                if not weights_list:
+                    raise ConfigError('Suitable weights is not detected')
+                if len(weights_list) != 1:
+                    raise ConfigError('Several suitable weights found, please specify required explicitly')
+                weights = weights_list[0]
+        weights = Path(weights)
+        accepted_suffixes = ['.prototxt']
+        if model.suffix not in accepted_suffixes:
+            raise ConfigError('Models with following suffixes are allowed: {}'.format(accepted_suffixes))
+        print_info('{} - Found model: {}'.format(self.default_model_name, model))
+        accepted_weights_suffixes = ['.caffemodel']
+        if weights.suffix not in accepted_weights_suffixes:
+            raise ConfigError('Weights with following suffixes are allowed: {}'.format(accepted_weights_suffixes))
+        print_info('{} - Found weights: {}'.format(self.default_model_name, weights))
+        return model, weights

tools/accuracy_checker/openvino/tools/accuracy_checker/evaluators/custom_evaluators/cocosnet_evaluator.py

@@ -19,7 +19,7 @@
 import cv2
 
 from .base_custom_evaluator import BaseCustomEvaluator
-from .base_models import BaseDLSDKModel, BaseOpenVINOModel, BaseCascadeModel
+from .base_models import BaseDLSDKModel, BaseOpenVINOModel, BaseCascadeModel, create_model
 from ...adapters import create_adapter
 from ...config import ConfigError
 from ...data_readers import DataRepresentation
@@ -117,21 +117,19 @@ def __init__(self, network_info, launcher, models_args, is_blob, delayed_model_l
             })
         if not contains_all(network_info, ['cocosnet_network']) and not delayed_model_loading:
             raise ConfigError('network_info should contain cocosnet_network field')
-        use_api2 = launcher.config['framework'] == 'openvino'
-
-        if not use_api2:
-            self.test_model = CocosnetModel(network_info.get('cocosnet_network', {}), launcher, 'cocosnet_network',
-                                            delayed_model_loading)
-        else:
-            self.test_model = CoCosNetModelOV(network_info.get('cocosnet_network', {}), launcher, 'cocosnet_network',
-                                              delayed_model_loading)
+        self._test_mapping = {
+            'dlsdk': CocosnetModel,
+            'openvino': CoCosNetModelOV
+        }
+        self._check_mapping = {
+            'dlsdk': GanCheckModel,
+            'openvino': GANCheckOVModel
+        }
+        self.test_model = create_model(network_info.get('cocosnet_network', {}), launcher, self._test_mapping,
+                                       'cocosnet_network', delayed_model_loading)
         if network_info.get('verification_network'):
-            if not use_api2:
-                self.check_model = GanCheckModel(network_info.get('verification_network', {}), launcher,
-                                                 'verification_network', delayed_model_loading)
-            else:
-                self.check_model = GANCheckOVModel(network_info.get('verification_network', {}), launcher,
-                                                   'verification_network', delayed_model_loading)
+            self.check_model = create_model(network_info.get('verification_network', {}), launcher, self._check_mapping,
+                                            'verification_network', delayed_model_loading)
         else:
             self.check_model = None
         self._part_by_name = {

tools/accuracy_checker/openvino/tools/accuracy_checker/evaluators/custom_evaluators/colorization_evaluator.py

@@ -18,7 +18,7 @@
 import cv2
 
 from .base_custom_evaluator import BaseCustomEvaluator
-from .base_models import BaseDLSDKModel, BaseCascadeModel, BaseOpenVINOModel
+from .base_models import BaseDLSDKModel, BaseCascadeModel, BaseOpenVINOModel, create_model
 from ...adapters import create_adapter
 from ...config import ConfigError
 from ...utils import extract_image_representations, contains_all, parse_partial_shape
@@ -63,37 +63,24 @@ def _process(self, output_callback, calculate_metrics, progress_reporter, metric
 class ColorizationCascadeModel(BaseCascadeModel):
     def __init__(self, network_info, launcher, models_args, is_blob, delayed_model_loading=False):
         super().__init__(network_info, launcher)
-        if models_args and not delayed_model_loading:
-            colorization_network = network_info.get('colorization_network', {})
-            verification_network = network_info.get('verification_network', {})
-            if 'model' not in colorization_network and models_args:
-                colorization_network['model'] = models_args[0]
-                colorization_network['_model_is_blob'] = is_blob
-            if 'model' not in verification_network and models_args:
-                verification_network['model'] = models_args[1 if len(models_args) > 1 else 0]
-                verification_network['_model_is_blob'] = is_blob
-            network_info.update({
-                'colorization_network': colorization_network,
-                'verification_network': verification_network
-            })
-        if not contains_all(network_info, ['colorization_network', 'verification_network']):
+        parts = ['colorization_network', 'verification_network']
+        network_info = self.fill_part_with_model(network_info, parts, models_args, is_blob, delayed_model_loading)
+        if not contains_all(network_info, parts) and not delayed_model_loading:
             raise ConfigError('configuration for colorization_network/verification_network does not exist')
-        use_api2 = launcher.config['framework'] == 'openvino'
-
-        if not use_api2:
-            self.test_model = ColorizationTestModel(network_info.get('colorization_network', {}), launcher,
-                                                    'colorization_network', delayed_model_loading)
-            self.check_model = ColorizationCheckModel(network_info.get('verification_network', {}), launcher,
-                                                      'verification_network', delayed_model_loading)
-        else:
-            self.test_model = ColorizationTestOVModel(network_info.get('colorization_network', {}), launcher,
-                                                      'colorization_network', delayed_model_loading)
-            self.check_model = ColorizationCheckOVModel(network_info.get('verification_network', {}), launcher,
-                                                        'verification_network', delayed_model_loading)
-        self._part_by_name = {
-            'colorization_network': self.test_model,
-            'verification_network': self.check_model
+
+        self._test_mapping = {
+            'dlsdk': ColorizationTestModel,
+            'openvino': ColorizationTestOVModel
+        }
+        self._check_mapping = {
+            'dlsdk': ColorizationCheckModel,
+            'openvino': ColorizationCheckOVModel
         }
+        self.test_model = create_model(network_info.get('colorization_network', {}), launcher, self._test_mapping,
+                                       'colorization_network', delayed_model_loading)
+        self.check_model = create_model(network_info.get('verification_network', {}), launcher, self._check_mapping,
+                                        'verification_network', delayed_model_loading)
+        self._part_by_name = {'colorization_network': self.test_model, 'verification_network': self.check_model}
 
     @property
     def adapter(self):