code: def _generate_jumpstart_model_versions(  # pylint: disable=redefined-builtin
      filter: Union[Operator, str] = Constant(BooleanValues.TRUE),
      region: str = JUMPSTART_DEFAULT_REGION_NAME,
      list_incomplete_models: bool = False,
      sagemaker_session: Session = DEFAULT_JUMPSTART_SAGEMAKER_SESSION,
  ) -> Generator:
      """Generate models for JumpStart, and optionally apply filters to result.
  
      Args:
          filter (Union[Operator, str]): Optional. The filter to apply to generate models. This can be
              either an ``Operator`` type filter (e.g. ``And("task == ic", "framework == pytorch")``),
              or simply a string filter which will get serialized into an Identity filter.
              (e.g. ``"task == ic"``). If this argument is not supplied, all models will be generated.
              (Default: Constant(BooleanValues.TRUE)).
          region (str): Optional. The AWS region from which to retrieve JumpStart metadata regarding
              models. (Default: JUMPSTART_DEFAULT_REGION_NAME).
          list_incomplete_models (bool): Optional. If a model does not contain metadata fields
              requested by the filter, and the filter cannot be resolved to a include/not include,
              whether the model should be included. By default, these models are omitted from
              results. (Default: False).
          sagemaker_session (sagemaker.session.Session): Optional. The SageMaker Session
              to use to perform the model search. (Default: DEFAULT_JUMPSTART_SAGEMAKER_SESSION).
      """
  
      models_manifest_list = (
          accessors.JumpStartModelsAccessor._get_manifest(
              region=region, s3_client=sagemaker_session.s3_client
          )
      )
  
      if isinstance(filter, str):
          filter = Identity(filter)
  
      manifest_keys = set(models_manifest_list[0].__slots__)
  
      all_keys: Set[str] = set()
  
      model_filters: Set[ModelFilter] = set()
  
      for operator in _model_filter_in_operator_generator(filter):
          model_filter = operator.unresolved_value
          key = model_filter.key
          all_keys.add(key)
          model_filters.add(model_filter)
  
      for key in all_keys:
          if "." in key:
              raise NotImplementedError(
                  f"No support for multiple level metadata indexing ('{key}')."
              )
  
      metadata_filter_keys = all_keys - SPECIAL_SUPPORTED_FILTER_KEYS
  
      required_manifest_keys = manifest_keys.intersection(
          metadata_filter_keys
      )
      possible_spec_keys = metadata_filter_keys - manifest_keys
  
      is_task_filter = SpecialSupportedFilterKeys.TASK in all_keys
      is_framework_filter = (
          SpecialSupportedFilterKeys.FRAMEWORK in all_keys
      )
  
      def evaluate_model(
          model_manifest: JumpStartModelHeader,
      ) -> Optional[Tuple[str, str]]:
          copied_filter = copy.deepcopy(filter)
  
          manifest_specs_cached_values: Dict[
              str, Union[bool, int, float, str, dict, list]
          ] = {}
  
          model_filters_to_resolved_values: Dict[
              ModelFilter, BooleanValues
          ] = {}
  
          for val in required_manifest_keys:
              manifest_specs_cached_values[val] = getattr(
                  model_manifest, val
              )
  
          if is_task_filter:
              manifest_specs_cached_values[
                  SpecialSupportedFilterKeys.TASK
              ] = extract_framework_task_model(model_manifest.model_id)[
                  1
              ]
  
          if is_framework_filter:
              manifest_specs_cached_values[
                  SpecialSupportedFilterKeys.FRAMEWORK
              ] = extract_framework_task_model(model_manifest.model_id)[
                  0
              ]
  
          if Version(model_manifest.min_version) > Version(
              get_sagemaker_version()
          ):
              return None
  
          _populate_model_filters_to_resolved_values(
              manifest_specs_cached_values,
              model_filters_to_resolved_values,
              model_filters,
          )
  
          _put_resolved_booleans_into_filter(
              copied_filter, model_filters_to_resolved_values
          )
  
          copied_filter.eval()
  
          if copied_filter.resolved_value in [
              BooleanValues.TRUE,
              BooleanValues.FALSE,
          ]:
              if copied_filter.resolved_value == BooleanValues.TRUE:
                  return (
                      model_manifest.model_id,
                      model_manifest.version,
                  )
              return None
  
          if copied_filter.resolved_value == BooleanValues.UNEVALUATED:
              raise RuntimeError(
                  "Filter expression in unevaluated state after using "
                  "values from model manifest. Model ID and version that "
                  f"is failing: {(model_manifest.model_id, model_manifest.version)}."
              )
          copied_filter_2 = copy.deepcopy(filter)
  
          # spec is downloaded to thread's memory. since each thread
          # accesses a unique s3 spec, there is no need to use the JS caching utils.
          # spec only stays in memory for lifecycle of thread.
          model_specs = JumpStartModelSpecs(
              json.loads(
                  sagemaker_session.read_s3_file(
                      get_jumpstart_content_bucket(region),
                      model_manifest.spec_key,
                  )
              )
          )
  
          for val in possible_spec_keys:
              if hasattr(model_specs, val):
                  manifest_specs_cached_values[val] = getattr(
                      model_specs, val
                  )
  
          _populate_model_filters_to_resolved_values(
              manifest_specs_cached_values,
              model_filters_to_resolved_values,
              model_filters,
          )
          _put_resolved_booleans_into_filter(
              copied_filter_2, model_filters_to_resolved_values
          )
  
          copied_filter_2.eval()
  
          if (
              copied_filter_2.resolved_value
              != BooleanValues.UNEVALUATED
          ):
              if (
                  copied_filter_2.resolved_value == BooleanValues.TRUE
                  or (BooleanValues.UNKNOWN and list_incomplete_models)
              ):
                  return (
                      model_manifest.model_id,
                      model_manifest.version,
                  )
              return None
  
          raise RuntimeError(
              "Filter expression in unevaluated state after using values from model specs. "
              "Model ID and version that is failing: "
              f"{(model_manifest.model_id, model_manifest.version)}."
          )
  
      with ThreadPoolExecutor(
          max_workers=MAX_SEARCH_WORKERS
      ) as executor:
          futures = []
          for header in models_manifest_list:
              futures.append(executor.submit(evaluate_model, header))
  
          for future in as_completed(futures):
              error = future.exception()
              if error:
                  raise error
              result = future.result()
              if result:
                  yield result
  
imports: from __future__ import absolute_import
  import copy
  from concurrent.futures import ThreadPoolExecutor, as_completed
  from functools import cmp_to_key
  import json
  from typing import Any, Generator, List, Optional, Tuple, Union, Set, Dict
  from packaging.version import Version
  from sagemaker.jumpstart import accessors
  from sagemaker.jumpstart.constants import DEFAULT_JUMPSTART_SAGEMAKER_SESSION, JUMPSTART_DEFAULT_REGION_NAME
  from sagemaker.jumpstart.enums import JumpStartScriptScope
  from sagemaker.jumpstart.filters import SPECIAL_SUPPORTED_FILTER_KEYS, BooleanValues, Identity, SpecialSupportedFilterKeys
  from sagemaker.jumpstart.filters import Constant, ModelFilter, Operator, evaluate_filter_expression
  from sagemaker.jumpstart.types import JumpStartModelHeader, JumpStartModelSpecs
  from sagemaker.jumpstart.utils import get_jumpstart_content_bucket, get_sagemaker_version, verify_model_region_and_return_specs
  from sagemaker.session import Session
label_desc: Use image node_id 10 for calling a global function with example usage: _generate_jumpstart_model_versions(filter, region, list_incomplete_models, sagemaker_session) and returns: None, None, None, model_manifest,  model_manifest, model_manifest,  model_manifest