test_geometric_mean_relative_absolute_error.py test_kendall_correlation.py test_manhattan_distance.py test_maximum_absolute_error.py

Author: BanzaiTokyo

tests/ignite/metrics/regression/test_canberra_metric.py

@@ -20,83 +20,83 @@ def test_wrong_input_shapes():
         m.update((torch.rand(4, 1), torch.rand(4)))
 
 
-def test_compute(available_device):
-    a = torch.randn(4)
-    b = torch.randn(4)
-    c = torch.randn(4)
-    d = torch.randn(4)
-    ground_truth = torch.randn(4)
+def test_compute():
+    a = np.random.randn(4)
+    b = np.random.randn(4)
+    c = np.random.randn(4)
+    d = np.random.randn(4)
+    ground_truth = np.random.randn(4)
 
-    m = CanberraMetric(device=available_device)
-    assert m._device == torch.device(available_device)
+    m = CanberraMetric()
 
     canberra = DistanceMetric.get_metric("canberra")
 
-    m.update((a, ground_truth))
-    np_sum = (torch.abs(ground_truth - a) / (torch.abs(a) + torch.abs(ground_truth))).sum()
+    m.update((torch.from_numpy(a), torch.from_numpy(ground_truth)))
+    np_sum = (np.abs(ground_truth - a) / (np.abs(a) + np.abs(ground_truth))).sum()
     assert m.compute() == pytest.approx(np_sum)
-    assert canberra.pairwise([a.cpu().numpy(), ground_truth.cpu().numpy()])[0][1] == pytest.approx(np_sum)
+    assert canberra.pairwise([a, ground_truth])[0][1] == pytest.approx(np_sum)
 
-    m.update((b, ground_truth))
-    np_sum += ((torch.abs(ground_truth - b)) / (torch.abs(b) + torch.abs(ground_truth))).sum()
+    m.update((torch.from_numpy(b), torch.from_numpy(ground_truth)))
+    np_sum += ((np.abs(ground_truth - b)) / (np.abs(b) + np.abs(ground_truth))).sum()
     assert m.compute() == pytest.approx(np_sum)
     v1 = np.hstack([a, b])
     v2 = np.hstack([ground_truth, ground_truth])
     assert canberra.pairwise([v1, v2])[0][1] == pytest.approx(np_sum)
 
-    m.update((c, ground_truth))
-    np_sum += ((torch.abs(ground_truth - c)) / (torch.abs(c) + torch.abs(ground_truth))).sum()
+    m.update((torch.from_numpy(c), torch.from_numpy(ground_truth)))
+    np_sum += ((np.abs(ground_truth - c)) / (np.abs(c) + np.abs(ground_truth))).sum()
     assert m.compute() == pytest.approx(np_sum)
     v1 = np.hstack([v1, c])
     v2 = np.hstack([v2, ground_truth])
     assert canberra.pairwise([v1, v2])[0][1] == pytest.approx(np_sum)
 
-    m.update((d, ground_truth))
-    np_sum += (torch.abs(ground_truth - d) / (torch.abs(d) + torch.abs(ground_truth))).sum()
+    m.update((torch.from_numpy(d), torch.from_numpy(ground_truth)))
+    np_sum += (np.abs(ground_truth - d) / (np.abs(d) + np.abs(ground_truth))).sum()
     assert m.compute() == pytest.approx(np_sum)
     v1 = np.hstack([v1, d])
     v2 = np.hstack([v2, ground_truth])
     assert canberra.pairwise([v1, v2])[0][1] == pytest.approx(np_sum)
 
 
-@pytest.mark.parametrize("n_times", range(3))
-@pytest.mark.parametrize(
-    "test_cases",
-    [
-        (torch.rand(size=(100,)), torch.rand(size=(100,)), 10),
-        (torch.rand(size=(100, 1)), torch.rand(size=(100, 1)), 20),
-    ],
-)
-def test_integration(n_times, test_cases, available_device):
-    y_pred, y, batch_size = test_cases
+def test_integration():
+    def _test(y_pred, y, batch_size):
+        def update_fn(engine, batch):
+            idx = (engine.state.iteration - 1) * batch_size
+            y_true_batch = np_y[idx : idx + batch_size]
+            y_pred_batch = np_y_pred[idx : idx + batch_size]
+            return torch.from_numpy(y_pred_batch), torch.from_numpy(y_true_batch)
 
-    def update_fn(engine, batch):
-        idx = (engine.state.iteration - 1) * batch_size
-        y_true_batch = y[idx : idx + batch_size]
-        y_pred_batch = y_pred[idx : idx + batch_size]
-        return y_pred_batch, y_true_batch
+        engine = Engine(update_fn)
 
-    engine = Engine(update_fn)
+        m = CanberraMetric()
+        m.attach(engine, "cm")
 
-    m = CanberraMetric(device=available_device)
-    assert m._device == torch.device(available_device)
+        np_y = y.numpy().ravel()
+        np_y_pred = y_pred.numpy().ravel()
 
-    m.attach(engine, "cm")
+        canberra = DistanceMetric.get_metric("canberra")
 
-    canberra = DistanceMetric.get_metric("canberra")
+        data = list(range(y_pred.shape[0] // batch_size))
+        cm = engine.run(data, max_epochs=1).metrics["cm"]
 
-    data = list(range(y_pred.shape[0] // batch_size))
-    cm = engine.run(data, max_epochs=1).metrics["cm"]
+        assert canberra.pairwise([np_y_pred, np_y])[0][1] == pytest.approx(cm)
 
-    pred_np = y_pred.cpu().numpy().reshape(len(y_pred), -1)
-    true_np = y.cpu().numpy().reshape(len(y), -1)
-    expected = np.sum(canberra.pairwise(pred_np, true_np).diagonal())
-    assert expected == pytest.approx(cm)
+    def get_test_cases():
+        test_cases = [
+            (torch.rand(size=(100,)), torch.rand(size=(100,)), 10),
+            (torch.rand(size=(100, 1)), torch.rand(size=(100, 1)), 20),
+        ]
+        return test_cases
 
+    for _ in range(5):
+        # check multiple random inputs as random exact occurencies are rare
+        test_cases = get_test_cases()
+        for y_pred, y, batch_size in test_cases:
+            _test(y_pred, y, batch_size)
 
-def test_error_is_not_nan(available_device):
-    m = CanberraMetric(device=available_device)
-    assert m._device == torch.device(available_device)
+
+def test_error_is_not_nan():
+    m = CanberraMetric()
     m.update((torch.zeros(4), torch.zeros(4)))
     assert not (torch.isnan(m._sum_of_errors).any() or torch.isinf(m._sum_of_errors).any()), m._sum_of_errors
 

tests/ignite/metrics/regression/test_fractional_absolute_error.py

@@ -28,65 +28,77 @@ def test_wrong_input_shapes():
         m.update((torch.rand(4, 1), torch.rand(4)))
 
 
-def test_compute(available_device):
-    a = torch.randn(4)
-    b = torch.randn(4)
-    c = torch.randn(4)
-    d = torch.randn(4)
-    ground_truth = torch.randn(4)
+def test_compute():
+    a = np.random.randn(4)
+    b = np.random.randn(4)
+    c = np.random.randn(4)
+    d = np.random.randn(4)
+    ground_truth = np.random.randn(4)
 
-    m = FractionalAbsoluteError(device=available_device)
-    assert m._device == torch.device(available_device)
+    m = FractionalAbsoluteError()
 
-    total_error = 0.0
-    total_len = 0
+    m.update((torch.from_numpy(a), torch.from_numpy(ground_truth)))
+    np_sum = (2 * np.abs((a - ground_truth)) / (np.abs(a) + np.abs(ground_truth))).sum()
+    np_len = len(a)
+    np_ans = np_sum / np_len
+    assert m.compute() == pytest.approx(np_ans)
 
-    for pred in [a, b, c, d]:
-        m.update((pred, ground_truth))
+    m.update((torch.from_numpy(b), torch.from_numpy(ground_truth)))
+    np_sum += (2 * np.abs((b - ground_truth)) / (np.abs(b) + np.abs(ground_truth))).sum()
+    np_len += len(b)
+    np_ans = np_sum / np_len
+    assert m.compute() == pytest.approx(np_ans)
 
-        # Compute fractional absolute error in PyTorch
-        error = 2 * torch.abs(pred - ground_truth) / (torch.abs(pred) + torch.abs(ground_truth))
-        total_error += error.sum().item()
-        total_len += len(pred)
+    m.update((torch.from_numpy(c), torch.from_numpy(ground_truth)))
+    np_sum += (2 * np.abs((c - ground_truth)) / (np.abs(c) + np.abs(ground_truth))).sum()
+    np_len += len(c)
+    np_ans = np_sum / np_len
+    assert m.compute() == pytest.approx(np_ans)
 
-        expected = total_error / total_len
-        assert m.compute() == pytest.approx(expected)
+    m.update((torch.from_numpy(d), torch.from_numpy(ground_truth)))
+    np_sum += (2 * np.abs((d - ground_truth)) / (np.abs(d) + np.abs(ground_truth))).sum()
+    np_len += len(d)
+    np_ans = np_sum / np_len
+    assert m.compute() == pytest.approx(np_ans)
 
 
-@pytest.mark.parametrize("n_times", range(5))
-@pytest.mark.parametrize(
-    "test_cases",
-    [
-        (torch.rand(size=(100,)), torch.rand(size=(100,)), 10),
-        (torch.rand(size=(100, 1)), torch.rand(size=(100, 1)), 20),
-    ],
-)
-def test_integration_fractional_absolute_error(n_times, test_cases, available_device):
-    y_pred, y, batch_size = test_cases
+def test_integration():
+    def _test(y_pred, y, batch_size):
+        def update_fn(engine, batch):
+            idx = (engine.state.iteration - 1) * batch_size
+            y_true_batch = np_y[idx : idx + batch_size]
+            y_pred_batch = np_y_pred[idx : idx + batch_size]
+            return torch.from_numpy(y_pred_batch), torch.from_numpy(y_true_batch)
 
-    np_y = y.numpy().ravel()
-    np_y_pred = y_pred.numpy().ravel()
+        engine = Engine(update_fn)
 
-    def update_fn(engine, batch):
-        idx = (engine.state.iteration - 1) * batch_size
-        y_true_batch = y[idx : idx + batch_size]
-        y_pred_batch = y_pred[idx : idx + batch_size]
-        return y_pred_batch, y_true_batch
+        m = FractionalAbsoluteError()
+        m.attach(engine, "fab")
 
-    engine = Engine(update_fn)
+        np_y = y.numpy().ravel()
+        np_y_pred = y_pred.numpy().ravel()
 
-    metric = FractionalAbsoluteError(device=available_device)
-    assert metric._device == torch.device(available_device)
+        data = list(range(y_pred.shape[0] // batch_size))
+        fab = engine.run(data, max_epochs=1).metrics["fab"]
 
-    metric.attach(engine, "fab")
+        np_sum = (2 * np.abs((np_y_pred - np_y)) / (np.abs(np_y_pred) + np.abs(np_y))).sum()
+        np_len = len(y_pred)
+        np_ans = np_sum / np_len
 
-    data = list(range(y_pred.shape[0] // batch_size))
-    fab = engine.run(data, max_epochs=1).metrics["fab"]
+        assert np_ans == pytest.approx(fab)
 
-    np_sum = (2 * np.abs(np_y_pred - np_y) / (np.abs(np_y_pred) + np.abs(np_y))).sum()
-    expected = np_sum / len(np_y)
+    def get_test_cases():
+        test_cases = [
+            (torch.rand(size=(100,)), torch.rand(size=(100,)), 10),
+            (torch.rand(size=(100, 1)), torch.rand(size=(100, 1)), 20),
+        ]
+        return test_cases
 
-    assert expected == pytest.approx(fab)
+    for _ in range(5):
+        # check multiple random inputs as random exact occurencies are rare
+        test_cases = get_test_cases()
+        for y_pred, y, batch_size in test_cases:
+            _test(y_pred, y, batch_size)
 
 
 def _test_distrib_compute(device):