Implement edge case test scenarios (#256)

* Implement edge case test scenarios

* Cleanup

* Add a test that causes switches, fix setup for unachieved recall values

python-sdk/nuscenes/eval/tracking/tests/scenarios.py

@@ -0,0 +1,93 @@
+from typing import List, Dict
+
+import numpy as np
+
+
+def get_scenarios() -> List[Dict[str, dict]]:
+    """ """
+
+    scenarios = []
+
+    # Scenario 1.
+    # Parallel motion 1 meter distance.
+    pos_gt = np.array([[(1, -3), (1, -2), (1, -1), (1, -0)],
+                      [(0, -3), (0, -2), (0, -1), (0, -0)], ]).astype(float)
+    pos_pred = pos_gt
+    sigma = 0.1
+    pos_pred += sigma * np.random.randn(*pos_pred.shape)
+
+    input_data = {'pos_gt':  pos_gt,
+                  'pos_pred': pos_pred}
+    output_data = {'ids': 0.0}
+
+    scenarios.append({'input': input_data, 'output': output_data})
+
+    # Scenario 2.
+    # Parallel motion bring closer predictions.
+    pos_gt = np.array([[(1, -3), (1, -2), (1, -1), (1, -0)],
+                      [(0, -3), (0, -2), (0, -1), (0, -0)], ]).astype(float)
+    pos_pred = pos_gt
+
+    pos_pred[0, :, 0] -= 0.3
+    pos_pred[1, :, 0] += 0.3
+    sigma = 0.1
+    pos_pred += sigma * np.random.randn(*pos_pred.shape)
+
+    input_data = {'pos_gt':  pos_gt,
+                  'pos_pred': pos_pred}
+    output_data = {'ids': 0.0}
+
+    scenarios.append({'input': input_data, 'output': output_data})
+
+    # Scenario 3.
+    # Parallel motion bring closer both ground truth and predictions.
+    pos_gt = np.array([[(1, -3), (1, -2), (1, -1), (1, -0)],
+                      [(0, -3), (0, -2), (0, -1), (0, -0)], ]).astype(float)
+    pos_pred = pos_gt
+
+    pos_gt[0, :, 0] -= 0.3
+    pos_gt[1, :, 0] += 0.3
+    pos_pred[0, :, 0] -= 0.3
+    pos_pred[1, :, 0] += 0.3
+    sigma = 0.1
+    pos_pred += sigma * np.random.randn(*pos_pred.shape)
+
+    input_data = {'pos_gt':  pos_gt,
+                  'pos_pred': pos_pred}
+    output_data = {'ids': 0.0}
+
+    scenarios.append({'input': input_data, 'output': output_data})
+
+    # Scenario 4.
+    # Crossing motion.
+    pos_gt = np.array([[(2, -3), (1, -2), (0, -1), (-1, -0)],
+                      [(-2, -3), (-1, -2), (0, -1), (1, -0)], ]).astype(float)
+    pos_pred = pos_gt
+    sigma = 0.1
+    pos_pred += sigma * np.random.randn(*pos_pred.shape)
+
+    input_data = {'pos_gt':  pos_gt,
+                  'pos_pred': pos_pred}
+    output_data = {'ids': 0.0}
+
+    scenarios.append({'input': input_data, 'output': output_data})
+
+    # Scenario 5.
+    # Identity switch due to a single misdetection (3rd timestamp).
+    pos_pred = np.array([
+        [(0, -2), (0, -1), (0, 0), (0, 1), (0, 2)],
+        [(-2, 0), (-1, 0), (3, 0), (1, 0), (2, 0)],
+    ]).astype(float)
+    pos_gt = np.array([
+        [(-2, 0), (-1, 0), (0, 0), (1, 0), (2, 0)],
+    ]).astype(float)
+    sigma = 0.1
+    pos_pred += sigma * np.random.randn(*pos_pred.shape)
+
+    input_data = {'pos_gt':  pos_gt,
+                  'pos_pred': pos_pred}
+    output_data = {'ids': 2}
+
+    scenarios.append({'input': input_data, 'output': output_data})
+
+    return scenarios

python-sdk/nuscenes/eval/tracking/tests/test_algo.py

@@ -9,6 +9,7 @@
 from nuscenes.eval.tracking.algo import TrackingEvaluation
 from nuscenes.eval.tracking.data_classes import TrackingMetricData, TrackingBox
 from nuscenes.eval.tracking.loaders import interpolate_tracks
+from nuscenes.eval.tracking.tests.scenarios import get_scenarios
 
 
 class TestAlgo(unittest.TestCase):
@@ -62,6 +63,8 @@ def test_gt_submission(self):
         assert np.all(md.ids == 0)
 
     def test_empty_submission(self):
+        """ Test a submission with no predictions. """
+
         # Get config.
         cfg = config_factory('tracking_nips_2019')
 
@@ -255,6 +258,40 @@ def test_drop_gt_interpolate(self):
         assert np.all(md.frag == 0)
         assert np.all(md.ids == 0)
 
+    def test_scenarios(self):
+        """ More flexible scenario test structure. """
+
+        def create_tracks(_scenario, tag=None):
+            tracks = {}
+            for entry_id, entry in enumerate(_scenario['input']['pos_'+tag]):
+                tracking_id = 'tag_{}'.format(entry_id)
+                for timestamp, pos in enumerate(entry):
+                    if timestamp not in tracks.keys():
+                        tracks[timestamp] = []
+                    box = TrackingBox(translation=(pos[0], pos[1], 0.0), tracking_id=tracking_id, tracking_name='car',
+                                      tracking_score=0.5)
+                    tracks[timestamp].append(box)
+
+            return tracks
+
+        # Get config.
+        cfg = config_factory('tracking_nips_2019')
+
+        for scenario in get_scenarios():
+            tracks_gt = {'scene-1': create_tracks(scenario, tag='gt')}
+            tracks_pred = {'scene-1': create_tracks(scenario, tag='pred')}
+
+            # Accumulate metrics.
+            ev = TrackingEvaluation(tracks_gt, tracks_pred, 'car', cfg.dist_fcn_callable,
+                                    cfg.dist_th_tp, cfg.min_recall, num_thresholds=TrackingMetricData.nelem,
+                                    metric_worst=cfg.metric_worst, verbose=False)
+            md = ev.accumulate()
+
+            for key, value in scenario['output'].items():
+                metric_values = getattr(md, key)
+                metric_values = metric_values[np.logical_not(np.isnan(metric_values))]
+                assert np.all(metric_values == value)
+
 
 if __name__ == '__main__':
     unittest.main()