You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
112 lines
3.9 KiB
112 lines
3.9 KiB
# 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 numpy as np |
|
import cv2 |
|
import argparse |
|
import time |
|
from utils import Raster, raster2uint8 |
|
|
|
try: |
|
from osgeo import gdal |
|
except ImportError: |
|
import gdal |
|
|
|
|
|
class MatchError (Exception): |
|
def __str__(self): |
|
return "Cannot match two images." |
|
|
|
|
|
def _calcu_tf(im1, im2): |
|
orb = cv2.AKAZE_create() |
|
kp1, des1 = orb.detectAndCompute(im1, None) |
|
kp2, des2 = orb.detectAndCompute(im2, None) |
|
bf = cv2.BFMatcher() |
|
mathces = bf.knnMatch(des2, des1, k=2) |
|
good_matches = [] |
|
for m, n in mathces: |
|
if m.distance < 0.75 * n.distance: |
|
good_matches.append([m]) |
|
if len(good_matches) < 4: |
|
raise MatchError() |
|
src_automatic_points = np.float32([kp2[m[0].queryIdx].pt \ |
|
for m in good_matches]).reshape(-1, 1, 2) |
|
den_automatic_points = np.float32([kp1[m[0].trainIdx].pt \ |
|
for m in good_matches]).reshape(-1, 1, 2) |
|
H, _ = cv2.findHomography(src_automatic_points, den_automatic_points, cv2.RANSAC, 5.0) |
|
return H |
|
|
|
|
|
def _get_match_img(raster, bands): |
|
if len(bands) not in [1, 3]: |
|
raise ValueError("The lenght of bands must be 1 or 3.") |
|
band_array = [] |
|
for b in bands: |
|
band_i = raster.GetRasterBand(b).ReadAsArray() |
|
band_array.append(band_i) |
|
if len(band_array) == 1: |
|
ima = raster2uint8(band_array) |
|
else: |
|
ima = raster2uint8(np.stack(band_array, axis=-1)) |
|
ima = cv2.cvtColor(ima, cv2.COLOR_RGB2GRAY) |
|
return ima |
|
|
|
|
|
def _img2tif(ima, save_path, proj, geot): |
|
if len(ima.shape) == 3: |
|
row, columns, bands = ima.shape |
|
else: |
|
row, columns = ima.shape |
|
bands = 1 |
|
driver = gdal.GetDriverByName("GTiff") |
|
dst_ds = driver.Create(save_path, columns, row, bands, gdal.GDT_UInt16) |
|
dst_ds.SetGeoTransform(geot) |
|
dst_ds.SetProjection(proj) |
|
if bands != 1: |
|
for b in range(bands): |
|
dst_ds.GetRasterBand(b + 1).WriteArray(ima[:, :, b]) |
|
else: |
|
dst_ds.GetRasterBand(1).WriteArray(ima) |
|
dst_ds.FlushCache() |
|
return dst_ds |
|
|
|
|
|
def matching(im1_path, im2_path, im1_bands=[1, 2, 3], im2_bands=[1, 2, 3]): |
|
im1_ras = Raster(im1_path) |
|
im2_ras = Raster(im2_path) |
|
im1 = _get_match_img(im1_ras._src_data, im1_bands) |
|
im2 = _get_match_img(im2_ras._src_data, im2_bands) |
|
H = _calcu_tf(im1, im2) |
|
# test |
|
# im2_t = cv2.warpPerspective(im2, H, (im1.shape[1], im1.shape[0])) |
|
# cv2.imwrite("B_M.png", cv2.cvtColor(im2_t, cv2.COLOR_RGB2BGR)) |
|
im2_arr_t = cv2.warpPerspective(im2_ras.getArray(), H, (im1_ras.width, im1_ras.height)) |
|
save_path = im2_ras.path.replace(("." + im2_ras.ext_type), "_M.tif") |
|
_img2tif(im2_arr_t, save_path, im1_ras.proj, im1_ras.geot) |
|
|
|
|
|
parser = argparse.ArgumentParser(description="input parameters") |
|
parser.add_argument("--im1_path", type=str, required=True, \ |
|
help="The path of time1 image (with geoinfo).") |
|
parser.add_argument("--im2_path", type=str, required=True, \ |
|
help="The path of time1 image.") |
|
|
|
|
|
if __name__ == "__main__": |
|
args = parser.parse_args() |
|
start_time = time.time() |
|
matching(args.im1_path, args.im2_path) |
|
end_time = time.time() |
|
print("Total time:", (end_time - start_time)) |