code: #  Copyright (c) 2021, Apple Inc. All rights reserved.
  #
  #  Use of this source code is governed by a BSD-3-clause license that can be
  #  found in the LICENSE.txt file or at https://opensource.org/licenses/BSD-3-Clause
  
  import numpy as np
  
  from coremltools.converters.mil.mil.passes.pass_registry import register_pass
  from coremltools.converters.mil.mil.passes.graph_pass import AbstractGraphPass
  from coremltools.converters.mil.mil import Builder as mb
  
  
  def _try_to_transform(conv_op, scale_op, block):
  
      # get the scale
      if scale_op.x.val is None and scale_op.y.val is None:
          return False
      scale_var = scale_op.x if scale_op.x.val is not None else scale_op.y
      scale = scale_var.val
  
      # for the scalar case, the scalar can be either
      # 1. a python int/float
      # 2. a 0d numpy array
      # 3. a 1d numpy array with shape (1,)
  
      is_scalar = True
      if isinstance(scale, np.ndarray):
          if scale.shape == ():
              scale = scale.tolist()
          elif scale.shape == (1) or scale.shape == (1,):
              scale = scale[0]
          else:
              is_scalar = False
  
      # get weight and bias and groups from conv layer
      if conv_op.weight.val is None:
          return False
      conv_weight = conv_op.weight.val
      conv_bias = conv_op.bias
      groups = conv_op.groups.val
  
      # get type of the conv layer
      is_deconv = conv_op.op_type == 'conv_transpose'
      is_conv_1d = len(conv_weight.shape) == 3
  
      # D_in denotes the spatial dimensions for conv kernel weight
      # for conv_transpose, conv_weight has shape [Cin, Cout / groups, *D_in]
      # for conv, conv_weight has shape [Cout, Cin / groups, *D_in]
      if is_deconv:
          Cout = conv_weight.shape[1] * groups
          Cin = conv_weight.shape[0]
      else:
          Cout = conv_weight.shape[0]
          Cin = conv_weight.shape[1] * groups
  
      # for the vector scale case, check if the shape is broacastable
      if not is_scalar:
          if not np.product(scale.shape) == Cout:
              return False
          if len(scale.shape) == len(conv_weight.shape):
              if not scale.shape[1] == Cout:
                  return False
          elif len(scale.shape) == len(conv_weight.shape) - 1:
              if not scale.shape[0] == Cout:
                  return False
          else:
              return False
  
      # transform the scale to 1./scale for the real_div case
      if scale_op.op_type == "real_div":
          scale = 1./scale
  
      # get the type of the conv weight
      conv_weight_type = conv_weight.dtype
  
      # create bias for conv if not exist
      if conv_bias is None:
          conv_bias = np.zeros(Cout)
      else:
          conv_bias = conv_bias.val
      conv_bias = conv_bias.astype(conv_weight_type)
  
      # get the original shape of weight and bias
      origin_weight_shape = conv_weight.shape
      origin_bias_shape = conv_bias.shape
  
      # update the weight/bias for conv layer
      if is_scalar:
          new_conv_bias = np.array(conv_bias * scale).astype(conv_weight_type)
          new_conv_weight = np.array(conv_weight * scale).astype(conv_weight_type)
  
      else:
          scale = np.reshape(scale, (Cout))
          new_conv_bias = np.array(conv_bias * scale).astype(conv_weight_type)
          new_conv_weight = []
          if is_deconv:
              conv_weight = np.transpose(conv_weight, [1, 0, 2] if is_conv_1d else [1, 0, 2, 3])
              conv_weight = np.reshape(conv_weight, [Cout, Cin // groups] + list(conv_weight.shape[2:]))
  
          for i in range(Cout):
              _conv_weight = conv_weight[i] * scale[i]
              new_conv_weight.append(_conv_weight)
          new_conv_weight = np.array(new_conv_weight).astype(conv_weight_type)
  
          if is_deconv:
              new_conv_weight = np.reshape(new_conv_weight, [Cout // groups, Cin] + list(new_conv_weight.shape[2:]))
              new_conv_weight = np.transpose(new_conv_weight, [1, 0, 2] if is_conv_1d else [1, 0, 2, 3])
  
      # make sure the updated weight and bias have the same shape as the original ones
      assert new_conv_weight.shape == origin_weight_shape, "conv weight should have the same shape before and after the fuse_conv_scale pass."
      assert new_conv_bias.shape == origin_bias_shape, "conv bias should have the same shape before and after the fuse_conv_scale pass."
  
      # create a new conv op with the new weight, bias value, copying rest of the attributes
      out_name = scale_op.outputs[0].name
      conv_kargs = {"weight": new_conv_weight, "bias": new_conv_bias, "name": out_name, "before_op": conv_op}
  
      for k, v in conv_op.inputs.items():
          if k in ["weight", "bias"]:
              continue
          conv_kargs[k] = v
  
      if is_deconv:
          x = mb.conv_transpose(**conv_kargs)
      else:
          x = mb.conv(**conv_kargs)
  
      scale_op.enclosing_block.replace_uses_of_var_after_op(
          anchor_op=scale_op, old_var=scale_op.outputs[0], new_var=x
      )
      # Remove all the ops at once
      block.remove_ops([conv_op, scale_op])
      return True
  
  @register_pass(namespace="common")
  class fuse_conv_scale(AbstractGraphPass):
      """
      Fold mul/div into conv/conv_transpose by updating the weight/bias of the convolution layers.
  
      The scale const can be a single number (scalar) or a vector with a broacasable shape,
      for instance, if the output of the conv/deconv layer is (B, Cout, H, W),
      const of shape (Cout, 1, 1) and (1, Cout, 1, 1) are allowed.
  
      Given:
          %2 = conv(%1)
          ...
          %3 = mul(%2, constant) # where constant is the scale constant
          ...
  
      Result:
          %3 = conv(%1)
          ...
  
      """
      def __init__(self):
          self.ops_to_skip = set()
  
      def set_ops_to_skip(self, prog):
          pass
  
      def _fuse_conv_scale_block(self, block):
  
          def _match_pattern(op):
              if op.op_type == "conv" or op.op_type == "conv_transpose":
                  # abort fusion if op output is also a block output
                  if op.outputs[0] in op.enclosing_block.outputs:
                      return None
                  # find batch_norm op
                  child_ops = op.outputs[0].child_ops
                  if len(child_ops) == 1:
                      scale_op_candidate = list(child_ops)[0]
                      if scale_op_candidate.op_type in ["mul", "real_div"]:
                          return scale_op_candidate
              return None
  
          fusion_occurred = False
          for op in list(block.operations):
              for b in op.blocks:
                  block_changed = True
                  while block_changed:
                      block_changed = self._fuse_conv_scale_block(b)
              if len(op.blocks) > 0:
                  # This op can't be conv or conv_transpose
                  continue
  
              scale_op = _match_pattern(op)
  
              if op in self.ops_to_skip or scale_op in self.ops_to_skip:
                  continue
  
              if scale_op is not None:
                  with block:
                      fusion_occurred = _try_to_transform(op, scale_op, block)
                  # has to break as the downstream iterator is affected.
                  if fusion_occurred:
                      return fusion_occurred
          return fusion_occurred
  
      def apply(self, prog):
          self.set_ops_to_skip(prog)
          for f in prog.functions.values():
              block_changed = True
              while block_changed:
                  block_changed = self._fuse_conv_scale_block(f)
   # Copyright (c) 2017, Apple Inc. All rights reserved.
  #
  # Use of this source code is governed by a BSD-3-clause license that can be
  # found in the LICENSE.txt file or at https://opensource.org/licenses/BSD-3-Clause
  
  import pandas as pd
  import numpy as np
  import random
  import tempfile
  import unittest
  import pytest
  
  from coremltools._deps import (
      _HAS_LIBSVM,
      MSG_LIBSVM_NOT_FOUND,
      _HAS_SKLEARN,
      MSG_SKLEARN_NOT_FOUND,
  )
  from coremltools.models.utils import evaluate_regressor, _macos_version, _is_macos
  
  if _HAS_LIBSVM:
      import svmutil
      import svm
      from coremltools.converters import libsvm
  
  if _HAS_SKLEARN:
      from sklearn.svm import SVR
      from sklearn.datasets import load_boston
      from coremltools.converters import sklearn as sklearn_converter
      from sklearn.preprocessing import OneHotEncoder
  
  
  @unittest.skipIf(not _HAS_SKLEARN, MSG_SKLEARN_NOT_FOUND)
  class SvrScikitTest(unittest.TestCase):
      """
      Unit test class for testing scikit-learn sklearn_converter.
      """
  
      @classmethod
      def setUpClass(self):
          """
          Set up the unit test by loading the dataset and training a model.
          """
          if not _HAS_SKLEARN:
              return
  
          scikit_data = load_boston()
          scikit_model = SVR(kernel="linear")
          scikit_model.fit(scikit_data["data"], scikit_data["target"])
  
          # Save the data and the model
          self.scikit_data = scikit_data
          self.scikit_model = scikit_model
  
      def test_conversion_bad_inputs(self):
          # Error on converting an untrained model
          with self.assertRaises(TypeError):
              model = SVR()
              spec = sklearn_converter.convert(model, "data", "out")
  
          # Check the expected class during covnersion.
          with self.assertRaises(TypeError):
              model = OneHotEncoder()
              spec = sklearn_converter.convert(model, "data", "out")
  
      @pytest.mark.slow
      def test_evaluation_stress_test(self):
          self._test_evaluation(allow_slow=True)
  
      def test_evaluation(self):
          self._test_evaluation(allow_slow=False)
  
      def _test_evaluation(self, allow_slow):
          """
          Test that the same predictions are made
          """
  
          # Generate some smallish (some kernels take too long on anything else) random data
          x, y = [], []
          for _ in range(50):
              cur_x1, cur_x2 = random.gauss(2, 3), random.gauss(-1, 2)
              x.append([cur_x1, cur_x2])
              y.append(1 + 2 * cur_x1 + 3 * cur_x2)
  
          input_names = ["x1", "x2"]
          df = pd.DataFrame(x, columns=input_names)
  
          # Parameters to test
          kernel_parameters = [
              {},
              {"kernel": "rbf", "gamma": 1.2},
              {"kernel": "linear"},
              {"kernel": "poly"},
              {"kernel": "poly", "degree": 2},
              {"kernel": "poly", "gamma": 0.75},
              {"kernel": "poly", "degree": 0, "gamma": 0.9, "coef0": 2},
              {"kernel": "sigmoid"},
              {"kernel": "sigmoid", "gamma": 1.3},
              {"kernel": "sigmoid", "coef0": 0.8},
              {"kernel": "sigmoid", "coef0": 0.8, "gamma": 0.5},
          ]
          non_kernel_parameters = [
              {},
              {"C": 1},
              {"C": 1.5, "epsilon": 0.5, "shrinking": True},
              {"C": 0.5, "epsilon": 1.5, "shrinking": False},
          ]
  
          # Test
          for param1 in non_kernel_parameters:
              for param2 in kernel_parameters:
                  cur_params = param1.copy()
                  cur_params.update(param2)
                  print("cur_params=" + str(cur_params))
  
                  cur_model = SVR(**cur_params)
                  cur_model.fit(x, y)
                  df["prediction"] = cur_model.predict(x)
  
                  spec = sklearn_converter.convert(cur_model, input_names, "target")
  
                  if _is_macos() and _macos_version() >= (10, 13):
                      metrics = evaluate_regressor(spec, df)
                      self.assertAlmostEqual(metrics["max_error"], 0)
  
                  if not allow_slow:
                      break
  
              if not allow_slow:
                  break
  
  
  @unittest.skipIf(not _HAS_LIBSVM, MSG_LIBSVM_NOT_FOUND)
  @unittest.skipIf(not _HAS_SKLEARN, MSG_SKLEARN_NOT_FOUND)
  class EpsilonSVRLibSVMTest(unittest.TestCase):
      """
      Unit test class for testing the libsvm sklearn converter.
      """
  
      @classmethod
      def setUpClass(self):
          """
          Set up the unit test by loading the dataset and training a model.
          """
          if not _HAS_SKLEARN:
              return
          if not _HAS_LIBSVM:
              return
  
          scikit_data = load_boston()
          prob = svmutil.svm_problem(scikit_data["target"], scikit_data["data"].tolist())
          param = svmutil.svm_parameter()
          param.svm_type = svmutil.EPSILON_SVR
          param.kernel_type = svmutil.LINEAR
          param.eps = 1
  
          self.libsvm_model = svmutil.svm_train(prob, param)
  
      def test_input_names(self):
          data = load_boston()
          df = pd.DataFrame({"input": data["data"].tolist()})
          df["input"] = df["input"].apply(np.array)
  
          # Default values
          spec = libsvm.convert(self.libsvm_model)
          if _is_macos() and _macos_version() >= (10, 13):
              (df["prediction"], _, _) = svmutil.svm_predict(
                  data["target"], data["data"].tolist(), self.libsvm_model
              )
              metrics = evaluate_regressor(spec, df)
              self.assertAlmostEqual(metrics["max_error"], 0)
  
          # One extra parameters. This is legal/possible.
          num_inputs = len(data["data"][0])
          spec = libsvm.convert(self.libsvm_model, input_length=num_inputs + 1)
  
          # Not enought input names.
          input_names = ["this", "is", "not", "enought", "names"]
          with self.assertRaises(ValueError):
              libsvm.convert(self.libsvm_model, input_names=input_names)
          with self.assertRaises(ValueError):
              libsvm.convert(self.libsvm_model, input_length=num_inputs - 1)
  
      def test_conversion_from_filesystem(self):
          libsvm_model_path = tempfile.mktemp(suffix="model.libsvm")
          svmutil.svm_save_model(libsvm_model_path, self.libsvm_model)
          spec = libsvm.convert(
              libsvm_model_path, input_names="data", target_name="target"
          )
  
      def test_conversion_bad_inputs(self):
          # Check the expected class during covnersion.
          with self.assertRaises(TypeError):
              model = OneHotEncoder()
              spec = libsvm.convert(model, "data", "out")
  
      @pytest.mark.slow
      def test_evaluation_stress_test(self):
          self._test_evaluation(allow_slow=True)
  
      def test_evaluation(self):
          self._test_evaluation(allow_slow=False)
  
      def _test_evaluation(self, allow_slow):
          """
          Test that the same predictions are made
          """
          from svm import svm_parameter, svm_problem
          from svmutil import svm_train, svm_predict
  
          # Generate some smallish (poly kernels take too long on anything else) random data
          x, y = [], []
          for _ in range(50):
              cur_x1, cur_x2 = random.gauss(2, 3), random.gauss(-1, 2)
              x.append([cur_x1, cur_x2])
              y.append(1 + 2 * cur_x1 + 3 * cur_x2)
  
          input_names = ["x1", "x2"]
          df = pd.DataFrame(x, columns=input_names)
          prob = svm_problem(y, x)
  
          # Parameters
          base_param = "-s 3"  # model type is epsilon SVR
          non_kernel_parameters = ["", "-c 1.5 -p 0.5 -h 1", "-c 0.5 -p 0.5 -h 0"]
          kernel_parameters = [
              "",
              "-t 2 -g 1.2",  # rbf kernel
              "-t 0",  # linear kernel
              "-t 1",
              "-t 1 -d 2",
              "-t 1 -g 0.75",
              "-t 1 -d 0 -g 0.9 -r 2",  # poly kernel
              "-t 3",
              "-t 3 -g 1.3",
              "-t 3 -r 0.8",
              "-t 3 -r 0.8 -g 0.5",  # sigmoid kernel
          ]
  
          for param1 in non_kernel_parameters:
              for param2 in kernel_parameters:
                  param_str = " ".join([base_param, param1, param2])
                  print(param_str)
                  param = svm_parameter(param_str)
  
                  model = svm_train(prob, param)
                  (df["prediction"], _, _) = svm_predict(y, x, model)
  
                  spec = libsvm.convert(
                      model, input_names=input_names, target_name="target"
                  )
  
                  if _is_macos() and _macos_version() >= (10, 13):
                      metrics = evaluate_regressor(spec, df)
                      self.assertAlmostEqual(metrics["max_error"], 0)
  
                  if not allow_slow:
                      break
  
              if not allow_slow:
                  break
  
apis: numpy.transpose numpy.zeros numpy.reshape numpy.product numpy.array
apis: sklearn.svm.SVR sklearn.datasets.load_boston pandas.DataFrame sklearn.preprocessing.OneHotEncoder