1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
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
|
import torch
from gds.common.metrics.metric import ElementwiseMetric, Metric, MultiTaskMetric
from gds.common.utils import maximum
class Loss(Metric):
def __init__(self, loss_fn, name=None):
self.loss_fn = loss_fn
if name is None:
name = 'loss'
super().__init__(name=name)
def _compute(self, y_pred, y_true):
"""
Helper for computing element-wise metric, implemented for each metric
Args:
- y_pred (Tensor): Predicted targets or model output
- y_true (Tensor): True targets
Output:
- element_wise_metrics (Tensor): tensor of size (batch_size, )
"""
return self.loss_fn(y_pred, y_true)
def worst(self, metrics):
"""
Given a list/numpy array/Tensor of metrics, computes the worst-case metric
Args:
- metrics (Tensor, numpy array, or list): Metrics
Output:
- worst_metric (float): Worst-case metric
"""
return maximum(metrics)
class ElementwiseLoss(ElementwiseMetric):
def __init__(self, loss_fn, name=None):
self.loss_fn = loss_fn
if name is None:
name = 'loss'
super().__init__(name=name)
def _compute_element_wise(self, y_pred, y_true):
"""
Helper for computing element-wise metric, implemented for each metric
Args:
- y_pred (Tensor): Predicted targets or model output
- y_true (Tensor): True targets
Output:
- element_wise_metrics (Tensor): tensor of size (batch_size, )
"""
# import pdb;pdb.set_trace()
if isinstance(self.loss_fn, torch.nn.BCEWithLogitsLoss):
return self.loss_fn(y_pred.float(), y_true.float()).squeeze(dim=-1)
elif isinstance(self.loss_fn, torch.nn.CrossEntropyLoss):
return self.loss_fn(y_pred, y_true)
else:
raise NotImplementedError
def worst(self, metrics):
"""
Given a list/numpy array/Tensor of metrics, computes the worst-case metric
Args:
- metrics (Tensor, numpy array, or list): Metrics
Output:
- worst_metric (float): Worst-case metric
"""
return maximum(metrics)
class MultiTaskLoss(MultiTaskMetric):
def __init__(self, loss_fn, name=None):
self.loss_fn = loss_fn # should be elementwise
if name is None:
name = 'loss'
super().__init__(name=name)
def _compute_flattened(self, flattened_y_pred, flattened_y_true):
if isinstance(self.loss_fn, torch.nn.BCEWithLogitsLoss):
flattened_y_pred = flattened_y_pred.float()
flattened_y_true = flattened_y_true.float()
elif isinstance(self.loss_fn, torch.nn.CrossEntropyLoss):
flattened_y_true = flattened_y_true.long()
flattened_loss = self.loss_fn(flattened_y_pred, flattened_y_true)
return flattened_loss
def worst(self, metrics):
"""
Given a list/numpy array/Tensor of metrics, computes the worst-case metric
Args:
- metrics (Tensor, numpy array, or list): Metrics
Output:
- worst_metric (float): Worst-case metric
"""
return maximum(metrics)
|