Files
ANSLibs/OpenVINO/python/openvino/tools/ovc/convert_impl.py

581 lines
23 KiB
Python

# Copyright (C) 2018-2025 Intel Corporation
# SPDX-License-Identifier: Apache-2.0
import argparse
import datetime
import logging as log
import os
import sys
import traceback
import tracemalloc
from collections import OrderedDict
from pathlib import Path
from collections.abc import Callable, Iterable
try:
import openvino_telemetry as tm
from openvino_telemetry.backend import backend_ga4
except ImportError:
import openvino.tools.ovc.telemetry_stub as tm
from openvino.tools.ovc.moc_frontend.check_config import any_extensions_used
from openvino.tools.ovc.moc_frontend.pipeline import moc_pipeline
from openvino.tools.ovc.moc_frontend.moc_emit_ir import moc_emit_ir
from openvino.tools.ovc.moc_frontend.type_utils import to_ov_type
from openvino.tools.ovc.cli_parser import get_available_front_ends, get_common_cli_options, depersonalize, \
get_mo_convert_params, input_to_input_cut_info, parse_inputs
from openvino.tools.ovc.help import get_convert_model_help_specifics
from openvino.tools.ovc.error import Error, FrameworkError
from openvino.tools.ovc.get_ov_update_message import get_compression_message
from openvino.tools.ovc.version import VersionChecker
from openvino.tools.ovc.utils import check_values_equal
from openvino.tools.ovc.logger import init_logger
from openvino.tools.ovc.telemetry_utils import send_params_info, send_conversion_result, \
init_ovc_telemetry
from openvino.tools.ovc.moc_frontend.pytorch_frontend_utils import get_pytorch_decoder, \
extract_input_info_from_example, get_pytorch_decoder_for_model_on_disk
from openvino.tools.ovc.moc_frontend.paddle_frontend_utils import paddle_frontend_converter
try:
from openvino.tools.ovc.moc_frontend.jax_frontend_utils import get_jax_decoder
except:
get_jax_decoder = None
# pylint: disable=no-name-in-module,import-error
from openvino.frontend import FrontEndManager, OpConversionFailure, TelemetryExtension
from openvino import get_version as get_rt_version
from openvino import PartialShape
try:
from openvino.frontend.tensorflow.utils import create_tf_graph_iterator, type_supported_by_tf_fe, \
extract_model_graph # pylint: disable=no-name-in-module,import-error
tf_frontend_with_python_bindings_installed = True
except (ModuleNotFoundError, ImportError):
tf_frontend_with_python_bindings_installed = False
def replace_ext(name: str, old: str, new: str):
base, ext = os.path.splitext(name)
log.debug("base: {}, ext: {}".format(base, ext))
if ext == old:
return base + new
def print_argv(argv: argparse.Namespace):
print('Model Conversion arguments:')
props = OrderedDict()
props['common_args'] = get_common_cli_options(argv, argv.is_python_api_used)
framework_specifics_map = {
'common_args': 'Common parameters:'
}
lines = []
for key in props:
lines.append(framework_specifics_map[key])
for (op, desc) in props[key].items():
if isinstance(desc, list):
lines.append('\t{}: \t{}'.format(desc[0], desc[1](getattr(argv, op, 'NONE'))))
else:
lines.append('\t{}: \t{}'.format(desc, getattr(argv, op, 'NONE')))
print('\n'.join(lines), flush=True)
def check_iterable(iterable: Iterable, func: Callable):
for element in iterable:
if not func(element):
return False
return True
def arguments_post_parsing(argv: argparse.Namespace):
# TODO: This function looks similar to another one. Check for code duplicates.
log.debug("Model Conversion API started")
if not argv.is_python_api_used:
log.debug('Output model name would be {}{{.xml, .bin}}'.format(argv.output_model))
if is_verbose(argv):
print_argv(argv)
import re
if argv.is_python_api_used and isinstance(argv.input, str):
argv.input = [argv.input]
if not argv.is_python_api_used and isinstance(argv.input, str):
argv.input = parse_inputs(argv.input)
normalize_inputs(argv)
log.debug("Placeholder shapes : {}".format(argv.placeholder_shapes))
if not hasattr(argv, 'output') or argv.output is None:
return argv
if argv.is_python_api_used:
error_msg = f"output '{argv.output}' is incorrect, it should be string or a list/tuple of strings"
assert isinstance(argv.output, (str, list, tuple)), error_msg
if isinstance(argv.output, list):
assert check_iterable(argv.output, lambda x: isinstance(x, str)), error_msg
else:
argv.output = [argv.output]
else:
assert isinstance(argv.output, str)
error_msg = f"output '{argv.output}' is incorrect, output names should not be empty or contain spaces"
processed_output = re.split(r'\s*,\s*', argv.output.strip())
assert check_iterable(processed_output, lambda x: x.find(' ') == -1), error_msg
assert check_iterable(processed_output, lambda x: len(x) > 0), error_msg
argv.output = processed_output
return argv
def get_moc_frontends(argv: argparse.Namespace):
fem = argv.feManager
if not fem:
return None, []
available_moc_front_ends = get_available_front_ends(fem)
if argv.framework:
moc_front_end = fem.load_by_framework(argv.framework)
return moc_front_end, available_moc_front_ends
if argv.input_model:
if isinstance(argv.input_model, (tuple, list)) and len(argv.input_model) == 2:
moc_front_end = fem.load_by_model(
[argv.input_model[0], argv.input_model[1]]) # TODO: Pass all input model parts
else:
moc_front_end = fem.load_by_model(argv.input_model)
if not moc_front_end:
return None, available_moc_front_ends
argv.framework = moc_front_end.get_name()
else:
return None, []
# This check as a workaround to skip IR frontend
if not moc_front_end.get_name() in available_moc_front_ends:
return None, available_moc_front_ends
return moc_front_end, available_moc_front_ends
def filtered_extensions(extensions):
try:
new_extensions = []
from openvino.frontend.pytorch.module_extension import ModuleExtension
for ext in extensions:
if not isinstance(ext, ModuleExtension):
new_extensions.append(ext)
return new_extensions
except:
return extensions
def prepare_ir(argv: argparse.Namespace):
argv = arguments_post_parsing(argv)
t = tm.Telemetry()
if isinstance(argv.input_model, (tuple, list)) and len(argv.input_model) == 1:
argv.input_model = argv.input_model[0]
moc_front_end, available_moc_front_ends = get_moc_frontends(argv)
if moc_front_end:
# TODO: Should be moved to the same place where paddle and pytorch handle their objects
if argv.framework == 'tf' and argv.is_python_object and type_supported_by_tf_fe(argv.input_model):
argv.input_model = create_tf_graph_iterator(argv.input_model,
argv.placeholder_shapes,
argv.placeholder_data_types,
getattr(argv, "example_input", None),
argv.share_weights)
t.send_event("ovc", "conversion_method", moc_front_end.get_name() + "_frontend")
moc_front_end.add_extension(TelemetryExtension("ovc", t.send_event, t.send_error, t.send_stack_trace))
if any_extensions_used(argv):
for extension in filtered_extensions(argv.extension):
moc_front_end.add_extension(extension)
ov_model = moc_pipeline(argv, moc_front_end)
return ov_model
if not argv.input_model:
raise Error('No input model is provided')
raise Error('Cannot recognize input model.')
def check_model_object(argv):
model = argv['input_model']
if 'tensorflow' in sys.modules:
if tf_frontend_with_python_bindings_installed and extract_model_graph(argv):
return "tf"
if 'torch' in sys.modules:
import torch
if isinstance(model, (torch.nn.Module, torch.jit.ScriptFunction)) or (
hasattr(torch, "export") and isinstance(model, (torch.export.ExportedProgram))):
return "pytorch"
try:
from openvino.frontend.pytorch.ts_decoder import TorchScriptPythonDecoder
from openvino.frontend.pytorch.fx_decoder import TorchFXPythonDecoder
if isinstance(model, (TorchScriptPythonDecoder, TorchFXPythonDecoder)):
return "pytorch"
except Exception as e:
pass
import io
# FIXME: Consuming any io.BytesIO object as an ONNX model is too dengerous and
# can conflict with others in the future (not future proof).
# TODO: Refer to https://onnx.ai/onnx/intro/python.html to find examples with
# real ONNX python objects. ONNX model has onnx.onnx_ml_pb2.ModelProto type.
if isinstance(model, io.BytesIO):
return 'onnx'
if 'paddle' in sys.modules:
import paddle
if isinstance(model, paddle.hapi.model.Model) or isinstance(model,
paddle.fluid.dygraph.layers.Layer) or isinstance(
model, paddle.fluid.executor.Executor):
return "paddle"
if 'jax' in sys.modules:
import jax
from packaging import version
if version.parse(jax.__version__) < version.parse("0.6.0"):
import jax as jex
import jax.core
else:
import jax.extend as jex
if isinstance(model, (jex.core.Jaxpr, jex.core.ClosedJaxpr)):
return "jax"
raise Error('Unknown model type: {}'.format(type(model)))
def driver(argv: argparse.Namespace, non_default_params: dict):
# Log dictionary with non-default cli parameters where complex classes are excluded.
log.debug(str(non_default_params))
ov_model = moc_emit_ir(prepare_ir(argv), argv)
return ov_model
def get_non_default_params(argv, cli_parser):
import numbers
import inspect
from openvino.tools.ovc import convert_model
signature = inspect.signature(convert_model)
# make dictionary with parameters which have non-default values to be serialized in IR in rt_info
non_default_params = {}
for arg, arg_value in vars(argv).items():
if arg in signature.parameters and check_values_equal(arg_value, signature.parameters[arg].default):
continue
if check_values_equal(arg_value, cli_parser.get_default(arg)):
continue
value = depersonalize(arg_value, arg)
# Skip complex classes in params to prevent
# serializing it to rt_info
if isinstance(value, (str, bool, numbers.Number)):
non_default_params[arg] = value
return non_default_params
def add_line_breaks(text: str, char_num: int, line_break: str):
words = text.replace('\n', "\n ").split(" ")
cnt = 0
for i, w in enumerate(words):
cnt += len(w)
if '\n' in w:
cnt = len(w) - w.find('\n') - 1
if cnt > char_num:
if words[i][-1] not in ['\n', '\t']:
words[i] = w + '\n'
cnt = 0
text = ' '.join(words).replace("\n ", "\n")
return line_break + text.replace("\n", line_break)
def show_mo_convert_help():
mo_convert_params = get_mo_convert_params()
for group_name, group in mo_convert_params.items():
print(group_name)
for param_name in group:
param_data = group[param_name]
text = param_data.description.replace(" ", '')
text = add_line_breaks(text, 56, "\n\t\t\t")
print(" :param {} {}".format(param_name, text))
print()
def input_model_is_object(input_model):
if input_model == ():
return False
if isinstance(input_model, (str, Path)):
return False
if isinstance(input_model, (tuple, list)):
return all(input_model_is_object(part) for part in input_model)
return True
def normalize_inputs(argv: argparse.Namespace):
"""
repacks params passed to convert_model and wraps resulting values into dictionaries or lists.
After working of this method following values are set in argv:
argv.input, argv.inputs_list - list of input names. Both values are used in some parts of MO.
Could be good to refactor it and use only one of these values.
argv.placeholder_shapes - dictionary where key is node name, value is PartialShape,
or list of PartialShape if node names were not set.
argv.placeholder_data_types - dictionary where key is node name, value is node np.type,
or list of np.types if node names were not set.
:param argv: OVC arguments
"""
# Parse input to list of InputCutInfo
inputs = input_to_input_cut_info(argv.input)
argv.input = inputs
# Make list of input names
input_names_list = []
for inp in inputs:
if inp.name is not None:
input_names_list.append(inp.name)
if len(input_names_list) > 0:
assert len(input_names_list) == len(inputs), "\"input\" parameter has unnamed inputs and named inputs. " \
"Please either set names for all inputs, " \
"or do not set names for all inputs."
if len(input_names_list) > 0:
# Named inputs case
shape_dict = {}
data_type_dict = {}
for inp in inputs:
if inp.shape is not None:
# Wrap shape to PartialShape for uniformity of stored values
shape_dict[inp.name] = PartialShape(inp.shape)
else:
shape_dict[inp.name] = None
if inp.type is not None:
# Convert type to ov.Type for uniformity of stored values
data_type_dict[inp.name] = to_ov_type(inp.type)
argv.placeholder_shapes = shape_dict if shape_dict else None
argv.placeholder_data_types = data_type_dict if data_type_dict else {}
else:
# Unnamed inputs case
shape_list = []
data_type_list = []
for inp in inputs:
if inp.shape is not None:
# Wrap shape to PartialShape for uniformity of stored values
shape_list.append(PartialShape(inp.shape))
if inp.type is not None:
# Convert type to ov.Type for uniformity of stored values
data_type_list.append(to_ov_type(inp.type))
argv.placeholder_shapes = shape_list if shape_list else None
argv.placeholder_data_types = data_type_list if data_type_list else {}
if hasattr(argv, "framework") and argv.framework == "pytorch" and getattr(argv, "example_input", None) is not None:
extract_input_info_from_example(argv, inputs)
def args_to_argv(**kwargs):
argv = argparse.Namespace()
args_specifics = get_convert_model_help_specifics()
import inspect
from openvino.tools.ovc import convert_model
signature = inspect.signature(convert_model)
for key, value in kwargs.items():
if value is None and key in signature.parameters:
setattr(argv, key, signature.parameters[key].default)
continue
if key in args_specifics:
param_specifics = args_specifics[key]
if 'action' in param_specifics and hasattr(param_specifics['action'], 'check_value'):
value = param_specifics['action'].check_value(value, key)
if 'type' in param_specifics:
value = param_specifics['type'](value)
setattr(argv, key, value)
return argv
def pack_params_to_args_namespace(args: dict, cli_parser: argparse.ArgumentParser, python_api_used):
if python_api_used:
argv = args_to_argv(**args)
# get list of all available params for convert_model()
all_params = {}
for key, value in get_mo_convert_params().items():
all_params.update(value)
# check that there are no unknown params provided
for key, value in args.items():
if key not in all_params.keys():
raise Error("Unrecognized argument: {}".format(key))
else:
argv = cli_parser.parse_args()
return argv
def is_verbose(argv, args=None):
if argv is not None and hasattr(argv, 'verbose') and argv.verbose:
return True
if args is not None and 'verbose' in args and args['verbose']:
return True
if '--verbose' in sys.argv:
return True
return False
def _convert(cli_parser: argparse.ArgumentParser, args, python_api_used):
start_time = datetime.datetime.now()
if is_verbose(None, args):
tracemalloc.start()
simplified_ie_version = VersionChecker().get_ie_simplified_version()
telemetry = init_ovc_telemetry()
telemetry.start_session('ovc')
telemetry.send_event('ovc', 'version', simplified_ie_version)
# Initialize logger with 'ERROR' as default level to be able to form nice messages
# before arg parser deliver log_level requested by user
verbose = False
if "verbose" in args and args["verbose"] or "--verbose" in sys.argv:
verbose = True
init_logger('ERROR', verbose, python_api_used)
argv = None
# Minimize modifications among other places in case if multiple pieces are passed as input_model
if python_api_used:
if 'input_model' not in args:
args['input_model'] = ()
if isinstance(args['input_model'], (tuple, list)) and len(args['input_model']) == 1:
args['input_model'] = args['input_model'][0]
try:
model_framework = None
inp_model_is_object = input_model_is_object(args['input_model']) if python_api_used else False
if inp_model_is_object:
model_framework = check_model_object(args)
if model_framework == "pytorch":
example_inputs = None
if 'example_input' in args and args['example_input'] is not None:
example_inputs = args['example_input']
elif 'example_inputs' in args:
raise AssertionError(
"'example_inputs' argument is not recognized, maybe you meant to provide 'example_input'?")
get_pytorch_decoder(args['input_model'], example_inputs, args)
if model_framework == "paddle":
example_inputs = None
if 'example_input' in args and args['example_input'] is not None:
example_inputs = args['example_input']
outputs = None
if 'output' in args and args['output'] is not None:
# Once the temporary PDPD model is generated. output can be dropped.
# Just swap outputs and args['output'] can reset the argv.output to `None`.
# It can avoid the following `output` negative effect.
outputs, args['output'] = args['output'], outputs
paddle_runtime_converter = paddle_frontend_converter(args['input_model'], example_inputs,
outputs)
pdmodel = paddle_runtime_converter.convert_paddle_to_pdmodel()
args['input_model'] = pdmodel
if model_framework == "jax":
if get_jax_decoder is not None:
get_jax_decoder(args['input_model'], args)
else:
raise Error("JAX Frontend is not available.")
if model_framework == "tf" and "nncf" in sys.modules:
try:
from nncf.tensorflow.strip import strip as nncf_tf_strip
args['input_model'] = nncf_tf_strip(args['input_model'])
except:
pass
argv = pack_params_to_args_namespace(args, cli_parser, python_api_used)
argv.framework = model_framework
argv.is_python_object = inp_model_is_object
argv.feManager = FrontEndManager()
non_default_params = get_non_default_params(argv, cli_parser)
argv.is_python_api_used = python_api_used
# send telemetry with params info
send_params_info(non_default_params)
argv.framework = model_framework
orig_input_model = argv.input_model
pytorch_model_on_disk = False
if argv.framework is None and get_pytorch_decoder_for_model_on_disk(argv, args):
# try to load a model from disk as TorchScript or ExportedProgram
# TorchScriptPythonDecoder or TorchFXPythonDecoder object will be assigned to argv.input_model
# saved TorchScript and ExportedModel model can be passed to both ovc tool and Python convert_model
pytorch_model_on_disk = True
ov_model = driver(argv, {"conversion_parameters": non_default_params})
if pytorch_model_on_disk:
# release memory allocated for temporal object
del argv.input_model
# restore original model name in arguments for tool reporting
argv.input_model = orig_input_model
if inp_model_is_object and model_framework == "paddle":
if paddle_runtime_converter:
paddle_runtime_converter.destroy()
# add OVC meta data to model
ov_model.set_rt_info(get_rt_version(), "Runtime_version")
for key, value in non_default_params.items():
ov_model.set_rt_info(str(value), ["conversion_parameters", str(key)])
if is_verbose(argv) or not python_api_used:
if 'compress_to_fp16' in argv and argv.compress_to_fp16:
print(get_compression_message())
send_conversion_result('success')
if is_verbose(argv):
elapsed_time = datetime.datetime.now() - start_time
print('[ SUCCESS ] Total execution time: {:.2f} seconds. '.format(elapsed_time.total_seconds()))
_, peak_size = tracemalloc.get_traced_memory()
print("[ SUCCESS ] Peak memory consumption (includes only memory allocated in Python): {:.2f} MB. ".format(
peak_size / (1024 * 1024)))
tracemalloc.stop()
return ov_model, argv
except Exception as e:
if is_verbose(argv) or not python_api_used:
if isinstance(e, (FileNotFoundError, NotADirectoryError)):
log.error('File {} was not found'.format(str(e).split('No such file or directory:')[1]))
log.debug(traceback.format_exc())
elif isinstance(e, (Error, OpConversionFailure)):
log.error(e)
log.debug(traceback.format_exc())
elif isinstance(e, FrameworkError):
log.error(e, extra={'framework_error': True})
log.debug(traceback.format_exc())
else:
log.error("-------------------------------------------------")
log.error("----------------- INTERNAL ERROR ----------------")
log.error("Unexpected exception happened.")
log.error("Please verify parameters and environment.")
log.error("If you think this is a bug, please create new ticket here: ")
log.error("https://github.com/openvinotoolkit/openvino/issues.")
log.error("-------------- DETAILED INFORMATION -------------")
log.error(str(e))
log.error(traceback.format_exc())
log.error("----------------- END OF REPORT -----------------")
log.error("-------------------------------------------------")
send_conversion_result('fail')
if python_api_used:
raise e
else:
return None, argv