File size: 4,691 Bytes
f343f06
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
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
"""
Unit tests for data preprocessing and dataset loading.
"""

import pytest
import torch
import numpy as np
from PIL import Image

import sys
sys.path.insert(0, '.')

from src.data.preprocessing import (
    preprocess_image,
    handle_channels,
    get_optical_transform,
    get_sar_transform
)
from src.data.dataset import (
    CrisisLandMarkDataset,
    create_splits
)


class TestPreprocessing:
    """Test preprocessing functions."""
    
    def test_optical_preprocessing(self):
        """Test optical RGB preprocessing."""
        dummy_image = Image.fromarray(
            np.random.randint(0, 255, (256, 256, 3), dtype=np.uint8)
        )
        
        result = preprocess_image(dummy_image, "optical", size=224)
        
        assert result.shape == (3, 224, 224)
        assert result.dtype == torch.float32
    
    def test_sar_preprocessing(self):
        """Test SAR preprocessing."""
        dummy_image = Image.fromarray(
            np.random.randint(0, 255, (256, 256, 2), dtype=np.uint8)
        )
        
        result = preprocess_image(dummy_image, "sar", size=224)
        
        assert result.shape == (2, 224, 224)
        assert result.dtype == torch.float32
    
    def test_multispectral_preprocessing(self):
        """Test multispectral preprocessing."""
        # Create 12-channel image
        dummy_array = np.random.randint(0, 255, (256, 256, 12), dtype=np.uint8)
        dummy_image = Image.fromarray(dummy_array[..., :3])  # PIL only supports 3 channels
        
        # For 12-channel, we'd need custom handling
        # This test verifies the function accepts the modality
        result = preprocess_image(dummy_image, "optical", size=224)
        
        assert result.shape[0] == 3  # Should be 3 channels from RGB
    
    def test_channel_handling(self):
        """Test channel mismatch handling."""
        # Create 4-channel image (should be trimmed to 3 for optical)
        image_4ch = np.random.randint(0, 255, (64, 64, 4), dtype=np.uint8)
        
        result = handle_channels(image_4ch, 3, "optical")
        
        assert result.shape[-1] == 3
    
    def test_invalid_modality(self):
        """Test invalid modality raises error."""
        dummy_image = Image.fromarray(
            np.random.randint(0, 255, (64, 64, 3), dtype=np.uint8)
        )
        
        with pytest.raises(ValueError):
            preprocess_image(dummy_image, "invalid_modality")


class TestDataset:
    """Test dataset class."""
    
    def test_dataset_creation(self):
        """Test dataset can be created."""
        dataset = CrisisLandMarkDataset(modality="optical")
        
        assert len(dataset) > 0
    
    def test_dataset_getitem(self):
        """Test dataset returns correct format."""
        dataset = CrisisLandMarkDataset(modality="optical")
        
        image, modality_label, class_label = dataset[0]
        
        assert isinstance(image, torch.Tensor)
        assert isinstance(modality_label, int)
        assert isinstance(class_label, int)
    
    def test_modality_labels(self):
        """Test modality labels are correct."""
        # Test optical and SAR (multispectral requires 12-channel images)
        for modality, expected_label in [("optical", 0), ("sar", 1)]:
            dataset = CrisisLandMarkDataset(modality=modality)
            _, modality_label, _ = dataset[0]
            
            assert modality_label == expected_label


class TestSplitting:
    """Test data splitting."""
    
    def test_split_no_overlap(self):
        """Test query/gallery split has no overlap."""
        dataset = CrisisLandMarkDataset(modality="optical")
        
        query_idx, gallery_idx = create_splits(dataset, query_ratio=0.2)
        
        # Check no overlap
        assert len(set(query_idx) & set(gallery_idx)) == 0
    
    def test_split_ratio(self):
        """Test split maintains approximate ratio."""
        dataset = CrisisLandMarkDataset(modality="optical")
        
        query_idx, gallery_idx = create_splits(dataset, query_ratio=0.2)
        
        total = len(query_idx) + len(gallery_idx)
        query_ratio = len(query_idx) / total
        
        # Allow some tolerance
        assert 0.15 <= query_ratio <= 0.25
    
    def test_split_reproducibility(self):
        """Test split is reproducible with same seed."""
        dataset = CrisisLandMarkDataset(modality="optical")
        
        query1, gallery1 = create_splits(dataset, seed=42)
        query2, gallery2 = create_splits(dataset, seed=42)
        
        assert query1 == query2
        assert gallery1 == gallery2


if __name__ == "__main__":
    pytest.main([__file__, "-v"])