docstr: PCQAT zero centroid masks stay the same and trainable variables are updating between epochs.
docstr_tok: ["PCQAT","zero","centroid","masks","stay","the","same","and","trainable","variables","are","updating","between","epochs","."]
code: def testTrainableWeightsBehaveCorrectlyDuringPCQAT(self):
      """PCQAT zero centroid masks stay the same and trainable variables are updating between epochs."""
      preserve_sparsity = True
      clustered_model = self._get_clustered_model(preserve_sparsity)
      clustered_model = cluster.strip_clustering(clustered_model)
  
      # Apply PCQAT
      quant_aware_annotate_model = quantize.quantize_annotate_model(
          clustered_model
      )
  
      quant_aware_model = quantize.quantize_apply(
          quant_aware_annotate_model,
          scheme=default_8bit_cluster_preserve_quantize_scheme.Default8BitClusterPreserveQuantizeScheme(
              True
          ),
      )
  
      quant_aware_model.compile(
          loss=tf.keras.losses.categorical_crossentropy,
          optimizer="adam",
          metrics=["accuracy"],
      )
  
      class CheckCentroidsAndTrainableVarsCallback(
          tf.keras.callbacks.Callback
      ):
          """Check the updates of trainable variables and centroid masks."""
  
          def on_epoch_begin(self, batch, logs=None):
              # Check cluster centroids have the zero in the right position
              vars_dictionary = self.model.layers[1]._weight_vars[0][2]
              self.centroid_mask = vars_dictionary["centroids_mask"]
              self.zero_centroid_index_begin = np.where(
                  self.centroid_mask == 0
              )[0]
  
              # Check trainable weights before training
              self.layer_kernel = (
                  self.model.layers[1].weights[3].numpy()
              )
              self.original_weight = vars_dictionary[
                  "ori_weights_vars_tf"
              ].numpy()
              self.centroids = vars_dictionary[
                  "cluster_centroids_tf"
              ].numpy()
  
          def on_epoch_end(self, batch, logs=None):
              # Check the index of the zero centroids are not changed after training
              vars_dictionary = self.model.layers[1]._weight_vars[0][2]
              self.zero_centroid_index_end = np.where(
                  vars_dictionary["centroids_mask"] == 0
              )[0]
              assert np.array_equal(
                  self.zero_centroid_index_begin,
                  self.zero_centroid_index_end,
              )
  
              # Check trainable variables after training are updated
              assert not np.array_equal(
                  self.layer_kernel,
                  self.model.layers[1].weights[3].numpy(),
              )
              assert not np.array_equal(
                  self.original_weight,
                  vars_dictionary["ori_weights_vars_tf"].numpy(),
              )
              assert not np.array_equal(
                  self.centroids,
                  vars_dictionary["cluster_centroids_tf"].numpy(),
              )
  
      # Use many epochs to verify layer's kernel weights are updating because
      # they can stay the same after being trained using only the first batch
      # of data for instance
      quant_aware_model.fit(
          np.random.rand(20, 10),
          tf.keras.utils.to_categorical(
              np.random.randint(5, size=(20, 1)), 5
          ),
          steps_per_epoch=5,
          epochs=3,
          callbacks=[CheckCentroidsAndTrainableVarsCallback()],
      )
  
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 19 for calling the ClusterPreserveIntegrationTest obj's underlying member method code with example usage: obj.testTrainableWeightsBehaveCorrectlyDuringPCQAT() without return types