Skip to content

rslearn.train.tasks.classification

classification

Classification task.

ClassificationTask

Bases: BasicTask

A window classification task.

Source code in rslearn/train/tasks/classification.py 
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
class ClassificationTask(BasicTask):
    """A window classification task."""

    def __init__(
        self,
        property_name: str,
        classes: list[str],
        filters: list[tuple[str, str]] = [],
        read_class_id: bool = False,
        allow_invalid: bool = False,
        skip_unknown_categories: bool = False,
        prob_property: str | None = None,
        metric_kwargs: dict[str, Any] = {},
        enable_f1_metric: bool = False,
        f1_metric_kwargs: dict[str, Any] = {},
        positive_class: str | None = None,
        positive_class_threshold: float = 0.5,
        enable_confusion_matrix: bool = False,
        **kwargs: Any,
    ):
        """Initialize a new ClassificationTask.

        Args:
            property_name: the property from which to extract the class name. The class
                is read from the first matching feature.
            classes: a list of class names.
            filters: optional list of (property_name, property_value) to only consider
                features with matching properties.
            read_class_id: whether to read an integer class ID instead of the class
                name.
            allow_invalid: instead of throwing error when no classification label is
                found at a window, simply mark the example invalid for this task
            skip_unknown_categories: whether to skip examples with categories that are
                not passed via classes, instead of throwing error
            prob_property: when predicting, write probabilities in addition to class ID
                under this property name.
            metric_kwargs: additional arguments to pass to underlying metric, see
                torchmetrics.classification.MulticlassAccuracy.
            enable_f1_metric: whether to compute F1 (default false)
            f1_metric_kwargs: extra arguments to pass to F1 metric.
            positive_class: positive class name.
            positive_class_threshold: threshold for classifying the positive class in
                binary classification (default 0.5).
            enable_confusion_matrix: whether to compute confusion matrix (default false).
                If true, it requires wandb to be initialized for logging.
            kwargs: other arguments to pass to BasicTask
        """
        super().__init__(**kwargs)
        self.property_name = property_name
        self.classes = classes
        self.filters = filters
        self.read_class_id = read_class_id
        self.allow_invalid = allow_invalid
        self.skip_unknown_categories = skip_unknown_categories
        self.prob_property = prob_property
        self.metric_kwargs = metric_kwargs
        self.enable_f1_metric = enable_f1_metric
        self.f1_metric_kwargs = f1_metric_kwargs
        self.positive_class = positive_class
        self.positive_class_threshold = positive_class_threshold
        self.enable_confusion_matrix = enable_confusion_matrix

        if self.positive_class_threshold != 0.5:
            # Must be binary classification
            if len(self.classes) != 2:
                raise ValueError(
                    "positive_class_threshold is only applicable for binary classification."
                )
            if positive_class is None:
                raise ValueError(
                    "Please set positive_class when using positive_class_threshold."
                )
            elif self.positive_class not in self.classes:
                raise ValueError(
                    f"Positive class '{self.positive_class}' not found in classes {self.classes}."
                )
            else:
                self.positive_class_id = self.classes.index(self.positive_class)

    def process_inputs(
        self,
        raw_inputs: dict[str, RasterImage | list[Feature]],
        metadata: SampleMetadata,
        load_targets: bool = True,
    ) -> tuple[dict[str, Any], dict[str, Any]]:
        """Processes the data into targets.

        Args:
            raw_inputs: raster or vector data to process
            metadata: metadata about the patch being read
            load_targets: whether to load the targets or only inputs

        Returns:
            tuple (input_dict, target_dict) containing the processed inputs and targets
                that are compatible with both metrics and loss functions
        """
        if not load_targets:
            return {}, {}

        data = raw_inputs["targets"]
        assert isinstance(data, list)
        for feat in data:
            if feat.properties is None:
                continue
            for property_name, property_value in self.filters:
                if feat.properties.get(property_name) != property_value:
                    continue
            if self.property_name not in feat.properties:
                continue

            v = feat.properties[self.property_name]
            if self.read_class_id:
                class_id = int(v)
            else:
                if v in self.classes:
                    class_id = self.classes.index(v)
                else:
                    class_id = -1

            if class_id < 0 or class_id >= len(self.classes):
                # Throw error if unknown categories are not acceptable.
                assert self.skip_unknown_categories
                # Otherwise, skip this example.
                continue

            return {}, {
                "class": torch.tensor(class_id, dtype=torch.int64),
                "valid": torch.tensor(1, dtype=torch.float32),
            }

        if not self.allow_invalid:
            raise Exception("no feature found providing class label")

        return {}, {
            "class": torch.tensor(0, dtype=torch.int64),
            "valid": torch.tensor(0, dtype=torch.float32),
        }

    def process_output(
        self, raw_output: Any, metadata: SampleMetadata
    ) -> list[Feature]:
        """Processes an output into raster or vector data.

        Args:
            raw_output: the output from prediction head, which must be a tensor
                containing output probabilities (one dimension).
            metadata: metadata about the patch being read

        Returns:
            a list with one Feature corresponding to the input patch extent with a
                property name containing the predicted class. It will have another
                property containing the probabilities if prob_property was set.
        """
        if not isinstance(raw_output, torch.Tensor) or len(raw_output.shape) != 1:
            raise ValueError(
                "expected output for ClassificationTask to be a Tensor with one dimension"
            )

        probs = raw_output.cpu().numpy()
        if len(self.classes) == 2 and self.positive_class_threshold != 0.5:
            positive_class_prob = probs[self.positive_class_id]
            if positive_class_prob >= self.positive_class_threshold:
                class_idx = self.positive_class_id
            else:
                class_idx = 1 - self.positive_class_id
        else:
            # For multiclass classification or when using the default threshold
            class_idx = probs.argmax().item()

        value: str | int
        if not self.read_class_id:
            value = self.classes[class_idx]  # type: ignore
        else:
            value = class_idx

        feature = Feature(
            STGeometry(
                metadata.projection,
                shapely.Point(metadata.crop_bounds[0], metadata.crop_bounds[1]),
                None,
            ),
            {
                self.property_name: value,
            },
        )
        if self.prob_property is not None and feature.properties is not None:
            feature.properties[self.prob_property] = probs.tolist()
        return [feature]

    def visualize(
        self,
        input_dict: dict[str, Any],
        target_dict: dict[str, Any] | None,
        output: Any,
    ) -> dict[str, npt.NDArray[Any]]:
        """Visualize the outputs and targets.

        Args:
            input_dict: the input dict from process_inputs
            target_dict: the target dict from process_inputs
            output: the prediction

        Returns:
            a dictionary mapping image name to visualization image
        """
        image = super().visualize(input_dict, target_dict, output)["image"]
        image = Image.fromarray(image)
        draw = ImageDraw.Draw(image)
        if target_dict is None:
            raise ValueError("target_dict is required for visualization")
        target_class = self.classes[target_dict["class"]]
        output_class = self.classes[output.argmax()]
        text = f"Label: {target_class}\nOutput: {output_class}"
        box = draw.textbbox(xy=(0, 0), text=text, font_size=12)
        draw.rectangle(xy=box, fill=(0, 0, 0))
        draw.text(xy=(0, 0), text=text, font_size=12, fill=(255, 255, 255))
        return {
            "image": np.array(image),
        }

    def get_metrics(self) -> MetricCollection:
        """Get the metrics for this task."""
        metrics = {}
        metric_kwargs = {"num_classes": len(self.classes)}
        metric_kwargs.update(self.metric_kwargs)
        metrics["accuracy"] = ClassificationMetric(MulticlassAccuracy(**metric_kwargs))

        if self.enable_f1_metric:
            kwargs = {
                "num_classes": len(self.classes),
                "average": None,
            }  # Default to per-class metrics
            kwargs.update(self.f1_metric_kwargs)

            # Check the 'average' parameter
            average_mode = kwargs.get("average", None)
            if average_mode in (None, "none"):
                # Generate per-class metrics with class-specific names
                for class_id, class_name in enumerate(self.classes):
                    metrics[f"{class_name}_recall"] = ClassificationMetric(
                        MulticlassRecall(**kwargs), class_id
                    )
                    metrics[f"{class_name}_precision"] = ClassificationMetric(
                        MulticlassPrecision(**kwargs), class_id
                    )
                    metrics[f"{class_name}_f1"] = ClassificationMetric(
                        MulticlassF1Score(**kwargs), class_id
                    )
            else:
                metrics["recall"] = ClassificationMetric(MulticlassRecall(**kwargs))
                metrics["precision"] = ClassificationMetric(
                    MulticlassPrecision(**kwargs)
                )
                metrics["f1"] = ClassificationMetric(MulticlassF1Score(**kwargs))

        if self.enable_confusion_matrix:
            metrics["confusion_matrix"] = ClassificationMetric(
                ConfusionMatrixMetric(
                    num_classes=len(self.classes),
                    class_names=self.classes,
                ),
            )

        return MetricCollection(metrics)

process_inputs 

process_inputs(raw_inputs: dict[str, RasterImage | list[Feature]], metadata: SampleMetadata, load_targets: bool = True) -> tuple[dict[str, Any], dict[str, Any]]

Processes the data into targets.

Parameters:

Name Type Description Default
raw_inputs dict[str, RasterImage | list[Feature]]

raster or vector data to process

 required
metadata SampleMetadata

metadata about the patch being read

 required
load_targets bool

whether to load the targets or only inputs

 True

Returns:

Type Description
tuple[dict[str, Any], dict[str, Any]]

tuple (input_dict, target_dict) containing the processed inputs and targets that are compatible with both metrics and loss functions
Source code in rslearn/train/tasks/classification.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def process_inputs(
    self,
    raw_inputs: dict[str, RasterImage | list[Feature]],
    metadata: SampleMetadata,
    load_targets: bool = True,
) -> tuple[dict[str, Any], dict[str, Any]]:
    """Processes the data into targets.

    Args:
        raw_inputs: raster or vector data to process
        metadata: metadata about the patch being read
        load_targets: whether to load the targets or only inputs

    Returns:
        tuple (input_dict, target_dict) containing the processed inputs and targets
            that are compatible with both metrics and loss functions
    """
    if not load_targets:
        return {}, {}

    data = raw_inputs["targets"]
    assert isinstance(data, list)
    for feat in data:
        if feat.properties is None:
            continue
        for property_name, property_value in self.filters:
            if feat.properties.get(property_name) != property_value:
                continue
        if self.property_name not in feat.properties:
            continue

        v = feat.properties[self.property_name]
        if self.read_class_id:
            class_id = int(v)
        else:
            if v in self.classes:
                class_id = self.classes.index(v)
            else:
                class_id = -1

        if class_id < 0 or class_id >= len(self.classes):
            # Throw error if unknown categories are not acceptable.
            assert self.skip_unknown_categories
            # Otherwise, skip this example.
            continue

        return {}, {
            "class": torch.tensor(class_id, dtype=torch.int64),
            "valid": torch.tensor(1, dtype=torch.float32),
        }

    if not self.allow_invalid:
        raise Exception("no feature found providing class label")

    return {}, {
        "class": torch.tensor(0, dtype=torch.int64),
        "valid": torch.tensor(0, dtype=torch.float32),
    }

process_output 

process_output(raw_output: Any, metadata: SampleMetadata) -> list[Feature]

Processes an output into raster or vector data.

Parameters:

Name Type Description Default
raw_output Any

the output from prediction head, which must be a tensor containing output probabilities (one dimension).

 required
metadata SampleMetadata

metadata about the patch being read

 required

Returns:

Type Description
list[Feature]

a list with one Feature corresponding to the input patch extent with a property name containing the predicted class. It will have another property containing the probabilities if prob_property was set.
Source code in rslearn/train/tasks/classification.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def process_output(
    self, raw_output: Any, metadata: SampleMetadata
) -> list[Feature]:
    """Processes an output into raster or vector data.

    Args:
        raw_output: the output from prediction head, which must be a tensor
            containing output probabilities (one dimension).
        metadata: metadata about the patch being read

    Returns:
        a list with one Feature corresponding to the input patch extent with a
            property name containing the predicted class. It will have another
            property containing the probabilities if prob_property was set.
    """
    if not isinstance(raw_output, torch.Tensor) or len(raw_output.shape) != 1:
        raise ValueError(
            "expected output for ClassificationTask to be a Tensor with one dimension"
        )

    probs = raw_output.cpu().numpy()
    if len(self.classes) == 2 and self.positive_class_threshold != 0.5:
        positive_class_prob = probs[self.positive_class_id]
        if positive_class_prob >= self.positive_class_threshold:
            class_idx = self.positive_class_id
        else:
            class_idx = 1 - self.positive_class_id
    else:
        # For multiclass classification or when using the default threshold
        class_idx = probs.argmax().item()

    value: str | int
    if not self.read_class_id:
        value = self.classes[class_idx]  # type: ignore
    else:
        value = class_idx

    feature = Feature(
        STGeometry(
            metadata.projection,
            shapely.Point(metadata.crop_bounds[0], metadata.crop_bounds[1]),
            None,
        ),
        {
            self.property_name: value,
        },
    )
    if self.prob_property is not None and feature.properties is not None:
        feature.properties[self.prob_property] = probs.tolist()
    return [feature]

visualize 

visualize(input_dict: dict[str, Any], target_dict: dict[str, Any] | None, output: Any) -> dict[str, NDArray[Any]]

Visualize the outputs and targets.

Parameters:

Name Type Description Default
input_dict dict[str, Any]

the input dict from process_inputs

 required
target_dict dict[str, Any] | None

the target dict from process_inputs

 required
output Any

the prediction

 required

Returns:

Type Description
dict[str, NDArray[Any]]

a dictionary mapping image name to visualization image
Source code in rslearn/train/tasks/classification.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def visualize(
    self,
    input_dict: dict[str, Any],
    target_dict: dict[str, Any] | None,
    output: Any,
) -> dict[str, npt.NDArray[Any]]:
    """Visualize the outputs and targets.

    Args:
        input_dict: the input dict from process_inputs
        target_dict: the target dict from process_inputs
        output: the prediction

    Returns:
        a dictionary mapping image name to visualization image
    """
    image = super().visualize(input_dict, target_dict, output)["image"]
    image = Image.fromarray(image)
    draw = ImageDraw.Draw(image)
    if target_dict is None:
        raise ValueError("target_dict is required for visualization")
    target_class = self.classes[target_dict["class"]]
    output_class = self.classes[output.argmax()]
    text = f"Label: {target_class}\nOutput: {output_class}"
    box = draw.textbbox(xy=(0, 0), text=text, font_size=12)
    draw.rectangle(xy=box, fill=(0, 0, 0))
    draw.text(xy=(0, 0), text=text, font_size=12, fill=(255, 255, 255))
    return {
        "image": np.array(image),
    }

get_metrics 

get_metrics() -> MetricCollection

Get the metrics for this task.

Source code in rslearn/train/tasks/classification.py 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
def get_metrics(self) -> MetricCollection:
    """Get the metrics for this task."""
    metrics = {}
    metric_kwargs = {"num_classes": len(self.classes)}
    metric_kwargs.update(self.metric_kwargs)
    metrics["accuracy"] = ClassificationMetric(MulticlassAccuracy(**metric_kwargs))

    if self.enable_f1_metric:
        kwargs = {
            "num_classes": len(self.classes),
            "average": None,
        }  # Default to per-class metrics
        kwargs.update(self.f1_metric_kwargs)

        # Check the 'average' parameter
        average_mode = kwargs.get("average", None)
        if average_mode in (None, "none"):
            # Generate per-class metrics with class-specific names
            for class_id, class_name in enumerate(self.classes):
                metrics[f"{class_name}_recall"] = ClassificationMetric(
                    MulticlassRecall(**kwargs), class_id
                )
                metrics[f"{class_name}_precision"] = ClassificationMetric(
                    MulticlassPrecision(**kwargs), class_id
                )
                metrics[f"{class_name}_f1"] = ClassificationMetric(
                    MulticlassF1Score(**kwargs), class_id
                )
        else:
            metrics["recall"] = ClassificationMetric(MulticlassRecall(**kwargs))
            metrics["precision"] = ClassificationMetric(
                MulticlassPrecision(**kwargs)
            )
            metrics["f1"] = ClassificationMetric(MulticlassF1Score(**kwargs))

    if self.enable_confusion_matrix:
        metrics["confusion_matrix"] = ClassificationMetric(
            ConfusionMatrixMetric(
                num_classes=len(self.classes),
                class_names=self.classes,
            ),
        )

    return MetricCollection(metrics)

ClassificationHead

Bases: Predictor

Head for classification task.

Source code in rslearn/train/tasks/classification.py 
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
class ClassificationHead(Predictor):
    """Head for classification task."""

    def forward(
        self,
        intermediates: Any,
        context: ModelContext,
        targets: list[dict[str, Any]] | None = None,
    ) -> ModelOutput:
        """Compute the classification outputs and loss from logits and targets.

        Args:
            intermediates: output from the previous model component, it should be a
                FeatureVector with a tensor that is (BatchSize, NumClasses) in shape.
            context: the model context.
            targets: must contain "class" key that stores the class label, along with
                "valid" key indicating whether the label is valid for each example.

        Returns:
            tuple of outputs and loss dict
        """
        if not isinstance(intermediates, FeatureVector):
            raise ValueError("the input to ClassificationHead must be a FeatureVector")

        logits = intermediates.feature_vector
        outputs = torch.nn.functional.softmax(logits, dim=1)

        losses = {}
        if targets:
            class_labels = torch.stack([target["class"] for target in targets], dim=0)
            mask = torch.stack([target["valid"] for target in targets], dim=0)
            loss = (
                torch.nn.functional.cross_entropy(
                    logits, class_labels, reduction="none"
                )
                * mask
            )
            losses["cls"] = torch.mean(loss)

        return ModelOutput(
            outputs=outputs,
            loss_dict=losses,
        )

forward 

forward(intermediates: Any, context: ModelContext, targets: list[dict[str, Any]] | None = None) -> ModelOutput

Compute the classification outputs and loss from logits and targets.

Parameters:

Name Type Description Default
intermediates Any

output from the previous model component, it should be a FeatureVector with a tensor that is (BatchSize, NumClasses) in shape.

 required
context ModelContext

the model context.

 required
targets list[dict[str, Any]] | None

must contain "class" key that stores the class label, along with "valid" key indicating whether the label is valid for each example.

 None

Returns:

Type Description
ModelOutput

tuple of outputs and loss dict
Source code in rslearn/train/tasks/classification.py 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
def forward(
    self,
    intermediates: Any,
    context: ModelContext,
    targets: list[dict[str, Any]] | None = None,
) -> ModelOutput:
    """Compute the classification outputs and loss from logits and targets.

    Args:
        intermediates: output from the previous model component, it should be a
            FeatureVector with a tensor that is (BatchSize, NumClasses) in shape.
        context: the model context.
        targets: must contain "class" key that stores the class label, along with
            "valid" key indicating whether the label is valid for each example.

    Returns:
        tuple of outputs and loss dict
    """
    if not isinstance(intermediates, FeatureVector):
        raise ValueError("the input to ClassificationHead must be a FeatureVector")

    logits = intermediates.feature_vector
    outputs = torch.nn.functional.softmax(logits, dim=1)

    losses = {}
    if targets:
        class_labels = torch.stack([target["class"] for target in targets], dim=0)
        mask = torch.stack([target["valid"] for target in targets], dim=0)
        loss = (
            torch.nn.functional.cross_entropy(
                logits, class_labels, reduction="none"
            )
            * mask
        )
        losses["cls"] = torch.mean(loss)

    return ModelOutput(
        outputs=outputs,
        loss_dict=losses,
    )

ClassificationMetric

Bases: Metric

Metric for classification task.

Source code in rslearn/train/tasks/classification.py 
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
class ClassificationMetric(Metric):
    """Metric for classification task."""

    def __init__(self, metric: Metric, class_idx: int | None = None):
        """Initialize a new ClassificationMetric.

        Args:
            metric: the metric to wrap
            class_idx: optional class index to return metric value for. If set, the
                metric should return a dict from class index to metric value.
        """
        super().__init__()
        self.metric = metric
        self.class_idx = class_idx

    def update(
        self, preds: list[Any] | torch.Tensor, targets: list[dict[str, Any]]
    ) -> None:
        """Update metric.

        Args:
            preds: the predictions
            targets: the targets
        """
        if not isinstance(preds, torch.Tensor):
            preds = torch.stack(preds)
        labels = torch.stack([target["class"] for target in targets])

        # Sub-select the valid labels.
        mask = torch.stack([target["valid"] > 0 for target in targets])
        preds = preds[mask]
        labels = labels[mask]
        if len(preds) == 0:
            return

        self.metric.update(preds, labels)

    def compute(self) -> Any:
        """Returns the computed metric."""
        result = self.metric.compute()
        if self.class_idx is None:
            return result
        else:
            return result[self.class_idx]

    def reset(self) -> None:
        """Reset metric."""
        super().reset()
        self.metric.reset()

    def plot(self, *args: list[Any], **kwargs: dict[str, Any]) -> Any:
        """Returns a plot of the metric."""
        return self.metric.plot(*args, **kwargs)

update 

update(preds: list[Any] | Tensor, targets: list[dict[str, Any]]) -> None

Update metric.

Parameters:

Name Type Description Default
preds list[Any] | Tensor

the predictions

 required
targets list[dict[str, Any]]

the targets

 required
Source code in rslearn/train/tasks/classification.py 
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
def update(
    self, preds: list[Any] | torch.Tensor, targets: list[dict[str, Any]]
) -> None:
    """Update metric.

    Args:
        preds: the predictions
        targets: the targets
    """
    if not isinstance(preds, torch.Tensor):
        preds = torch.stack(preds)
    labels = torch.stack([target["class"] for target in targets])

    # Sub-select the valid labels.
    mask = torch.stack([target["valid"] > 0 for target in targets])
    preds = preds[mask]
    labels = labels[mask]
    if len(preds) == 0:
        return

    self.metric.update(preds, labels)

compute 

compute() -> Any

Returns the computed metric.

Source code in rslearn/train/tasks/classification.py 
379
380
381
382
383
384
385
def compute(self) -> Any:
    """Returns the computed metric."""
    result = self.metric.compute()
    if self.class_idx is None:
        return result
    else:
        return result[self.class_idx]

reset 

reset() -> None

Reset metric.

Source code in rslearn/train/tasks/classification.py 
387
388
389
390
def reset(self) -> None:
    """Reset metric."""
    super().reset()
    self.metric.reset()

plot 

plot(*args: list[Any], **kwargs: dict[str, Any]) -> Any

Returns a plot of the metric.

Source code in rslearn/train/tasks/classification.py 
392
393
394
def plot(self, *args: list[Any], **kwargs: dict[str, Any]) -> Any:
    """Returns a plot of the metric."""
    return self.metric.plot(*args, **kwargs)