Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/tensorflow_addons/metrics/cohens_kappa.py: 28%
89 statements
« prev ^ index » next coverage.py v7.4.0, created at 2024-01-03 07:57 +0000
1# Copyright 2019 The TensorFlow Authors. All Rights Reserved.
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14# ==============================================================================
15"""Implements Cohen's Kappa."""
17import tensorflow as tf
18import numpy as np
19import tensorflow.keras.backend as K
20from tensorflow.keras.metrics import Metric
21from tensorflow_addons.utils.types import AcceptableDTypes, FloatTensorLike
23from typeguard import typechecked
24from typing import Optional
27@tf.keras.utils.register_keras_serializable(package="Addons")
28class CohenKappa(Metric):
29 """Computes Kappa score between two raters.
31 The score lies in the range `[-1, 1]`. A score of -1 represents
32 complete disagreement between two raters whereas a score of 1
33 represents complete agreement between the two raters.
34 A score of 0 means agreement by chance.
36 Note: As of now, this implementation considers all labels
37 while calculating the Cohen's Kappa score.
39 Args:
40 num_classes: Number of unique classes in your dataset.
41 weightage: (optional) Weighting to be considered for calculating
42 kappa statistics. A valid value is one of
43 [None, 'linear', 'quadratic']. Defaults to `None`
44 sparse_labels: (bool) Valid only for multi-class scenario.
45 If True, ground truth labels are expected to be integers
46 and not one-hot encoded.
47 regression: (bool) If set, that means the problem is being treated
48 as a regression problem where you are regressing the predictions.
49 **Note:** If you are regressing for the values, the the output layer
50 should contain a single unit.
51 name: (optional) String name of the metric instance
52 dtype: (optional) Data type of the metric result. Defaults to `None`.
54 Raises:
55 ValueError: If the value passed for `weightage` is invalid
56 i.e. not any one of [None, 'linear', 'quadratic'].
58 Usage:
60 >>> y_true = np.array([4, 4, 3, 4, 2, 4, 1, 1], dtype=np.int32)
61 >>> y_pred = np.array([4, 4, 3, 4, 4, 2, 1, 1], dtype=np.int32)
62 >>> weights = np.array([1, 1, 2, 5, 10, 2, 3, 3], dtype=np.int32)
63 >>> metric = tfa.metrics.CohenKappa(num_classes=5, sparse_labels=True)
64 >>> metric.update_state(y_true , y_pred)
65 <tf.Tensor: shape=(5, 5), dtype=float32, numpy=
66 array([[0., 0., 0., 0., 0.],
67 [0., 2., 0., 0., 0.],
68 [0., 0., 0., 0., 1.],
69 [0., 0., 0., 1., 0.],
70 [0., 0., 1., 0., 3.]], dtype=float32)>
71 >>> result = metric.result()
72 >>> result.numpy()
73 0.61904764
74 >>> # To use this with weights, sample_weight argument can be used.
75 >>> metric = tfa.metrics.CohenKappa(num_classes=5, sparse_labels=True)
76 >>> metric.update_state(y_true , y_pred , sample_weight=weights)
77 <tf.Tensor: shape=(5, 5), dtype=float32, numpy=
78 array([[ 0., 0., 0., 0., 0.],
79 [ 0., 6., 0., 0., 0.],
80 [ 0., 0., 0., 0., 10.],
81 [ 0., 0., 0., 2., 0.],
82 [ 0., 0., 2., 0., 7.]], dtype=float32)>
83 >>> result = metric.result()
84 >>> result.numpy()
85 0.37209308
87 Usage with `tf.keras` API:
89 >>> inputs = tf.keras.Input(shape=(10,))
90 >>> x = tf.keras.layers.Dense(10)(inputs)
91 >>> outputs = tf.keras.layers.Dense(1)(x)
92 >>> model = tf.keras.models.Model(inputs=inputs, outputs=outputs)
93 >>> model.compile('sgd', loss='mse', metrics=[tfa.metrics.CohenKappa(num_classes=3, sparse_labels=True)])
94 """
96 @typechecked
97 def __init__(
98 self,
99 num_classes: FloatTensorLike,
100 name: str = "cohen_kappa",
101 weightage: Optional[str] = None,
102 sparse_labels: bool = False,
103 regression: bool = False,
104 dtype: AcceptableDTypes = None,
105 ):
106 """Creates a `CohenKappa` instance."""
107 super().__init__(name=name, dtype=dtype)
109 if weightage not in (None, "linear", "quadratic"):
110 raise ValueError("Unknown kappa weighting type.")
112 if num_classes == 2:
113 self._update = self._update_binary_class_model
114 elif num_classes > 2:
115 self._update = self._update_multi_class_model
116 else:
117 raise ValueError(
118 """Number of classes must be
119 greater than or euqal to two"""
120 )
122 self.weightage = weightage
123 self.num_classes = num_classes
124 self.regression = regression
125 self.sparse_labels = sparse_labels
126 self.conf_mtx = self.add_weight(
127 "conf_mtx",
128 shape=(self.num_classes, self.num_classes),
129 initializer=tf.keras.initializers.zeros,
130 dtype=tf.float32,
131 )
133 def update_state(self, y_true, y_pred, sample_weight=None):
134 """Accumulates the confusion matrix condition statistics.
136 Args:
137 y_true: Labels assigned by the first annotator with shape
138 `[num_samples,]`.
139 y_pred: Labels assigned by the second annotator with shape
140 `[num_samples,]`. The kappa statistic is symmetric,
141 so swapping `y_true` and `y_pred` doesn't change the value.
142 sample_weight (optional): for weighting labels in confusion matrix
143 Defaults to `None`. The dtype for weights should be the same
144 as the dtype for confusion matrix. For more details,
145 please check `tf.math.confusion_matrix`.
147 Returns:
148 Update op.
149 """
150 return self._update(y_true, y_pred, sample_weight)
152 def _update_binary_class_model(self, y_true, y_pred, sample_weight=None):
153 y_true = tf.cast(y_true, dtype=tf.int64)
154 y_pred = tf.cast(y_pred, dtype=tf.float32)
155 y_pred = tf.cast(y_pred > 0.5, dtype=tf.int64)
156 return self._update_confusion_matrix(y_true, y_pred, sample_weight)
158 @tf.function
159 def _update_multi_class_model(self, y_true, y_pred, sample_weight=None):
160 v = tf.argmax(y_true, axis=1) if not self.sparse_labels else y_true
161 y_true = tf.cast(v, dtype=tf.int64)
163 y_pred = self._cast_ypred(y_pred)
165 return self._update_confusion_matrix(y_true, y_pred, sample_weight)
167 @tf.function
168 def _cast_ypred(self, y_pred):
169 if tf.rank(y_pred) > 1:
170 if not self.regression:
171 y_pred = tf.cast(tf.argmax(y_pred, axis=-1), dtype=tf.int64)
172 else:
173 y_pred = tf.math.round(tf.math.abs(y_pred))
174 y_pred = tf.cast(y_pred, dtype=tf.int64)
175 else:
176 y_pred = tf.cast(y_pred, dtype=tf.int64)
177 return y_pred
179 @tf.function
180 def _safe_squeeze(self, y):
181 y = tf.squeeze(y)
183 # Check for scalar result
184 if tf.rank(y) == 0:
185 y = tf.expand_dims(y, 0)
187 return y
189 def _update_confusion_matrix(self, y_true, y_pred, sample_weight):
190 y_true = self._safe_squeeze(y_true)
191 y_pred = self._safe_squeeze(y_pred)
193 new_conf_mtx = tf.math.confusion_matrix(
194 labels=y_true,
195 predictions=y_pred,
196 num_classes=self.num_classes,
197 weights=sample_weight,
198 dtype=tf.float32,
199 )
201 return self.conf_mtx.assign_add(new_conf_mtx)
203 def result(self):
204 nb_ratings = tf.shape(self.conf_mtx)[0]
205 weight_mtx = tf.ones([nb_ratings, nb_ratings], dtype=tf.float32)
207 # 2. Create a weight matrix
208 if self.weightage is None:
209 diagonal = tf.zeros([nb_ratings], dtype=tf.float32)
210 weight_mtx = tf.linalg.set_diag(weight_mtx, diagonal=diagonal)
211 else:
212 weight_mtx += tf.cast(tf.range(nb_ratings), dtype=tf.float32)
213 weight_mtx = tf.cast(weight_mtx, dtype=self.dtype)
215 if self.weightage == "linear":
216 weight_mtx = tf.abs(weight_mtx - tf.transpose(weight_mtx))
217 else:
218 weight_mtx = tf.pow((weight_mtx - tf.transpose(weight_mtx)), 2)
220 weight_mtx = tf.cast(weight_mtx, dtype=self.dtype)
222 # 3. Get counts
223 actual_ratings_hist = tf.reduce_sum(self.conf_mtx, axis=1)
224 pred_ratings_hist = tf.reduce_sum(self.conf_mtx, axis=0)
226 # 4. Get the outer product
227 out_prod = pred_ratings_hist[..., None] * actual_ratings_hist[None, ...]
229 # 5. Normalize the confusion matrix and outer product
230 conf_mtx = self.conf_mtx / tf.reduce_sum(self.conf_mtx)
231 out_prod = out_prod / tf.reduce_sum(out_prod)
233 conf_mtx = tf.cast(conf_mtx, dtype=self.dtype)
234 out_prod = tf.cast(out_prod, dtype=self.dtype)
236 # 6. Calculate Kappa score
237 numerator = tf.reduce_sum(conf_mtx * weight_mtx)
238 denominator = tf.reduce_sum(out_prod * weight_mtx)
239 return tf.cond(
240 tf.math.is_nan(denominator),
241 true_fn=lambda: 0.0,
242 false_fn=lambda: 1 - (numerator / denominator),
243 )
245 def get_config(self):
246 """Returns the serializable config of the metric."""
248 config = {
249 "num_classes": self.num_classes,
250 "weightage": self.weightage,
251 "sparse_labels": self.sparse_labels,
252 "regression": self.regression,
253 }
254 base_config = super().get_config()
255 return {**base_config, **config}
257 def reset_state(self):
258 """Resets all of the metric state variables."""
260 for v in self.variables:
261 K.set_value(
262 v,
263 np.zeros((self.num_classes, self.num_classes), v.dtype.as_numpy_dtype),
264 )
266 def reset_states(self):
267 # Backwards compatibility alias of `reset_state`. New classes should
268 # only implement `reset_state`.
269 # Required in Tensorflow < 2.5.0
270 return self.reset_state()