diff --git a/src/sasctl/pzmm/__init__.py b/src/sasctl/pzmm/__init__.py index 4667bc65..bf341760 100644 --- a/src/sasctl/pzmm/__init__.py +++ b/src/sasctl/pzmm/__init__.py @@ -8,4 +8,5 @@ from .pickle_model import PickleModel from .write_json_files import JSONFiles from .write_score_code import ScoreCode +from .write_score_wrapper import ScoreWrapper from .zip_model import ZipModel diff --git a/src/sasctl/pzmm/import_model.py b/src/sasctl/pzmm/import_model.py index dc66deab..02b654f6 100644 --- a/src/sasctl/pzmm/import_model.py +++ b/src/sasctl/pzmm/import_model.py @@ -199,6 +199,7 @@ def import_model( target_values: Optional[List[str]] = None, overwrite_project_properties: Optional[bool] = False, target_index: Optional[int] = None, + score_wrapper: Optional[bool] = False, **kwargs, ) -> Tuple[RestObj, Union[dict, str, Path]]: """ @@ -286,6 +287,9 @@ def import_model( index should match the index of the target outcome in target_values. If target_values are not given, this index should indicate whether the the target probability variable is the first or second variable returned by the model. The default value is 1. + score_wrapper : bool, optional + if the score wrapper method is being used, it shouldn't write the score code as the score wrapper function will handle + the formulation of the score code. kwargs : dict, optional Other keyword arguments are passed to the following function: * sasctl.pzmm.ScoreCode.write_score_code(..., @@ -305,6 +309,7 @@ def import_model( model_files : dict, str, or Path Dictionary representation of all files or the path the model files were generated from. + :param score_wrapper: """ # For mlflow models, overwrite the provided or default pickle_type if mlflow_details: @@ -345,6 +350,38 @@ def import_model( model_prefix, project, zip_io_file, version=project_version ) return model, model_files + elif score_wrapper: + print("The user has made use of the score_wrapper argument, score code has already been generated and will not be generated again.") + + if isinstance(model_files, dict): + zip_io_file = zm.zip_files(model_files, model_prefix, is_viya4=True) + else: + zip_io_file = zm.zip_files( + Path(model_files), model_prefix, is_viya4=True + ) + if cls.notebook_output: + print(f"All model files were zipped to {Path(model_files)}.") + + project_response = mr.get_project(project) + project = project_exists( + project, + project_response, + target_values, + model_files, + overwrite_project_properties, + ) + + # Check if model with same name already exists in project. + model_exists( + project, model_prefix, overwrite_model, version_name=project_version + ) + + model = mr.import_model_from_zip( + model_prefix, project, zip_io_file, version=project_version + ) + + return model, model_files + # For SAS Viya 4, the score code can be written beforehand and imported with # all the model files elif current_session().version_info() == 4: diff --git a/src/sasctl/pzmm/write_score_wrapper.py b/src/sasctl/pzmm/write_score_wrapper.py new file mode 100644 index 00000000..583da030 --- /dev/null +++ b/src/sasctl/pzmm/write_score_wrapper.py @@ -0,0 +1,170 @@ +# Copyright (c) 2024, SAS Institute Inc., Cary, NC, USA. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 +from typing import List +from pathlib import Path +import re +import importlib.util +import inspect +import importlib +import inspect +import sys +import os +import textwrap +import ast + + +class ScoreWrapper: + #score_wrapper string to hold score code + score_wrapper: str = "" + + @classmethod + def write_score_wrapper_function_input(cls, + imports: List[str], + score_function_definition: str, + score_function_body: str, + score_function_exception_body:str, + items_to_return: str, + model_load: str, + model_name_with_file_extension: str, + output_variables: List[str]): + """ + Method to generate scoring code from a function and add it to cls.score_wrapper. + + Parameters: + imports (List[str]): List of modules to import. + function_definition (str): Function definition. + function_body (str): Function body. + model_load (str): Name of the model to load. + model_name_with_file_extension (str): Name of the model file with extension. + output_variables (List[str]): List of output variables to define in score function + + Returns: + cls.score_wrapper (str): The scoring code. + """ + # Import Modules + for module in imports: + cls.score_wrapper += f"import {module}\n" + # Import pathlib and settings + cls.score_wrapper += "from pathlib import Path\n" + cls.score_wrapper += "import settings\n\n\n" + + # Load the model + cls.score_wrapper += f"model_path = Path(settings.pickle_path) / '{model_name_with_file_extension}'\n\n" + cls.score_wrapper += f"with open(Path(settings.pickle_path) / '{model_name_with_file_extension}', 'rb') as f:\n\tmodel = {model_load}\n\n" + + # Define the score function and add the function body specified + cls.score_wrapper += f"{score_function_definition}:\n" + cls.score_wrapper += '\t"' + cls.score_wrapper += "Output: " + ", ".join(output_variables) # Join output variables with comma + cls.score_wrapper += '"\n' + cls.score_wrapper += "\tglobal model\n" + cls.score_wrapper += "\ttry:" + cls.score_wrapper += f"\n{score_function_body}\n" + cls.score_wrapper += "\n\texcept Exception as e:\n" + cls.score_wrapper += f"\t\t{score_function_exception_body}\n" + cls.score_wrapper += f"\treturn {items_to_return}\n" + + return cls.score_wrapper + + # # Validate syntax before returning + # if not cls.validate_score_wrapper_syntax(cls.score_wrapper): + # raise SyntaxError("Syntax error in generated code.") + + # return cls.score_wrapper + + @classmethod + def write_score_wrapper_file_input(cls, + imports: List[str], + file_path: str, + model_load: str, + model_name_with_file_extension: str, + score_function_body: str, + score_function_input_parameters: str, + output_variables: List[str], + score_function_exception_body: str, + items_to_return: str, + ): + """ + Method to generate scoring code from a file and add it to cls.score_wrapper. + + Parameters: + imports (List[str]): List of modules to import. + file_path (str): Path to the file containing the additional classes and functions necessary for scoring. + model_load (str): Name of the model to load. + model_name_with_file_extension (str): Name of the model file with extension. + score_function_body (str): The code needed to evaluate the model. + output_variables (List[str]): List of output variables to define in score function + + Returns: + cls.score_wrapper (str): The scoring code. + """ + # write imports to file + for module in imports: + cls.score_wrapper += f"import {module}\n" + cls.score_wrapper += "from pathlib import Path\n" + cls.score_wrapper += "import settings\n\n\n" + + # open a file and read all of the class definitions and functions, and write them to cls.score_wrapper maintain spacing of the code being written + with open(file_path, "r") as file: + code = file.read() + cls.score_wrapper += code + + # load model + cls.score_wrapper += f"\n\nmodel_path = Path(settings.pickle_path) / '{model_name_with_file_extension}'\n" + cls.score_wrapper += f"{model_load}\n\n" + + # define the generic score function, and append the score_function_body to evaluate the model. + cls.score_wrapper += f"def score({score_function_input_parameters}):\n" + cls.score_wrapper += '\t"' + cls.score_wrapper += "Output: " + ", ".join(output_variables) # Join output variables with comma + cls.score_wrapper += '"\n' + + cls.score_wrapper += "\tglobal model\n" + cls.score_wrapper += "\ttry:" + cls.score_wrapper += f"\n{score_function_body}\n" + cls.score_wrapper += "\n\texcept Exception as e:\n" + cls.score_wrapper += f"\t\t{score_function_exception_body}\n" + cls.score_wrapper += f"\treturn {items_to_return}\n" + + # Validate Syntax before returning + if not cls.validate_score_wrapper_syntax(cls.score_wrapper): + raise SyntaxError("Syntax error in generated code.") + + return cls.score_wrapper + + @classmethod + def write_wrapper_to_file(cls, + path: str, + model_prefix: str, + file_name: str) -> None: + """ + Method to write the score wrapper to a file. + + Parameters: + file_name (str): Name of the file to write the score wrapper to. + + Returns: + None + """ + score_wrapper_path = Path(path) / f"score_{model_prefix}.py" + with open(score_wrapper_path, "w") as score_wrapper_file: + score_wrapper_file.write(cls.score_wrapper) + + @classmethod + def validate_score_wrapper_syntax(cls, code: str) -> bool: + """ + Method to perform a syntax check on the provided code. + + Parameters: + code (str): Code to be checked for syntax. + + Returns: + bool: True if syntax is correct, False otherwise. + """ + try: + ast.parse(code) + return True + except SyntaxError: + return False + + diff --git a/tests/unit/test_write_score_wrapper.py b/tests/unit/test_write_score_wrapper.py new file mode 100644 index 00000000..8c7306b5 --- /dev/null +++ b/tests/unit/test_write_score_wrapper.py @@ -0,0 +1,38 @@ +import unittest +from unittest.mock import Mock +from pathlib import Path +from sasctl import pzmm + +class TestScoreWrapper(unittest.TestCase): + def test_write_score_wrapper_function_input(self): + # mock imports + imports = ['torch', 'pandas'] + score_function_definition = 'def score_function():' + score_function_body = 'print("Scoring")' + score_function_exception_body = 'print("Error in scoring")' + items_to_return = 'result' + model_load = 'model = pickle.load(f)' + model_name_with_file_extension = 'model.pkl' + output_variables = ['output_1', 'output_2'] + + # mok the write_score_wrapper_function_input method + result = pzmm.ScoreWrapper.write_score_wrapper_function_input(imports, score_function_definition, score_function_body, score_function_exception_body, items_to_return, model_load, model_name_with_file_extension, output_variables) + + # assert + self.assertIn('import torch', result) + self.assertIn('import pandas', result) + self.assertIn('def score_function():', result) + self.assertIn('print("Scoring")', result) + self.assertIn('result', result) + self.assertIn('model = pickle.load(f)', result) + self.assertIn('model.pkl', result) + self.assertIn('output_1, output_2', result) + + def test_validate_score_wrapper_syntax(self): + # mock code + code = 'def score_function():\n\tprint("Scoring")\n\tresult = model.predict(data)\n\treturn result\n' + #send through validate score_wrapper_syntax + result = pzmm.ScoreWrapper.validate_score_wrapper_syntax(code) + #assert + self.assertTrue(result) +