lichen/detecttracking
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update for bakeup

Browse Source
This commit is contained in:
王庆刚
2024-08-06 20:00:29 +08:00
parent 27d57b21d4
commit 5109400a57
19 changed files with 497 additions and 352 deletions
Download Patch File Download Diff File Expand all files Collapse all files

264
tracking/module_analysis.py
View File

@ -10,6 +10,7 @@ import os
import cv2
import numpy as np
from pathlib import Path
import warnings
import sys
sys.path.append(r"D:\DetectTracking")
@ -203,7 +204,7 @@ def do_tracker_tracking(fpath, save_dir):
def do_tracking(fpath, savedir):
def do_tracking(fpath, savedir, event_name='images'):
'''
fpath: 算法各模块输出的data文件地址，匹配；
savedir: 对 fpath 各模块输出的复现；
@ -212,53 +213,59 @@ def do_tracking(fpath, savedir):
# fpath = r'D:\contrast\dataset\1_to_n\709\20240709-102758_6971558612189\1_track.data'
# savedir = r'D:\contrast\dataset\result\20240709-102843_6958770005357_6971558612189\error_6971558612189'
'''1.1 构造 0/1_tracking_output.data 文件地址，读取文件数据'''
imgpath, dfname = os.path.split(fpath)
CamerType = dfname.split('_')[0]
'''1.1 构造 0/1_tracking_output.data 文件地址，读取文件数据'''
tracking_output_path = os.path.join(imgpath, CamerType + '_tracking_output.data')
basename = os.path.basename(imgpath)
if not os.path.isfile(fpath):
print(f"Can't find {dfname} file!")
return
print(f"{basename}: Can't find {dfname} file!")
return None, None
if not os.path.isfile(tracking_output_path):
print(f"Can't find {CamerType}_tracking_output.data file!")
return
print(f"{basename}: Can't find {CamerType}_tracking_output.data file!")
return None, None
bboxes, ffeats, trackerboxes, tracker_feat_dict, trackingboxes, tracking_feat_dict = extract_data(fpath)
tracking_output_boxes, _ = read_tracking_output(tracking_output_path)
'''1.2 分别构造 2 个文件夹，(1) 存储画框后的图像; (2) 运动轨迹对应的 boxes子图'''
save_dir, basename = os.path.split(savedir)
if not os.path.exists(savedir):
os.makedirs(savedir)
subimg_dir = os.path.join(save_dir, basename.split('_')[0] + '_subimgs')
save_dir = os.path.join(savedir, event_name)
subimg_dir = os.path.join(savedir, event_name + '_subimgs')
if not os.path.exists(save_dir):
os.makedirs(save_dir)
if not os.path.exists(subimg_dir):
os.makedirs(subimg_dir)
os.makedirs(subimg_dir)
'''1.3 读取 fpath 中 track.data 文件对应的图像 imgs '''
imgs = read_imgs(imgpath, CamerType)
'''2. 执行轨迹分析, 保存轨迹分析前后的对比图示'''
traj_graphic = basename + '_' + CamerType
traj_graphic = event_name + '_' + CamerType
if CamerType == '1':
vts = doFrontTracks(trackerboxes, tracker_feat_dict)
vts.classify()
plt = plot_frameID_y2(vts)
# ftpath = os.path.join(save_dir, f"{traj_graphic}_front_y2.png")
# ftpath = os.path.join(savedir, f"{traj_graphic}_front_y2.png")
# plt.savefig(ftpath)
plt.close()
edgeline = cv2.imread("./shopcart/cart_tempt/board_ftmp_line.png")
img_tracking = draw_all_trajectories(vts, edgeline, savedir, CamerType, draw5p=True)
elif CamerType == '0':
vts = doBackTracks(trackerboxes, tracker_feat_dict)
vts.classify()
edgeline = cv2.imread("./shopcart/cart_tempt/edgeline.png")
img = draw_all_trajectories(vts, edgeline, save_dir, traj_graphic)
img_tracking = draw_all_trajectories(vts, edgeline, savedir, CamerType, draw5p=True)
# imgpth = os.path.join(save_dir, f"{traj_graphic}_.png")
# imgpth = os.path.join(savedir, f"{traj_graphic}_.png")
# cv2.imwrite(str(imgpth), img)
else:
print("Please check data file!")
@ -291,56 +298,131 @@ def do_tracking(fpath, savedir):
abH, abW = abimg.shape[:2]
cv2.line(abimg, (int(abW/2), 0), (int(abW/2), abH), (128, 255, 128), 2)
# algpath = os.path.join(save_dir, f"{traj_graphic}_alg.png")
# algpath = os.path.join(savedir, f"{traj_graphic}_alg.png")
# cv2.imwrite(str(algpath), abimg)
'''4. 画框后的图像和子图保存，若imgs数与tracker中fid数不匹配，只保存原图，不保存子图'''
'''4.0 读取 fpath 中对应的图像 imgs '''
imgs = read_imgs(imgpath, CamerType)
'''4.1 imgs数 < trackerboxes 的 max(fid)，返回原图'''
if len(imgs) < np.max(trackerboxes[:,7]):
for i in range(len(imgs)):
img_savepath = os.path.join(savedir, CamerType + "_" + f"{i}.png")
img_savepath = os.path.join(save_dir, CamerType + "_" + f"{i}.png")
cv2.imwrite(img_savepath, imgs[i])
print(f"fpath: {fpath}, len(imgs) = {len(imgs)} < Tracker max(fid) = {int(np.max(trackerboxes[:,7]))}, 无法匹配画框")
return
print(f"{basename}: len(imgs) = {len(imgs)} < Tracker max(fid) = {int(np.max(trackerboxes[:,7]))}, 无法匹配画框")
return img_tracking, abimg
'''4.2 在 imgs 上画框并保存'''
imgs_dw = draw_tracking_boxes(imgs, trackerboxes)
for fid, img in imgs_dw:
img_savepath = os.path.join(savedir, CamerType + "_fid_" + f"{fid}.png")
img_savepath = os.path.join(save_dir, CamerType + "_fid_" + f"{fid}.png")
cv2.imwrite(img_savepath, img)
# =============================================================================
# '''4.3.1 保存轨迹分析对应的子图'''
# for track in vts.Residual:
# for *xyxy, tid, conf, cls, fid, bid in track.boxes:
# img = imgs[int(fid-1)]
# x1, y1, x2, y2 = int(xyxy[0]/2), int(xyxy[1]/2), int(xyxy[2]/2), int(xyxy[3]/2)
# subimg = img[y1:y2, x1:x2]
#
# subimg_path = os.path.join(subimg_dir, f'{CamerType}_tid{int(tid)}_{int(fid-1)}_{int(bid)}.png' )
# cv2.imwrite(subimg_path, subimg)
# =============================================================================
'''4.3.2 保存轨迹选择对应的子图'''
for track in tracking_output_boxes:
for *xyxy, tid, conf, cls, fid, bid in track:
# for track in tracking_output_boxes:
for track in vts.Residual:
for *xyxy, tid, conf, cls, fid, bid in track.boxes:
img = imgs[int(fid-1)]
x1, y1, x2, y2 = int(xyxy[0]/2), int(xyxy[1]/2), int(xyxy[2]/2), int(xyxy[3]/2)
subimg = img[y1:y2, x1:x2]
subimg_path = os.path.join(subimg_dir, f'{CamerType}_tid{int(tid)}_{int(fid-1)}_{int(bid)}.png' )
cv2.imwrite(subimg_path, subimg)
# for track in tracking_output_boxes:
# for *xyxy, tid, conf, cls, fid, bid in track:
# img = imgs[int(fid-1)]
# x1, y1, x2, y2 = int(xyxy[0]/2), int(xyxy[1]/2), int(xyxy[2]/2), int(xyxy[3]/2)
# subimg = img[y1:y2, x1:x2]
# subimg_path = os.path.join(subimg_dir, f'{CamerType}_tid{int(tid)}_{int(fid-1)}_{int(bid)}_x.png' )
# cv2.imwrite(subimg_path, subimg)
return abimg
return img_tracking, abimg
def tracking_simulate(eventpath, savepath):
'''args:
eventpath: 时间文件夹
savepath: 存储文件夹
'''
'''1. 获取事件名'''
event_names = os.path.basename(eventpath).strip().split('_')
if len(event_names)==2 and len(event_names[1])>=8:
enent_name = event_names[1]
elif len(event_names)==2 and len(event_names[1])==0:
enent_name = event_names[0]
else:
return
'''2. 依次读取 0/1_track.data 中数据，进行仿真'''
illu_tracking, illu_select = [], []
for filename in os.listdir(eventpath):
# filename = '1_track.data'
if filename.find("track.data") <= 0: continue
fpath = os.path.join(eventpath, filename)
if not os.path.isfile(fpath): continue
img_tracking, img_select = do_tracking(fpath, savepath, enent_name)
if img_select is not None:
illu_select.append(img_select)
if img_tracking is not None:
illu_tracking.append(img_tracking)
'''3. 前、后摄，原始轨迹、本地tracking输出、现场算法轨迹选择前、后，共幅8图'''
if len(illu_select)==2:
Img_s = np.concatenate((illu_select[0], illu_select[1]), axis = 0)
H, W = Img_s.shape[:2]
cv2.line(Img_s, (0, int(H/2)), (int(W), int(H/2)), (128, 255, 128), 2)
elif len(illu_select)==1:
Img_s = illu_select[0]
else:
Img_s = None
if len(illu_tracking)==2:
Img_t = np.concatenate((illu_tracking[0], illu_tracking[1]), axis = 0)
H, W = Img_t.shape[:2]
cv2.line(Img_t, (0, int(H/2)), (int(W), int(H/2)), (128, 255, 128), 2)
elif len(illu_tracking)==1:
Img_t = illu_tracking[0]
else:
Img_t = None
'''3.1 单独另存保存完好的 8 轨迹图'''
basepath, _ = os.path.split(savepath)
trajpath = os.path.join(basepath, 'trajs')
if not os.path.exists(trajpath):
os.makedirs(trajpath)
traj_path = os.path.join(trajpath, enent_name+'.png')
imgpath_tracking = os.path.join(savepath, enent_name + '_ing.png')
imgpath_select = os.path.join(savepath, enent_name + '_slt.png')
imgpath_ts = os.path.join(savepath, enent_name + '_ts.png')
if Img_t is not None and Img_s is not None and np.all(Img_s.shape==Img_t.shape):
Img_ts = np.concatenate((Img_t, Img_s), axis = 1)
H, W = Img_ts.shape[:2]
cv2.line(Img_ts, (int(W/2), 0), (int(W/2), int(H)), (0, 0, 255), 4)
cv2.imwrite(imgpath_ts, Img_ts)
cv2.imwrite(traj_path, Img_ts)
else:
if Img_s: cv2.imwrite(imgpath_select, Img_s)
if Img_t: cv2.imwrite(imgpath_tracking, Img_t)
# warnings.simplefilter("error", category=np.VisibleDeprecationWarning)
def main_loop():
del_barcode_file = r'\\192.168.1.28\share\测试_202406\0723\0723_2\deletedBarcode.txt'
basepath = r'\\192.168.1.28\share\测试_202406\0723\0723_2' # 测试数据文件夹地址
del_barcode_file = r'\\192.168.1.28\share\测试_202406\0723\0723_3\deletedBarcode.txt'
basepath = r'\\192.168.1.28\share\测试_202406\0723\0723_3' # 测试数据文件夹地址
SavePath = r'D:\contrast\dataset\resultx' # 结果保存地址
prefix = ["getout_", "input_", "error_"]
# prefix = ["getout_", "input_", "error_"]
'''获取性能测试数据相关路径'''
relative_paths = contrast_analysis(del_barcode_file, basepath, SavePath)
@ -349,7 +431,7 @@ def main_loop():
k = 0
for tuple_paths in relative_paths:
'''生成文件夹存储结果图像的文件夹'''
'''1. 生成存储结果图像的文件夹'''
namedirs = []
for data_path in tuple_paths:
base_name = os.path.basename(data_path).strip().split('_')
@ -358,76 +440,56 @@ def main_loop():
else:
name = base_name[0]
namedirs.append(name)
sdir = "_".join(namedirs)
savepath = os.path.join(SavePath, sdir)
# if os.path.exists(savepath):
# continue
if not os.path.exists(savepath):
os.makedirs(savepath)
os.makedirs(savepath)
for path in tuple_paths:
'''============= 分别指定指定存储、读取对应的文件夹 ============='''
# if sdir.find('094631_6904724022444_6976075000082') < 0: continue
# if path.find('094631_') < 0: continue
imgs = []
for filename in os.listdir(path):
# filename = '1_track.data'
fpath = os.path.join(path, filename)
if os.path.isfile(fpath) and filename.find("track.data")>0:
enent_name = ''
'''构建结果保存文件名前缀'''
for i, name in enumerate(namedirs):
if fpath.find(name)>0:
enent_name = prefix[i] + name
break
spath = os.path.join(savepath, enent_name)
# abimg = do_tracking(fpath, spath)
# imgs.append(abimg)
try:
abimg = do_tracking(fpath, spath)
imgs.append(abimg)
if len(imgs) == 2:
Img = np.concatenate((imgs[0], imgs[1]), axis = 0)
H, W = Img.shape[:2]
cv2.line(Img, (0, int(H/2)), (int(W), int(H/2)), (128, 255, 128), 2)
else:
Img = imgs[0]
imgpath = os.path.join(savepath, enent_name + '_alg.png')
cv2.imwrite(imgpath, Img)
except Exception as e:
print(f'Error! {fpath}, {e}')
'''2. 循环执行操作事件：取出、放入、错误匹配'''
for eventpath in tuple_paths:
try:
tracking_simulate(eventpath, savepath)
except Exception as e:
print(f'Error! {eventpath}, {e}')
# k +=1
# if k==1:
# break
def main():
'''
fpath: data文件地址，该 data 文件包括 Pipeline 各模块输出
save_dir：需包含二级目录，其中一级目录为轨迹图像；
二级目录为与data文件对应的序列图像存储地址。
eventpath: data文件地址，该 data 文件包括 Pipeline 各模块输出
savepath: 包含二级目录，一级目录为轨迹图像;二级目录为与data文件对应的序列图像存储地址。
'''
EventPaths = r'\\192.168.1.28\share\测试_202406\0723\0723_2'
SavePath = r'D:\contrast\dataset\result'
k=0
for pathname in os.listdir(EventPaths):
# pathname = "20240723-094731_6903148242797"
eventpath = os.path.join(EventPaths, pathname)
savepath = os.path.join(SavePath, pathname)
if not os.path.exists(savepath):
os.makedirs(savepath)
# tracking_simulate(eventpath, savepath)
try:
tracking_simulate(eventpath, savepath)
except Exception as e:
print(f'Error! {eventpath}, {e}')
# k += 1
# if k==10:
# break
fpath = r'\\192.168.1.28\share\测试_202406\0723\0723_1\20240723-101506_6906839615771\1_track.data'
save_dir = r'D:\contrast\dataset\result\20240723-101506_\images'
do_tracking(fpath, save_dir)
if __name__ == "__main__":
main_loop()
# main()
if __name__ == "__main__":
# main_loop()
main()
# try:
# main_loop()
# except Exception as e: