1.root_path = './'
raw_data_file = osp.join(root_path, 'raw_data', 'raw_skes_data.pkl')
save_path = osp.join(root_path, 'denoised_data')-》載入上一個檔案處理後的.pkl檔案路徑
2.if not osp.exists(save_path):
os.mkdir(save_path)-》如果儲存路徑不存在則建立此路徑
3.rgb_ske_path = osp.join(save_path, 'rgb+ske')
if not osp.exists(rgb_ske_path):
os.mkdir(rgb_ske_path)-》建立save_path路徑下的rgb+ske檔案夾
4.actors_info_dir = osp.join(save_path, 'actors_info')
if not osp.exists(actors_info_dir):
os.mkdir(actors_info_dir)-》建立save_path路徑下的actors_info檔案夾
5.missing_count = 0
noise_len_thres = 11
noise_spr_thres1 = 0.8
noise_spr_thres2 = 0.69754
noise_mot_thres_lo = 0.089925
noise_mot_thres_hi = 2-》建立各種參數
6.noise_len_logger = logging.getLogger('noise_length')-》建立日志
noise_len_logger.setLevel(logging.INFO)-》設定日志等級
noise_len_logger.addHandler(logging.FileHandler(osp.join(save_path, 'noise_length.log')))-》存儲日志
noise_len_logger.info('{:^20}\t{:^17}\t{:^8}\t{}'.format('Skeleton', 'bodyID', 'Motion', 'Length'))-》日志資訊格式
首先介紹get_raw_denoised_data方法
1.此函數功能：從原始骨架序列中擷取去噪資料(關節位置和顔色位置)。
解釋:對于骨骼序列的每一幀，都表示一個演員的25個關節的三維位置通過一個二維數組(形狀:25 x 3)被重新塑造成一個75-維矢量通過連接配接每個3-維(x, y, z)坐标沿行維按關節順序排列。每一幀包含兩個演員的關節位置構成一個150-維向量。如果隻有一個演員，然後最後75個值被0填充。否則，選擇主參與者和第二個演員基于動作量。将每個150-維向量作為行向量放入一個二維numpy數組，其中行數等于有效幀數。所有這些2D數組被放到一個清單中，最後這個清單被序列化到一個cPickle檔案中。對于包含兩個或更多參與者的骨架序列(大多對應于最後11個類)，檔案名和參與者的資訊被記錄到日志檔案中。為了更好地了解，還可以生成可視化的RGB+骨架視訊。
2.with open(raw_data_file, 'rb') as fr: # load raw skeletons data
raw_skes_data = pickle.load(fr)-》打開.pkl檔案
3.num_skes = len(raw_skes_data)
print('Found %d available skeleton sequences.' % num_skes)-》列印檔案有多少資料總量
3.raw_denoised_joints = []
raw_denoised_colors = []
frames_cnt = []-》建立3個空數組
4. for (idx, bodies_data) in enumerate(raw_skes_data):-》周遊.pkl檔案取出id和骨骼名
ske_name = bodies_data['name']-》取出bodies_data中ID為name這一列的值
print('Processing %s' % ske_name)-》列印
num_bodies = len(bodies_data['data'])-》取出“”data“”列的資料長度
5.if num_bodies == 1: # only 1 actor-》如果幀中隻有一個人物
num_frames = bodies_data['num_frames']-》擷取此人的動作幀數
body_data = list(bodies_data['data'].values())[0]-》取出骨骼資料的第一行
joints, colors = get_one_actor_points(body_data, num_frames)-》擷取關節和像素值
else: -》 超過一個演員，選擇兩個主要演員
joints, colors = get_two_actors_points(bodies_data)-》擷取兩個演員的關節
# Remove missing frames
joints, colors = remove_missing_frames(ske_name, joints, colors)-》移除掉有丢失的幀的關節像素。
num_frames = joints.shape[0] # 更新關節總數
6. raw_denoised_joints.append(joints)
raw_denoised_colors.append(colors)
frames_cnt.append(num_frames)-》把joints，colors，num_frames傳入數組變量
7.if (idx + 1) % 1000 == 0:
print('Processed: %.2f%% (%d / %d), ' % \
(100.0 * (idx + 1) / num_skes, idx + 1, num_skes) + \
'Missing count: %d' % missing_count)-》每處理1000次列印一次處理資訊
8.
raw_skes_joints_pkl = osp.join(save_path, 'raw_denoised_joints.pkl')
with open(raw_skes_joints_pkl, 'wb') as f:
pickle.dump(raw_denoised_joints, f, pickle.HIGHEST_PROTOCOL)
raw_skes_colors_pkl = osp.join(save_path, 'raw_denoised_colors.pkl')
with open(raw_skes_colors_pkl, 'wb') as f:
pickle.dump(raw_denoised_colors, f, pickle.HIGHEST_PROTOCOL)-》打開raw_denoised_joints.pkl和raw_denoised_colors.pkl檔案
9.
frames_cnt = np.array(frames_cnt, dtype=np.int)
np.savetxt(osp.join(save_path, 'frames_cnt.txt'), frames_cnt, fmt='%d')
print('Saved raw denoised positions of {} frames into {}'.format(np.sum(frames_cnt),
raw_skes_joints_pkl))
print('Found %d files that have missing data' % missing_count)-》儲存總幀數到指定路徑的txt檔案裡面去。
解讀get_two_actors_points方法：
作用：獲得第一個和第二個演員的關節位置和顔色位置。
資料構成：bodies_data (dict): 3個鍵值對:'name'， 'data'， 'num_frames'。
bodies_data['data'] 也是一個字典, 當這個鍵是身體ID, 值是對應的身體資料（關節像素），它也是一個有4個鍵的字典:
關節:原始的三維關節位置。形狀:(num_frames x 25,3)
顔色:原始的2D顔色位置。形狀:(num_frames, 25,2)
間隔:記錄幀索引的清單。
運動:運動數量
傳回的是關節,顔色坐标。
1.ske_name = bodies_data['name']-》提取骨骼名
2.label = int(ske_name[-2:])-》取出所有name作為标簽。
3.num_frames = bodies_data['num_frames']-》去幀總數
4.bodies_info = get_bodies_info(bodies_data['data'])-》取body資料
5. bodies_data, noise_info = denoising_bodies_data(bodies_data) -》把body資料放入去噪方法中處理
bodies_info += noise_info-》作body資訊和噪聲資訊的和
6. bodies_data = list(bodies_data)-》放入清單
7.if len(bodies_data) == 1: -》判斷body資料是不是一個人的
if label >= 50: -》雙主體動作去噪失敗
fail_logger_2.info(ske_name)-》記錄日志
bodyID, body_data = bodies_data[0]-》取body資料第一個的ID和資料
joints, colors = get_one_actor_points(body_data, num_frames)-》将body資料和總幀數傳入get_one_actor_points取出關節和像素
bodies_info += 'Main actor: %s' % bodyID-》将ID加入到bodies_info 中
else:
if label < 50: -》單主體動作去噪失敗
fail_logger_1.info(ske_name)-》記錄日志
8.joints = np.zeros((num_frames, 150), dtype=np.float32)-》建立一個行數為總幀數，150列的填充0的矩陣
9.colors = np.ones((num_frames, 2, 25, 2), dtype=np.float32) * np.nan-》建立一個帶總幀數的以1為填充的4維矩陣。
10.bodyID, actor1 = bodies_data[0]-》取body資料第一個的ID和資料
11.start1, end1 = actor1['interval'][0], actor1['interval'][-1]-》記錄bodies_data[0]行為資料的第一行和最後一行
12.joints[start1:end1 + 1, :75] = actor1['joints'].reshape(-1, 75)--》取出行為資料中的關節資料
13.colors[start1:end1 + 1, 0] = actor1['colors']--》取出行為資料中的像素資料
14.actor1_info = '{:^17}\t{}\t{:^8}\n'.format('Actor1', 'Interval', 'Motion') + \
'{}\t{:^8}\t{:f}\n'.format(bodyID, str([start1, end1]), actor1['motion'])-》比對正規表達式的字元串
15. del bodies_data[0]-》删除bodies_data[0]對象
16.actor2_info = '{:^17}\t{}\t{:^8}\n'.format('Actor2', 'Interval', 'Motion')-》-》比對正規表達式的字元串
17.start2, end2 = [0, 0] -》actor2的初始間隔(虛拟)
18.while len(bodies_data) > 0:-》如果資料不為空
bodyID, actor = bodies_data[0]-》取body資料第一個的ID和資料
start, end = actor['interval'][0], actor['interval'][-1]-》記錄actor中間隔資料的第一行和最後一行
19.if min(end1, end) - max(start1, start) <= 0: -》與actor1沒有重疊
joints[start:end + 1, :75] = actor['joints'].reshape(-1, 75)-》取出關節資料并重新定形
colors[start:end + 1, 0] = actor['colors']-》取出像素資料
actor1_info += '{}\t{:^8}\t{:f}\n'.format(bodyID, str([start, end]), actor['motion']) -》 更新actor1的間隔
start1 = min(start, start1)-》更新start1值
end1 = max(end, end1)-》更新end1值
20.del bodies_data[0]-》删除bodies_data值
21.bodies_info += ('\n' + actor1_info + '\n' + actor2_info)-》更新body資訊，加入actor1_info和actor2_info資訊
22. with open(osp.join(actors_info_dir, ske_name + '.txt'), 'w') as fw:
fw.write(bodies_info + '\n')-》寫入txt檔案
解讀get_bodies_info方法：
1.bodies_info = '{:^17}\t{}\t{:^8}\n'.format('bodyID', 'Interval', 'Motion')-》規定bodies_info格式
2.for (bodyID, body_data) in bodies_data.items():-》循環周遊出bodyID, body_data
start, end = body_data['interval'][0], body_data['interval'][-1]-》取出第一個和最後一個索引
bodies_info += '{}\t{:^8}\t{:f}\n'.format(bodyID, str([start, end]), body_data['motion'])-》添加到bodies_info
3.return bodies_info + '\n'-》傳回bodies_info資訊
解讀remove_missing_frames方法：
作用：切掉所有關節位置為0的缺失幀 。
對于有2個演員資料的序列，也記錄缺失幀數actor1和actor2(用于調試)。
1.num_frames = joints.shape[0]-》總幀數
num_bodies = colors.shape[1] -》主體數
2.if num_bodies == 2: -》如果主體數為2
missing_indices_1 = np.where(joints[:, :75].sum(axis=1) == 0)[0]-》查詢主體1丢失的索引
missing_indices_2 = np.where(joints[:, 75:].sum(axis=1) == 0)[0]-》查詢主體2丢失的索引
cnt1 = len(missing_indices_1)-》記錄主體1丢失的索引的長度
cnt2 = len(missing_indices_2)-》記錄主體2丢失的索引的長度
3.start = 1 if 0 in missing_indices_1 else 0-》判斷start 值
end = 1 if num_frames - 1 in missing_indices_1 else 0-》判斷end的值
4.if max(cnt1, cnt2) > 0:-》寫入日志
if cnt1 > cnt2:
info = '{}\t{:^10d}\t{:^6d}\t{:^6d}\t{:^5d}\t{:^3d}'.format(ske_name, num_frames,
cnt1, cnt2, start, end)
missing_skes_logger1.info(info)
else:
info = '{}\t{:^10d}\t{:^6d}\t{:^6d}'.format(ske_name, num_frames, cnt1, cnt2)
missing_skes_logger2.info(info)
找到資料沒有丢失或丢失的有效架構索引
對于雙主體動作，這意味着角色1和角色2的資料都缺失了。
5.valid_indices = np.where(joints.sum(axis=1) != 0)[0]-》基于指數
missing_indices = np.where(joints.sum(axis=1) == 0)[0]-》查詢索引
num_missing = len(missing_indices)-》記錄長度
6.if num_missing > 0: -》更新關節和顔色
joints = joints[valid_indices]
colors[missing_indices] = np.nan
global missing_count
missing_count += 1
missing_skes_logger.info('{}\t{:^10d}\t{:^11d}'.format(ske_name, num_frames, num_missing))-》寫入日志
解讀get_one_actor_points方法：
作用：隻為一個演員找關節和顔色。
對于關節，每個坐标系包含75個X-Y-Z坐标。
對于顔色，每個幀包含25 x 2 (x, Y)坐标。
1.joints = np.zeros((num_frames, 75), dtype=np.float32)-》建立關節數組
2.colors = np.ones((num_frames, 1, 25, 2), dtype=np.float32) * np.nan-》建立像素數組
3. start, end = body_data['interval'][0], body_data['interval'][-1]-》記錄start, end值
4. joints[start:end + 1] = body_data['joints'].reshape(-1, 75)-》載入關節值
5. colors[start:end + 1, 0] = body_data['colors']-》載入像素值
解讀denoising_bodies_data方法
作用：基于一些啟發式方法去噪資料，不一定對所有樣本都正确。
傳回:元組:(bodyID, body_data)。
1.ske_name = bodies_data['name']-》取出name值
2.bodies_data = bodies_data['data']-》取出data值
3.bodies_data, noise_info_len = denoising_by_length(ske_name, bodies_data)-》步驟1:基于幀長去噪。
4.if len(bodies_data) == 1: -》步驟1後隻剩下一個bodyID
return bodies_data.items(), noise_info_len-》傳回可周遊的(鍵, 值) 元組數組bodies_data和長度。
基于擴散的去噪：
5.bodies_data, noise_info_spr, denoised_by_spr = denoising_by_spread(ske_name, bodies_data)-》獲得body資料、噪聲資料、去燥資料
6. if len(bodies_data) == 1:-》如果bodies_data長度為1
return bodies_data.items(), noise_info_len + noise_info_spr-》傳回可周遊的(鍵, 值) 元組數組bodies_data和噪聲資料噪聲長度。
7.bodies_motion = dict() -》讓身體運動
8.for (bodyID, body_data) in bodies_data.items():-》取出bodyID和body資料
bodies_motion[bodyID] = body_data['motion']-》取出對于ID的運動資訊
9.bodies_motion = sorted(bodies_motion.items(), key=lambda x: x[1], reverse=True)-》基于運動對主體進行排序
denoised_bodies_data = list()-》建立一個空清單
10.for (bodyID, _) in bodies_motion:-》周遊運動資訊
denoised_bodies_data.append((bodyID, bodies_data[bodyID]))-》将ID和ID對應bodies_data加入在去噪資料中
return denoised_bodies_data, noise_info_len + noise_info_spr-》傳回去燥資料，噪聲資訊
解讀denoising_by_motion方法：
作用過濾出運動超出預定義間隔範圍的bodyID
1.bodies_motion = sorted(bodies_motion.items(), key=lambda x: x[1], reverse=True)-》根據動作對主體進行排序，傳回元組清單
2.denoised_bodies_data = [(bodies_motion[0][0], bodies_data[bodies_motion[0][0]])]-》保留最大動作的身體資料
noise_info = str() -》将對象轉化為适于人閱讀的形式。
3. for (bodyID, motion) in bodies_motion[1:]:-》周遊取出bodyID和動作資訊
if (motion < noise_mot_thres_lo) or (motion > noise_mot_thres_hi):-》如果motion 不正常的話
noise_info += 'Filter out: %s, %.6f (motion).\n' % (bodyID, motion)-》加入噪聲資訊
noise_mot_logger.info('{}\t{}\t{:.6f}'.format(ske_name, bodyID, motion))-》加入日志
else:
denoised_bodies_data.append((bodyID, bodies_data[bodyID]))-》否則正常去燥body資料
解讀denoising_by_spread方法
作用：根據Y值和X值的擴散對資料進行去噪。濾除噪聲幀比高于預先定義的bodyID。
1.noise_info = str()
2.denoised_by_spr = False-》标記該序列是否已被擴充處理。
3.new_bodies_data = bodies_data.copy()-》複制bodies_data資料
4.for (bodyID, body_data) in new_bodies_data.items():-》周遊得到bodyID, body_data
if len(bodies_data) == 1:-》若隻有一條資料跳出
break
5.valid_frames = get_valid_frames_by_spread(body_data['joints'].reshape(-1, 25, 3))-》得到正常幀
6. num_frames = len(body_data['interval'])-》得到總幀數
7. num_noise = num_frames - len(valid_frames)-》得到總噪聲數
8. if num_noise == 0:-》如果沒有噪聲，繼續執行
continue
9.ratio = num_noise / float(num_frames)-》得到噪聲率
10.motion = body_data['motion']-》取出body_data裡面的動作資料
11.if ratio >= noise_spr_thres2: -》如果噪聲率大于0.69754
del bodies_data[bodyID]-》删除bodies_data[bodyID]
denoised_by_spr = True-》更新denoised_by_spr 為true
noise_info += 'Filter out: %s (spread rate >= %.2f).\n' % (bodyID, noise_spr_thres2)-》添加噪聲資訊
noise_spr_logger.info('%s\t%s\t%.6f\t%.6f' % (ske_name, bodyID, motion, ratio))-》添加噪聲日志
12.else: -》否則更新動作
joints = body_data['joints'].reshape(-1, 25, 3)[valid_frames]-》将關節資料重新組織指派給joint
body_data['motion'] = min(motion, np.sum(np.var(joints.reshape(-1, 3), axis=0)))-》求關節的方差的和與動作值的最小值指派給body_data['motion']
noise_info += '%s: motion %.6f -> %.6f\n' % (bodyID, motion, body_data['motion'])-》更新noise_info
解讀get_valid_frames_by_spread方法
作用：根據X和Y的擴充找到有效的(或合理的)幀(索引)。
1.num_frames = points.shape[0]-》取出總幀數
valid_frames = []
2.for i in range(num_frames):-》周遊
x = points[i, :, 0]
y = points[i, :, 1]
3. if (x.max() - x.min()) <= noise_spr_thres1 * (y.max() - y.min()): -》若x小于等于0.8y
valid_frames.append(i)-》清單末尾添加新的對象i。
return valid_frames
解讀denoising_by_length方法：
作用：根據每個bodyID的幀長度去噪資料。過濾長度小于或等于預定義門檻值的bodyID
1.noise_info = str()
new_bodies_data = bodies_data.copy()-》複制資料
2.for (bodyID, body_data) in new_bodies_data.items():
length = len(body_data['interval'])
if length <= noise_len_thres:-》若長度小于11
noise_info += 'Filter out: %s, %d (length).\n' % (bodyID, length)-》添加噪聲資訊
noise_len_logger.info('{}\t{}\t{:.6f}\t{:^6d}'.format(ske_name, bodyID,-》寫入日志
body_data['motion'], length))
del bodies_data[bodyID]-》删除對應資料
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
import os
import os.path as osp
import numpy as np
import pickle
import logging
root_path = './'
raw_data_file = osp.join(root_path, 'raw_data', 'raw_skes_data.pkl')
save_path = osp.join(root_path, 'denoised_data')
if not osp.exists(save_path):
os.mkdir(save_path)
rgb_ske_path = osp.join(save_path, 'rgb+ske')
if not osp.exists(rgb_ske_path):
os.mkdir(rgb_ske_path)
actors_info_dir = osp.join(save_path, 'actors_info')
if not osp.exists(actors_info_dir):
os.mkdir(actors_info_dir)
missing_count = 0
noise_len_thres = 11
noise_spr_thres1 = 0.8
noise_spr_thres2 = 0.69754
noise_mot_thres_lo = 0.089925
noise_mot_thres_hi = 2
noise_len_logger = logging.getLogger('noise_length')
noise_len_logger.setLevel(logging.INFO)
noise_len_logger.addHandler(logging.FileHandler(osp.join(save_path, 'noise_length.log')))
noise_len_logger.info('{:^20}\t{:^17}\t{:^8}\t{}'.format('Skeleton', 'bodyID', 'Motion', 'Length'))
noise_spr_logger = logging.getLogger('noise_spread')
noise_spr_logger.setLevel(logging.INFO)
noise_spr_logger.addHandler(logging.FileHandler(osp.join(save_path, 'noise_spread.log')))
noise_spr_logger.info('{:^20}\t{:^17}\t{:^8}\t{:^8}'.format('Skeleton', 'bodyID', 'Motion', 'Rate'))
noise_mot_logger = logging.getLogger('noise_motion')
noise_mot_logger.setLevel(logging.INFO)
noise_mot_logger.addHandler(logging.FileHandler(osp.join(save_path, 'noise_motion.log')))
noise_mot_logger.info('{:^20}\t{:^17}\t{:^8}'.format('Skeleton', 'bodyID', 'Motion'))
fail_logger_1 = logging.getLogger('noise_outliers_1')
fail_logger_1.setLevel(logging.INFO)
fail_logger_1.addHandler(logging.FileHandler(osp.join(save_path, 'denoised_failed_1.log')))
fail_logger_2 = logging.getLogger('noise_outliers_2')
fail_logger_2.setLevel(logging.INFO)
fail_logger_2.addHandler(logging.FileHandler(osp.join(save_path, 'denoised_failed_2.log')))
missing_skes_logger = logging.getLogger('missing_frames')
missing_skes_logger.setLevel(logging.INFO)
missing_skes_logger.addHandler(logging.FileHandler(osp.join(save_path, 'missing_skes.log')))
missing_skes_logger.info('{:^20}\t{}\t{}'.format('Skeleton', 'num_frames', 'num_missing'))
missing_skes_logger1 = logging.getLogger('missing_frames_1')
missing_skes_logger1.setLevel(logging.INFO)
missing_skes_logger1.addHandler(logging.FileHandler(osp.join(save_path, 'missing_skes_1.log')))
missing_skes_logger1.info('{:^20}\t{}\t{}\t{}\t{}\t{}'.format('Skeleton', 'num_frames', 'Actor1',
'Actor2', 'Start', 'End'))
missing_skes_logger2 = logging.getLogger('missing_frames_2')
missing_skes_logger2.setLevel(logging.INFO)
missing_skes_logger2.addHandler(logging.FileHandler(osp.join(save_path, 'missing_skes_2.log')))
missing_skes_logger2.info('{:^20}\t{}\t{}\t{}'.format('Skeleton', 'num_frames', 'Actor1', 'Actor2'))
def denoising_by_length(ske_name, bodies_data):
"""
Denoising data based on the frame length for each bodyID.
Filter out the bodyID which length is less or equal than the predefined threshold.
"""
noise_info = str()
new_bodies_data = bodies_data.copy()
for (bodyID, body_data) in new_bodies_data.items():
length = len(body_data['interval'])
if length <= noise_len_thres:
noise_info += 'Filter out: %s, %d (length).\n' % (bodyID, length)
noise_len_logger.info('{}\t{}\t{:.6f}\t{:^6d}'.format(ske_name, bodyID,
body_data['motion'], length))
del bodies_data[bodyID]
if noise_info != '':
noise_info += '\n'
return bodies_data, noise_info
def get_valid_frames_by_spread(points):
"""
Find the valid (or reasonable) frames (index) based on the spread of X and Y.
:param points: joints or colors
"""
num_frames = points.shape[0]
valid_frames = []
for i in range(num_frames):
x = points[i, :, 0]
y = points[i, :, 1]
if (x.max() - x.min()) <= noise_spr_thres1 * (y.max() - y.min()): # 0.8
valid_frames.append(i)
return valid_frames
def denoising_by_spread(ske_name, bodies_data):
"""
Denoising data based on the spread of Y value and X value.
Filter out the bodyID which the ratio of noisy frames is higher than the predefined
threshold.
bodies_data: contains at least 2 bodyIDs
"""
noise_info = str()
denoised_by_spr = False # mark if this sequence has been processed by spread.
new_bodies_data = bodies_data.copy()
# for (bodyID, body_data) in bodies_data.items():
for (bodyID, body_data) in new_bodies_data.items():
if len(bodies_data) == 1:
break
valid_frames = get_valid_frames_by_spread(body_data['joints'].reshape(-1, 25, 3))
num_frames = len(body_data['interval'])
num_noise = num_frames - len(valid_frames)
if num_noise == 0:
continue
ratio = num_noise / float(num_frames)
motion = body_data['motion']
if ratio >= noise_spr_thres2: # 0.69754
del bodies_data[bodyID]
denoised_by_spr = True
noise_info += 'Filter out: %s (spread rate >= %.2f).\n' % (bodyID, noise_spr_thres2)
noise_spr_logger.info('%s\t%s\t%.6f\t%.6f' % (ske_name, bodyID, motion, ratio))
else: # Update motion
joints = body_data['joints'].reshape(-1, 25, 3)[valid_frames]
body_data['motion'] = min(motion, np.sum(np.var(joints.reshape(-1, 3), axis=0)))
noise_info += '%s: motion %.6f -> %.6f\n' % (bodyID, motion, body_data['motion'])
# TODO: Consider removing noisy frames for each bodyID
if noise_info != '':
noise_info += '\n'
return bodies_data, noise_info, denoised_by_spr
def denoising_by_motion(ske_name, bodies_data, bodies_motion):
"""
Filter out the bodyID which motion is out of the range of predefined interval
"""
# Sort bodies based on the motion, return a list of tuples
# bodies_motion = sorted(bodies_motion.items(), key=lambda x, y: cmp(x[1], y[1]), reverse=True)
bodies_motion = sorted(bodies_motion.items(), key=lambda x: x[1], reverse=True)
# Reserve the body data with the largest motion
denoised_bodies_data = [(bodies_motion[0][0], bodies_data[bodies_motion[0][0]])]
noise_info = str()
for (bodyID, motion) in bodies_motion[1:]:
if (motion < noise_mot_thres_lo) or (motion > noise_mot_thres_hi):
noise_info += 'Filter out: %s, %.6f (motion).\n' % (bodyID, motion)
noise_mot_logger.info('{}\t{}\t{:.6f}'.format(ske_name, bodyID, motion))
else:
denoised_bodies_data.append((bodyID, bodies_data[bodyID]))
if noise_info != '':
noise_info += '\n'
return denoised_bodies_data, noise_info
def denoising_bodies_data(bodies_data):
"""
Denoising data based on some heuristic methods, not necessarily correct for all samples.
Return:
denoised_bodies_data (list): tuple: (bodyID, body_data).
"""
ske_name = bodies_data['name']
bodies_data = bodies_data['data']
# Step 1: Denoising based on frame length.
bodies_data, noise_info_len = denoising_by_length(ske_name, bodies_data)
if len(bodies_data) == 1: # only has one bodyID left after step 1
return bodies_data.items(), noise_info_len
# Step 2: Denoising based on spread.
bodies_data, noise_info_spr, denoised_by_spr = denoising_by_spread(ske_name, bodies_data)
if len(bodies_data) == 1:
return bodies_data.items(), noise_info_len + noise_info_spr
bodies_motion = dict() # get body motion
for (bodyID, body_data) in bodies_data.items():
bodies_motion[bodyID] = body_data['motion']
# Sort bodies based on the motion
# bodies_motion = sorted(bodies_motion.items(), key=lambda x, y: cmp(x[1], y[1]), reverse=True)
bodies_motion = sorted(bodies_motion.items(), key=lambda x: x[1], reverse=True)
denoised_bodies_data = list()
for (bodyID, _) in bodies_motion:
denoised_bodies_data.append((bodyID, bodies_data[bodyID]))
return denoised_bodies_data, noise_info_len + noise_info_spr
# TODO: Consider denoising further by integrating motion method
# if denoised_by_spr: # this sequence has been denoised by spread
# bodies_motion = sorted(bodies_motion.items(), lambda x, y: cmp(x[1], y[1]), reverse=True)
# denoised_bodies_data = list()
# for (bodyID, _) in bodies_motion:
# denoised_bodies_data.append((bodyID, bodies_data[bodyID]))
# return denoised_bodies_data, noise_info
# Step 3: Denoising based on motion
# bodies_data, noise_info = denoising_by_motion(ske_name, bodies_data, bodies_motion)
# return bodies_data, noise_info
def get_one_actor_points(body_data, num_frames):
"""
Get joints and colors for only one actor.
For joints, each frame contains 75 X-Y-Z coordinates.
For colors, each frame contains 25 x 2 (X, Y) coordinates.
"""
joints = np.zeros((num_frames, 75), dtype=np.float32)
colors = np.ones((num_frames, 1, 25, 2), dtype=np.float32) * np.nan
start, end = body_data['interval'][0], body_data['interval'][-1]
joints[start:end + 1] = body_data['joints'].reshape(-1, 75)
colors[start:end + 1, 0] = body_data['colors']
return joints, colors
def remove_missing_frames(ske_name, joints, colors):
"""
Cut off missing frames which all joints positions are 0s
For the sequence with 2 actors' data, also record the number of missing frames for
actor1 and actor2, respectively (for debug).
"""
num_frames = joints.shape[0]
num_bodies = colors.shape[1] # 1 or 2
if num_bodies == 2: # DEBUG
missing_indices_1 = np.where(joints[:, :75].sum(axis=1) == 0)[0]
missing_indices_2 = np.where(joints[:, 75:].sum(axis=1) == 0)[0]
cnt1 = len(missing_indices_1)
cnt2 = len(missing_indices_2)
start = 1 if 0 in missing_indices_1 else 0
end = 1 if num_frames - 1 in missing_indices_1 else 0
if max(cnt1, cnt2) > 0:
if cnt1 > cnt2:
info = '{}\t{:^10d}\t{:^6d}\t{:^6d}\t{:^5d}\t{:^3d}'.format(ske_name, num_frames,
cnt1, cnt2, start, end)
missing_skes_logger1.info(info)
else:
info = '{}\t{:^10d}\t{:^6d}\t{:^6d}'.format(ske_name, num_frames, cnt1, cnt2)
missing_skes_logger2.info(info)
# Find valid frame indices that the data is not missing or lost
# For two-subjects action, this means both data of actor1 and actor2 is missing.
valid_indices = np.where(joints.sum(axis=1) != 0)[0] # 0-based index
missing_indices = np.where(joints.sum(axis=1) == 0)[0]
num_missing = len(missing_indices)
if num_missing > 0: # Update joints and colors
joints = joints[valid_indices]
colors[missing_indices] = np.nan
global missing_count
missing_count += 1
missing_skes_logger.info('{}\t{:^10d}\t{:^11d}'.format(ske_name, num_frames, num_missing))
return joints, colors
def get_bodies_info(bodies_data):
bodies_info = '{:^17}\t{}\t{:^8}\n'.format('bodyID', 'Interval', 'Motion')
for (bodyID, body_data) in bodies_data.items():
start, end = body_data['interval'][0], body_data['interval'][-1]
bodies_info += '{}\t{:^8}\t{:f}\n'.format(bodyID, str([start, end]), body_data['motion'])
return bodies_info + '\n'
def get_two_actors_points(bodies_data):
"""
Get the first and second actor's joints positions and colors locations.
# Arguments:
bodies_data (dict): 3 key-value pairs: 'name', 'data', 'num_frames'.
bodies_data['data'] is also a dict, while the key is bodyID, the value is
the corresponding body_data which is also a dict with 4 keys:
- joints: raw 3D joints positions. Shape: (num_frames x 25, 3)
- colors: raw 2D color locations. Shape: (num_frames, 25, 2)
- interval: a list which records the frame indices.
- motion: motion amount
# Return:
joints, colors.
"""
ske_name = bodies_data['name']
label = int(ske_name[-2:])
num_frames = bodies_data['num_frames']
bodies_info = get_bodies_info(bodies_data['data'])
bodies_data, noise_info = denoising_bodies_data(bodies_data) # Denoising data
bodies_info += noise_info
bodies_data = list(bodies_data)
if len(bodies_data) == 1: # Only left one actor after denoising
if label >= 50: # DEBUG: Denoising failed for two-subjects action
fail_logger_2.info(ske_name)
bodyID, body_data = bodies_data[0]
joints, colors = get_one_actor_points(body_data, num_frames)
bodies_info += 'Main actor: %s' % bodyID
else:
if label < 50: # DEBUG: Denoising failed for one-subject action
fail_logger_1.info(ske_name)
joints = np.zeros((num_frames, 150), dtype=np.float32)
colors = np.ones((num_frames, 2, 25, 2), dtype=np.float32) * np.nan
bodyID, actor1 = bodies_data[0] # the 1st actor with largest motion
start1, end1 = actor1['interval'][0], actor1['interval'][-1]
joints[start1:end1 + 1, :75] = actor1['joints'].reshape(-1, 75)
colors[start1:end1 + 1, 0] = actor1['colors']
actor1_info = '{:^17}\t{}\t{:^8}\n'.format('Actor1', 'Interval', 'Motion') + \
'{}\t{:^8}\t{:f}\n'.format(bodyID, str([start1, end1]), actor1['motion'])
del bodies_data[0]
actor2_info = '{:^17}\t{}\t{:^8}\n'.format('Actor2', 'Interval', 'Motion')
start2, end2 = [0, 0] # initial interval for actor2 (virtual)
while len(bodies_data) > 0:
bodyID, actor = bodies_data[0]
start, end = actor['interval'][0], actor['interval'][-1]
if min(end1, end) - max(start1, start) <= 0: # no overlap with actor1
joints[start:end + 1, :75] = actor['joints'].reshape(-1, 75)
colors[start:end + 1, 0] = actor['colors']
actor1_info += '{}\t{:^8}\t{:f}\n'.format(bodyID, str([start, end]), actor['motion'])
# Update the interval of actor1
start1 = min(start, start1)
end1 = max(end, end1)
elif min(end2, end) - max(start2, start) <= 0: # no overlap with actor2
joints[start:end + 1, 75:] = actor['joints'].reshape(-1, 75)
colors[start:end + 1, 1] = actor['colors']
actor2_info += '{}\t{:^8}\t{:f}\n'.format(bodyID, str([start, end]), actor['motion'])
# Update the interval of actor2
start2 = min(start, start2)
end2 = max(end, end2)
del bodies_data[0]
bodies_info += ('\n' + actor1_info + '\n' + actor2_info)
with open(osp.join(actors_info_dir, ske_name + '.txt'), 'w') as fw:
fw.write(bodies_info + '\n')
return joints, colors
def get_raw_denoised_data():
"""
Get denoised data (joints positions and color locations) from raw skeleton sequences.
For each frame of a skeleton sequence, an actor's 3D positions of 25 joints represented
by an 2D array (shape: 25 x 3) is reshaped into a 75-dim vector by concatenating each
3-dim (x, y, z) coordinates along the row dimension in joint order. Each frame contains
two actor's joints positions constituting a 150-dim vector. If there is only one actor,
then the last 75 values are filled with zeros. Otherwise, select the main actor and the
second actor based on the motion amount. Each 150-dim vector as a row vector is put into
a 2D numpy array where the number of rows equals the number of valid frames. All such
2D arrays are put into a list and finally the list is serialized into a cPickle file.
For the skeleton sequence which contains two or more actors (mostly corresponds to the
last 11 classes), the filename and actors' information are recorded into log files.
For better understanding, also generate RGB+skeleton videos for visualization.
"""
with open(raw_data_file, 'rb') as fr: # load raw skeletons data
raw_skes_data = pickle.load(fr)
num_skes = len(raw_skes_data)
print('Found %d available skeleton sequences.' % num_skes)
raw_denoised_joints = []
raw_denoised_colors = []
frames_cnt = []
for (idx, bodies_data) in enumerate(raw_skes_data):
ske_name = bodies_data['name']
print('Processing %s' % ske_name)
num_bodies = len(bodies_data['data'])
if num_bodies == 1: # only 1 actor
num_frames = bodies_data['num_frames']
body_data = list(bodies_data['data'].values())[0]
joints, colors = get_one_actor_points(body_data, num_frames)
else: # more than 1 actor, select two main actors
joints, colors = get_two_actors_points(bodies_data)
# Remove missing frames
joints, colors = remove_missing_frames(ske_name, joints, colors)
num_frames = joints.shape[0] # Update
# Visualize selected actors' skeletons on RGB videos.
raw_denoised_joints.append(joints)
raw_denoised_colors.append(colors)
frames_cnt.append(num_frames)
if (idx + 1) % 1000 == 0:
print('Processed: %.2f%% (%d / %d), ' % \
(100.0 * (idx + 1) / num_skes, idx + 1, num_skes) + \
'Missing count: %d' % missing_count)
raw_skes_joints_pkl = osp.join(save_path, 'raw_denoised_joints.pkl')
with open(raw_skes_joints_pkl, 'wb') as f:
pickle.dump(raw_denoised_joints, f, pickle.HIGHEST_PROTOCOL)
raw_skes_colors_pkl = osp.join(save_path, 'raw_denoised_colors.pkl')
with open(raw_skes_colors_pkl, 'wb') as f:
pickle.dump(raw_denoised_colors, f, pickle.HIGHEST_PROTOCOL)
frames_cnt = np.array(frames_cnt, dtype=np.int)
np.savetxt(osp.join(save_path, 'frames_cnt.txt'), frames_cnt, fmt='%d')
print('Saved raw denoised positions of {} frames into {}'.format(np.sum(frames_cnt),
raw_skes_joints_pkl))
print('Found %d files that have missing data' % missing_count)
if __name__ == '__main__':
get_raw_denoised_data()
https://github.com/microsoft/SGN
![2021年危險化學品經營機關安全管理人員考試題庫及危險化學品經營機關安全管理人員考試技巧[圖]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)