Skip to content

rslearn.train.tasks.regression

regression

Regression task.

RegressionTask

Bases: BasicTask

A window regression task.

Source code in rslearn/train/tasks/regression.py 
 28
 29
 30
 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
class RegressionTask(BasicTask):
    """A window regression task."""

    def __init__(
        self,
        property_name: str,
        filters: list[tuple[str, str]] | None = None,
        allow_invalid: bool = False,
        scale_factor: float = 1,
        metric_mode: (
            Literal["mse", "rmse", "l1", "r2", "mape"]
            | Sequence[Literal["mse", "rmse", "l1", "r2", "mape"]]
            | None
        ) = None,
        use_accuracy_metric: bool = False,
        within_factor: float = 0.1,
        metrics: Sequence[str] | None = None,
        **kwargs: Any,
    ) -> None:
        """Initialize a new RegressionTask.

        Args:
            property_name: the property from which to extract the ground truth
                regression value. The value is read from the first matching feature.
            filters: optional list of (property_name, property_value) to only consider
                features with matching properties.
            allow_invalid: instead of throwing error when no regression label is found
                at a window, simply mark the example invalid for this task
            scale_factor: multiply the label value by this factor for training
            metric_mode: deprecated; use metrics instead. Will be removed after
                2026-06-01.
            use_accuracy_metric: include metric that reports percentage of
                examples where output is within a factor of the ground truth.
            within_factor: the factor for accuracy metric. If it's 0.2, and ground
                truth is 5.0, then values from 5.0*0.8 to 5.0*1.2 are accepted.
            metrics: metric(s) to compute. Supported values: "mse", "rmse", "l1",
                "r2", "mape".
            kwargs: other arguments to pass to BasicTask
        """
        super().__init__(**kwargs)
        self.property_name = property_name
        self.filters = filters
        self.allow_invalid = allow_invalid
        self.scale_factor = scale_factor

        if metrics is not None:
            metric_names = list(metrics)
            if metric_mode is not None:
                warnings.warn(
                    "RegressionTask.metric_mode is deprecated and ignored when "
                    "`metrics` is set. It will be removed after 2026-06-01.",
                    FutureWarning,
                    stacklevel=2,
                )
        elif metric_mode is not None:
            warnings.warn(
                "RegressionTask.metric_mode is deprecated; use `metrics` instead. "
                "It will be removed after 2026-06-01.",
                FutureWarning,
                stacklevel=2,
            )
            if isinstance(metric_mode, str):
                metric_names = [metric_mode]
            else:
                metric_names = list(metric_mode)
        else:
            metric_names = ["mse"]

        if len(metric_names) == 0:
            raise ValueError("metrics must contain at least one metric")
        allowed = {"mse", "rmse", "l1", "r2", "mape"}
        invalid = [m for m in metric_names if m not in allowed]
        if invalid:
            raise ValueError(f"invalid metrics entries: {invalid}")
        self.metrics = metric_names

        self.use_accuracy_metric = use_accuracy_metric
        self.within_factor = within_factor

        if not self.filters:
            self.filters = []

    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 or self.filters 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
            value = float(feat.properties[self.property_name]) * self.scale_factor
            return {}, {
                "value": torch.tensor(value, dtype=torch.float32),
                "valid": torch.tensor(1, dtype=torch.float32),
            }

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

        return {}, {
            "value": torch.tensor(0, dtype=torch.float32),
            "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 scalar tensor.
            metadata: metadata about the patch being read

        Returns:
            a list with a single Feature corresponding to the patch extent and with a
                property containing the predicted value.
        """
        if not isinstance(raw_output, torch.Tensor) or len(raw_output.shape) != 0:
            raise ValueError("output for RegressionTask must be a scalar Tensor")

        output = raw_output.item() / self.scale_factor
        feature = Feature(
            STGeometry(
                metadata.projection,
                shapely.Point(metadata.crop_bounds[0], metadata.crop_bounds[1]),
                None,
            ),
            {
                self.property_name: output,
            },
        )
        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 = target_dict["value"] / self.scale_factor
        output = output / self.scale_factor
        text = f"Label: {target:.2f}\nOutput: {output:.2f}"
        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."""
        metric_dict: dict[str, Metric] = {}

        for metric_name in self.metrics:
            if metric_name == "mse":
                metric_dict["mse"] = RegressionMetricWrapper(
                    metric=torchmetrics.MeanSquaredError(),
                    scale_factor=self.scale_factor,
                )
            elif metric_name == "rmse":
                metric_dict["rmse"] = RegressionMetricWrapper(
                    metric=torchmetrics.MeanSquaredError(squared=False),
                    scale_factor=self.scale_factor,
                )
            elif metric_name == "l1":
                metric_dict["l1"] = RegressionMetricWrapper(
                    metric=torchmetrics.MeanAbsoluteError(),
                    scale_factor=self.scale_factor,
                )
            elif metric_name == "r2":
                metric_dict["r2"] = RegressionMetricWrapper(
                    metric=torchmetrics.R2Score(),
                    scale_factor=self.scale_factor,
                )
            elif metric_name == "mape":
                metric_dict["mape"] = RegressionMetricWrapper(
                    metric=torchmetrics.MeanAbsolutePercentageError(),
                    scale_factor=self.scale_factor,
                )
            else:
                raise ValueError(f"unknown metric {metric_name}")

        if self.use_accuracy_metric:
            metric_dict["accuracy"] = RegressionMetricWrapper(
                metric=RegressionAccuracy(self.within_factor),
                scale_factor=self.scale_factor,
            )

        return MetricCollection(metric_dict)

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/regression.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
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 or self.filters 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
        value = float(feat.properties[self.property_name]) * self.scale_factor
        return {}, {
            "value": torch.tensor(value, dtype=torch.float32),
            "valid": torch.tensor(1, dtype=torch.float32),
        }

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

    return {}, {
        "value": torch.tensor(0, dtype=torch.float32),
        "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 scalar tensor.

 required
metadata SampleMetadata

metadata about the patch being read

 required

Returns:

Type Description
list[Feature]

a list with a single Feature corresponding to the patch extent and with a property containing the predicted value.
Source code in rslearn/train/tasks/regression.py 
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
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 scalar tensor.
        metadata: metadata about the patch being read

    Returns:
        a list with a single Feature corresponding to the patch extent and with a
            property containing the predicted value.
    """
    if not isinstance(raw_output, torch.Tensor) or len(raw_output.shape) != 0:
        raise ValueError("output for RegressionTask must be a scalar Tensor")

    output = raw_output.item() / self.scale_factor
    feature = Feature(
        STGeometry(
            metadata.projection,
            shapely.Point(metadata.crop_bounds[0], metadata.crop_bounds[1]),
            None,
        ),
        {
            self.property_name: output,
        },
    )
    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/regression.py 
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
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 = target_dict["value"] / self.scale_factor
    output = output / self.scale_factor
    text = f"Label: {target:.2f}\nOutput: {output:.2f}"
    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/regression.py 
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
def get_metrics(self) -> MetricCollection:
    """Get the metrics for this task."""
    metric_dict: dict[str, Metric] = {}

    for metric_name in self.metrics:
        if metric_name == "mse":
            metric_dict["mse"] = RegressionMetricWrapper(
                metric=torchmetrics.MeanSquaredError(),
                scale_factor=self.scale_factor,
            )
        elif metric_name == "rmse":
            metric_dict["rmse"] = RegressionMetricWrapper(
                metric=torchmetrics.MeanSquaredError(squared=False),
                scale_factor=self.scale_factor,
            )
        elif metric_name == "l1":
            metric_dict["l1"] = RegressionMetricWrapper(
                metric=torchmetrics.MeanAbsoluteError(),
                scale_factor=self.scale_factor,
            )
        elif metric_name == "r2":
            metric_dict["r2"] = RegressionMetricWrapper(
                metric=torchmetrics.R2Score(),
                scale_factor=self.scale_factor,
            )
        elif metric_name == "mape":
            metric_dict["mape"] = RegressionMetricWrapper(
                metric=torchmetrics.MeanAbsolutePercentageError(),
                scale_factor=self.scale_factor,
            )
        else:
            raise ValueError(f"unknown metric {metric_name}")

    if self.use_accuracy_metric:
        metric_dict["accuracy"] = RegressionMetricWrapper(
            metric=RegressionAccuracy(self.within_factor),
            scale_factor=self.scale_factor,
        )

    return MetricCollection(metric_dict)

RegressionHead

Bases: Predictor

Head for regression task.

Source code in rslearn/train/tasks/regression.py 
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
295
296
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
340
class RegressionHead(Predictor):
    """Head for regression task."""

    def __init__(
        self,
        loss_mode: Literal["mse", "l1", "huber"] = "mse",
        use_sigmoid: bool = False,
        huber_delta: float = 1.0,
    ):
        """Initialize a new RegressionHead.

        Args:
            loss_mode: the loss function to use: "mse" (default), "l1", or "huber".
            use_sigmoid: whether to apply a sigmoid activation on the output. This
                requires targets to be between 0-1.
            huber_delta: delta parameter for Huber loss (only used when
                loss_mode="huber").
        """
        super().__init__()
        self.loss_mode = loss_mode
        self.use_sigmoid = use_sigmoid
        self.huber_delta = huber_delta

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

        Args:
            intermediates: output from previous model component, which must be a
                FeatureVector with channel dimension 1 (Bx1).
            context: the model context.
            targets: target dicts, which each must contain a "value" key containing the
                regression label, along with a "valid" key containing a flag indicating
                whether each example is valid for this task.

        Returns:
            the model outputs. The output is a B tensor so that it is split up into a
                scalar for each example.
        """
        if not isinstance(intermediates, FeatureVector):
            raise ValueError("the input to RegressionHead must be a FeatureVector")

        features = intermediates.feature_vector

        if features.shape[1] != 1:
            raise ValueError(
                f"the input to RegressionHead must have channel dimension size 1, but got shape {features.shape}"
            )

        logits = features[:, 0]

        if self.use_sigmoid:
            outputs = torch.nn.functional.sigmoid(logits)
        else:
            outputs = logits

        losses = {}
        if targets:
            labels = torch.stack([target["value"] for target in targets])
            mask = torch.stack([target["valid"] for target in targets])
            if self.loss_mode == "mse":
                losses["regress"] = torch.mean(torch.square(outputs - labels) * mask)
            elif self.loss_mode == "l1":
                losses["regress"] = torch.mean(torch.abs(outputs - labels) * mask)
            elif self.loss_mode == "huber":
                losses["regress"] = torch.mean(
                    torch.nn.functional.huber_loss(
                        outputs,
                        labels,
                        reduction="none",
                        delta=self.huber_delta,
                    )
                    * mask
                )
            else:
                raise ValueError(f"unknown loss mode {self.loss_mode}")

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

forward 

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

Compute the regression outputs and loss from logits and targets.

Parameters:

Name Type Description Default
intermediates Any

output from previous model component, which must be a FeatureVector with channel dimension 1 (Bx1).

 required
context ModelContext

the model context.

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

target dicts, which each must contain a "value" key containing the regression label, along with a "valid" key containing a flag indicating whether each example is valid for this task.

 None

Returns:

Type Description
ModelOutput

the model outputs. The output is a B tensor so that it is split up into a scalar for each example.
Source code in rslearn/train/tasks/regression.py 
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
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
340
def forward(
    self,
    intermediates: Any,
    context: ModelContext,
    targets: list[dict[str, Any]] | None = None,
) -> ModelOutput:
    """Compute the regression outputs and loss from logits and targets.

    Args:
        intermediates: output from previous model component, which must be a
            FeatureVector with channel dimension 1 (Bx1).
        context: the model context.
        targets: target dicts, which each must contain a "value" key containing the
            regression label, along with a "valid" key containing a flag indicating
            whether each example is valid for this task.

    Returns:
        the model outputs. The output is a B tensor so that it is split up into a
            scalar for each example.
    """
    if not isinstance(intermediates, FeatureVector):
        raise ValueError("the input to RegressionHead must be a FeatureVector")

    features = intermediates.feature_vector

    if features.shape[1] != 1:
        raise ValueError(
            f"the input to RegressionHead must have channel dimension size 1, but got shape {features.shape}"
        )

    logits = features[:, 0]

    if self.use_sigmoid:
        outputs = torch.nn.functional.sigmoid(logits)
    else:
        outputs = logits

    losses = {}
    if targets:
        labels = torch.stack([target["value"] for target in targets])
        mask = torch.stack([target["valid"] for target in targets])
        if self.loss_mode == "mse":
            losses["regress"] = torch.mean(torch.square(outputs - labels) * mask)
        elif self.loss_mode == "l1":
            losses["regress"] = torch.mean(torch.abs(outputs - labels) * mask)
        elif self.loss_mode == "huber":
            losses["regress"] = torch.mean(
                torch.nn.functional.huber_loss(
                    outputs,
                    labels,
                    reduction="none",
                    delta=self.huber_delta,
                )
                * mask
            )
        else:
            raise ValueError(f"unknown loss mode {self.loss_mode}")

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

RegressionMetricWrapper

Bases: Metric

Metric for regression task.

Source code in rslearn/train/tasks/regression.py 
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
class RegressionMetricWrapper(Metric):
    """Metric for regression task."""

    def __init__(self, metric: Metric, scale_factor: float, **kwargs: Any) -> None:
        """Initialize a new RegressionMetricWrapper.

        Args:
            metric: the underlying torchmetric to apply, which should accept a flat
                tensor of predicted values followed by a flat tensor of target values
            scale_factor: scale factor to undo so that metric is based on original
                values
            kwargs: other arguments to pass to super constructor
        """
        super().__init__(**kwargs)
        self.metric = metric
        self.scale_factor = scale_factor

    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["value"] for target in targets])

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

        self.metric.update(preds, labels)

    def compute(self) -> Any:
        """Returns the computed metric."""
        return self.metric.compute()

    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/regression.py 
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
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["value"] for target in targets])

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

    self.metric.update(preds, labels)

compute 

compute() -> Any

Returns the computed metric.

Source code in rslearn/train/tasks/regression.py 
382
383
384
def compute(self) -> Any:
    """Returns the computed metric."""
    return self.metric.compute()

reset 

reset() -> None

Reset metric.

Source code in rslearn/train/tasks/regression.py 
386
387
388
389
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/regression.py 
391
392
393
def plot(self, *args: list[Any], **kwargs: dict[str, Any]) -> Any:
    """Returns a plot of the metric."""
    return self.metric.plot(*args, **kwargs)

RegressionAccuracy

Bases: Metric

Percentage of examples with estimate within some factor of ground truth.

Source code in rslearn/train/tasks/regression.py 
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
class RegressionAccuracy(Metric):
    """Percentage of examples with estimate within some factor of ground truth."""

    def __init__(self, factor: float) -> None:
        """Initialize a new RegressionAccuracy.

        Args:
            factor: the factor so if estimate is within this much of ground truth then
                it is marked correct.
        """
        super().__init__()
        self.factor = factor
        self.correct = 0
        self.total = 0

    def update(self, preds: torch.Tensor, labels: torch.Tensor) -> None:
        """Update metric.

        Args:
            preds: the predictions
            labels: the ground truth data
        """
        decisions = (preds >= labels * (1 - self.factor)) & (
            preds <= labels * (1 + self.factor)
        )
        self.correct += torch.count_nonzero(decisions)
        self.total += len(decisions)

    def compute(self) -> Any:
        """Returns the computed metric."""
        return torch.tensor(self.correct / self.total)

    def reset(self) -> None:
        """Reset metric."""
        super().reset()
        self.correct = 0
        self.total = 0

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

update 

update(preds: Tensor, labels: Tensor) -> None

Update metric.

Parameters:

Name Type Description Default
preds Tensor

the predictions

 required
labels Tensor

the ground truth data

 required
Source code in rslearn/train/tasks/regression.py 
411
412
413
414
415
416
417
418
419
420
421
422
def update(self, preds: torch.Tensor, labels: torch.Tensor) -> None:
    """Update metric.

    Args:
        preds: the predictions
        labels: the ground truth data
    """
    decisions = (preds >= labels * (1 - self.factor)) & (
        preds <= labels * (1 + self.factor)
    )
    self.correct += torch.count_nonzero(decisions)
    self.total += len(decisions)

compute 

compute() -> Any

Returns the computed metric.

Source code in rslearn/train/tasks/regression.py 
424
425
426
def compute(self) -> Any:
    """Returns the computed metric."""
    return torch.tensor(self.correct / self.total)

reset 

reset() -> None

Reset metric.

Source code in rslearn/train/tasks/regression.py 
428
429
430
431
432
def reset(self) -> None:
    """Reset metric."""
    super().reset()
    self.correct = 0
    self.total = 0

plot 

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

Returns a plot of the metric.

Source code in rslearn/train/tasks/regression.py 
434
435
436
def plot(self, *args: list[Any], **kwargs: dict[str, Any]) -> Any:
    """Returns a plot of the metric."""
    return None