464 lines
18 KiB
Python
464 lines
18 KiB
Python
# -*- coding: utf-8 -*-
|
|
# Copyright (C) 2018-2025 Intel Corporation
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Factory functions for all openvino ops."""
|
|
from typing import Optional, Union
|
|
|
|
import numpy as np
|
|
from functools import partial
|
|
|
|
from openvino import Node, Shape
|
|
from openvino.op import Constant, Parameter
|
|
from openvino.utils.decorators import binary_op, nameable_op, unary_op
|
|
from openvino.utils.input_validation import (
|
|
assert_list_of_ints,
|
|
check_valid_attributes,
|
|
is_non_negative_value,
|
|
is_positive_value,
|
|
)
|
|
from openvino.utils.node_factory import NodeFactory, _get_node_factory
|
|
from openvino.utils.types import (
|
|
NodeInput,
|
|
NumericData,
|
|
NumericType,
|
|
ScalarData,
|
|
TensorShape,
|
|
as_node,
|
|
as_nodes,
|
|
get_dtype,
|
|
get_element_type,
|
|
get_element_type_str,
|
|
make_constant_node,
|
|
)
|
|
|
|
_get_node_factory_opset4 = partial(_get_node_factory, "opset4")
|
|
|
|
# -------------------------------------------- ops ------------------------------------------------
|
|
|
|
|
|
@nameable_op
|
|
def ctc_loss(
|
|
logits: NodeInput,
|
|
logit_length: NodeInput,
|
|
labels: NodeInput,
|
|
label_length: NodeInput,
|
|
blank_index: Optional[NodeInput] = None,
|
|
preprocess_collapse_repeated: bool = False,
|
|
ctc_merge_repeated: bool = True,
|
|
unique: bool = False,
|
|
name: Optional[str] = None,
|
|
) -> Node:
|
|
"""Return a node which performs CTCLoss.
|
|
|
|
:param logits: 3-D tensor of logits.
|
|
:param logit_length: 1-D tensor of lengths for each object from a batch.
|
|
:param labels: 2-D tensor of labels for which likelihood is estimated using logits.
|
|
:param label_length: 1-D tensor of length for each label sequence.
|
|
:param blank_index: Scalar used to mark a blank index.
|
|
:param preprocess_collapse_repeated: Flag for preprocessing labels before loss calculation.
|
|
:param ctc_merge_repeated: Flag for merging repeated characters in a potential alignment.
|
|
:param unique: Flag to find unique elements in a target.
|
|
:return: The new node which performs CTCLoss
|
|
"""
|
|
if blank_index is not None:
|
|
inputs = as_nodes(logits, logit_length, labels, label_length, blank_index, name=name)
|
|
else:
|
|
inputs = as_nodes(logits, logit_length, labels, label_length, name=name)
|
|
|
|
attributes = {
|
|
"preprocess_collapse_repeated": preprocess_collapse_repeated,
|
|
"ctc_merge_repeated": ctc_merge_repeated,
|
|
"unique": unique,
|
|
}
|
|
|
|
return _get_node_factory_opset4().create("CTCLoss", inputs, attributes)
|
|
|
|
|
|
@nameable_op
|
|
def non_max_suppression(
|
|
boxes: NodeInput,
|
|
scores: NodeInput,
|
|
max_output_boxes_per_class: Optional[NodeInput] = None,
|
|
iou_threshold: Optional[NodeInput] = None,
|
|
score_threshold: Optional[NodeInput] = None,
|
|
box_encoding: str = "corner",
|
|
sort_result_descending: bool = True,
|
|
output_type: str = "i64",
|
|
name: Optional[str] = None,
|
|
) -> Node:
|
|
"""Return a node which performs NonMaxSuppression.
|
|
|
|
:param boxes: Tensor with box coordinates.
|
|
:param scores: Tensor with box scores.
|
|
:param max_output_boxes_per_class: Tensor Specifying maximum number of boxes
|
|
to be selected per class.
|
|
:param iou_threshold: Tensor specifying intersection over union threshold
|
|
:param score_threshold: Tensor specifying minimum score to consider box for the processing.
|
|
:param box_encoding: Format of boxes data encoding.
|
|
:param sort_result_descending: Flag that specifies whenever it is necessary to sort selected
|
|
boxes across batches or not.
|
|
:param output_type: Output element type.
|
|
:return: The new node which performs NonMaxSuppression
|
|
"""
|
|
if max_output_boxes_per_class is None:
|
|
max_output_boxes_per_class = make_constant_node(0, np.int64)
|
|
if iou_threshold is None:
|
|
iou_threshold = make_constant_node(0, np.float32)
|
|
if score_threshold is None:
|
|
score_threshold = make_constant_node(0, np.float32)
|
|
|
|
inputs = as_nodes(boxes, scores, max_output_boxes_per_class, iou_threshold, score_threshold, name=name)
|
|
attributes = {
|
|
"box_encoding": box_encoding,
|
|
"sort_result_descending": sort_result_descending,
|
|
"output_type": output_type,
|
|
}
|
|
|
|
return _get_node_factory_opset4().create("NonMaxSuppression", inputs, attributes)
|
|
|
|
|
|
@nameable_op
|
|
def softplus(data: NodeInput, name: Optional[str] = None) -> Node:
|
|
"""Apply SoftPlus operation on each element of input tensor.
|
|
|
|
:param data: The tensor providing input data.
|
|
:return: The new node with SoftPlus operation applied on each element.
|
|
"""
|
|
return _get_node_factory_opset4().create("SoftPlus", as_nodes(data, name=name), {})
|
|
|
|
|
|
@nameable_op
|
|
def mish(data: NodeInput, name: Optional[str] = None) -> Node:
|
|
"""Return a node which performs Mish.
|
|
|
|
:param data: Tensor with input data floating point type.
|
|
:return: The new node which performs Mish
|
|
"""
|
|
return _get_node_factory_opset4().create("Mish", as_nodes(data, name=name), {})
|
|
|
|
|
|
@nameable_op
|
|
def hswish(data: NodeInput, name: Optional[str] = None) -> Node:
|
|
"""Return a node which performs HSwish (hard version of Swish).
|
|
|
|
:param data: Tensor with input data floating point type.
|
|
:return: The new node which performs HSwish
|
|
"""
|
|
return _get_node_factory_opset4().create("HSwish", as_nodes(data, name=name), {})
|
|
|
|
|
|
@nameable_op
|
|
def swish(
|
|
data: NodeInput,
|
|
beta: Optional[NodeInput] = None,
|
|
name: Optional[str] = None,
|
|
) -> Node:
|
|
"""Return a node which performing Swish activation function Swish(x, beta=1.0) = x * sigmoid(x * beta)).
|
|
|
|
:param data: Tensor with input data floating point type.
|
|
:return: The new node which performs Swish
|
|
"""
|
|
if beta is None:
|
|
beta = make_constant_node(1.0, np.float32)
|
|
return _get_node_factory_opset4().create("Swish", as_nodes(data, beta, name=name), {})
|
|
|
|
|
|
@nameable_op
|
|
def acosh(node: NodeInput, name: Optional[str] = None) -> Node:
|
|
"""Apply hyperbolic inverse cosine function on the input node element-wise.
|
|
|
|
:param node: One of: input node, array or scalar.
|
|
:param name: Optional new name for output node.
|
|
:return: New node with arccosh operation applied on it.
|
|
"""
|
|
return _get_node_factory_opset4().create("Acosh", as_nodes(node, name=name))
|
|
|
|
|
|
@nameable_op
|
|
def asinh(node: NodeInput, name: Optional[str] = None) -> Node:
|
|
"""Apply hyperbolic inverse sinus function on the input node element-wise.
|
|
|
|
:param node: One of: input node, array or scalar.
|
|
:param name: Optional new name for output node.
|
|
:return: New node with arcsinh operation applied on it.
|
|
"""
|
|
return _get_node_factory_opset4().create("Asinh", as_nodes(node, name=name))
|
|
|
|
|
|
@nameable_op
|
|
def atanh(node: NodeInput, name: Optional[str] = None) -> Node:
|
|
"""Apply hyperbolic inverse tangent function on the input node element-wise.
|
|
|
|
:param node: One of: input node, array or scalar.
|
|
:param name: Optional new name for output node.
|
|
:return: New node with arctanh operation applied on it.
|
|
"""
|
|
return _get_node_factory_opset4().create("Atanh", as_nodes(node, name=name))
|
|
|
|
|
|
@nameable_op
|
|
def proposal(
|
|
class_probs: Node,
|
|
bbox_deltas: Node,
|
|
image_shape: NodeInput,
|
|
attrs: dict,
|
|
name: Optional[str] = None,
|
|
) -> Node:
|
|
"""Filter bounding boxes and outputs only those with the highest prediction confidence.
|
|
|
|
:param class_probs: 4D input floating point tensor with class prediction scores.
|
|
:param bbox_deltas: 4D input floating point tensor with corrected predictions of bounding boxes
|
|
:param image_shape: The 1D input tensor with 3 or 4 elements describing image shape.
|
|
:param attrs: The dictionary containing key, value pairs for attributes.
|
|
:param name: Optional name for the output node.
|
|
|
|
* base_size The size of the anchor to which scale and ratio attributes are applied.
|
|
Range of values: a positive unsigned integer number
|
|
Default value: None
|
|
Required: yes
|
|
* pre_nms_topn The number of bounding boxes before the NMS operation.
|
|
Range of values: a positive unsigned integer number
|
|
Default value: None
|
|
Required: yes
|
|
* post_nms_topn The number of bounding boxes after the NMS operation.
|
|
Range of values: a positive unsigned integer number
|
|
Default value: None
|
|
Required: yes
|
|
* nms_thresh The minimum value of the proposal to be taken into consideration.
|
|
Range of values: a positive floating-point number
|
|
Default value: None
|
|
Required: yes
|
|
* feat_stride The step size to slide over boxes (in pixels).
|
|
Range of values: a positive unsigned integer
|
|
Default value: None
|
|
Required: yes
|
|
* min_size The minimum size of box to be taken into consideration.
|
|
Range of values: a positive unsigned integer number
|
|
Default value: None
|
|
Required: yes
|
|
* ratio The ratios for anchor generation.
|
|
Range of values: a list of floating-point numbers
|
|
Default value: None
|
|
Required: yes
|
|
* scale The scales for anchor generation.
|
|
Range of values: a list of floating-point numbers
|
|
Default value: None
|
|
Required: yes
|
|
* clip_before_nms The flag that specifies whether to perform clip bounding boxes before
|
|
non-maximum suppression or not.
|
|
Range of values: True or False
|
|
Default value: True
|
|
Required: no
|
|
* clip_after_nms The flag that specifies whether to perform clip bounding boxes after
|
|
non-maximum suppression or not.
|
|
Range of values: True or False
|
|
Default value: False
|
|
Required: no
|
|
* normalize The flag that specifies whether to perform normalization of output boxes to
|
|
[0,1] interval or not.
|
|
Range of values: True or False
|
|
Default value: False
|
|
Required: no
|
|
* box_size_scale Specifies the scale factor applied to logits of box sizes before decoding.
|
|
Range of values: a positive floating-point number
|
|
Default value: 1.0
|
|
Required: no
|
|
* box_coordinate_scale Specifies the scale factor applied to logits of box coordinates
|
|
before decoding.
|
|
Range of values: a positive floating-point number
|
|
Default value: 1.0
|
|
Required: no
|
|
* framework Specifies how the box coordinates are calculated.
|
|
Range of values: "" (empty string) - calculate box coordinates like in Caffe*
|
|
tensorflow - calculate box coordinates like in the TensorFlow*
|
|
Object Detection API models
|
|
|
|
Default value: "" (empty string)
|
|
Required: no
|
|
|
|
Example of attribute dictionary:
|
|
|
|
.. code-block:: python
|
|
|
|
# just required ones
|
|
attrs = {
|
|
'base_size': 85,
|
|
'pre_nms_topn': 10,
|
|
'post_nms_topn': 20,
|
|
'nms_thresh': 0.34,
|
|
'feat_stride': 16,
|
|
'min_size': 32,
|
|
'ratio': [0.1, 1.5, 2.0, 2.5],
|
|
'scale': [2, 3, 3, 4],
|
|
}
|
|
|
|
Optional attributes which are absent from dictionary will be set with corresponding default.
|
|
:return: Node representing Proposal operation.
|
|
"""
|
|
requirements = [
|
|
("base_size", True, np.unsignedinteger, is_positive_value),
|
|
("pre_nms_topn", True, np.unsignedinteger, is_positive_value),
|
|
("post_nms_topn", True, np.unsignedinteger, is_positive_value),
|
|
("nms_thresh", True, np.floating, is_positive_value),
|
|
("feat_stride", True, np.unsignedinteger, is_positive_value),
|
|
("min_size", True, np.unsignedinteger, is_positive_value),
|
|
("ratio", True, np.floating, None),
|
|
("scale", True, np.floating, None),
|
|
("clip_before_nms", False, np.bool_, None),
|
|
("clip_after_nms", False, np.bool_, None),
|
|
("normalize", False, np.bool_, None),
|
|
("box_size_scale", False, np.floating, is_positive_value),
|
|
("box_coordinate_scale", False, np.floating, is_positive_value),
|
|
("framework", False, np.str_, None),
|
|
]
|
|
|
|
check_valid_attributes("Proposal", attrs, requirements)
|
|
|
|
return _get_node_factory_opset4().create(
|
|
"Proposal",
|
|
[class_probs, bbox_deltas, as_node(image_shape, name=name)],
|
|
attrs,
|
|
)
|
|
|
|
|
|
@nameable_op
|
|
def reduce_l1(
|
|
node: NodeInput,
|
|
reduction_axes: NodeInput,
|
|
keep_dims: bool = False,
|
|
name: Optional[str] = None,
|
|
) -> Node:
|
|
"""L1-reduction operation on input tensor, eliminating the specified reduction axes.
|
|
|
|
:param node: The tensor we want to mean-reduce.
|
|
:param reduction_axes: The axes to eliminate through mean operation.
|
|
:param keep_dims: If set to True it holds axes that are used for reduction
|
|
:param name: Optional name for output node.
|
|
:return: The new node performing mean-reduction operation.
|
|
"""
|
|
return _get_node_factory_opset4().create(
|
|
"ReduceL1",
|
|
as_nodes(node, reduction_axes, name=name),
|
|
{"keep_dims": keep_dims},
|
|
)
|
|
|
|
|
|
@nameable_op
|
|
def reduce_l2(
|
|
node: NodeInput,
|
|
reduction_axes: NodeInput,
|
|
keep_dims: bool = False,
|
|
name: Optional[str] = None,
|
|
) -> Node:
|
|
"""L2-reduction operation on input tensor, eliminating the specified reduction axes.
|
|
|
|
:param node: The tensor we want to mean-reduce.
|
|
:param reduction_axes: The axes to eliminate through mean operation.
|
|
:param keep_dims: If set to True it holds axes that are used for reduction
|
|
:param name: Optional name for output node.
|
|
:return: The new node performing mean-reduction operation.
|
|
"""
|
|
return _get_node_factory_opset4().create(
|
|
"ReduceL2",
|
|
as_nodes(node, reduction_axes, name=name),
|
|
{"keep_dims": keep_dims},
|
|
)
|
|
|
|
|
|
@nameable_op
|
|
def lstm_cell(
|
|
X: NodeInput,
|
|
initial_hidden_state: NodeInput,
|
|
initial_cell_state: NodeInput,
|
|
W: NodeInput,
|
|
R: NodeInput,
|
|
B: NodeInput,
|
|
hidden_size: int,
|
|
activations: Optional[list[str]] = None,
|
|
activations_alpha: Optional[list[float]] = None,
|
|
activations_beta: Optional[list[float]] = None,
|
|
clip: float = 0.0,
|
|
name: Optional[str] = None,
|
|
) -> Node:
|
|
"""Return a node which performs LSTMCell operation.
|
|
|
|
:param X: The input tensor with shape: [batch_size, input_size].
|
|
:param initial_hidden_state: The hidden state tensor with shape: [batch_size, hidden_size].
|
|
:param initial_cell_state: The cell state tensor with shape: [batch_size, hidden_size].
|
|
:param W: The weight tensor with shape: [4*hidden_size, input_size].
|
|
:param R: The recurrence weight tensor with shape: [4*hidden_size, hidden_size].
|
|
:param B: The bias tensor for gates with shape: [4*hidden_size].
|
|
:param hidden_size: Specifies hidden state size.
|
|
:param activations: The list of three activation functions for gates.
|
|
:param activations_alpha: The list of alpha parameters for activation functions.
|
|
:param activations_beta: The list of beta parameters for activation functions.
|
|
:param clip: Specifies bound values [-C, C] for tensor clipping performed before activations.
|
|
:param name: An optional name of the output node.
|
|
|
|
:return: The new node represents LSTMCell. Node outputs count: 2.
|
|
"""
|
|
if activations is None:
|
|
activations = ["sigmoid", "tanh", "tanh"]
|
|
if activations_alpha is None:
|
|
activations_alpha = []
|
|
if activations_beta is None:
|
|
activations_beta = []
|
|
|
|
node_inputs = as_nodes(X, initial_hidden_state, initial_cell_state, W, R, B, name=name)
|
|
|
|
attributes = {
|
|
"hidden_size": hidden_size,
|
|
"activations": activations,
|
|
"activations_alpha": activations_alpha,
|
|
"activations_beta": activations_beta,
|
|
"clip": clip,
|
|
}
|
|
return _get_node_factory_opset4().create("LSTMCell", node_inputs, attributes)
|
|
|
|
|
|
@nameable_op
|
|
def range(
|
|
start: Node,
|
|
stop: NodeInput,
|
|
step: NodeInput,
|
|
output_type: str,
|
|
name: Optional[str] = None,
|
|
) -> Node:
|
|
"""Return a node which produces the Range operation.
|
|
|
|
:param start: The start value of the generated range.
|
|
:param stop: The stop value of the generated range.
|
|
:param step: The step value for the generated range.
|
|
:param output_type: The output tensor type.
|
|
:param name: Optional name for output node.
|
|
:return: Range node
|
|
"""
|
|
return _get_node_factory_opset4().create(
|
|
"Range",
|
|
as_nodes(start, stop, step, name=name),
|
|
{
|
|
"output_type": output_type,
|
|
},
|
|
)
|
|
|
|
|
|
@nameable_op
|
|
def scatter_nd_update(
|
|
data: NodeInput,
|
|
indices: NodeInput,
|
|
updates: NodeInput,
|
|
name: Optional[str] = None,
|
|
) -> Node:
|
|
"""Return a node which performs ScatterNDUpdate.
|
|
|
|
:param data: Node input representing the tensor to be updated.
|
|
:param indices: Node input representing the indices at which updates will be applied.
|
|
:param updates: Node input representing the updates to be applied.
|
|
:param name: Optional name for the output node.
|
|
:return: New node performing the ScatterNDUpdate.
|
|
"""
|
|
inputs = as_nodes(data, indices, updates, name=name)
|
|
|
|
return _get_node_factory_opset4().create("ScatterNDUpdate", inputs, {})
|