-
Notifications
You must be signed in to change notification settings - Fork 2
/
tartan_azure_helpers.py
202 lines (165 loc) · 7.08 KB
/
tartan_azure_helpers.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
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
from azure.storage.blob import ContainerClient
import numpy as np
import io
import cv2
from PIL import Image
import os
from tqdm import tqdm
import time
import re
import timeout_decorator
# Dataset website: http://theairlab.org/tartanair-dataset/
account_url = 'https://tartanair.blob.core.windows.net/'
container_name = 'tartanair-release1'
container_client = ContainerClient(account_url=account_url,
container_name=container_name,
credential=None,
read_timeout=5,
connection_timeout=5)
print("timeouts set to 5")
def get_environment_list():
'''
List all the environments shown in the root directory
'''
env_gen = container_client.walk_blobs()
envlist = []
for env in env_gen:
envlist.append(env.name)
return envlist
def get_trajectory_list(envname, easy_hard = 'Easy'):
'''
List all the trajectory folders, which is named as 'P0XX'
'''
assert(easy_hard=='Easy' or easy_hard=='Hard')
traj_gen = container_client.walk_blobs(name_starts_with=envname + '/' + easy_hard+'/')
trajlist = []
for traj in traj_gen:
trajname = traj.name
trajname_split = trajname.split('/')
trajname_split = [tt for tt in trajname_split if len(tt)>0]
if trajname_split[-1][0] == 'P':
trajlist.append(trajname)
return trajlist
def _list_blobs_in_folder(folder_name):
"""
List all blobs in a virtual folder in an Azure blob container
"""
files = []
generator = container_client.list_blobs(name_starts_with=folder_name)
for blob in generator:
files.append(blob.name)
return files
def get_image_list(trajdir, left_right = 'left'):
assert(left_right == 'left' or left_right == 'right')
files = _list_blobs_in_folder(trajdir + '/image_' + left_right + '/')
files = [fn for fn in files if fn.endswith('.png')]
return files
def get_depth_list(trajdir, left_right = 'left'):
assert(left_right == 'left' or left_right == 'right')
files = _list_blobs_in_folder(trajdir + '/depth_' + left_right + '/')
files = [fn for fn in files if fn.endswith('.npy')]
return files
def get_flow_list(trajdir, ):
files = _list_blobs_in_folder(trajdir + '/flow/')
files = [fn for fn in files if fn.endswith('flow.npy')]
return files
def get_flow_mask_list(trajdir, ):
files = _list_blobs_in_folder(trajdir + '/flow/')
files = [fn for fn in files if fn.endswith('mask.npy')]
return files
def get_posefile(trajdir, left_right = 'left'):
assert(left_right == 'left' or left_right == 'right')
return trajdir + '/pose_' + left_right + '.txt'
def get_seg_list(trajdir, left_right = 'left'):
assert(left_right == 'left' or left_right == 'right')
files = _list_blobs_in_folder(trajdir + '/seg_' + left_right + '/')
files = [fn for fn in files if fn.endswith('.npy')]
return files
def read_numpy_file(numpy_file,):
'''
return a numpy array given the file path
'''
bc = container_client.get_blob_client(blob=numpy_file)
data = bc.download_blob()
ee = io.BytesIO(data.content_as_bytes())
ff = np.load(ee)
return ff
@timeout_decorator.timeout(5, timeout_exception = StopIteration)
def read_image_file(image_file,):
'''
return a uint8 numpy array given the file path
'''
bc = container_client.get_blob_client(blob=image_file)
data = bc.download_blob(timeout=5)
ee = io.BytesIO(data.content_as_bytes())
img=cv2.imdecode(np.asarray(bytearray(ee.read()),dtype=np.uint8), cv2.IMREAD_COLOR)
im_rgb = img[:, :, [2, 1, 0]] # BGR2RGB
return im_rgb
def download_image_directory(image_list,):
if not os.path.exists(re.sub("t/(.*png)", "t/", image_list[0])):
os.makedirs(re.sub("t/(.*png)", "t/", image_list[0]))
for image_name in tqdm(image_list):
if os.path.isfile(image_name):
continue
im = Image.fromarray(read_image_file(image_name))
im.save(image_name)
def read_text_file(text_file,):
'''
return a uint8 numpy array given the file path
'''
bc = container_client.get_blob_client(blob=text_file)
data = bc.download_blob()
ee = data.content_as_text()
return ee
def download_pose_file(pose_file):
if not os.path.exists(re.sub("/p.*txt", "", pose_file)):
os.makedirs(re.sub("/p.*txt", "", pose_file))
pose = read_text_file(pose_file)
fpr = open(pose_file, "wb")
fpr.write(pose)
fpr.close()
def depth2vis(depth, maxthresh = 50):
depthvis = np.clip(depth,0,maxthresh)
depthvis = depthvis/maxthresh*255
depthvis = depthvis.astype(np.uint8)
depthvis = np.tile(depthvis.reshape(depthvis.shape+(1,)), (1,1,3))
return depthvis
def seg2vis(segnp):
colors = [(205, 92, 92), (0, 255, 0), (199, 21, 133), (32, 178, 170), (233, 150, 122), (0, 0, 255), (128, 0, 0), (255, 0, 0), (255, 0, 255), (176, 196, 222), (139, 0, 139), (102, 205, 170), (128, 0, 128), (0, 255, 255), (0, 255, 255), (127, 255, 212), (222, 184, 135), (128, 128, 0), (255, 99, 71), (0, 128, 0), (218, 165, 32), (100, 149, 237), (30, 144, 255), (255, 0, 255), (112, 128, 144), (72, 61, 139), (165, 42, 42), (0, 128, 128), (255, 255, 0), (255, 182, 193), (107, 142, 35), (0, 0, 128), (135, 206, 235), (128, 0, 0), (0, 0, 255), (160, 82, 45), (0, 128, 128), (128, 128, 0), (25, 25, 112), (255, 215, 0), (154, 205, 50), (205, 133, 63), (255, 140, 0), (220, 20, 60), (255, 20, 147), (95, 158, 160), (138, 43, 226), (127, 255, 0), (123, 104, 238), (255, 160, 122), (92, 205, 92),]
segvis = np.zeros(segnp.shape+(3,), dtype=np.uint8)
for k in range(256):
mask = segnp==k
colorind = k % len(colors)
if np.sum(mask)>0:
segvis[mask,:] = colors[colorind]
return segvis
def _calculate_angle_distance_from_du_dv(du, dv, flagDegree=False):
a = np.arctan2( dv, du )
angleShift = np.pi
if ( True == flagDegree ):
a = a / np.pi * 180
angleShift = 180
# print("Convert angle from radian to degree as demanded by the input file.")
d = np.sqrt( du * du + dv * dv )
return a, d, angleShift
def flow2vis(flownp, maxF=500.0, n=8, mask=None, hueMax=179, angShift=0.0):
"""
Show a optical flow field as the KITTI dataset does.
Some parts of this function is the transform of the original MATLAB code flow_to_color.m.
"""
ang, mag, _ = _calculate_angle_distance_from_du_dv( flownp[:, :, 0], flownp[:, :, 1], flagDegree=False )
# Use Hue, Saturation, Value colour model
hsv = np.zeros( ( ang.shape[0], ang.shape[1], 3 ) , dtype=np.float32)
am = ang < 0
ang[am] = ang[am] + np.pi * 2
hsv[ :, :, 0 ] = np.remainder( ( ang + angShift ) / (2*np.pi), 1 )
hsv[ :, :, 1 ] = mag / maxF * n
hsv[ :, :, 2 ] = (n - hsv[:, :, 1])/n
hsv[:, :, 0] = np.clip( hsv[:, :, 0], 0, 1 ) * hueMax
hsv[:, :, 1:3] = np.clip( hsv[:, :, 1:3], 0, 1 ) * 255
hsv = hsv.astype(np.uint8)
rgb = cv2.cvtColor(hsv, cv2.COLOR_HSV2RGB)
if ( mask is not None ):
mask = mask > 0
rgb[mask] = np.array([0, 0 ,0], dtype=np.uint8)
return rgb