-
Notifications
You must be signed in to change notification settings - Fork 1
/
curate_FLIR_data.py
187 lines (130 loc) · 7.31 KB
/
curate_FLIR_data.py
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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
import os
import json
import cv2
import imutils
import numpy as np
"""
credit:
https://medium.com/@aswinvb/how-to-perform-thermal-to-visible-image-registration-c18a34894866
"""
def align_images(therm: np.ndarray, vis: np.ndarray, output: str, dims: tuple = (512, 640)):
"""
Aligns (1024, 1224) visible image onto (512, 640) thermal image. Thermal image is subset of visible, so this
function identifies overlapping pixels and restricts to these.
:param therm: np.ndarray, thermal image
:param vis: np.ndarray, matched visible image
:param output: str, full path and name of file to output
:param dims: tuple, (height, width) dimensions of desired output image
:return:
None
Writes (as default) 512 x 1280 concatenated image to disk
"""
therm = cv2.cvtColor(therm, cv2.COLOR_BGR2GRAY)
vis = cv2.cvtColor(vis, cv2.COLOR_BGR2GRAY)
# histogram equalization
clahe = cv2.createCLAHE(clipLimit=1.0, tileGridSize=(15, 15))
therm = clahe.apply(therm)
vis = clahe.apply(vis)
# Apply Canny line sequence detection algorithm
therm_canny = cv2.Canny(therm, 100, 200)
# bookkeeping variable to keep track of the matched region
found = None
# loop over the scales of the image
for scale in np.linspace(0.2, 1.0, 20)[::-1]:
# resize the image according to the scale, and keep track
# of the ratio of the resizing
resized = imutils.resize(vis, width=int(vis.shape[1] * scale))
r = vis.shape[1] / float(resized.shape[1])
# if the resized image is smaller than the template, then break
# from the loop
if resized.shape[0] < therm.shape[0] or resized.shape[1] < therm.shape[1]:
break
# detect edges in the resized, grayscale image and apply template
# matching to find the template in the image
edged = cv2.Canny(resized, 100, 200)
result = cv2.matchTemplate(edged, therm_canny, cv2.TM_CCOEFF)
(_, maxVal, _, maxLoc) = cv2.minMaxLoc(result)
# if we have found a new maximum correlation value, then update bookkeeping variable
if found is None or maxVal > found[0]:
found = (maxVal, maxLoc, r)
# unpack the bookkeeping variable and compute the (x, y) coordinates
# of the bounding box based on the resized ratio
(_, maxLoc, r) = found
(startX, startY) = (int(maxLoc[0] * r), int(maxLoc[1] * r))
(endX, endY) = (int((maxLoc[0] + therm.shape[1]) * r), int((maxLoc[1] + therm.shape[0]) * r))
# draw a bounding box around the detected result and display the image
cv2.rectangle(vis, (startX, startY), (endX, endY), (0, 0, 255), 2)
crop_img = vis[startY:endY, startX:endX]
# resize both images to desired dimensions
resized_therm = cv2.resize(therm, (dims[1], dims[0]))
resized_vis = cv2.resize(crop_img, (dims[1], dims[0]))
# blur to smooth noise
therm_enhanced = cv2.GaussianBlur(resized_therm, (0,0), sigmaX=0.5, sigmaY=0.5)
# Sharpen thermal image
kernel = np.array([[0, -1, 0],
[-1, 5, -1],
[0, -1, 0]])
therm_enhanced = cv2.filter2D(src=therm_enhanced, ddepth=-1, kernel=kernel)
# Concat horizontally
concatenated = cv2.hconcat([therm_enhanced, resized_vis])
cv2.imwrite(output, concatenated)
if __name__ == '__main__':
output_dir = '/Users/josephking/Documents/sponsored_projects/MERGEN/data/FLIR_matched_gray_thermal'
################################
###### Images from Europe ######
################################
europe = '/Users/josephking/Documents/sponsored_projects/MERGEN/data/FLIR_ADAS_DATASET/Europe_1_0_2_full/Europe_1_0_2'
for subset in ['val', 'train']:
# Crosswalk
with open(os.path.join(europe, subset, 'europe_thermal_to_rgb.json')) as f:
crosswalk = json.load(f)
crosswalk = crosswalk['thermal_to_rgb_ordered_frames'] # converts to list
for matched_pair in range(len(crosswalk)):
# See if matched images exist, read in if they do
therm_path = os.path.join(europe, subset, 'thermal_8_bit', crosswalk[matched_pair]['thermal_filename'])
vis_path = os.path.join(europe, subset, 'RGB', crosswalk[matched_pair]['rgb_filename'])
if (os.path.exists(therm_path)) and (os.path.exists(vis_path)):
therm = cv2.imread(therm_path)
vis = cv2.imread(vis_path)
if (isinstance(therm, np.ndarray)) and (isinstance(vis, np.ndarray)): # ensure they read in correctly
name = f"europe_{subset}_{matched_pair}.png"
align_images(therm=therm, vis=vis, output=os.path.join(output_dir, name))
else:
if subset == 'val': # check if image in val_video folder instead
# See if matched images exist, read in if they do
therm_path = os.path.join(europe, 'val_video', 'thermal_8_bit', crosswalk[matched_pair]['thermal_filename'])
vis_path = os.path.join(europe, 'val_video', 'RGB', crosswalk[matched_pair]['rgb_filename'])
if (os.path.exists(therm_path)) and (os.path.exists(vis_path)):
therm = cv2.imread(therm_path)
vis = cv2.imread(vis_path)
if (isinstance(therm, np.ndarray)) and (isinstance(vis, np.ndarray)): # ensure they read in correctly
name = f"europe_video_{matched_pair}.png"
align_images(therm=therm, vis=vis, output=os.path.join(output_dir, name))
################################
### Images from San Francisco ##
################################
sf = '/Users/josephking/Documents/sponsored_projects/MERGEN/data/FLIR_ADAS_DATASET/FLIR_ADAS_SF_1_0_0'
for subset in ['val', 'train']:
files = [i for i in os.listdir(os.path.join(sf, subset, 'thermal_8_bit'))]
for file in files:
# See if matched images exist, read in if they do
therm_path = os.path.join(sf, subset, 'thermal_8_bit', file)
vis_path = os.path.join(sf, subset, 'RGB', file)
if (os.path.exists(therm_path)) and (os.path.exists(vis_path)):
therm = cv2.imread(therm_path)
vis = cv2.imread(vis_path)
if (isinstance(therm, np.ndarray)) and (isinstance(vis, np.ndarray)): # ensure they read in correctly
name = f"sf_{subset}_{file[5:-5]}.png"
align_images(therm=therm, vis=vis, output=os.path.join(output_dir, name))
##### video #####
files = [i for i in os.listdir(os.path.join(sf, 'video', 'thermal_8_bit'))]
for file in files:
# See if matched images exist, read in if they do
therm_path = os.path.join(sf, 'video', 'thermal_8_bit', file)
vis_path = os.path.join(sf, 'video', 'RGB', file.replace('jpeg', 'jpg')) # RGB files use JPG suffix
if (os.path.exists(therm_path)) and (os.path.exists(vis_path)):
therm = cv2.imread(therm_path)
vis = cv2.imread(vis_path)
if (isinstance(therm, np.ndarray)) and (isinstance(vis, np.ndarray)): # ensure they read in correctly
name = f"sf_video_{file[5:-5]}.png"
align_images(therm=therm, vis=vis, output=os.path.join(output_dir, name))