def evaluate_model()

in src/sagemaker/jumpstart/notebook_utils.py [0:0]

• 66 lines of code
• 16 McCabe index (conditional complexity)

    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 is_model_type_filter:
            manifest_specs_cached_values[SpecialSupportedFilterKeys.MODEL_TYPE] = (
                extract_model_type_filter_representation(model_manifest.spec_key)
            )

        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)}."
        )