def validate_model_spec()

in scripts/azureml-assets/azureml/assets/validate_assets.py [0:0]

• 84 lines of code
• 26 McCabe index (conditional complexity)

def validate_model_spec(asset_config: assets.AssetConfig) -> int:
    """Validate model spec.

    Args:
        asset_config (assets.AssetConfig): asset config for model spec

    Returns:
        int: error count
    """
    error_count = 0
    model = model_config = None

    try:
        model = load_model(asset_config.spec_with_path)
    except Exception:
        _log_error(asset_config.file_name_with_path, "Invalid spec file")
        return 1

    try:
        model_config: assets.ModelConfig = asset_config.extra_config_as_object()
    except Exception:
        _log_error(asset_config.file_name_with_path, "Invalid model config")
        return 1

    if model_config.type != assets.config.ModelType.MLFLOW:
        logger.print(
            f"Bypass validation for {asset_config.name} as model type is: {model_config.type.value}"
        )
        return 0

    # confirm must have
    if not model.tags.get(MLFlowModelTags.TASK):
        _log_error(asset_config.file_name_with_path, f"{MLFlowModelTags.TASK} missing")
        error_count += 1

    if not model.tags.get(MLFlowModelTags.LICENSE):
        _log_error(asset_config.file_name_with_path, f"{MLFlowModelTags.LICENSE} missing")
        error_count += 1

    # TODO: skip hidden layer valdn till most models are scanned
    # if MLFlowModelTags.HIDDEN_LAYERS_SCANNED not in model.tags:
    #     _log_error(
    #         asset_config.file_name_with_path,
    #         (
    #             "`hiddenlayerscanned` tag missing. "
    #             "Model is not scanned by HiddenLayer ModelScanner tool. Please scan the model and retry"
    #         )
    #     )
    #     error_count += 1

    # shared compute check
    if MLFlowModelTags.SHARED_COMPUTE_CAPACITY not in model.tags:
        _log_error(asset_config.file_name_with_path, f"Tag {MLFlowModelTags.SHARED_COMPUTE_CAPACITY} missing")
        error_count += 1

    if not model.properties.get(MLFlowModelProperties.SHARED_COMPUTE_CAPACITY, False):
        _log_error(
            asset_config.file_name_with_path, f"Property {MLFlowModelProperties.SHARED_COMPUTE_CAPACITY} is not set"
        )
        error_count += 1

    # if any of the relevant tags or properies is present
    # assume support as true and then fail in valdn
    supports_eval = (
        MLFlowModelTags.EVALUATION_COMPUTE_ALLOWLIST in model.tags
        or MLFlowModelProperties.EVALUATION_RECOMMENDED_SKU in model.properties
        or MLFlowModelProperties.EVALUATION_MIN_SKU_SPEC in model.properties
    )

    # If any of the relevant tags or properies is present
    # assume support as true and then fail in valdn
    supports_ft = (
        MLFlowModelTags.FINETUNE_COMPUTE_ALLOWLIST in model.tags
        or MLFlowModelProperties.FINETUNE_RECOMMENDED_SKU in model.properties
        or MLFlowModelProperties.FINETUNE_MIN_SKU_SPEC in model.properties
        or MLFlowModelProperties.FINETUNING_TASKS in model.properties
    )

    # validate inference compute req.
    error_count += validate_model_scenario(
        asset_config.file_name_with_path,
        model,
        MLFlowModelProperties.INFERENCE_MIN_SKU_SPEC,
        MLFlowModelProperties.INFERENCE_RECOMMENDED_SKU,
        MLFlowModelTags.INFERENCE_COMPUTE_ALLOWLIST,
    )

    # check valid computes for inference
    with open(SUPPORTED_INFERENCE_SKU_FILE_PATH) as f:
        supported_inference_skus = set(json.load(f))
        unsupported_skus_in_spec = [
            sku
            for sku in model.tags.get(MLFlowModelTags.INFERENCE_COMPUTE_ALLOWLIST, [])
            if sku not in supported_inference_skus
        ]
        if unsupported_skus_in_spec:
            _log_error(asset_config.file_name_with_path,
                       f"Unsupported inference SKU in spec: {unsupported_skus_in_spec}")
            error_count += 1

    if supports_eval:
        error_count += validate_model_scenario(
            asset_config.file_name_with_path,
            model,
            MLFlowModelProperties.EVALUATION_MIN_SKU_SPEC,
            MLFlowModelProperties.EVALUATION_RECOMMENDED_SKU,
            MLFlowModelTags.EVALUATION_COMPUTE_ALLOWLIST,
        )

    if supports_ft:
        if not model.properties.get(MLFlowModelProperties.FINETUNING_TASKS):
            _log_error(
                asset_config.file_name_with_path,
                f"{MLFlowModelProperties.FINETUNING_TASKS} not set for supporting finetuning scenario"
            )
            error_count += 1

        error_count += validate_model_scenario(
            asset_config.file_name_with_path,
            model,
            MLFlowModelProperties.FINETUNE_MIN_SKU_SPEC,
            MLFlowModelProperties.FINETUNE_RECOMMENDED_SKU,
            MLFlowModelTags.FINETUNE_COMPUTE_ALLOWLIST,
        )

    return error_count