430 lines
15 KiB
Python
430 lines
15 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
Created on Thu May 30 14:03:03 2024
|
||
|
||
现场测试性能分析
|
||
|
||
@author: ym
|
||
"""
|
||
import os
|
||
import cv2
|
||
import numpy as np
|
||
from pathlib import Path
|
||
import warnings
|
||
import sys
|
||
sys.path.append(r"D:\DetectTracking")
|
||
|
||
|
||
from tracking.utils.plotting import Annotator, colors, draw_tracking_boxes
|
||
from tracking.utils import Boxes, IterableSimpleNamespace, yaml_load
|
||
from tracking.trackers import BOTSORT, BYTETracker
|
||
from tracking.dotrack.dotracks_back import doBackTracks
|
||
from tracking.dotrack.dotracks_front import doFrontTracks
|
||
from tracking.utils.drawtracks import plot_frameID_y2, draw_all_trajectories
|
||
|
||
from tracking.utils.read_data import extract_data, read_deletedBarcode_file, read_tracking_output
|
||
|
||
from contrast_analysis import contrast_analysis
|
||
from tracking.utils.annotator import TrackAnnotator
|
||
|
||
W, H = 1024, 1280
|
||
Mode = 'front' #'back'
|
||
ImgFormat = ['.jpg', '.jpeg', '.png', '.bmp']
|
||
|
||
|
||
'''调用tracking()函数，利用本地跟踪算法获取各目标轨迹，可以比较本地跟踪算法与现场跟踪算法的区别。'''
|
||
def init_tracker(tracker_yaml = None, bs=1):
|
||
"""
|
||
Initialize tracker for object tracking during prediction.
|
||
"""
|
||
TRACKER_MAP = {'bytetrack': BYTETracker, 'botsort': BOTSORT}
|
||
cfg = IterableSimpleNamespace(**yaml_load(tracker_yaml))
|
||
|
||
tracker = TRACKER_MAP[cfg.tracker_type](args=cfg, frame_rate=30)
|
||
|
||
return tracker
|
||
|
||
def tracking(bboxes, ffeats):
|
||
tracker_yaml = r"./trackers/cfg/botsort.yaml"
|
||
tracker = init_tracker(tracker_yaml)
|
||
|
||
TrackBoxes = np.empty((0, 9), dtype = np.float32)
|
||
TracksDict = {}
|
||
|
||
'''========================== 执行跟踪处理 ============================='''
|
||
# dets 与 feats 应保持严格对应
|
||
for dets, feats in zip(bboxes, ffeats):
|
||
det_tracking = Boxes(dets).cpu().numpy()
|
||
tracks = tracker.update(det_tracking, features=feats)
|
||
|
||
|
||
'''tracks: [x1, y1, x2, y2, track_id, score, cls, frame_index, box_index]
|
||
0 1 2 3 4 5 6 7 8
|
||
这里，frame_index 也可以用视频的 帧ID 代替, box_index 保持不变
|
||
'''
|
||
|
||
if len(tracks):
|
||
TrackBoxes = np.concatenate([TrackBoxes, tracks], axis=0)
|
||
|
||
FeatDict = {}
|
||
for track in tracks:
|
||
tid = int(track[8])
|
||
FeatDict.update({tid: feats[tid, :]})
|
||
|
||
frameID = tracks[0, 7]
|
||
|
||
# print(f"frameID: {int(frameID)}")
|
||
assert len(tracks) == len(FeatDict), f"Please check the func: tracker.update() at frameID({int(frameID)})"
|
||
|
||
TracksDict[f"frame_{int(frameID)}"] = {"feats":FeatDict}
|
||
|
||
|
||
return TrackBoxes, TracksDict
|
||
|
||
def read_imgs(imgspath, CamerType):
|
||
'''
|
||
inputs:
|
||
imgspath；序列图像地址
|
||
CamerType：相机类型，0：后摄，1：前摄
|
||
outputs：
|
||
imgs：图像序列
|
||
功能：
|
||
根据CamerType类型读取imgspath文件夹中的图像，并根据帧索引进行排序。
|
||
do_tracking()中调用该函数，实现（1）读取imgs并绘制各目标轨迹框；（2）获取subimgs
|
||
'''
|
||
imgs, frmIDs = [], []
|
||
for filename in os.listdir(imgspath):
|
||
file, ext = os.path.splitext(filename)
|
||
flist = file.split('_')
|
||
if len(flist)==4 and ext in ImgFormat:
|
||
camID, frmID = flist[0], int(flist[-1])
|
||
if camID==CamerType:
|
||
img = cv2.imread(os.path.join(imgspath, filename))
|
||
imgs.append(img)
|
||
frmIDs.append(frmID)
|
||
if len(frmIDs):
|
||
indice = np.argsort(np.array(frmIDs))
|
||
imgs = [imgs[i] for i in indice]
|
||
|
||
return imgs
|
||
|
||
def do_tracking(fpath, savedir, event_name='images'):
|
||
'''
|
||
args:
|
||
fpath: 算法各模块输出的data文件地址，匹配；
|
||
savedir: 对 fpath 各模块输出的复现；
|
||
分析具体视频时，需指定 fpath 和 savedir
|
||
outputs:
|
||
img_tracking：目标跟踪轨迹、本地轨迹分析算法的轨迹对比图
|
||
abimg：现场轨迹分析算法、轨迹选择输出的对比图
|
||
'''
|
||
# 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'
|
||
|
||
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"{basename}: Can't find {dfname} file!")
|
||
return None, None
|
||
if not os.path.isfile(tracking_output_path):
|
||
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 利用本地跟踪算法生成各商品轨迹'''
|
||
# trackerboxes, tracker_feat_dict = tracking(bboxes, ffeats)
|
||
|
||
'''1.3 分别构造 2 个文件夹，(1) 存储画框后的图像; (2) 运动轨迹对应的 boxes子图'''
|
||
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)
|
||
|
||
|
||
'''2. 执行轨迹分析, 保存轨迹分析前后的对比图示'''
|
||
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(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_tracking = draw_all_trajectories(vts, edgeline, savedir, CamerType, draw5p=True)
|
||
|
||
# imgpth = os.path.join(savedir, f"{traj_graphic}_.png")
|
||
# cv2.imwrite(str(imgpth), img)
|
||
else:
|
||
print("Please check data file!")
|
||
|
||
|
||
'''3 tracking() 算法输出后多轨迹选择问题分析'''
|
||
if CamerType == '1':
|
||
aline = cv2.imread("./shopcart/cart_tempt/board_ftmp_line.png")
|
||
elif CamerType == '0':
|
||
aline = cv2.imread("./shopcart/cart_tempt/edgeline.png")
|
||
else:
|
||
print("Please check data file!")
|
||
|
||
bline = aline.copy()
|
||
|
||
annotator = TrackAnnotator(aline, line_width=2)
|
||
for track in trackingboxes:
|
||
annotator.plotting_track(track)
|
||
aline = annotator.result()
|
||
|
||
annotator = TrackAnnotator(bline, line_width=2)
|
||
if not isinstance(tracking_output_boxes, list):
|
||
tracking_output_boxes = [tracking_output_boxes]
|
||
|
||
for track in tracking_output_boxes:
|
||
annotator.plotting_track(track)
|
||
bline = annotator.result()
|
||
|
||
abimg = np.concatenate((aline, bline), axis = 1)
|
||
abH, abW = abimg.shape[:2]
|
||
cv2.line(abimg, (int(abW/2), 0), (int(abW/2), abH), (128, 255, 128), 2)
|
||
|
||
# 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(save_dir, CamerType + "_" + f"{i}.png")
|
||
cv2.imwrite(img_savepath, imgs[i])
|
||
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(save_dir, CamerType + "_fid_" + f"{int(fid)}.png")
|
||
cv2.imwrite(img_savepath, img)
|
||
|
||
'''4.3.2 保存轨迹选择对应的子图'''
|
||
# 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)}_{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 img_tracking, abimg
|
||
|
||
|
||
def tracking_simulate(eventpath, savepath):
|
||
'''args:
|
||
eventpath: 事件文件夹
|
||
savepath: 存储文件夹
|
||
遍历eventpath
|
||
'''
|
||
|
||
# =============================================================================
|
||
# '''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
|
||
# =============================================================================
|
||
bname = os.path.basename(eventpath)
|
||
idx = bname.find('2024')
|
||
enent_name = bname[idx:(idx+15)]
|
||
|
||
'''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. 共幅8图，上下子图显示的是前后摄，每一行4个子图，分别为：
|
||
(1) tracker输出原始轨迹; (2)本地tracking输出; (3)现场算法轨迹选择前轨迹; (4)现场算法轨迹选择后的轨迹
|
||
'''
|
||
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_3\deletedBarcode.txt'
|
||
basepath = r'\\192.168.1.28\share\测试_202406\0723\0723_3' # 测试数据文件夹地址
|
||
SavePath = r'D:\contrast\dataset\resultx' # 结果保存地址
|
||
# prefix = ["getout_", "input_", "error_"]
|
||
|
||
'''获取性能测试数据相关路径'''
|
||
relative_paths = contrast_analysis(del_barcode_file, basepath, SavePath)
|
||
|
||
'''开始循环执行每次测试过任务'''
|
||
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('_')
|
||
if len(base_name[-1]):
|
||
name = base_name[-1]
|
||
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)
|
||
|
||
'''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():
|
||
'''
|
||
eventPaths: data文件地址，该 data 文件包括 Pipeline 各模块输出
|
||
SavePath: 包含二级目录，一级目录为轨迹图像;二级目录为与data文件对应的序列图像存储地址。
|
||
'''
|
||
# eventPaths = r'\\192.168.1.28\share\测试_202406\0723\0723_3'
|
||
eventPaths = r"D:\DetectTracking\tracking\images"
|
||
|
||
savePath = r'D:\contrast\dataset\result'
|
||
k=0
|
||
for pathname in os.listdir(eventPaths):
|
||
pathname = "20240925-142635-3e3cb61a-8bbe-45f2-aed7-a40de7f2d624_6924743924161"
|
||
|
||
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==1:
|
||
break
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# main_loop()
|
||
main()
|
||
# try:
|
||
# main_loop()
|
||
# except Exception as e:
|
||
# print(f'Error: {e}')
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|