docstr: Runs CQAT end to end for the model that is clustered per channel.
docstr_tok: ["Runs","CQAT","end","to","end","for","the","model","that","is","clustered","per","channel","."]
code: def testEndToEndClusterPreserveQATClusteredPerChannel(
      self, data_format="channels_last"
  ):
      """Runs CQAT end to end for the model that is clustered per channel."""
  
      nr_of_channels = 12
      nr_of_clusters = 4
  
      clustered_model = self._get_conv_clustered_model(
          nr_of_channels,
          nr_of_clusters,
          data_format,
          preserve_sparsity=False,
      )
      stripped_model = cluster.strip_clustering(clustered_model)
  
      # Save the kernel weights
      conv2d_layer = stripped_model.layers[2]
      self.assertEqual(conv2d_layer.name, "conv2d")
  
      # should be nr_of_channels * nr_of_clusters
      nr_unique_weights = -1
  
      for weight in conv2d_layer.weights:
          if "kernel" in weight.name:
              nr_unique_weights = len(np.unique(weight.numpy()))
              self.assertLessEqual(
                  nr_unique_weights, nr_of_clusters * nr_of_channels
              )
  
      quant_aware_annotate_model = quantize.quantize_annotate_model(
          stripped_model
      )
  
      quant_aware_model = quantize.quantize_apply(
          quant_aware_annotate_model,
          scheme=default_8bit_cluster_preserve_quantize_scheme.Default8BitClusterPreserveQuantizeScheme(),
      )
  
      # Lets train for more epochs to have a chance to scatter clusters
      model = self._compile_and_fit_conv_model(quant_aware_model, 3)
  
      stripped_cqat_model = strip_clustering_cqat(model)
  
      # Check the unique weights of a certain layer of
      # clustered_model and pcqat_model
      layer_nr = 3
      num_of_unique_weights_cqat = self._get_number_of_unique_weights(
          stripped_cqat_model, layer_nr, "kernel"
      )
      self.assertLessEqual(
          num_of_unique_weights_cqat, nr_unique_weights
      )
  
      # We need to do tighter check: we check that the number of unique
      # weights per channel is less than the given nr_of_channels
      layer = stripped_cqat_model.layers[layer_nr]
      weight_to_check = None
      if isinstance(layer, quantize.quantize_wrapper.QuantizeWrapper):
          for weight_item in layer.trainable_weights:
              if "kernel" in weight_item.name:
                  weight_to_check = weight_item
  
      assert weight_to_check is not None
  
      for i in range(nr_of_channels):
          nr_unique_weights_per_channel = len(
              np.unique(weight_to_check[:, :, :, i])
          )
          assert nr_unique_weights_per_channel == nr_of_clusters
  
imports: from absl.testing import parameterized
  import numpy
  import tensorflow
  from tensorflow_model_optimization.python.core.clustering.keras import cluster
  from tensorflow_model_optimization.python.core.clustering.keras import cluster_config
  from tensorflow_model_optimization.python.core.clustering.keras.experimental import cluster
  from tensorflow_model_optimization.python.core.quantization.keras import quantize
  from tensorflow_model_optimization.python.core.quantization.keras.collab_opts.cluster_preserve import default_8bit_cluster_preserve_quantize_scheme
  from tensorflow_model_optimization.python.core.quantization.keras.collab_opts.cluster_preserve.cluster_utils import strip_clustering_cqat
label_desc: Use image node_id 14 for calling the ClusterPreserveIntegrationTest obj's underlying member method code with example usage: obj.testEndToEndClusterPreserveQATClusteredPerChannel(data_format) without return types