class_docstr: TFX infra validator executor.
class_docstr_tok: ["TFX","infra","validator","executor","."]
docstr: Contract for running InfraValidator Executor.
  
  Args:
    input_dict:
      - `model`: Single `Model` artifact that we're validating.
      - `examples`: `Examples` artifacts to be used for test requests.
    output_dict:
      - `blessing`: Single `InfraBlessing` artifact containing the validated
        result and optinally validated model if warmup requests are appended.
        Artifact URI includes an empty file with the name either of
        INFRA_BLESSED or INFRA_NOT_BLESSED.
    exec_properties:
      - `serving_spec`: Serialized `ServingSpec` configuration.
      - `validation_spec`: Serialized `ValidationSpec` configuration.
      - `request_spec`: Serialized `RequestSpec` configuration.
code: def Do(
      self,
      input_dict: Dict[str, List[types.Artifact]],
      output_dict: Dict[str, List[types.Artifact]],
      exec_properties: Dict[str, Any],
  ) -> None:
      """Contract for running InfraValidator Executor.
  
      Args:
        input_dict:
          - `model`: Single `Model` artifact that we're validating.
          - `examples`: `Examples` artifacts to be used for test requests.
        output_dict:
          - `blessing`: Single `InfraBlessing` artifact containing the validated
            result and optinally validated model if warmup requests are appended.
            Artifact URI includes an empty file with the name either of
            INFRA_BLESSED or INFRA_NOT_BLESSED.
        exec_properties:
          - `serving_spec`: Serialized `ServingSpec` configuration.
          - `validation_spec`: Serialized `ValidationSpec` configuration.
          - `request_spec`: Serialized `RequestSpec` configuration.
      """
      self._log_startup(input_dict, output_dict, exec_properties)
  
      model = artifact_utils.get_single_instance(
          input_dict[standard_component_specs.MODEL_KEY]
      )
      blessing = artifact_utils.get_single_instance(
          output_dict[standard_component_specs.BLESSING_KEY]
      )
  
      if input_dict.get(standard_component_specs.EXAMPLES_KEY):
          examples = artifact_utils.get_single_instance(
              input_dict[standard_component_specs.EXAMPLES_KEY]
          )
      else:
          examples = None
  
      serving_spec = infra_validator_pb2.ServingSpec()
      proto_utils.json_to_proto(
          exec_properties[standard_component_specs.SERVING_SPEC_KEY],
          serving_spec,
      )
      if not serving_spec.model_name:
          serving_spec.model_name = _DEFAULT_MODEL_NAME
  
      validation_spec = infra_validator_pb2.ValidationSpec()
      if exec_properties.get(
          standard_component_specs.VALIDATION_SPEC_KEY
      ):
          proto_utils.json_to_proto(
              exec_properties[
                  standard_component_specs.VALIDATION_SPEC_KEY
              ],
              validation_spec,
          )
      if not validation_spec.num_tries:
          validation_spec.num_tries = _DEFAULT_NUM_TRIES
      if not validation_spec.max_loading_time_seconds:
          validation_spec.max_loading_time_seconds = (
              _DEFAULT_MAX_LOADING_TIME_SEC
          )
  
      if exec_properties.get(standard_component_specs.REQUEST_SPEC_KEY):
          request_spec = infra_validator_pb2.RequestSpec()
          proto_utils.json_to_proto(
              exec_properties[
                  standard_component_specs.REQUEST_SPEC_KEY
              ],
              request_spec,
          )
      else:
          request_spec = None
  
      with self._InstallGracefulShutdownHandler():
          self._Do(
              model=model,
              examples=examples,
              blessing=blessing,
              serving_spec=serving_spec,
              validation_spec=validation_spec,
              request_spec=request_spec,
          )
  
imports: import contextlib
  import functools
  import os
  import signal
  import threading
  import time
  from typing import Any, Dict, List, Optional
  from absl import logging
  from tfx import types
  from tfx.components.infra_validator import error_types
  from tfx.components.infra_validator import request_builder
  from tfx.components.infra_validator import serving_bins
  from tfx.components.infra_validator import types
  from tfx.components.infra_validator.model_server_runners import kubernetes_runner
  from tfx.components.infra_validator.model_server_runners import local_docker_runner
  from tfx.dsl.components.base import base_executor
  from tfx.proto import infra_validator_pb2
  from tfx.types import artifact_utils
  from tfx.types import standard_component_specs
  from tfx.utils import io_utils
  from tfx.utils import path_utils
  from tfx.utils import proto_utils
  from tfx.utils.model_paths import tf_serving_flavor
  from tensorflow_serving.apis import classification_pb2
  from tensorflow_serving.apis import predict_pb2
  from tensorflow_serving.apis import prediction_log_pb2
  from tensorflow_serving.apis import regression_pb2
label_desc: Use image node_id 4 for calling the Executor obj's underlying member method code with example usage: obj.Do(input_dict, output_dict, exec_properties) without return types