Spaces:
Sleeping
Sleeping
| """ | |
| Tests for evaluation metrics and ground truth. | |
| These tests verify: | |
| - F1@K computation correctness | |
| - Ground truth creation and validation | |
| - Save/load roundtrip | |
| - Per-modality breakdown | |
| """ | |
| import pytest | |
| import tempfile | |
| from pathlib import Path | |
| from src.evaluation.metrics import EvaluationMetrics, EvaluationResult | |
| from src.evaluation.ground_truth import ( | |
| GroundTruth, | |
| GroundTruthPair, | |
| create_ground_truth_from_matches, | |
| ) | |
| def evaluation_metrics(): | |
| """Create fresh EvaluationMetrics.""" | |
| return EvaluationMetrics() | |
| def sample_predictions(): | |
| """Sample prediction lists.""" | |
| return [ | |
| [0, 1, 2, 3, 4, 5, 6, 7, 8, 9], | |
| [1, 2, 3, 4, 5, 6, 7, 8, 9, 10], | |
| [2, 3, 4, 5, 6, 7, 8, 9, 10, 11], | |
| ] | |
| def sample_ground_truth(): | |
| """Sample ground truth lists.""" | |
| return [ | |
| [0, 2, 4, 6, 8], | |
| [1, 3, 5, 7, 9], | |
| [2, 4, 6, 8, 10], | |
| ] | |
| class TestEvaluationMetrics: | |
| """Tests for EvaluationMetrics class.""" | |
| def test_f1_perfect(self, evaluation_metrics): | |
| """Test F1=1 when predictions match ground truth.""" | |
| predicted = [0, 1, 2, 3, 4] | |
| ground_truth = [0, 1, 2, 3, 4] | |
| f1 = evaluation_metrics.compute_f1_at_k(predicted, ground_truth, k=5) | |
| assert f1 == 1.0 | |
| def test_f1_partial(self, evaluation_metrics): | |
| """Test F1 for partial overlap.""" | |
| predicted = [0, 1, 2, 3, 4] | |
| ground_truth = [0, 2, 4, 6, 8] | |
| f1 = evaluation_metrics.compute_f1_at_k(predicted, ground_truth, k=5) | |
| # 3 relevant in top 5, precision=3/5, recall=3/5 | |
| # F1 = 2 * (3/5) * (3/5) / (3/5 + 3/5) = 3/5 = 0.6 | |
| assert abs(f1 - 0.6) < 0.01 | |
| def test_f1_no_overlap(self, evaluation_metrics): | |
| """Test F1=0 when no overlap.""" | |
| predicted = [10, 11, 12, 13, 14] | |
| ground_truth = [0, 1, 2, 3, 4] | |
| f1 = evaluation_metrics.compute_f1_at_k(predicted, ground_truth, k=5) | |
| assert f1 == 0.0 | |
| def test_f1_empty_predictions(self, evaluation_metrics): | |
| """Test F1 with empty predictions.""" | |
| predicted = [] | |
| ground_truth = [0, 1, 2] | |
| f1 = evaluation_metrics.compute_f1_at_k(predicted, ground_truth, k=5) | |
| assert f1 == 0.0 | |
| def test_f1_empty_ground_truth(self, evaluation_metrics): | |
| """Test F1 with empty ground truth.""" | |
| predicted = [0, 1, 2] | |
| ground_truth = [] | |
| f1 = evaluation_metrics.compute_f1_at_k(predicted, ground_truth, k=5) | |
| assert f1 == 0.0 | |
| def test_f1_at_5_vs_10(self, evaluation_metrics): | |
| """Test F1@5 vs F1@10.""" | |
| predicted = list(range(20)) | |
| ground_truth = list(range(10, 15)) # Only in second half | |
| f1_5 = evaluation_metrics.compute_f1_at_k(predicted, ground_truth, k=5) | |
| f1_10 = evaluation_metrics.compute_f1_at_k(predicted, ground_truth, k=10) | |
| # F1@10 should be better since ground truth is in positions 10-14 | |
| assert f1_10 >= f1_5 | |
| def test_batch_f1(self, evaluation_metrics, sample_predictions, sample_ground_truth): | |
| """Test batch F1 computation.""" | |
| batch_f1 = evaluation_metrics.compute_f1_at_k_batch( | |
| sample_predictions, sample_ground_truth, k=5 | |
| ) | |
| assert 0.0 <= batch_f1 <= 1.0 | |
| def test_timing_stats(self, evaluation_metrics): | |
| """Test timing statistics.""" | |
| times = [10.0, 20.0, 30.0, 40.0, 50.0] | |
| for t in times: | |
| evaluation_metrics.add_query_time(t) | |
| stats = evaluation_metrics.get_timing_stats() | |
| assert stats["mean_ms"] == 30.0 | |
| assert stats["median_ms"] == 30.0 | |
| assert stats["p95_ms"] == 50.0 # With only 5 samples, p95 = max | |
| def test_timing_stats_empty(self, evaluation_metrics): | |
| """Test timing stats with no data.""" | |
| stats = evaluation_metrics.get_timing_stats() | |
| assert stats["mean_ms"] == 0.0 | |
| assert stats["median_ms"] == 0.0 | |
| def test_full_evaluation( | |
| self, evaluation_metrics, sample_predictions, sample_ground_truth | |
| ): | |
| """Test full evaluation pipeline.""" | |
| query_times = [15.0, 25.0, 35.0] | |
| modality_labels = ["optical", "sar", "optical"] | |
| result = evaluation_metrics.evaluate_retrieval( | |
| sample_predictions, sample_ground_truth, | |
| query_times=query_times, | |
| modality_labels=modality_labels | |
| ) | |
| assert isinstance(result, EvaluationResult) | |
| assert 0.0 <= result.f1_at_5 <= 1.0 | |
| assert 0.0 <= result.f1_at_10 <= 1.0 | |
| assert result.mean_time_ms > 0 | |
| assert "optical" in result.modality_results | |
| assert "sar" in result.modality_results | |
| class TestGroundTruth: | |
| """Tests for GroundTruth class.""" | |
| def test_add_pair(self): | |
| """Test adding ground truth pairs.""" | |
| gt = GroundTruth() | |
| gt.add_pair(0, [1, 2], "optical", "optical") | |
| assert gt.n_pairs == 1 | |
| assert gt.get_gallery_ids(0) == [1, 2] | |
| def test_get_gallery_ids(self): | |
| """Test getting gallery IDs for query.""" | |
| gt = GroundTruth() | |
| gt.add_pair(0, [1, 2, 3], "optical", "sar") | |
| ids = gt.get_gallery_ids(0) | |
| assert ids == [1, 2, 3] | |
| def test_get_gallery_ids_unknown(self): | |
| """Test getting gallery IDs for unknown query.""" | |
| gt = GroundTruth() | |
| ids = gt.get_gallery_ids(999) | |
| assert ids == [] | |
| def test_same_modal_pairs(self): | |
| """Test filtering same-modal pairs.""" | |
| gt = GroundTruth() | |
| gt.add_pair(0, [1], "optical", "optical") | |
| gt.add_pair(1, [2], "optical", "sar") | |
| gt.add_pair(2, [3], "sar", "sar") | |
| same_modal = gt.get_same_modal_pairs() | |
| assert len(same_modal) == 2 | |
| def test_cross_modal_pairs(self): | |
| """Test filtering cross-modal pairs.""" | |
| gt = GroundTruth() | |
| gt.add_pair(0, [1], "optical", "optical") | |
| gt.add_pair(1, [2], "optical", "sar") | |
| gt.add_pair(2, [3], "sar", "optical") | |
| cross_modal = gt.get_cross_modal_pairs() | |
| assert len(cross_modal) == 2 | |
| def test_validate_valid(self): | |
| """Test validation passes for valid ground truth.""" | |
| gt = GroundTruth() | |
| gt.add_pair(0, [1, 2], "optical", "optical") | |
| gt.add_pair(1, [3, 4], "sar", "sar") | |
| assert gt.validate() | |
| def test_validate_invalid_modality(self): | |
| """Test validation fails for invalid modality.""" | |
| gt = GroundTruth() | |
| gt.add_pair(0, [1], "invalid", "optical") | |
| assert not gt.validate() | |
| def test_save_load_roundtrip(self): | |
| """Test save then load returns same data.""" | |
| gt = GroundTruth() | |
| gt.add_pair(0, [1, 2], "optical", "optical") | |
| gt.add_pair(1, [3, 4], "sar", "sar") | |
| with tempfile.TemporaryDirectory() as tmpdir: | |
| save_path = Path(tmpdir) / "ground_truth.json" | |
| gt.save(save_path) | |
| loaded_gt = GroundTruth.load(save_path) | |
| assert loaded_gt.n_pairs == gt.n_pairs | |
| assert loaded_gt.validate() | |
| # Verify specific pairs | |
| assert loaded_gt.get_gallery_ids(0) == [1, 2] | |
| assert loaded_gt.get_gallery_ids(1) == [3, 4] | |
| class TestCreateGroundTruthFromMatches: | |
| """Tests for create_ground_truth_from_matches.""" | |
| def test_create_from_matches(self): | |
| """Test creating ground truth from matches dict.""" | |
| matches = { | |
| ("optical", "optical"): [(0, 1), (0, 2), (1, 3)], | |
| ("sar", "sar"): [(2, 4), (2, 5)], | |
| } | |
| gt = create_ground_truth_from_matches(matches) | |
| # 3 unique query_ids: 0, 1, 2 | |
| assert gt.n_pairs == 3 | |
| assert gt.get_gallery_ids(0) == [1, 2] | |
| assert gt.get_gallery_ids(1) == [3] | |
| assert gt.get_gallery_ids(2) == [4, 5] | |
| # Self-check | |
| if __name__ == "__main__": | |
| pytest.main([__file__, "-v"]) | |