diff --git a/tests/functional/constants/metrics.py b/tests/functional/constants/metrics.py index 401d8191f1..82a51b83b9 100644 --- a/tests/functional/constants/metrics.py +++ b/tests/functional/constants/metrics.py @@ -69,13 +69,9 @@ class Metric: Method_infer = "ModelInfer" - Method_getmodelstatus = "GetModelStatus" - Method_getmodelmetadata = "GetModelMetadata" - Method_predict = "Predict" Method_modelready = "ModelReady" Method_modelmetadata = "ModelMetadata" - TensorFlowServing = "TensorFlowServing" KServe = "KServe" Type_counter = "counter" @@ -118,12 +114,9 @@ class Metric: DescType = {"requests_success", "counter", "request_fail", "counter"} - Protocol = [KServe, TensorFlowServing] + Protocol = [KServe] Methods = [ - Method_getmodelstatus, - Method_getmodelmetadata, - Method_predict, Method_infer, Method_modelready, Method_modelmetadata, @@ -133,9 +126,6 @@ class Metric: Method_modelready: KServe, Method_modelmetadata: KServe, Method_infer: KServe, - Method_getmodelstatus: TensorFlowServing, - Method_getmodelmetadata: TensorFlowServing, - Method_predict: TensorFlowServing, } Histogram_bucket_len_list = [ @@ -194,7 +184,7 @@ def create_method_metrics(model, base_name): "name": model.name, } - if method not in [Metric.Method_getmodelstatus, Metric.Method_modelready]: + if method not in [Metric.Method_modelready]: content["version"] = str(model.version) result.append(Metric(metric_name=base_name, content=content)) @@ -384,19 +374,15 @@ def create_from_model_list(model_list, ovms_run=None, metrics=None): """ The following metrics are not multiplied for each model version (should occur once for single model name) - ovms_requests_success[{'api': 'TensorFlowServing', 'interface': 'gRPC', 'method': 'GetModelStatus', 'name': 'resnet-50-tf'}] 0 - ovms_requests_success[{'api': 'TensorFlowServing', 'interface': 'REST', 'method': 'GetModelStatus', 'name': 'resnet-50-tf'}] 0 ovms_requests_success[{'api': 'KServe', 'interface': 'gRPC', 'method': 'ModelReady', 'name': 'resnet-50-tf'}] 0 ovms_requests_success[{'api': 'KServe', 'interface': 'REST', 'method': 'ModelReady', 'name': 'resnet-50-tf'}] 0 - ovms_requests_fail[{'api': 'TensorFlowServing', 'interface': 'gRPC', 'method': 'GetModelStatus', 'name': 'resnet-50-tf'}] 0 - ovms_requests_fail[{'api': 'TensorFlowServing', 'interface': 'REST', 'method': 'GetModelStatus', 'name': 'resnet-50-tf'}] 0 ovms_requests_fail[{'api': 'KServe', 'interface': 'gRPC', 'method': 'ModelReady', 'name': 'resnet-50-tf'}] 0 ovms_requests_fail[{'api': 'KServe', 'interface': 'REST', 'method': 'ModelReady', 'name': 'resnet-50-tf'}] 0 """ metrics_to_remove = [] model_unique_metrics = [] for metric in metric_list: - if metric.content.get("method", None) in [Metric.Method_getmodelstatus, Metric.Method_modelready]: + if metric.content.get("method", None) in [Metric.Method_modelready]: if metric.to_str() in model_unique_metrics: metrics_to_remove.append(metric) else: @@ -541,40 +527,12 @@ def __init__(self): # protocol="kserve", # version="1"} 0 # ovms_requests_success{ -# interface="rest", -# method="getmodelstatus", -# name="ssdlite_mobilenet_v2_ov", -# protocol="tensorflowserving"} 0 -# ovms_requests_success{ -# interface="rest", -# method="getmodelmetadata", -# name="ssdlite_mobilenet_v2_ov", -# protocol="tensorflowserving", -# version="1"} 0 -# ovms_requests_success{ -# interface="rest", -# method="predict", -# name="ssdlite_mobilenet_v2_ov", -# protocol="tensorflowserving", -# version="1"} 0 -# ovms_requests_success{ # interface="grpc", # method="modelinfer", # name="ssdlite_mobilenet_v2_ov", # protocol="kserve", # version="1"} 0 # ovms_requests_success{ -# interface="grpc", -# method="getmodelstatus", -# name="ssdlite_mobilenet_v2_ov", -# protocol="tensorflowserving"} 0 -# ovms_requests_success{ -# interface="grpc", -# method="getmodelmetadata", -# name="ssdlite_mobilenet_v2_ov", -# protocol="tensorflowserving", -# version="1"} 0 -# ovms_requests_success{ # interface="rest", # method="modelready", # name="ssdlite_mobilenet_v2_ov", @@ -585,12 +543,6 @@ def __init__(self): # name="ssdlite_mobilenet_v2_ov", # protocol="kserve", # version="1"} 0 -# ovms_requests_success{ -# interface="grpc", -# method="predict", -# name="ssdlite_mobilenet_v2_ov", -# protocol="tensorflowserving", -# version="1"} 0 # # HELP ovms_requests_fail Number of failed requests to a model or a DAG. # # TYPE ovms_requests_fail counter # ovms_requests_fail{ @@ -618,53 +570,17 @@ def __init__(self): # protocol="kserve", # version="1"} 0 # ovms_requests_fail{ -# interface="rest", -# method="getmodelstatus", -# name="ssdlite_mobilenet_v2_ov", -# protocol="tensorflowserving", -# version="1"} 0 -# ovms_requests_fail{ -# interface="rest", -# method="getmodelmetadata", -# name="ssdlite_mobilenet_v2_ov", -# protocol="tensorflowserving", -# version="1"} 0 -# ovms_requests_fail{ -# interface="rest", -# method="predict", -# name="ssdlite_mobilenet_v2_ov", -# protocol="tensorflowserving", -# version="1"} 0 -# ovms_requests_fail{ # interface="grpc", # method="modelmetadata", # name="ssdlite_mobilenet_v2_ov", # protocol="kserve", # version="1"} 0 # ovms_requests_fail{ -# interface="grpc", -# method="getmodelstatus", -# name="ssdlite_mobilenet_v2_ov", -# protocol="tensorflowserving", -# version="1"} 0 -# ovms_requests_fail{ -# interface="grpc", -# method="getmodelmetadata", -# name="ssdlite_mobilenet_v2_ov", -# protocol="tensorflowserving", -# version="1"} 0 -# ovms_requests_fail{ # interface="rest", # method="modelmetadata", # name="ssdlite_mobilenet_v2_ov", # protocol="kserve", # version="1"} 0 -# ovms_requests_fail{ -# interface="grpc", -# method="predict", -# name="ssdlite_mobilenet_v2_ov", -# protocol="tensorflowserving", -# version="1"} 0 # # HELP ovms_streams Number of OpenVINO execution streams. # # TYPE ovms_streams gauge # ovms_streams{name="ssdlite_mobilenet_v2_ov",version="1"} 4 diff --git a/tests/functional/constants/ovms.py b/tests/functional/constants/ovms.py index 9069f41318..e19e6131d8 100644 --- a/tests/functional/constants/ovms.py +++ b/tests/functional/constants/ovms.py @@ -18,7 +18,6 @@ import re from enum import Enum -from tensorflow_serving.apis.get_model_status_pb2 import ModelVersionStatus from tests.functional.constants.os_type import OsType @@ -137,12 +136,12 @@ class Ovms: class ModelStatus(Enum): UNDEFINED = None - UNKNOWN = ModelVersionStatus.UNKNOWN - START = ModelVersionStatus.START - LOADING = ModelVersionStatus.LOADING - AVAILABLE = ModelVersionStatus.AVAILABLE - UNLOADING = ModelVersionStatus.UNLOADING - END = ModelVersionStatus.END + UNKNOWN = 0 + START = 10 + LOADING = 20 + AVAILABLE = 30 + UNLOADING = 40 + END = 50 LAYOUT_NHWC = "NHWC:NCHW" LAYOUT_NCHW = "NCHW:NCHW" diff --git a/tests/functional/fixtures/api_type.py b/tests/functional/fixtures/api_type.py index b1d5b27da8..8eac952f9e 100644 --- a/tests/functional/fixtures/api_type.py +++ b/tests/functional/fixtures/api_type.py @@ -21,7 +21,7 @@ from tests.functional.constants.ovms_type import OvmsType from tests.functional.utils.inference.communication import GRPC, REST from tests.functional.utils.inference.inference_client_factory import InferenceClientFactory -from tests.functional.utils.inference.serving import KFS, OPENAI, TFS, TRITON, COHERE +from tests.functional.utils.inference.serving import KFS, OPENAI, TRITON, COHERE def api_type_non_fixture(serving, communication, ovms_type=None): @@ -41,18 +41,13 @@ def api_type(request): return api_type_non_fixture(*request.param, ovms_type=None) -@pytest.fixture(scope="session", params=itertools.product([TFS], [GRPC, REST]), ids=lambda x: f":".join(x).upper()) -def tfs_api_type(request): - return api_type_non_fixture(*request.param) - - -@pytest.fixture(scope="session", params=[(TFS, REST)], ids=lambda x: f":".join(x).upper()) -def tfs_rest_api_type(request): +@pytest.fixture(scope="session", params=itertools.product([KFS], [REST]), ids=lambda x: f":".join(x).upper()) +def rest_api_type(request): return api_type_non_fixture(*request.param) -@pytest.fixture(scope="session", params=[(TFS, GRPC)], ids=lambda x: f":".join(x).upper()) -def tfs_grpc_api_type(request): +@pytest.fixture(scope="session", params=itertools.product([KFS], [GRPC]), ids=lambda x: f":".join(x).upper()) +def grpc_api_type(request): return api_type_non_fixture(*request.param) @@ -94,13 +89,3 @@ def triton_grpc_api_type(request): @pytest.fixture(scope="session", params=[(TRITON, REST)], ids=lambda x: f":".join(x).upper()) def triton_rest_api_type(request): return api_type_non_fixture(*request.param) - - -@pytest.fixture(scope="session", params=itertools.product([KFS, TFS], [REST]), ids=lambda x: f":".join(x).upper()) -def rest_api_type(request): - return api_type_non_fixture(*request.param) - - -@pytest.fixture(scope="session", params=itertools.product([KFS, TFS], [GRPC]), ids=lambda x: f":".join(x).upper()) -def grpc_api_type(request): - return api_type_non_fixture(*request.param) diff --git a/tests/functional/object_model/inference_helpers.py b/tests/functional/object_model/inference_helpers.py index da04acc62a..b368a24ecf 100644 --- a/tests/functional/object_model/inference_helpers.py +++ b/tests/functional/object_model/inference_helpers.py @@ -40,9 +40,6 @@ from openai import OpenAI from pydantic import BaseModel from retry.api import retry_call -from tensorflow import make_tensor_proto -from tensorflow_serving.apis import get_model_status_pb2 -from tensorflow_serving.apis.predict_pb2 import PredictRequest from tritonclient.grpc import service_pb2, service_pb2_grpc from tritonclient.grpc.service_pb2 import ModelInferRequest from tritonclient.utils import InferenceServerException, deserialize_bytes_tensor, serialize_byte_tensor @@ -62,7 +59,6 @@ AudioApi, ResponsesApi, ) -from tests.functional.utils.inference.serving.tf import TensorFlowServingWrapper from tests.functional.utils.logger import get_logger from tests.functional.utils.test_framework import FrameworkMessages, skip_if_runtime from tests.functional.utils.generative_ai.validation_utils import GenerativeAIValidationUtils @@ -223,14 +219,6 @@ def prepare_request_to_send(self, client, input_data): request = {"request": request} elif isinstance(client, KserveWrapper) and isinstance(client, RestCommunicationInterface): request = self._create_kfs_post_request(input_data) - elif isinstance(client, TensorFlowServingWrapper) and isinstance(client, RestCommunicationInterface): - request = self._create_post_request(self.model.input_names, input_data, request_format=self.layout) - elif isinstance(client, TensorFlowServingWrapper) and isinstance(client, GrpcCommunicationInterface): - request = PredictRequest() - request.model_spec.name = self.model.name - for input_name, input_object in input_data.items(): - request.inputs[input_name].CopyFrom(make_tensor_proto(input_object, shape=[len(input_object)])) - request = {"request": request} else: raise NotImplementedError return request @@ -824,78 +812,22 @@ def prepare_and_run_set_of_predict_requests(ovms: OvmsInstance, models, api_type def get_model_status(client, accepted_model_states=None, model_version=None, port=None): model_state = None port = port if port is not None else client.port - if client.serving == KFS: - if accepted_model_states is not None: - for elem in accepted_model_states: - is_ready = True if elem == Ovms.ModelStatus.AVAILABLE else False - try: - model_state = check_model_readiness(client.model, port, type(client), timeout=30, is_ready=is_ready) - except ModelNotReadyException: - logger.info(f"Model state not in accepted state: {elem}") - finally: - break - else: - raise ModelNotReadyException(f"Failed to check model: {client.model}") + if accepted_model_states is not None: + for elem in accepted_model_states: + is_ready = True if elem == Ovms.ModelStatus.AVAILABLE else False + try: + model_state = check_model_readiness(client.model, port, type(client), timeout=30, is_ready=is_ready) + except ModelNotReadyException: + logger.info(f"Model state not in accepted state: {elem}") + finally: + break else: - model_state = check_model_readiness(client.model, port, type(client)) + raise ModelNotReadyException(f"Failed to check model: {client.model}") else: - status = client.get_model_status() - logger.debug(f"status: {status}") - if model_version is None: - model_state = Ovms.ModelStatus(status.model_version_status[0].state) - else: - for model_version_status in status.model_version_status: - if model_version_status.version == model_version: - model_state = Ovms.ModelStatus(model_version_status.state) - break - if accepted_model_states: - if model_state not in accepted_model_states: - model_str_name = client.model_name - raise ValueError(f"Incorrect state of {model_str_name}: {model_state}") + model_state = check_model_readiness(client.model, port, type(client)) return model_state -def get_and_validate_model_status(inference, expected_models_status): - status = inference.get_model_status() - if expected_models_status is not None: - assert len(expected_models_status) == len(status.model_version_status) - - for i, model_version_status in enumerate(status.model_version_status): - model_state = model_version_status.state - error_message = model_version_status.status.error_message - version = model_version_status.version - - if expected_models_status is None or expected_models_status[i].get("accepted_states", None) is None: - model_accepted_states = [ - get_model_status_pb2.ModelVersionStatus.START, - get_model_status_pb2.ModelVersionStatus.AVAILABLE, - get_model_status_pb2.ModelVersionStatus.UNLOADING, - get_model_status_pb2.ModelVersionStatus.LOADING, - get_model_status_pb2.ModelVersionStatus.END, - ] - else: - model_accepted_states = expected_models_status[i]["accepted_states"] - - if model_state not in model_accepted_states: - raise ValueError(f"Incorrect model state: {model_state}") - - if expected_models_status is None or expected_models_status[i].get("accepted_error_messages", None) is None: - if model_state == get_model_status_pb2.ModelVersionStatus.LOADING: - model_accepted_error_messages = ["OK", "UNKNOWN"] - else: - model_accepted_error_messages = ["OK"] - else: - model_accepted_error_messages = expected_models_status[i]["accepted_error_messages"] - - if error_message not in model_accepted_error_messages: - raise ValueError(f"Incorrect error message: {model_state}") - - if expected_models_status is not None: - assert version == expected_models_status[i]["version"] - - return status - - def get_multiple_model_status(models_and_expected_state): for client, state in models_and_expected_state: try: diff --git a/tests/functional/utils/inference/inference_client_factory.py b/tests/functional/utils/inference/inference_client_factory.py index 89b84d5449..f6111951c2 100644 --- a/tests/functional/utils/inference/inference_client_factory.py +++ b/tests/functional/utils/inference/inference_client_factory.py @@ -22,7 +22,6 @@ from tests.functional.utils.inference.serving.cohere import COHERE, CohereWrapper from tests.functional.utils.inference.serving.kf import KFS, KserveWrapper from tests.functional.utils.inference.serving.openai import OPENAI, OpenAIWrapper -from tests.functional.utils.inference.serving.tf import TFS, TensorFlowServingWrapper from tests.functional.utils.inference.serving.triton import TRITON, TritonServingWrapper from tests.functional.constants.ovms_type import OvmsType from tests.functional.object_model.ovsa import OvsaCerts @@ -35,9 +34,7 @@ def get_client(serving, communication, ovms_type=None): if ovms_type == OvmsType.CAPI: communication_class = CapiServingWrapper else: - if serving == TFS: - serving_class = TensorFlowServingWrapper - elif serving == KFS: + if serving == KFS: serving_class = KserveWrapper elif serving == TRITON: serving_class = TritonServingWrapper @@ -46,15 +43,15 @@ def get_client(serving, communication, ovms_type=None): elif serving == COHERE: serving_class = CohereWrapper else: - raise Exception + raise NotImplementedError(f"Serving not supported: {serving}") - if serving in [TFS, KFS, TRITON, OPENAI, COHERE]: + if serving in [KFS, TRITON, OPENAI, COHERE]: if communication == REST: communication_class = RestCommunicationInterface elif communication == GRPC: communication_class = GrpcCommunicationInterface else: - raise Exception + raise NotImplementedError(f"Communication interface not supported: {communication}") # pylint: disable=too-many-arguments def common_inference_client_init(self, diff --git a/tests/functional/utils/inference/serving/tf.py b/tests/functional/utils/inference/serving/tf.py index e36387336a..0a9952d287 100644 --- a/tests/functional/utils/inference/serving/tf.py +++ b/tests/functional/utils/inference/serving/tf.py @@ -14,414 +14,8 @@ # limitations under the License. # -import json - -import numpy as np -import tensorflow -from google.protobuf.json_format import MessageToJson, Parse -from tensorboard.util.tensor_util import make_ndarray -from tensorflow import make_tensor_proto -from tensorflow.core.framework import types_pb2 -from tensorflow_serving.apis import ( - get_model_metadata_pb2, get_model_status_pb2, model_service_pb2_grpc, predict_pb2, prediction_service_pb2_grpc,) - -from tests.functional.utils.assertions import InvalidMetadataException, NotSupported -from tests.functional.utils.http.base import HttpMethod -from tests.functional.utils.inference.communication.grpc import GRPC_TIMEOUT -from tests.functional.utils.inference.serving.base import AbstractServingWrapper from tests.functional.utils.logger import get_logger -from tests.functional.constants.metrics import Metric -from tests.functional.constants.ovms import Ovms logger = get_logger(__name__) TFS = "TFS" - - -class TensorFlowServingWrapper(AbstractServingWrapper): - REST_VERSION = "v1" - PREDICT = ":predict" - - METRICS_PROTOCOL = Metric.TensorFlowServing - - - def set_grpc_stubs(self): - """ - Assigns objects for inference purposes. - """ - self.predict_stub = prediction_service_pb2_grpc.PredictionServiceStub(self.channel) - self.model_service_stub = model_service_pb2_grpc.ModelServiceStub(self.channel) - - def get_model_status_grpc_request(self, model_name=None, version=None): - request = get_model_status_pb2.GetModelStatusRequest() - request.model_spec.name = self.model_name if model_name is None else model_name - if version is not None: - request.model_spec.version.value = int(version) - - if self.model_version is not None: - request.model_spec.version.value = int(self.model_version) - return request - - def get_model_meta_grpc_request(self, model_name=None): - metadata_field = "signature_def" - request = get_model_metadata_pb2.GetModelMetadataRequest() - request.model_spec.name = self.model_name - if self.model_version is not None: - request.model_spec.version.value = int(self.model_version) - request.metadata_field.append(metadata_field) - return request - - def send_model_status_grpc_request(self, request): - response = self.model_service_stub.GetModelStatus( - request, wait_for_ready=True, timeout=GRPC_TIMEOUT - ) - return response - - def send_model_meta_grpc_request(self, request): - response = self.predict_stub.GetModelMetadata( - request, wait_for_ready=True, timeout=GRPC_TIMEOUT - ) - return response - - def get_predict_grpc_request(self, input_objects, raw=False, mediapipe_name=None): - request = predict_pb2.PredictRequest() - request.model_spec.name = self.model_name - if self.model_version is not None: - request.model_spec.version.value = int(self.model_version) - for input_name, input_object in input_objects.items(): - request.inputs[input_name].CopyFrom( - make_tensor_proto(input_object, shape=input_object.shape) - ) - return request - - @staticmethod - def process_predict_grpc_output(result, **kwargs): - outputs = { - output_name: make_ndarray(output) for output_name, output in result.outputs.items() - } - return outputs - - def create_inference(self): - """ - Assigns objects for inference purposes. - """ - # method from brother class (multiple inheritance) - self.communication_service = self.create_communication_service() - return self.communication_service - - def send_predict_grpc_request(self, request, timeout=GRPC_TIMEOUT): - return self.predict_stub.Predict(request, wait_for_ready=True, timeout=timeout) - - def predict(self, request, timeout=60, raw=False): - result = self.send_predict_request(request, timeout) - outputs = self.process_predict_output(result) - return outputs - - def get_rest_path(self, operation=None, model_version=None, model_name=None): - """ - Expect 2 REST path formats for TF format: - - GET: (METADATA, MODELS) - http://{REST_URL}:{REST_PORT}/v1/models/{MODEL_NAME}/versions/{MODEL_VERSION}/{OPERATION} - - POST: (PREDICT) - http://{REST_URL}:{REST_PORT}/v1/models/{MODEL_NAME}/versions/{MODEL_VERSION}:predict - """ - model_name = model_name if model_name is not None else (self.model.name if self.model else self.model_name) - assert model_name - rest_path = [self.REST_VERSION, self.MODELS, model_name] - if model_version is not None: - rest_path.append(self.VERSIONS) - rest_path.append(str(model_version)) - if operation == self.PREDICT: - rest_path[-1] = "".join([rest_path[-1], operation]) - elif operation not in [self.STATUS, None]: - rest_path.append(operation) - rest_path = "/".join(rest_path) - return rest_path - - - @staticmethod - def prepare_body_dict(input_objects: dict, request_format=Ovms.BINARY_IO_LAYOUT_ROW_NAME, **kwargs): - """ - Prepare HTTP request's body in given format: - - row_name, - - column_name, - - row_noname, - - column_noname - """ - signature = "serving_default" - if request_format == Ovms.BINARY_IO_LAYOUT_ROW_NAME: - instances = [] - for input_name, input_object in input_objects.items(): - if input_object.shape: - for i in range(0, input_object.shape[0], 1): - input_data = input_object[i].decode() if ( - input_object.dtype == np.object_) else input_object[i].tolist() - instances.append({input_name: input_data}) - else: - instances.append({input_name: str(input_object[()])}) - # https://numpy.org/doc/stable/reference/arrays.scalars.html#indexing - data_obj = {"signature_name": signature, "instances": instances} - elif request_format == Ovms.BINARY_IO_LAYOUT_ROW_NONAME: - instances = [] - for input_object in input_objects.values(): - if input_object.shape: - for i in range(0, input_object.shape[0]): - input_data = input_object[i].decode() if ( - input_object.dtype == np.object_) else input_object[i].tolist() - instances.append(input_data) - else: - # https://numpy.org/doc/stable/reference/arrays.scalars.html#indexing - instances.append([str(input_object[()])]) - data_obj = {"signature_name": signature, 'instances': instances} - elif request_format == Ovms.BINARY_IO_LAYOUT_COLUMN_NAME: - inputs = {} - for input_name, input_object in input_objects.items(): - inputs[input_name] = [x.decode() for x in input_object.tolist()] if input_object.dtype == np.object_ \ - else input_object.tolist() - data_obj = {"signature_name": signature, 'inputs': inputs} - elif request_format == Ovms.BINARY_IO_LAYOUT_COLUMN_NONAME: - assert len(input_objects) == 1, \ - f"Only single input is required if {Ovms.BINARY_IO_LAYOUT_COLUMN_NONAME} format is used" - input_object = list(input_objects.items())[0][1] - _input = [x.decode() for x in input_object.tolist()] if input_object.dtype == np.object_ \ - else input_object.tolist() - data_obj = {"signature_name": signature, 'inputs': _input} - else: - raise ValueError(f"Unknown response format: {request_format}") - return data_obj - - def get_inputs_outputs_from_response(self, response): - # expect content to be dictionary encoded as bytes: - # model.content == b'{\n "modelSpec": {\n "name": "resnet-50-tf",\n "signatureName": "" ... - model_specification = json.loads(response.text) - - serving_default = model_specification['metadata']['signature_def']['signatureDef']['serving_default'] - - self.model.inputs = {} - self.model.outputs = {} - - for name, details in serving_default['inputs'].items(): - self.model.inputs[details['name']] = { - 'shape': [int(x['size']) for x in details["tensorShape"]["dim"]], - 'dtype': tensorflow.dtypes.as_dtype(getattr(types_pb2, details['dtype'])) - } - - for name, details in serving_default['outputs'].items(): - self.model.outputs[details['name']] = { - 'shape': [int(x['size']) for x in details["tensorShape"]["dim"]], - 'dtype': tensorflow.dtypes.as_dtype(getattr(types_pb2, details['dtype'])) - } - - def process_json_output(self, result_dict): - """ - Converts predict result to output as a numpy array. - Input: - result_dict = {'predictions': []} - Output: - = {ndarray: (1, 1001)} - """ - output = {} - if "outputs" in result_dict: - key_name = "outputs" - if isinstance(result_dict[key_name], dict): - for output_tensor in self.output_names: - output[output_tensor] = np.asarray(result_dict[key_name][output_tensor]) - else: - output[self.output_names[0]] = np.asarray(result_dict[key_name]) - elif "predictions" in result_dict: - key_name = "predictions" - if isinstance(result_dict[key_name][0], dict): - for row in result_dict[key_name]: - for output_tensor in self.output_names: - if output_tensor not in output: - output[output_tensor] = [] - output[output_tensor].append(row[output_tensor]) - for output_tensor in self.output_names: - output[output_tensor] = np.asarray(output[output_tensor]) - else: - output[self.output_names[0]] = np.asarray(result_dict[key_name]) - else: - logger.error(f"Missing required response in {result_dict}") - return output - - def set_serving_inputs_outputs_grpc(self, response, **kwargs): - """ - Sets inference response inputs and outputs. - Parameters: - response (GetModelMetadataResponse): inference response - """ - signature_def = response.metadata['signature_def'] - signature_map = get_model_metadata_pb2.SignatureDefMap() - signature_map.ParseFromString(signature_def.value) - serving_default = signature_map.ListFields()[0][1]['serving_default'] - - inputs = {} - outputs = {} - - for name, details in serving_default.inputs.items(): - inputs[name] = { - 'shape': [x.size for x in details.tensor_shape.dim], - 'dtype': tensorflow.dtypes.as_dtype(details.dtype) - } - - for name, details in serving_default.outputs.items(): - outputs[name] = { - 'shape': [x.size for x in details.tensor_shape.dim], - 'dtype': tensorflow.dtypes.as_dtype(details.dtype) - } - - self.set_inputs(inputs) - self.set_outputs(outputs) - - - @staticmethod - def get_data_type(data_type): - """ - Converts given data_type to numpy format. - Parameters: - data_type (int) - Returns: - result (np) - """ - result = None - if data_type == 6: - result = np.int8 - elif data_type == 3: - result = np.int32 - elif data_type == 9: - result = np.int64 - elif data_type == 1: - result = np.float32 - else: - raise NotImplementedError() - return result - - def get_model_status_rest(self, timeout=60, version=None, model_name=None): - rest_path = self.get_rest_path(None, model_version=version, model_name=model_name) - response = self.client.request(HttpMethod.GET, path=rest_path, timeout=timeout, raw_response=True) - - # Transform JSON friendly output to protobuf compatible object required by callers. - status_pb = get_model_status_pb2.GetModelStatusResponse() - response = Parse(response.text, status_pb, ignore_unknown_fields=False) - - return response - - # KFS not supported API calls: - def is_server_live_grpc(self): - raise NotSupported("is_server_live is not available in TFS") - - def is_server_live_rest(self): - raise NotSupported("is_server_live is not available in TFS") - - def is_server_ready_grpc(self): - raise NotSupported("is_server_live is not available in TFS") - - def is_server_ready_rest(self): - raise NotSupported("is_server_live is not available in TFS") - - def is_model_ready_grpc(self, model_name, model_version=""): - """ - Gets information about model readiness (specific only for KFS - gRPC or REST). - GET http://${REST_URL}:${REST_PORT}/v2/models/${MODEL_NAME}[/versions/${MODEL_VERSION}]/ready - Response: True (ready) or False (not ready) - """ - raise NotSupported() - - def is_model_ready_rest(self, model_name, model_version=""): - """ - Gets information about model readiness (specific only for KFS - gRPC or REST). - GET http://${REST_URL}:${REST_PORT}/v2/models/${MODEL_NAME}[/versions/${MODEL_VERSION}]/ready - Response: True (ready) or False (not ready) - """ - raise NotSupported() - - def cast_type_to_string(self, data_type): - # https://github.com/openvinotoolkit/model_server/blob/main/src/tfs_frontend/tfs_utils.cpp - if data_type == np.float32: - result = 'DT_FLOAT' - elif data_type == np.int32: - result = 'DT_INT32' - elif data_type == np.int64: - result = 'DT_INT64' - elif data_type == str: - result = 'DT_STRING' - elif data_type == np.uint8: - result = 'DT_UINT8' - else: - raise NotImplementedError() - return result - - def validate_meta_grpc(self, model, meta): - """ - Validates model metadata. - Parameters: - model (ModelInfo): model class object - meta (ModelMetadataResponse): model metadata - """ - json_data = json.loads(MessageToJson(meta)) - - assert meta.model_spec.name == model.name, \ - f"Unexpected model name (expected: {model.name}, " \ - f"detected: {meta.model_spec.name})" - assert meta.model_spec.version.value == model.version - assert "signature_def" in meta.metadata - assert meta.metadata['signature_def'].type_url == \ - 'type.googleapis.com/tensorflow.serving.SignatureDefMap' - def validate(test_data, val_shapes, val_types): - assert len(test_data) == len(val_shapes), \ - f"Unexpected argument list (shapes; expect: {len(val_shapes)}, " \ - f"detect: {len(test_data)})" - assert len(test_data) == len(val_types), \ - f"Unexpected argument list (shapes; expect: {len(val_types)}, " \ - f"detect: {len(test_data)})" - for arg_name, arg_data in test_data.items(): - for test, val_dim in zip(arg_data['tensorShape']['dim'], val_shapes[arg_name]): - if int(test['size']) != val_dim: - raise InvalidMetadataException( - f"Unexpected shape (expected: {val_shapes[arg_name]}, " \ - f"detected: {arg_data['tensorShape']['dim']})") - val_type = self.cast_type_to_string(val_types[arg_name]) - assert arg_data['dtype'] == val_type, \ - f"Unexpected type (expected: {val_type}, detected: {arg_data['dtype']}" - - data = json_data['metadata']['signature_def']['signatureDef']['serving_default'] - validate( - test_data=data['inputs'], val_shapes=model.input_shapes, val_types=model.input_types - ) - validate( - test_data=data['outputs'], val_shapes=model.output_shapes, val_types=model.output_types - ) - - def validate_meta_rest(self, model, response): - metadata = json.loads(response.text) - assert model.name == metadata['modelSpec']['name'] - assert model.version == int(metadata['modelSpec']['version']) - - metadata_inputs = metadata['metadata']['signature_def']['signatureDef']['serving_default']['inputs'] - metadata_outputs = metadata['metadata']['signature_def']['signatureDef']['serving_default']['outputs'] - - for name, description in model.inputs.items(): - assert name in metadata_inputs - assert model.inputs[name]['shape'] == [ - int(x['size']) for x in metadata_inputs[name]['tensorShape']['dim'] - ] - assert self.cast_type_to_string(model.inputs[name]['dtype']) == metadata_inputs[name]['dtype'] - - for name, description in model.outputs.items(): - assert name in metadata_outputs - assert model.outputs[name]['shape'] == [ - int(x['size']) for x in metadata_outputs[name]['tensorShape']['dim'] - ] - assert self.cast_type_to_string(model.outputs[name]['dtype']) == metadata_outputs[name]['dtype'] - - def validate_status(self, model, status): - to_check = status.model_version_status[0] - assert model.version == to_check.version, f"Unexpected version (detected: {to_check.version}, expected: "\ - f"{model.version})" - expected_res = get_model_status_pb2.ModelVersionStatus.State.AVAILABLE - state_map = get_model_status_pb2.ModelVersionStatus.State.items() - assert to_check.state == expected_res, f"Unexpected state (detected: {to_check.state}, expected "\ - f"{expected_res} - map: {state_map})" - assert to_check.status.error_message == 'OK', f"Unexpected error msg (detected: "\ - f"{to_check.status.error_message}, expected: OK" -