test_COREG.py 9.43 KB
Newer Older
1
2
3
4
5
6
#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""Tests for the global co-registration module of AROSICS."""

import unittest
7
8
import shutil
import os
9
import numpy as np
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

# custom
from .cases import test_cases
from arosics import COREG
from geoarray import GeoArray


class COREG_GLOBAL_init(unittest.TestCase):
    """Test case on object initialization of COREG_LOCAL."""

    def setUp(self):
        self.ref_path = test_cases['INTER1']['ref_path']
        self.tgt_path = test_cases['INTER1']['tgt_path']
        self.coreg_kwargs = test_cases['INTER1']['kwargs_global']
        self.coreg_kwargs['wp'] = test_cases['INTER1']['wp_inside']

    def test_coreg_init_from_disk(self):
        self.CRL = COREG(self.ref_path, self.tgt_path, **self.coreg_kwargs)

    def test_coreg_init_from_inMem_GeoArray(self):
        # get GeoArray instances
        self.ref_gA = GeoArray(self.ref_path)
        self.tgt_gA = GeoArray(self.tgt_path)

        # assure the raster data are in-memory
        self.ref_gA.to_mem()
        self.tgt_gA.to_mem()

        # get instance of COREG_LOCAL object
        self.CRL = COREG(self.ref_gA, self.tgt_gA, **self.coreg_kwargs)


class CompleteWorkflow_INTER1_S2A_S2A(unittest.TestCase):
    """Test case for the complete workflow of global co-registration based on two Sentinel-2 datasets, one with
    ~25% cloud cover, the other one without any clouds. The subsets cover the S2A tiles only partly (nodata areas
    are present).
    """

    def setUp(self):
        self.ref_path = test_cases['INTER1']['ref_path']
        self.tgt_path = test_cases['INTER1']['tgt_path']
        self.coreg_kwargs = test_cases['INTER1']['kwargs_global']

    def tearDown(self):
54
55
56
57
        """Delete output."""
        dir_out = os.path.dirname(self.coreg_kwargs['path_out'])
        if os.path.isdir(dir_out):
            shutil.rmtree(dir_out)
58

59
    def run_shift_detection_correction(self, ref, tgt, **params):
60
        # get instance of COREG_LOCAL object
61
        CR = COREG(ref, tgt, **params)
62
63
64
65
66
67
68

        # calculate global X/Y shift
        CR.calculate_spatial_shifts()

        # test shift correction and output writer
        CR.correct_shifts()

69
70
71
72
73
74
75
76
77
        self.assertTrue(
            os.path.exists(params['path_out']),
            'Output of global co-registration has not been written.')

        return CR

    def test_shift_calculation_with_default_params(self):
        """Test with default parameters - should compute X/Y shifts properly ad write the de-shifted target image."""

78
79
80
81
82
        CR = self.run_shift_detection_correction(self.ref_path, self.tgt_path,
                                                 **dict(self.coreg_kwargs,
                                                        footprint_poly_ref=None,
                                                        footprint_poly_tgt=None))
        self.assertTrue(CR.success)
83

Daniel Scheffler's avatar
Daniel Scheffler committed
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
    def test_shift_calculation_with_image_coords_only(self):
        """Test with default parameters - should compute X/Y shifts properly ad write the de-shifted target image."""

        # overwrite gt and prj
        ref = GeoArray(self.ref_path)
        ref.to_mem()
        ref.filePath = None
        ref.gt = [0, 1, 0, 0, 0, -1]
        ref.prj = ''
        tgt = GeoArray(self.tgt_path)
        tgt.to_mem()
        tgt.filePath = None
        tgt.gt = [0, 1, 0, 0, 0, -1]
        tgt.prj = ''

        CR = self.run_shift_detection_correction(ref, tgt,
                                                 **dict(self.coreg_kwargs,
                                                        wp=(1200, -1600),
                                                        footprint_poly_ref=None,
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
                                                        footprint_poly_tgt=None))
        self.assertTrue(CR.success)

    def test_shift_calculation_with_float_coords(self):
        """Test with default parameters - should compute X/Y shifts properly ad write the de-shifted target image."""

        # overwrite gt and prj
        ref = GeoArray(self.ref_path)
        ref.to_mem()
        ref.filePath = None
        ref.gt = [330000.00000001, 10.1, 0.0, 5862000.0000001, 0.0, -10.1]
        tgt = GeoArray(self.tgt_path)
        tgt.to_mem()
        tgt.filePath = None
        tgt.gt = [335440.0000001, 10.1, 0.0, 5866490.0000001, 0.0, -10.1]

        CR = self.run_shift_detection_correction(ref, tgt,
                                                 **dict(self.coreg_kwargs,
                                                        wp=(341500.0, 5861440.0),
                                                        footprint_poly_ref=None,
Daniel Scheffler's avatar
Daniel Scheffler committed
123
124
125
                                                        footprint_poly_tgt=None))
        self.assertTrue(CR.success)

126
127
128
129
130
131
132
133
134
135
136
137
    def test_shift_calculation_inmem_gAs_path_out_auto(self):
        """Test input parameter path_out='auto' in case input reference/ target image are in-memory GeoArrays."""
        ref = GeoArray(np.random.randint(1, 100, (1000, 1000)))
        tgt = GeoArray(np.random.randint(1, 100, (1000, 1000)))

        with self.assertRaises(ValueError):
            self.run_shift_detection_correction(ref, tgt,
                                                **dict(self.coreg_kwargs,
                                                       path_out='auto',
                                                       fmt_out='ENVI',
                                                       v=True))

Daniel Scheffler's avatar
Daniel Scheffler committed
138
    # @unittest.SkipTest
139
140
141
    def test_shift_calculation_verboseMode(self):
        """Test the verbose mode - runs the functions of the plotting submodule."""

142
143
144
        CR = self.run_shift_detection_correction(self.ref_path, self.tgt_path,
                                                 **dict(self.coreg_kwargs, v=True))
        self.assertTrue(CR.success)
145
146

    def test_shift_calculation_windowCoveringNodata(self):
147
148
        """Test shift detection in case the given matching window (defined by 'wp' and 'ws' covers the nodata area
         of an input image.
149

150
151
152
153
154
155
156
        Detected subpixel shifts (X/Y): 0.280572488796/-0.11016529071
        Calculated map shifts (X,Y): -7.19427511207/-18.8983470928
        """

        # TODO compare to expected results
        CR = self.run_shift_detection_correction(self.ref_path, self.tgt_path,
                                                 **dict(self.coreg_kwargs,
Daniel Scheffler's avatar
Daniel Scheffler committed
157
158
                                                        wp=test_cases['INTER1']['wp_covering_nodata'],
                                                        ws=(256, 256)))
159
        self.assertTrue(CR.success)
160
161

    def test_shift_calculation_windowAtImageEdge(self):
162
163
        """Test shift detection in case the given matching window is close to an image edge without covering any nodata
        area.
164

165
166
167
168
169
170
171
        Detected subpixel shifts (X/Y): 0.34361492307/-0.320197995758
        Calculated map shifts (X,Y): -6.56385076931/-16.7980200425
        """

        # TODO compare to expected results
        CR = self.run_shift_detection_correction(self.ref_path, self.tgt_path,
                                                 **dict(self.coreg_kwargs,
Daniel Scheffler's avatar
Daniel Scheffler committed
172
                                                        wp=test_cases['INTER1']['wp_close_to_edge'], ws=(256, 256)))
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
        self.assertTrue(CR.success)

    def test_shift_calculation_windowOutside(self):
        """Test if shift computation properly raises a ValueError if the given window position is outside of the image
        overlap."""

        with self.assertRaises(ValueError):
            self.run_shift_detection_correction(self.ref_path, self.tgt_path,
                                                **dict(self.coreg_kwargs,
                                                       wp=test_cases['INTER1']['wp_outside']))

    def test_shift_calculation_windowAtClouds(self):
        """Test if shift computation properly raises a RunTimeError if the matching window is centered at a cloudy
        image position.
        """

        with self.assertRaises(RuntimeError):
            self.run_shift_detection_correction(self.ref_path, self.tgt_path,
                                                **dict(self.coreg_kwargs,
Daniel Scheffler's avatar
Daniel Scheffler committed
192
                                                       wp=test_cases['INTER1']['wp_cloudy'], ws=(256, 256)))
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208

    def test_shift_calculation_differentInputGrids(self):
        """"""

        self.skipTest('Not yet implemented.')

    def test_shift_calculation_withoutPyFFTW(self):
        """"""

        self.skipTest('Not yet implemented.')

    def test_shift_calculation_SSIMdecreases(self):
        """"""

        self.skipTest('Not yet implemented.')

Daniel Scheffler's avatar
Daniel Scheffler committed
209
210
211
212
    # @unittest.SkipTest
    def test_plotting_after_shift_calculation(self):  # , mock_show):
        """Test plotting functionality."""
        # mock_show.return_value = None  # probably not necessary here in your case
213
        CR = self.run_shift_detection_correction(self.ref_path, self.tgt_path, **self.coreg_kwargs)
214
215
216
        self.assertTrue(CR.success)

        # test all the visualization functions
217
218
        CR.show_cross_power_spectrum()
        CR.show_cross_power_spectrum(interactive=True)
Daniel Scheffler's avatar
Daniel Scheffler committed
219
        CR.show_matchWin(interactive=False, after_correction=None)
220
        CR.show_matchWin(interactive=False, after_correction=True)
Daniel Scheffler's avatar
Daniel Scheffler committed
221
222
223
224
        CR.show_matchWin(interactive=False, after_correction=False)
        CR.show_matchWin(interactive=True, after_correction=None)  # only works if test is started with ipython
        CR.show_matchWin(interactive=True, after_correction=True)
        CR.show_matchWin(interactive=True, after_correction=False)
225
        CR.show_image_footprints()
Daniel Scheffler's avatar
Daniel Scheffler committed
226

Daniel Scheffler's avatar
Daniel Scheffler committed
227
# if __name__ == '__main__':
Daniel Scheffler's avatar
Daniel Scheffler committed
228
#    unittest.main(argv=['first-arg-is-ignored'],exit=False, verbosity=2)