diff --git a/tests/functional/constants/pipelines.py b/tests/functional/constants/pipelines.py index 49d25b3528..36a7b47f78 100644 --- a/tests/functional/constants/pipelines.py +++ b/tests/functional/constants/pipelines.py @@ -22,46 +22,15 @@ import numpy as np -from tests.functional.config import datasets_path -from ovms.constants.model_dataset import RandomDataset from tests.functional.models.models import ModelInfo -from ovms.constants.models import ( - ArgMax, - Dummy, - DummyAdd2Inputs, - DummyIncrement, - DummyIncrementDecrement, - GoogleNetV2Fp32, - Increment4d, - Resnet, - ResnetWrongInputShapeDim, - ResnetWrongInputShapes, - VehicleAttributesRecognition, - VehicleDetection, -) from tests.functional.constants.ovms import Config from tests.functional.constants.paths import Paths -from tests.functional.object_model.custom_node import ( - CustomNode, - CustomNodeAddSub, - CustomNodeChooseMaximum, - CustomNodeDemultiply, - CustomNodeDifferentOperations, - CustomNodeDynamicDemultiplex, - CustomNodeElastic1T, - CustomNodeVehicles, -) -from tests.functional.object_model.mediapipe_calculators import ( - CorruptedFileCalculator, - MediaPipeCalculator, - OpenVINOInferenceCalculator, - OpenVINOModelServerSessionCalculator, - PythonCalculator, -) +from tests.functional.object_model.custom_node import CustomNode +from tests.functional.object_model.mediapipe_calculators import MediaPipeCalculator, PythonCalculator, \ + OpenVINOModelServerSessionCalculator, OpenVINOInferenceCalculator class NodesConnection: - def __init__(self, target_node, target_node_input_id, source_node, source_output_id): self.target_node = target_node self.target_node_input_id = target_node_input_id @@ -100,7 +69,6 @@ class NodeType(Enum): class Node: - def __init__( self, name, @@ -292,7 +260,6 @@ def __init__( class Pipeline(ModelInfo): - def __init__(self, name=None, **kwargs): self.name = name self.child_nodes = [] @@ -554,17 +521,13 @@ def change_output_name(self, old_name, new_name): class SimplePipeline(Pipeline): - def __init__(self, model=None, demultiply_count=None, name=None, **kwargs): + def __init__(self, model, demultiply_count=None, name=None, **kwargs): name = "single_model_pipeline" if name is None else f"single_model_pipeline_{name}" super().__init__(name=name, **kwargs) self.demultiply_count = demultiply_count - - if model is None: - model = Resnet() - self._initialize([model]) - def _create_nodes(self, models): + def _create_nodes(self, models=None): model = models[0] node1 = Node("node_1", model) @@ -577,461 +540,6 @@ def _create_nodes(self, models): return [request, node1, output] -class MultipleInputsOutputsPipeline(Pipeline): - - def __init__(self, **kwargs): - super().__init__("multiple_inputs_outputs_pipeline", **kwargs) - self._initialize() - - def _create_nodes(self, models=None): - node1 = Node("node_1", DummyIncrementDecrement()) - - request = Node("request", node_type=NodeType.Input) - output = Node("output", node_type=NodeType.Output) - - NodesConnection.connect(node1, 0, request, 0) - NodesConnection.connect(node1, 1, request, 1) - NodesConnection.connect(output, 0, node1, 0) - NodesConnection.connect(output, 1, node1, 1) - - nodes = [request, node1, output] - return nodes - - def get_expected_output(self, input_data: dict, client_type: str = None): - model_output = self.get_models()[0].get_expected_output(input_data) - return self.map_model_output_to_pipeline_output(model_output) - - -class InputNotConnectedPipeline(Pipeline): - - def __init__(self, **kwargs): - super().__init__("single_input_not_nonnected_pipeline", **kwargs) - self._initialize() - - def _create_nodes(self, models=None): - node1 = Node("node_1", DummyIncrementDecrement()) - - request = Node("request", node_type=NodeType.Input) - output = Node("output", node_type=NodeType.Output) - - NodesConnection.connect(node1, 0, request, 0) - NodesConnection.connect(output, 0, node1, 0) - NodesConnection.connect(output, 1, node1, 1) - - nodes = [request, node1, output] - return nodes - - -class ComplexDummyPipeline(Pipeline): - - def __init__(self, **kwargs): - super().__init__("complex_pipeline", **kwargs) - self._initialize() - - def get_pipeline_transpose_axes(self): - return None - - def get_pipeline_datasets(self): - return { - "input1": os.path.join(datasets_path, RandomDataset.name), - "input2": os.path.join(datasets_path, RandomDataset.name), - } - - def _create_nodes(self, models=None): - node1 = Node("node_1", DummyIncrementDecrement()) - node2 = Node("node_2", DummyIncrement()) - node3 = Node("node_3", DummyAdd2Inputs()) - node4 = Node("node_4", DummyAdd2Inputs()) - node5 = Node("node_5", DummyIncrement()) - node6 = Node("node_6", DummyIncrement()) - - request = Node("request", node_type=NodeType.Input) - output = Node("output", node_type=NodeType.Output) - - NodesConnection.connect(node1, 0, request, 0) - NodesConnection.connect(node1, 1, request, 1) - NodesConnection.connect(node2, 0, node1, 0) - NodesConnection.connect(node3, 0, node1, 1) - NodesConnection.connect(node3, 1, node6, 0) - NodesConnection.connect(node4, 0, node1, 0) - NodesConnection.connect(node4, 1, node2, 0) - NodesConnection.connect(node5, 0, request, 1) - NodesConnection.connect(node6, 0, request, 1) - NodesConnection.connect(output, 0, node4, 0) - NodesConnection.connect(output, 1, node3, 0) - NodesConnection.connect(output, 2, node5, 0) - NodesConnection.connect(output, 3, node1, 0) - - nodes = [request, node1, node2, node3, node4, node5, node6, output] - return nodes - - -class DemultiplyPipeline(Pipeline): - - def __init__(self, demultiply_value, **kwargs): - super().__init__("demultiply_pipeline", **kwargs) - self.demultiply_node = Node( - "demultiply", CustomNodeDemultiply(demultiply_value), NodeType.Custom, demultiply_count=-1 - ) - self.resnet_node = Node("resnet", Resnet()) - self._initialize() - self.update_demultiply_value(demultiply_value) - - def update_demultiply_value(self, new_demultiply_value): - self.demultiply_node.model.demultiply_size = new_demultiply_value - self.set_expected_output_shape() - - def set_expected_output_shape(self): - self.outputs["result"]["shape"] = [self.demultiply_node.model.demultiply_size] + self.resnet_node.model.outputs[ - "softmax_tensor" - ]["shape"] - - def _create_nodes(self, models=None): - request = Node("request", node_type=NodeType.Input) - output = Node("output", node_type=NodeType.Output, input_names=["result"]) - - NodesConnection.connect(self.demultiply_node, 0, request, 0) - NodesConnection.connect(self.resnet_node, 0, self.demultiply_node, 0) - NodesConnection.connect(output, 0, self.resnet_node, 0) - - return [request, self.demultiply_node, self.resnet_node, output] - - -class ElasticPipeline(Pipeline): - - def __init__(self, input_shape, output_shape, demultiply_count=None, **kwargs): - super().__init__("elastic_pipeline", **kwargs) - self.custom_node = Node( - "elastic_node", - CustomNodeElastic1T(input_shape, output_shape), - NodeType.Custom, - demultiply_count=demultiply_count, - ) - self._request = Node("request", node_type=NodeType.Input) - self.model_node = Node("resnet", Resnet()) - for key in self.model_node.model.inputs: - self.model_node.model.inputs[key]["shape"] = None - self._initialize() - self.set_expected_output_shape() - - def set_expected_output_shape(self): - demultiply_value = self.custom_node.model.outputs["tensor_out"]["shape"][0] - self.outputs["result"]["shape"] = [demultiply_value] + self.model_node.model.outputs["softmax_tensor"]["shape"] - - def _create_nodes(self, models): - output = Node("output", node_type=NodeType.Output, input_names=["result"]) - - NodesConnection.connect(self.custom_node, 0, self._request, 0) - NodesConnection.connect(self.model_node, 0, self.custom_node, 0) - NodesConnection.connect(output, 0, self.model_node, 0) - return [self._request, self.custom_node, self.model_node, output] - - -class ElasticBatchSizePipeline(Pipeline): - def __init__(self, node_batch_configuration_list, **kwargs): - super().__init__(name="misconfigurated_pipeline", **kwargs) - self.node_batch_cfg_list = node_batch_configuration_list - self._initialize() - - def _create_nodes(self, models): - request = Node("request", node_type=NodeType.Input, output_names=["input"]) - output = Node("output", node_type=NodeType.Output, input_names=["output"]) - - nodes = [request] - for node_name, batch_size in self.node_batch_cfg_list: - model = Dummy(batch_size=batch_size) - model.name = f"{model.name}_{node_name}" - nodes.append(Node(node_name, model)) - nodes.append(output) - - for i in range(1, len(nodes)): - NodesConnection.connect(nodes[i], 0, nodes[i - 1], 0) - return nodes - - -class CustomNodesConnectedToEachOtherPipeline(Pipeline): - - def __init__(self, **kwargs): - super().__init__("custom_nodes_connected_to_each_other_pipeline", **kwargs) - self.custom_node_a = Node("node_1", CustomNodeAddSub(1.5, 0.7), NodeType.Custom) - self.custom_node_b = Node("node_2", CustomNodeAddSub(2.4, 1.2), NodeType.Custom) - self._initialize() - - def set_expected_output_shape(self): - arg_name = list(self.custom_node_a.model.outputs.keys())[0] - self.outputs["output_0"]["shape"] = self.custom_node_a.model.outputs[arg_name]["shape"] - - def _create_nodes(self, models=None): - request = Node("request", node_type=NodeType.Input) - output = Node("output", node_type=NodeType.Output) - nodes = [request, self.custom_node_a, self.custom_node_b, output] - - for i in range(1, len(nodes)): - NodesConnection.connect(nodes[i], 0, nodes[i - 1], 0) - return nodes - - def get_expected_output(self, input_data, client_type: str = None): - custom_nodes = self.get_custom_nodes() - node1_out = custom_nodes[0].get_expected_output(input_data) - node2_out = custom_nodes[1].get_expected_output(node1_out) - return self.map_model_output_to_pipeline_output(node2_out) - - -class CustomNodeNotAllOutputsConnectedPipeline(Pipeline): - def __init__(self, **kwargs): - super().__init__("custom_node_not_all_outputs_connected_pipeline", **kwargs) - self._initialize() - - def _create_nodes(self, models=None): - node1 = Node("node_1", CustomNodeDifferentOperations(), NodeType.Custom) - - request = Node("request", node_type=NodeType.Input, output_names=self.input_names) - output = Node("output", node_type=NodeType.Output, input_names=self.output_names) - - NodesConnection.connect(node1, 0, request, 0) - NodesConnection.connect(node1, 1, request, 1) - NodesConnection.connect(output, 0, node1, 0) - - nodes = [request, node1, output] - return nodes - - -class CustomNodeNotAllInputsConnectedPipeline(Pipeline): - - def __init__(self, **kwargs): - super().__init__("custom_node_not_all_inputs_connected_pipeline", **kwargs) - self._initialize() - - def _create_nodes(self, models=None): - node1 = Node("node_1", CustomNodeDifferentOperations(), NodeType.Custom) - - request = Node("request", node_type=NodeType.Input, output_names=self.input_names) - output = Node("output", node_type=NodeType.Output, input_names=self.output_names) - - NodesConnection.connect(node1, 1, request, 1) - NodesConnection.connect(output, 0, node1, 0) - NodesConnection.connect(output, 1, node1, 1) - - nodes = [request, node1, output] - return nodes - - -class VehiclesAnalysisPipeline(Pipeline): - - def __init__(self, **kwargs): - super().__init__("multiple_vehicle_recognition", **kwargs) - self._initialize() - - def _create_nodes(self, models=None): - detection_model = VehicleDetection() - recognition_model = VehicleAttributesRecognition() - vehicles_custom_node = CustomNodeVehicles() - - vehicle_detection_node = Node("vehicle_detection_node", detection_model, output_names=["detection_out"]) - extract_node = Node( - "extract_node", - vehicles_custom_node, - NodeType.Custom, - demultiply_count=0, - output_names=["vehicle_images", "vehicle_coordinates", "confidence_levels"], - ) - vehicle_recognition_node = Node("vehicle_recognition_node", recognition_model, output_names=["color", "type"]) - - request = Node("request", node_type=NodeType.Input, output_names=["image"]) - output = Node( - "output", - node_type=NodeType.Output, - input_names=["vehicle_images", "vehicle_coordinates", "confidence_levels", "colors", "types"], - ) - - NodesConnection.connect(vehicle_detection_node, 0, request, 0) - - NodesConnection.connect(extract_node, 0, request, 0) - NodesConnection.connect(extract_node, 1, vehicle_detection_node, 0) - - NodesConnection.connect(vehicle_recognition_node, 0, extract_node, 0) - - NodesConnection.connect(output, 0, extract_node, 0) - NodesConnection.connect(output, 1, extract_node, 1) - NodesConnection.connect(output, 2, extract_node, 2) - - NodesConnection.connect(output, 3, vehicle_recognition_node, 0) - NodesConnection.connect(output, 4, vehicle_recognition_node, 1) - - nodes = [request, vehicle_detection_node, extract_node, vehicle_recognition_node, output] - return nodes - - -class TenDummySerialPipeline(Pipeline): - - def __init__(self, **kwargs): - super().__init__("ten_dummy_serial", **kwargs) - self._initialize() - - def _create_nodes(self, models=None): - dummy = Dummy() - dummy.inputs["b"]["shape"] = [1, 150528] - - node1 = Node("node_1", dummy) - node2 = Node("node_2", dummy) - node3 = Node("node_3", dummy) - node4 = Node("node_4", dummy) - node5 = Node("node_5", dummy) - node6 = Node("node_6", dummy) - node7 = Node("node_7", dummy) - node8 = Node("node_8", dummy) - node9 = Node("node_9", dummy) - node10 = Node("node_10", dummy) - - request = Node("request", node_type=NodeType.Input) - output = Node("output", node_type=NodeType.Output) - - NodesConnection.connect(node1, 0, request, 0) - NodesConnection.connect(node2, 0, node1, 0) - NodesConnection.connect(node3, 0, node2, 0) - NodesConnection.connect(node4, 0, node3, 0) - NodesConnection.connect(node5, 0, node4, 0) - NodesConnection.connect(node6, 0, node5, 0) - NodesConnection.connect(node7, 0, node6, 0) - NodesConnection.connect(node8, 0, node7, 0) - NodesConnection.connect(node9, 0, node8, 0) - NodesConnection.connect(node10, 0, node9, 0) - NodesConnection.connect(output, 0, node10, 0) - - nodes = [request, node1, node2, node3, node4, node5, node6, node7, node8, node9, node10, output] - return nodes - - -class DummyDiffOpsMaxPipeline(Pipeline): - - def __init__(self, **kwargs): - super().__init__("dummy_diff_ops_max_pipeline", **kwargs) - self._initialize() - - def _create_nodes(self, models=None): - request = Node("request", node_type=NodeType.Input, output_names=self.input_names) - dummy = Dummy() - node_diff = Node("node_diff", CustomNodeDifferentOperations(), NodeType.Custom, demultiply_count=4) - node_dummy_1 = Node("node_d1", dummy) - node_max = Node("node_max", CustomNodeChooseMaximum(), NodeType.Custom, gather_from_node="node_diff") - node_max.model.selection_criteria = CustomNodeChooseMaximum.Method.MAXIMUM_MAXIMUM - node_dummy_2 = Node("node_d2", dummy, output_names=self.input_names) - output = Node("output", node_type=NodeType.Output, input_names=self.output_names) - - NodesConnection.connect(node_diff, 0, request, 0) - NodesConnection.connect(node_diff, 1, request, 1) - NodesConnection.connect(node_dummy_1, 0, node_diff, 0) - NodesConnection.connect(node_max, 0, node_dummy_1, 0) - NodesConnection.connect(node_dummy_2, 0, node_max, 0) - NodesConnection.connect(output, 0, node_dummy_2, 0) - nodes = [request, node_dummy_1, node_dummy_2, node_max, node_diff, output] - return nodes - - -class DummyDynamicDemuxPipeline(Pipeline): - - def __init__(self, **kwargs): - super().__init__("dummy_dag_pipeline", **kwargs) - self._initialize() - - def _create_nodes(self, models=None): - self.demultiply_count = -1 - - request = Node("request", node_type=NodeType.Input, output_names=self.input_names) - output = Node("output", node_type=NodeType.Output, input_names=self.output_names) - - dummy = Dummy() - node_dummy = Node("node_d", dummy) - - NodesConnection.connect(node_dummy, 0, request, 0) - NodesConnection.connect(output, 0, node_dummy, 0) - nodes = [request, node_dummy, output] - - return nodes - - -class DifferentDemultiplyValuesPipeline(Pipeline): - - def __init__(self, **kwargs): - super().__init__("dynamic_demultiplex_pipeline", **kwargs) - self.demux_node = Node( - "diff_node", CustomNodeDynamicDemultiplex(), node_type=NodeType.Custom, demultiply_count=-1 - ) - self._initialize() - self.set_expected_output_shape() - - def set_expected_output_shape(self): - self.outputs["output_0"]["shape"] = self.demux_node.model.outputs["dynamic_demultiplex_results"]["shape"] - - def _create_nodes(self, models=None): - request = Node("request", node_type=NodeType.Input, input_names=self.input_names) - dummy_node = Node("dummy_node", Dummy()) - output = Node("output", node_type=NodeType.Output, output_names=self.output_names) - - NodesConnection.connect(self.demux_node, 0, request, 0) - NodesConnection.connect(dummy_node, 0, self.demux_node, 0) - NodesConnection.connect(output, 0, dummy_node, 0) - - nodes = [request, self.demux_node, dummy_node, output] - return nodes - - -class SingleLevelPipeline(Pipeline): - def __init__(self, list_of_models, predict_shape, **kwargs): - super().__init__("single_level_pipeline", **kwargs) - self.predict_shape = predict_shape - self.models = list_of_models - self._initialize() - - def _create_nodes(self, model=None): - names = [f"img_{i}" for i in range(len(self.models))] - request = Node("request", node_type=NodeType.Input, output_names=["img"]) - output = Node("output", node_type=NodeType.Output, input_names=names) - model_nodes = [] - for idx, model in enumerate(self.models): - model_node = Node(f"model_{idx}", model) - NodesConnection.connect(model_node, 0, request, 0) - NodesConnection.connect(output, idx, model_node, 0) - model_nodes.append(model_node) - return [request] + model_nodes + [output] - - def prepare_input_data(self, batch_size=None, random_data=False, input_key=None): - result = {} - for in_name, in_data in self.models[0].inputs.items(): - shape = self.predict_shape.copy() - if batch_size is not None: - shape[0] = batch_size - result[in_name] = np.ones(shape, dtype=in_data["dtype"]) - - return self.map_inputs(result) - - -class MultiLevelPipeline(Pipeline): - def __init__(self, shape_model_list, **kwargs): - super().__init__("multi_level_pipeline", **kwargs) - self._vertical_shape_list = shape_model_list - self._initialize() - - def _create_nodes(self, model=None): - request = Node("request", node_type=NodeType.Input, output_names=["img"]) - - model_nodes = [] - for idx, shape in enumerate(self._vertical_shape_list): - model = Increment4d() - model.name = f"{model.name}_{idx}" - model.update_shapes(shape) - model.set_input_shape_for_ovms(shape) - model_nodes.append(Node(f"model_{idx}", model)) - - output = Node("output", node_type=NodeType.Output, input_names=[model_nodes[-1].name]) - node_list = [request] + model_nodes + [output] - - for i in range(1, len(node_list)): - NodesConnection.connect(node_list[i], 0, node_list[i - 1], 0) - - return node_list - - class MediaPipe(Pipeline): name = "MediaPipe" is_mediapipe = True @@ -1200,10 +708,8 @@ def graph_refresh(self): class SimpleMediaPipe(MediaPipe): - def __init__(self, model=None, demultiply_count=None, **kwargs): + def __init__(self, model, demultiply_count=None, **kwargs): pipeline = SimplePipeline - if model is None: - model = Resnet() super().__init__(model, pipeline, demultiply_count, **kwargs) self.calculators = [OpenVINOModelServerSessionCalculator(model=self), OpenVINOInferenceCalculator(model=self)] self._initialize([model]) @@ -1232,311 +738,3 @@ def _create_nodes(self, models=None): NodesConnection.connect(output, 0, inference_node, 0) return [request, session_node, inference_node, output] - - -class FailedToLoadModelMediaPipe(SimpleMediaPipe): - def __init__(self, model=None, demultiply_count=None, **kwargs): - super().__init__(model, demultiply_count, **kwargs) - self.regular_models = self.get_regular_models() - side_feed_calculator = OpenVINOInferenceCalculator() - api_session_calculator = OpenVINOModelServerSessionCalculator( - model=self, session=MediaPipeCalculator.get_valid_model_name(self.regular_models[0]), model_name=self.name - ) - self.calculators = [side_feed_calculator, api_session_calculator] - - -class ImageClassificationMediaPipe(MediaPipe): - def __init__(self, **kwargs): - Pipeline.__init__(self, "image_classification_pipeline", **kwargs) - super().__init__() - self._initialize() - self.regular_models = self.get_regular_models() - - def _create_nodes(self, models=None): - session_calculator = OpenVINOModelServerSessionCalculator() - inference_calculator = OpenVINOInferenceCalculator() - - googlenet_session_node = MediaPipeGraphNode( - "googlenet_session_node", GoogleNetV2Fp32(), calculator=session_calculator - ) - resnet_session_node = MediaPipeGraphNode("resnet_session_node", Resnet(), calculator=session_calculator) - argmax_session_node = MediaPipeGraphNode("argmax_session_node", ArgMax(), calculator=session_calculator) - - googlenet_inference_node = MediaPipeGraphNode( - "googlenet_inference_node", - GoogleNetV2Fp32(), - calculator=inference_calculator, - input_stream="GOOGLE_INPUT:input_0", - output_stream="GOOGLE_OUTPUT:google_output", - ) - resnet_inference_node = MediaPipeGraphNode( - "resnet_inference_node", - Resnet(), - calculator=inference_calculator, - input_stream="RESNET_INPUT:input_0", - output_stream="RESNET_OUTPUT:resnet_output", - ) - argmax_inference_node = MediaPipeGraphNode( - "argmax_inference_node", - ArgMax(), - calculator=inference_calculator, - input_stream=["ARGMAX_INPUT1:google_output", "ARGMAX_INPUT2:resnet_output"], - output_stream="ARGMAX_OUTPUT:argmax_0", - ) - - request = MediaPipeGraphNode("request", node_type=NodeType.Input, output_names=["input_0"]) - output = MediaPipeGraphNode("output", node_type=NodeType.Output, input_names=["argmax_0"]) - - NodesConnection.connect(googlenet_session_node, 0, googlenet_inference_node, 0) - NodesConnection.connect(resnet_session_node, 0, resnet_inference_node, 0) - NodesConnection.connect(argmax_session_node, 0, argmax_inference_node, 0) - - NodesConnection.connect(googlenet_inference_node, 0, request, 0) - NodesConnection.connect(resnet_inference_node, 0, request, 0) - NodesConnection.connect(argmax_inference_node, 0, googlenet_inference_node, 0) - NodesConnection.connect(argmax_inference_node, 1, resnet_inference_node, 0) - NodesConnection.connect(output, 0, argmax_inference_node, 0) - - nodes = [ - request, - googlenet_session_node, - resnet_session_node, - argmax_session_node, - googlenet_inference_node, - resnet_inference_node, - argmax_inference_node, - output, - ] - return nodes - - -class SimpleModelMediaPipe(MediaPipe): - def __init__(self, model=None, use_mapping=False, batch_size=None, single_mediapipe_model_mode=False, - pbtxt_name=None): - model = Resnet(batch_size=batch_size) if model is None else model - self.__dict__.update(model.__dict__) - super().__init__(model) - self.calculators = [OpenVINOInferenceCalculator(), OpenVINOModelServerSessionCalculator()] - self.regular_models = model.get_regular_models() - assert not self.name.endswith("_mediapipe") - self.name += "_mediapipe" - self.pbtxt_name = pbtxt_name - self.single_mediapipe_model_mode = single_mediapipe_model_mode - if self.single_mediapipe_model_mode: - self.base_path = os.path.join(Paths.MODELS_PATH_INTERNAL, self.name) - - def replace_input_data_names(self, input_data, input_key=None): - new_data = {} - for i, key in enumerate(list(input_data.keys()), start=0): - new_input_key = input_key if input_key is not None else f"in_{i}" - new_data.update({new_input_key: input_data[key]}) - return new_data - - def prepare_input_data(self, batch_size=None, input_key=None): - data = super().prepare_model_input_data(batch_size) - new_data = self.replace_input_data_names(data, input_key) - return new_data - - def prepare_input_data_from_model_datasets(self, batch_size=None, input_key=None): - result = ModelInfo.prepare_input_data_from_model_datasets(self, batch_size) - new_data = self.replace_input_data_names(result, input_key) - return new_data - - def get_expected_model_output_data(self): - expected_model_output_data = {} - for i, key in enumerate(list(self.outputs.keys()), start=0): - new_output_key = f"out_{i}" - expected_model_output_data.update({new_output_key: self.outputs[key]}) - return expected_model_output_data - - def validate_outputs(self, outputs, expected_output_shapes=None, provided_input=None): - assert outputs, "Prediction returned no output" - if expected_output_shapes is None: - expected_output_shapes = list(self.output_shapes.values()) - for i, shape in enumerate(expected_output_shapes): # Check for dynamic shape - for j, val in enumerate(shape): - if val == -1: - expected_output_shapes[i][j] = 1 - - expected_outputs = self.get_expected_model_output_data() - for output_name in expected_outputs: - assert ( - output_name in outputs - ), f"Incorrect output name, expected: {output_name}, found: {', '.join(outputs.keys())}" - output_shapes = [list(o.shape) for o in outputs.values()] - assert any( - shape in expected_output_shapes for shape in output_shapes - ), f"Incorrect output shape, expected: {expected_output_shapes}, found: {output_shapes}." - - @staticmethod - def is_pipeline(): - return False - - def get_models(self): - return [self] - - def prepare_resources(self, base_location): - return super().prepare_model_resources(base_location) - - def get_demultiply_count(self): - return None - - -class SimpleDynamicModelMediaPipe(SimpleModelMediaPipe): - def __init__(self, model, **kwargs): - super().__init__(model=model, **kwargs) - self._initialize(models=[model]) - - def _create_nodes(self, models=None): - nodes = [] - if models is not None: - assert len(models) == 1, f"Currently only single model in {self.__class__.__name__} is supported." - model = models[0] - session_calculator = OpenVINOModelServerSessionCalculator() - inference_calculator = OpenVINOInferenceCalculator() - valid_model_name = MediaPipeCalculator.get_upper_model_name(model) - dummy_node = MediaPipeGraphNode(model.name, model, calculator=session_calculator) - dummy_inference_node = MediaPipeGraphNode( - f"{model.name}_inference_node", - model, - calculator=inference_calculator, - input_stream=f"{valid_model_name}:in_0", - output_stream=f"{valid_model_name}:out_0", - ) - - request = MediaPipeGraphNode("request", node_type=NodeType.Input, output_names=[f"{valid_model_name}:in_0"]) - output = MediaPipeGraphNode("output", node_type=NodeType.Output, input_names=[f"{valid_model_name}:out_0"]) - - NodesConnection.connect(dummy_node, 0, dummy_inference_node, 0) - - nodes = [request, dummy_node, dummy_inference_node, output] - return nodes - - -class SimpleModelMediaPipeResnetWrongInputShapes(SimpleModelMediaPipe): - def __init__(self, model=None, use_mapping=False, batch_size=None): - model = ResnetWrongInputShapes() - super().__init__(model, use_mapping, batch_size) - - -class SimpleModelMediaPipeResnetWrongInputShapeDim(SimpleModelMediaPipe): - def __init__(self, model=None, use_mapping=False, batch_size=None): - model = ResnetWrongInputShapeDim() - super().__init__(model, use_mapping, batch_size) - - -class CorruptedFileModelMediaPipe(SimpleModelMediaPipe): - def __init__(self, model=None): - super().__init__(model) - self.calculators = [CorruptedFileCalculator()] - - -class SameModelsMediaPipe(MediaPipe): - def __init__(self, **kwargs): - Pipeline.__init__(self, "same_models_mediapipe", **kwargs) - super().__init__() - self._initialize() - self.regular_models = self.get_regular_models() - - def _create_nodes(self, models=None): - model = Dummy() - dummy_session_name = f"{model.name}_session" - session1_calculator = OpenVINOModelServerSessionCalculator(session=dummy_session_name) - inference1_calculator = OpenVINOInferenceCalculator(session=dummy_session_name) - inference2_calculator = OpenVINOInferenceCalculator(session=dummy_session_name) - - dummy_session_node = MediaPipeGraphNode("dummy_session_node", model, calculator=session1_calculator) - - dummy1_inference_node = MediaPipeGraphNode( - "dummy1_inference_node", - model, - calculator=inference1_calculator, - input_stream="DUMMY1_INPUT:input", - output_stream="DUMMY1_OUTPUT:dummy1_output", - ) - dummy2_inference_node = MediaPipeGraphNode( - "dummy2_inference_node", - model, - calculator=inference2_calculator, - input_stream="DUMMY2_INPUT:dummy1_output", - output_stream="DUMMY2_OUTPUT:output", - ) - - request = MediaPipeGraphNode("request", node_type=NodeType.Input, output_names=["input"]) - output = MediaPipeGraphNode("output", node_type=NodeType.Output, input_names=["output"]) - - NodesConnection.connect(dummy_session_node, 0, dummy1_inference_node, 0) - NodesConnection.connect(dummy_session_node, 0, dummy2_inference_node, 0) - - NodesConnection.connect(dummy1_inference_node, 0, request, 0) - NodesConnection.connect(dummy2_inference_node, 0, dummy1_inference_node, 0) - NodesConnection.connect(output, 0, dummy2_inference_node, 0) - - nodes = [request, dummy_session_node, dummy1_inference_node, dummy2_inference_node, output] - return nodes - - -class ModelsChainMediaPipe(MediaPipe): - def __init__(self, models=None, demultiply_count=None, **kwargs): - Pipeline.__init__(self, "models_chain_mediapipe", **kwargs) - super().__init__() - if models is None: - models = [Dummy()] - self.chain_length = len(models) - self.demultiply_count = demultiply_count - self._initialize(models) - self.regular_models = self.get_regular_models() - - def _create_nodes(self, models=None): - model = models[0] - inference_nodes = [] - - model_name = MediaPipeCalculator.get_valid_model_name(model) - final_input_name = "input" - final_output_name = "output" - session_calculator = OpenVINOModelServerSessionCalculator(session=model_name) - session_node = MediaPipeGraphNode(f"{model_name}_session_node", model, calculator=session_calculator) - request = MediaPipeGraphNode("request", node_type=NodeType.Input, output_names=[final_input_name]) - output = MediaPipeGraphNode("output", node_type=NodeType.Output, input_names=[final_output_name]) - inference_calculators = [OpenVINOInferenceCalculator(session=model_name) for i in range(self.chain_length)] - - for i, inf_calc in enumerate(inference_calculators): - output_stream = f"{model_name}_{i}_output" - inf_node_name = f"{model_name}_{i}_inference_node" - if i == 0: - inf_node = MediaPipeGraphNode( - inf_node_name, - model, - calculator=inf_calc, - input_stream=final_input_name, - output_stream=output_stream, - ) - elif i == (len(inference_calculators) - 1): - inf_node = MediaPipeGraphNode( - inf_node_name, - model, - calculator=inf_calc, - input_stream=inference_nodes[i - 1].output_stream, - output_stream=final_output_name, - ) - elif 0 < i < len(inference_calculators): - inf_node = MediaPipeGraphNode( - inf_node_name, - model, - calculator=inf_calc, - input_stream=inference_nodes[i - 1].output_stream, - output_stream=output_stream, - ) - inference_nodes.append(inf_node) - - for i, inf_node in enumerate(inference_nodes): - NodesConnection.connect(session_node, 0, inf_node, 0) - if i == 0: - NodesConnection.connect(inf_node, 0, request, 0) - elif i == (len(inference_nodes) - 1): - NodesConnection.connect(output, 0, inference_nodes[i - 1], 0) - elif 0 < i < len(inference_nodes): - NodesConnection.connect(inf_node, 0, inference_nodes[i - 1], 0) - - nodes = [request, output, session_node] + inference_nodes - return nodes diff --git a/tests/functional/models/models.py b/tests/functional/models/models.py index 8f94e1f006..6ea12654ea 100644 --- a/tests/functional/models/models.py +++ b/tests/functional/models/models.py @@ -13,11 +13,11 @@ # See the License for the specific language governing permissions and # limitations under the License. # - # pylint: disable=too-many-instance-attributes # pylint: disable=too-many-public-methods # pylint: disable=unused-argument +import copy import json import math import os @@ -371,7 +371,8 @@ def clone(self, clone_model_name=None, model_path_on_host=None): def create_new_version(self, container_folder, new_version, copy_from_host_path=False, model_name=None): model_name = model_name if model_name is not None else self.name - result = type(self)() + # Use deepcopy instead of type(self)() to avoid calling __init__ which may require arguments + result = copy.deepcopy(self) if copy_from_host_path: source = self.model_path_on_host diff --git a/tests/functional/object_model/custom_node.py b/tests/functional/object_model/custom_node.py index ccc8100866..52022c296b 100644 --- a/tests/functional/object_model/custom_node.py +++ b/tests/functional/object_model/custom_node.py @@ -268,40 +268,6 @@ def get_parameters(self): "debug": "true", } -@dataclass -class CustomNodeDemultiply(OvmsTestDevCustomNode): - ORIGINAL_DEMULTIPLY_COUNT = 3 - - def __init__(self, demultiply_size=None, **kwargs): - super().__init__( - name="demultiply", - inputs={"tensor": {"shape": [1, 3, 224, 224], "dtype": np.float32}}, - outputs={"tensor_out": {"shape": [demultiply_size, 1, 3, 224, 224], "dtype": np.float32}}, - **kwargs, - ) - - self.demultiply_size = demultiply_size - - def get_parameters(self): - return {"demultiply_size": str(self.demultiply_size)} - - -@dataclass -class CustomNodeElastic1T(OvmsTestDevCustomNode): - - def __init__(self, input_shape=None, output_shape=None, **kwargs): - super().__init__( - name="elastic_in_1t_out_1t", - inputs={"tensor_in": {"shape": input_shape, "dtype": np.float32}}, - outputs={"tensor_out": {"shape": output_shape, "dtype": np.float32}}, - **kwargs, - ) - self.input_shape = input_shape - self.output_shape = output_shape - - def get_parameters(self): - return {"input_shape": str(self.input_shape), "output_shape": str(self.output_shape)} - @dataclass class CustomNodeDemultiplyGather(OvmsTestDevCustomNode):