Updates to score code testing; executing snippets

Author: smlindauer

src/sasctl/pzmm/write_score_code.py

@@ -812,7 +812,7 @@ def _predict_method(
                 f"{'':4}h2o_array = h2o.H2OFrame(input_array, "
                 f"column_types=column_types)\n{'':4}prediction = "
                 f"model.{method.__name__}(h2o_array)\n{'':4}prediction"
-                f" = h2o.as_list(prediction, use_pandas=False)\n"
+                f" = h2o.as_list(prediction, use_pandas=prediction.shape[0]>1)\n"
             )
             """
     input_array = pd.DataFrame(
@@ -821,7 +821,7 @@ def _predict_method(
     input_array = impute_missing_values(input_array)
     column_types = {"var1": "string", "var2": "numeric", "var3": "numeric"}
     h2o_array = h2o.H2OFrame(input_array, column_types=column_types)
-    prediction = model.predict(h2o.array)
+    prediction = model.predict(h2o_array)
     prediction = h2o.as_list(prediction, use_pandas=False)
             """
         # Statsmodels models
@@ -1167,30 +1167,29 @@ def _no_targets_no_thresholds(
                     f"{'':4}if input_array.shape[0] == 1:\n"
                     f"{'':8}{metrics} = prediction[1][0]\n{'':8}return {metrics}\n"
                     f"{'':4}else:\n"
-                    f"{'':8}prediction = pd.Dataframe(prediction[1:], columns=prediction[0])\n"
-                    f"{'':8}return pd.DataFrame({{'{metrics}': prediction.iloc[:,0]}})"
+                    f"{'':8}return prediction.iloc[:, 0]\n"
                 )
                 """
     if input_array.shape[0] == 1:
         output_variable = prediction[1][0]
         return output_variable
     else:
-        prediction = pd.Dataframe(prediction[1:], columns=prediction[0])
-        return pd.DataFrame({'output_variable': prediction.iloc[:,0]})
+        return prediction.iloc[:, 0]
                 """
             else:
                 cls.score_code += (
                     f"{'':4}if input_array.shape[0] == 1:\n"
                     f"{'':8}{metrics} = prediction[0]\n{'':8}return {metrics}\n"
                     f"{'':4}else:\n"
-                    f"{'':8}output_table = pd.DataFrame('{metrics}': prediction)\n{'':8}return output_table"
+                    f"{'':8}output_table = pd.DataFrame({{'{metrics}': prediction}})\n"
+                    f"{'':8}return output_table"
                 )
                 """
     if input_array.shape[0] == 1:
         output_variable = prediction[0]
         return output_variable
     else:
-        output_table = pd.DataFrame('output_variable': prediction)
+        output_table = pd.DataFrame({'output_variable': prediction})
         return output_table
                 """
         else:
@@ -1204,40 +1203,42 @@ def _no_targets_no_thresholds(
                         f"{'':8}{metrics[i + 1]} = float(prediction[1][{i + 1}])\n"
                     )
                 cls.score_code += (
+                    f"{'':8}return {', '.join(metrics)}\n"
                     f"{'':4}else:\n"
-                    f"{'':8}output_table = pd.DataFrame(prediction[1:], columns={metrics})\n"
+                    f"{'':8}output_table = prediction\n"
+                    f"{'':8}output_table.columns = {metrics}\n"
                     f"{'':8}return output_table\n"
                 )
                 """
     if input_array.shape[0] == 1:
         classification_variable = prediction[1][0]
         prediction_variable_1 = float(prediction[1][1])
         prediction_variable_2 = float(prediction[1][2])
+        return classification_variable, prediction_variable_1, prediction_variable_2
     else:
-        output_table = pd.DataFrame(prediction[1:], columns=[classification_variable,variable_1,variable_2])
+        output_table = pd.DataFrame(prediction[1:], columns=[classification_variable, prediction_variable_1, prediction_variable_2])
         return output_table
                 """
             else:
                 cls.score_code += f"{'':4}if input_array.shape[0] == 1:\n"
                 for i in range(len(metrics)):
                     cls.score_code += f"{'':8}{metrics[i]} = prediction[{i}]\n"
+                cls.score_code += f"\n{'':8}return {', '.join(metrics)}\n"
                 cls.score_code += (
                     f"{'':4}else:\n"
-                    f"{'':8}output_table = pd.DataFrame(prediction, columns={metrics})\n"
-                    f"{'':8}return output_table\n"
+                    f"{'':8}output_table = pd.DataFrame(prediction, columns={metrics})"
+                    f"\n{'':8}return output_table\n"
                 )
-            cls.score_code += f"\n{'':4}return {', '.join(metrics)}"
             """
     if input_array.shape[0] == 1:
         classification_variable = prediction[0]
         prediction_variable_1 = prediction[1]
         prediction_variable_2 = prediction[2]
     else:
-        output_table = pd.DataFrame(prediction, columns=["classification_variable","prediction_variable_1","prediction_variable_2"])
+        output_table = pd.DataFrame(prediction, columns=["classification_variable", "prediction_variable_1", "prediction_variable_2"])
         return output_table
             """
 
-
     @classmethod
     def _binary_target(
         cls,

tests/unit/test_write_score_code.py

@@ -9,6 +9,7 @@
 import pickle
 import random
 import tempfile
+import unittest
 from pathlib import Path
 from unittest.mock import DEFAULT, MagicMock, patch
 
@@ -20,6 +21,7 @@
 from sasctl import current_session
 from sasctl.core import RestObj, VersionInfo
 from sasctl.pzmm.write_score_code import ScoreCode as sc
+from sasctl.pzmm.write_score_code import ScoreCode
 
 
 @pytest.fixture()
@@ -301,50 +303,174 @@ def test_determine_score_metrics():
     ) == ["P_A", "P_B", "P_C"]
 
 
-def test_no_targets_no_thresholds():
-    """
-    Test Cases:
-    - len(metrics) == 1
-        - non-h2o
-        - h2o
-    - len(metrics) > 1
-        - non-h2o
-        - h2o
-    - raise error for invalid config (returns - metrics != 0)
-    """
-    metrics = "Classification"
-    returns = [1, "A"]
-    with pytest.raises(ValueError):
-        sc._no_targets_no_thresholds(metrics, returns)
+class TestNoTargetsNoThresholds(unittest.TestCase):
+    def setUp(self):
+        self.sc = ScoreCode
 
-    returns = [1]
-    sc._no_targets_no_thresholds(metrics, returns)
-    assert "Classification = prediction" in sc.score_code
-    sc.score_code = ""
+    def tearDown(self):
+        self.sc.score_code = ""
 
-    sc._no_targets_no_thresholds(metrics, returns, h2o_model=True)
-    assert "Classification = prediction[1][0]"
-    sc.score_code = ""
+    def execute_snippet(self, *args):
+        scope = {}
+        exec(self.sc.score_code, scope)
+        test_snippet = scope["test_snippet"]
+        return test_snippet(*args)
 
-    metrics = ["Classification", "Proba_A", "Proba_B", "Proba_C"]
-    returns = ["I", 1, 2, 3]
-    sc._no_targets_no_thresholds(metrics, returns)
-    assert (
-        sc.score_code == f"{'':4}Classification = prediction[0]\n"
-        f"{'':4}Proba_A = prediction[1]\n"
-        f"{'':4}Proba_B = prediction[2]\n"
-        f"{'':4}Proba_C = prediction[3]\n\n"
-        f"{'':4}return Classification, Proba_A, Proba_B, Proba_C"
-    )
-    sc.score_code = ""
-    sc._no_targets_no_thresholds(metrics, returns, h2o_model=True)
-    assert (
-        sc.score_code == f"{'':4}Classification = prediction[1][0]\n"
-        f"{'':4}Proba_A = float(prediction[1][1])\n"
-        f"{'':4}Proba_B = float(prediction[1][2])\n"
-        f"{'':4}Proba_C = float(prediction[1][3])\n\n"
-        f"{'':4}return Classification, Proba_A, Proba_B, Proba_C"
-    )
+    def test_improper_arguments(self):
+        metrics = "Classification"
+        returns = [1, "A"]
+        with pytest.raises(ValueError):
+            self.sc._no_targets_no_thresholds(metrics, returns)
+
+    def test_single_metric(self):
+        metrics = "Classification"
+        returns = [1]
+        self.sc.score_code += "import pandas as pd\n" \
+                              "def test_snippet(input_array, prediction):\n"
+        self.sc._no_targets_no_thresholds(metrics, returns)
+        # Single row
+        input_array = pd.DataFrame([[1]], columns=["A"], index=[0])
+        prediction = [.5]
+        self.assertEqual(self.execute_snippet(input_array, prediction), .5)
+        # Multi row
+        input_array = pd.DataFrame({"A": [.9, 1, 1.1]})
+        prediction = [.3, .4, .5]
+        pd.testing.assert_frame_equal(
+            self.execute_snippet(input_array, prediction),
+            pd.DataFrame({metrics: prediction})
+        )
+
+    def test_single_metric_h2o(self):
+        metrics = "Classification"
+        returns = [1]
+        self.sc.score_code += "import pandas as pd\n" \
+                              "def test_snippet(input_array, prediction):\n"
+        self.sc._no_targets_no_thresholds(metrics, returns, h2o_model=True)
+        # Single row
+        input_array = pd.DataFrame([[1]], columns=["A"], index=[0])
+        prediction = [[], [.5]]
+        self.assertEqual(self.execute_snippet(input_array, prediction), .5)
+        # Multi row
+        input_array = pd.DataFrame({"A": [.9, 1, 1.1]})
+        prediction = pd.DataFrame({"predict": [0, 1, 1], "p0": [.3, .4, .5]})
+        pd.testing.assert_series_equal(
+            self.execute_snippet(input_array, prediction),
+            pd.Series([0, 1, 1], name="predict")
+        )
+
+    def test_multi_metric(self):
+        metrics = ["Classification", "Proba_A", "Proba_B", "Proba_C"]
+        returns = ["I", 1, 2, 3]
+        self.sc.score_code += "import pandas as pd\n" \
+                              "def test_snippet(input_array, prediction):\n"
+        self.sc._no_targets_no_thresholds(metrics, returns)
+        # Single row
+        input_array = pd.DataFrame([[1]], columns=["A"], index=[0])
+        prediction = ["i", .3, .4, .5]
+        self.assertEqual(
+            self.execute_snippet(input_array, prediction),
+            ("i", .3, .4, .5)
+        )
+        # Multi row
+        input_array = pd.DataFrame({"A": [1, 0, 1]})
+        prediction = pd.DataFrame({
+            "Classification": ["i", "j", "k"],
+            "Proba_A": [.1, .2, .3],
+            "Proba_B": [.4, .5, .6],
+            "Proba_C": [.7, .8, .9]
+        })
+        pd.testing.assert_frame_equal(
+            self.execute_snippet(input_array, prediction),
+            prediction
+        )
+
+    def test_multi_metric_h2o(self):
+        metrics = ["Classification", "Proba_A", "Proba_B", "Proba_C"]
+        returns = ["I", 1, 2, 3]
+        self.sc.score_code += "import pandas as pd\n" \
+                              "def test_snippet(input_array, prediction):\n"
+        self.sc._no_targets_no_thresholds(metrics, returns, h2o_model=True)
+        # Single row
+        input_array = pd.DataFrame([[1]], columns=["A"], index=[0])
+        prediction = [[], ["i", .3, .4, .5]]
+        self.assertEqual(
+            self.execute_snippet(input_array, prediction),
+            ("i", .3, .4, .5)
+        )
+        # Multi row
+        input_array = pd.DataFrame({"A": [1, 0, 1]})
+        prediction = pd.DataFrame({
+            "Classification": ["i", "j", "k"],
+            "Proba_A": [.1, .2, .3],
+            "Proba_B": [.4, .5, .6],
+            "Proba_C": [.7, .8, .9]
+        })
+        pd.testing.assert_frame_equal(
+            self.execute_snippet(input_array, prediction),
+            prediction
+        )
+
+
+class TestBinaryTarget(unittest.TestCase):
+    def setUp(self):
+        self.sc = ScoreCode
+        self.target_values = ["A", "B"]
+
+    def tearDown(self):
+        self.sc.score_code = ""
+
+    def execute_snippet(self, *args):
+        scope = {}
+        exec(self.sc.score_code, scope)
+        test_snippet = scope["test_snippet"]
+        return test_snippet(*args)
+
+    def test_improper_arguments(self):
+        with pytest.raises(ValueError):
+            sc._binary_target([], [], ["A", 1, 2, 3])
+        with pytest.raises(ValueError):
+            sc._binary_target([], [], ["A", "B"])
+        with pytest.raises(ValueError):
+            sc._binary_target(["A", "B", "C", "D"], [], [])
+
+    def test_one_metric_one_return(self):
+        metrics = "Classification"
+        returns = [""]
+        self.sc.score_code += "import pandas as pd\n" \
+                              "def test_snippet(input_array, prediction):\n"
+        self.sc._binary_target(metrics, self.target_values, returns)
+        # Single row
+        input_array = pd.DataFrame([[1]], columns=["A"], index=[0])
+        prediction = .5
+        self.assertEqual(self.execute_snippet(input_array, prediction), .5)
+        # Multi row
+        input_array = pd.DataFrame({"A": [.9, 1, 1.1]})
+        prediction = [.3, .4, .5]
+        pd.testing.assert_frame_equal(
+            self.execute_snippet(input_array, prediction),
+            pd.DataFrame({metrics: prediction})
+        )
+
+    def test_one_metric_two_returns(self):
+        pass
+
+    def test_one_metric_three_returns(self):
+        pass
+
+    def test_two_metrics_one_return(self):
+        pass
+
+    def test_two_metrics_two_returns(self):
+        pass
+
+    def test_two_metrics_three_returns(self):
+        pass
+
+    def test_three_metrics_one_return(self):
+        pass
+
+    def test_three_metrics_three_returns(self):
+        pass
 
 
 def test_binary_target():
@@ -398,14 +524,6 @@ def test_binary_target():
         - sum(returns) >= 2
         - len(metrics) > 3
     """
-    # Initial errors
-    with pytest.raises(ValueError):
-        sc._binary_target([], [], ["A", 1, 2, 3])
-    with pytest.raises(ValueError):
-        sc._binary_target([], [], ["A", "B"])
-    with pytest.raises(ValueError):
-        sc._binary_target(["A", "B", "C", "D"], [], [])
-
     # # metrics == 1
     metrics = "Classification"
     sc._binary_target(metrics, ["A", "B"], [""], h2o_model=True)
@@ -504,6 +622,20 @@ def test_binary_target():
         sc._binary_target(metrics, ["A", "B"], ["1", 2, 3])
 
 
+class TestNonbinaryTargets(unittest.TestCase):
+    def setUp(self):
+        self.sc = ScoreCode
+
+    def tearDown(self):
+        self.sc.score_code = ""
+
+    def execute_snippet(self, *args):
+        scope = {}
+        exec(self.sc.score_code, scope)
+        test_snippet = scope["test_snippet"]
+        return test_snippet(*args)
+
+
 def test_nonbinary_targets():
     """
     Test Cases: