paddlers/tests/tasks/test_slider_predict.py

# Copyright (c) 2022 PaddlePaddle Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import os.path as osp
import tempfile

import paddlers as pdrs
import paddlers.transforms as T
from testing_utils import CommonTest


class _TestSliderPredictNamespace:
    class TestSliderPredict(CommonTest):
        def test_blocksize_and_overlap_whole(self):
            # Original image size (256, 256)
            with tempfile.TemporaryDirectory() as td:
                # Whole-image inference using predict()
                pred_whole = self.model.predict(self.image_path,
                                                self.transforms)
                pred_whole = pred_whole['label_map']

                # Whole-image inference using slider_predict()
                save_dir = osp.join(td, 'pred1')
                self.model.slider_predict(self.image_path, save_dir, 256, 0,
                                          self.transforms)
                pred1 = T.decode_image(
                    osp.join(save_dir, self.basename),
                    read_raw=True,
                    decode_sar=False)
                self.check_output_equal(pred1.shape, pred_whole.shape)

                # `block_size` == `overlap`
                save_dir = osp.join(td, 'pred2')
                with self.assertRaises(ValueError):
                    self.model.slider_predict(self.image_path, save_dir, 128,
                                              128, self.transforms)

                # `block_size` is a tuple
                save_dir = osp.join(td, 'pred3')
                self.model.slider_predict(self.image_path, save_dir, (128, 32),
                                          0, self.transforms)
                pred3 = T.decode_image(
                    osp.join(save_dir, self.basename),
                    read_raw=True,
                    decode_sar=False)
                self.check_output_equal(pred3.shape, pred_whole.shape)

                # `block_size` and `overlap` are both tuples
                save_dir = osp.join(td, 'pred4')
                self.model.slider_predict(self.image_path, save_dir, (128, 100),
                                          (10, 5), self.transforms)
                pred4 = T.decode_image(
                    osp.join(save_dir, self.basename),
                    read_raw=True,
                    decode_sar=False)
                self.check_output_equal(pred4.shape, pred_whole.shape)

                # `block_size` larger than image size
                save_dir = osp.join(td, 'pred5')
                with self.assertRaises(ValueError):
                    self.model.slider_predict(self.image_path, save_dir, 512, 0,
                                              self.transforms)

        def test_eager_load(self):
            with tempfile.TemporaryDirectory() as td:
                # Lazy
                save_dir = osp.join(td, 'lazy')
                self.model.slider_predict(self.image_path, save_dir, 128, 64,
                                          self.transforms)
                pred_lazy = T.decode_image(
                    osp.join(save_dir, self.basename),
                    read_raw=True,
                    decode_sar=False)

                # Eager
                save_dir = osp.join(td, 'eager')
                self.model.slider_predict(
                    self.image_path,
                    save_dir,
                    128,
                    64,
                    self.transforms,
                    eager_load=True)
                pred_eager = T.decode_image(
                    osp.join(save_dir, self.basename),
                    read_raw=True,
                    decode_sar=False)

                self.check_output_equal(pred_lazy, pred_eager)

        def test_merge_strategy(self):
            with tempfile.TemporaryDirectory() as td:
                # Whole-image inference using predict()
                pred_whole = self.model.predict(self.image_path,
                                                self.transforms)
                pred_whole = pred_whole['label_map']

                # 'keep_first'
                save_dir = osp.join(td, 'keep_first')
                self.model.slider_predict(
                    self.image_path,
                    save_dir,
                    128,
                    64,
                    self.transforms,
                    merge_strategy='keep_first')
                pred_keepfirst = T.decode_image(
                    osp.join(save_dir, self.basename),
                    read_raw=True,
                    decode_sar=False)
                self.check_output_equal(pred_keepfirst.shape, pred_whole.shape)

                # 'keep_last'
                save_dir = osp.join(td, 'keep_last')
                self.model.slider_predict(
                    self.image_path,
                    save_dir,
                    128,
                    64,
                    self.transforms,
                    merge_strategy='keep_last')
                pred_keeplast = T.decode_image(
                    osp.join(save_dir, self.basename),
                    read_raw=True,
                    decode_sar=False)
                self.check_output_equal(pred_keeplast.shape, pred_whole.shape)

                # 'accum'
                save_dir = osp.join(td, 'accum')
                self.model.slider_predict(
                    self.image_path,
                    save_dir,
                    128,
                    64,
                    self.transforms,
                    merge_strategy='accum')
                pred_accum = T.decode_image(
                    osp.join(save_dir, self.basename),
                    read_raw=True,
                    decode_sar=False)
                self.check_output_equal(pred_accum.shape, pred_whole.shape)

                # 'swell'
                save_dir = osp.join(td, 'swell')
                self.model.slider_predict(
                    self.image_path,
                    save_dir,
                    128,
                    64,
                    self.transforms,
                    merge_strategy='swell')
                pred_swell = T.decode_image(
                    osp.join(save_dir, self.basename),
                    read_raw=True,
                    decode_sar=False)
                self.check_output_equal(pred_swell.shape, pred_whole.shape)

        def test_geo_info(self):
            with tempfile.TemporaryDirectory() as td:
                _, geo_info_in = T.decode_image(
                    self.ref_path, read_geo_info=True)
                self.model.slider_predict(self.image_path, td, 128, 0,
                                          self.transforms)
                _, geo_info_out = T.decode_image(
                    osp.join(td, self.basename), read_geo_info=True)
                self.assertEqual(geo_info_out['geo_trans'],
                                 geo_info_in['geo_trans'])
                self.assertEqual(geo_info_out['geo_proj'],
                                 geo_info_in['geo_proj'])

        def test_batch_size(self):
            with tempfile.TemporaryDirectory() as td:
                # batch_size = 1
                save_dir = osp.join(td, 'bs1')
                self.model.slider_predict(
                    self.image_path,
                    save_dir,
                    128,
                    64,
                    self.transforms,
                    merge_strategy='keep_first',
                    batch_size=1)
                pred_bs1 = T.decode_image(
                    osp.join(save_dir, self.basename),
                    read_raw=True,
                    decode_sar=False)

                # batch_size = 4
                save_dir = osp.join(td, 'bs4')
                self.model.slider_predict(
                    self.image_path,
                    save_dir,
                    128,
                    64,
                    self.transforms,
                    merge_strategy='keep_first',
                    batch_size=4)
                pred_bs4 = T.decode_image(
                    osp.join(save_dir, self.basename),
                    read_raw=True,
                    decode_sar=False)
                self.check_output_equal(pred_bs4, pred_bs1)

                # batch_size = 8
                save_dir = osp.join(td, 'bs4')
                self.model.slider_predict(
                    self.image_path,
                    save_dir,
                    128,
                    64,
                    self.transforms,
                    merge_strategy='keep_first',
                    batch_size=8)
                pred_bs8 = T.decode_image(
                    osp.join(save_dir, self.basename),
                    read_raw=True,
                    decode_sar=False)
                self.check_output_equal(pred_bs8, pred_bs1)


class TestSegSliderPredict(_TestSliderPredictNamespace.TestSliderPredict):
    def setUp(self):
        self.model = pdrs.tasks.seg.UNet(in_channels=10)
        self.transforms = [T.Normalize([0.5] * 10, [0.5] * 10)]
        self.image_path = "data/ssst/multispectral.tif"
        self.ref_path = self.image_path
        self.basename = osp.basename(self.ref_path)


class TestCDSliderPredict(_TestSliderPredictNamespace.TestSliderPredict):
    def setUp(self):
        self.model = pdrs.tasks.cd.BIT(in_channels=10)
        self.transforms = [T.Normalize([0.5] * 10, [0.5] * 10)]
        self.image_path = ("data/ssmt/multispectral_t1.tif",
                           "data/ssmt/multispectral_t2.tif")
        self.ref_path = self.image_path[0]
        self.basename = osp.basename(self.ref_path)