Files
detecttracking/contrast/contrast_one2one.py
2024-09-05 19:01:49 +08:00

709 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# -*- coding: utf-8 -*-
"""
Created on Fri Aug 30 17:53:03 2024
1. 确认在相同CamerType下track.data 中 CamerID 项数量 = 图像数 = 帧ID数 = 最大帧ID
2. 读取0/1_tracking_output.data 中数据boxes、featslen(boxes)=len(feats)
帧ID约束
3. 优先选择前摄
4. 保存图像数据
5. 一次购物事件类型
shopEvent: {barcode:
type: getout, input
front_traj:[{imgpath: str,
box: arrar(1, 9),
feat: array(1, 256)
}]
back_traj: [{imgpath: str,
box: arrar(1, 9),
feat: array(1, 256)
}]
}
@author: ym
"""
import numpy as np
import cv2
import os
import sys
import random
import pickle
import torch
import time
import json
from pathlib import Path
from scipy.spatial.distance import cdist
import matplotlib.pyplot as plt
import shutil
from datetime import datetime
# Vit版resnet, 和现场特征不一致需将resnet_vit中文件提出
# from config import config as conf
# from model import resnet18
# from inference import load_contrast_model
# from inference import featurize
# embedding_size = conf.embedding_size
# img_size = conf.img_size
# device = conf.device
# model = load_contrast_model()
sys.path.append(r"D:\DetectTracking")
from tracking.utils.read_data import extract_data, read_tracking_output, read_deletedBarcode_file
from config import config as conf
from model import resnet18 as resnet18
from test_ori import inference_image
IMG_FORMAT = ['.bmp', '.jpg', '.jpeg', '.png']
model = resnet18().to(conf.device)
print('load model {} '.format(conf.testbackbone))
# model = nn.DataParallel(model).to(conf.device)
model.load_state_dict(torch.load(conf.test_model, map_location=conf.device))
model.eval()
def creat_shopping_event(eventPath, subimgPath=False):
'''构造放入商品事件字典,这些事件需满足条件:
1) 前后摄至少有一条轨迹输出
2) 保存有帧图像,以便裁剪出 boxe 子图
'''
# filename = "20240723-155413_6904406215720"
'''filename下为一次购物事件'''
eventName = os.path.basename(eventPath)
'''================ 0. 检查 filename 及 eventPath 正确性和有效性 ================'''
nmlist = eventName.split('_')
if eventName.find('2024')<0 or len(nmlist)!=2 or len(nmlist[0])!=15 or len(nmlist[1])<11:
return
if not os.path.isdir(eventPath):
return
'''================ 1. 构造事件描述字典,暂定 9 items ==============='''
event = {}
event['barcode'] = eventName.split('_')[1]
event['type'] = 'input'
event['filepath'] = eventPath
event['back_imgpaths'] = []
event['front_imgpaths'] = []
event['back_boxes'] = np.empty((0, 9), dtype=np.float64)
event['front_boxes'] = np.empty((0, 9), dtype=np.float64)
event['back_feats'] = np.empty((0, 256), dtype=np.float64)
event['front_feats'] = np.empty((0, 256), dtype=np.float64)
event['feats_compose'] = np.empty((0, 256), dtype=np.float64)
# event['feats_select'] = np.empty((0, 256), dtype=np.float64)
'''================= 2. 读取 data 文件 ============================='''
for dataname in os.listdir(eventPath):
# filename = '1_track.data'
datapath = os.path.join(eventPath, dataname)
if not os.path.isfile(datapath): continue
CamerType = dataname.split('_')[0]
''' 2.1 读取 0/1_track.data 中数据,暂不考虑'''
# if dataname.find("_track.data")>0:
# bboxes, ffeats, trackerboxes, tracker_feat_dict, trackingboxes, tracking_feat_dict = extract_data(datapath)
''' 2.2 读取 0/1_tracking_output.data 中数据'''
if dataname.find("_tracking_output.data")>0:
tracking_output_boxes, tracking_output_feats = read_tracking_output(datapath)
if len(tracking_output_boxes) != len(tracking_output_feats): continue
if CamerType == '0':
event['back_boxes'] = tracking_output_boxes
event['back_feats'] = tracking_output_feats
elif CamerType == '1':
event['front_boxes'] = tracking_output_boxes
event['front_feats'] = tracking_output_feats
if len(event['back_boxes'])==0 or len(event['front_boxes'])==0:
return None
'''2.3 事件的特征表征方式: 特征选择、特征集成'''
bk_feats = event['back_feats']
ft_feats = event['front_feats']
'''2.3.1 特征集成'''
feats_compose = np.empty((0, 256), dtype=np.float64)
if len(ft_feats):
feats_compose = np.concatenate((feats_compose, ft_feats), axis=0)
if len(bk_feats):
feats_compose = np.concatenate((feats_compose, bk_feats), axis=0)
event['feats_compose'] = feats_compose
'''2.3.1 特征选择'''
if len(ft_feats):
event['feats_select'] = ft_feats
# pickpath = os.path.join(savePath, f"{filename}.pickle")
# with open(pickpath, 'wb') as f:
# pickle.dump(event, f)
# print(f"Event: {filename}")
# if subimgPath==False:
# eventList.append(event)
# continue
'''================ 2. 读取图像文件地址并按照帧ID排序 ============='''
frontImgs, frontFid = [], []
backImgs, backFid = [], []
for imgname in os.listdir(eventPath):
name, ext = os.path.splitext(imgname)
if ext not in IMG_FORMAT or name.find('frameId')<0: continue
CamerType = name.split('_')[0]
frameId = int(name.split('_')[3])
imgpath = os.path.join(eventPath, imgname)
if CamerType == '0':
backImgs.append(imgpath)
backFid.append(frameId)
if CamerType == '1':
frontImgs.append(imgpath)
frontFid.append(frameId)
frontIdx = np.argsort(np.array(frontFid))
backIdx = np.argsort(np.array(backFid))
'''2.1 生成依据帧 ID 排序的前后摄图像地址列表'''
frontImgs = [frontImgs[i] for i in frontIdx]
backImgs = [backImgs[i] for i in backIdx]
'''2.2 将前、后摄图像路径添加至事件字典'''
bfid = event['back_boxes'][:, 7].astype(np.int64)
ffid = event['front_boxes'][:, 7].astype(np.int64)
if len(bfid) and max(bfid) <= len(backImgs):
event['back_imgpaths'] = [backImgs[i-1] for i in bfid]
if len(ffid) and max(ffid) <= len(frontImgs):
event['front_imgpaths'] = [frontImgs[i-1] for i in ffid]
'''================ 3. 判断当前事件有效性,并添加至事件列表 =========='''
condt1 = len(event['back_imgpaths'])==0 or len(event['front_imgpaths'])==0
condt2 = len(event['front_feats'])==0 and len(event['back_feats'])==0
if condt1 or condt2:
print(f"Event: {eventName}, Error, condt1: {condt1}, condt2: {condt2}")
return None
'''构造放入商品事件列表,暂不处理'''
# delepath = os.path.join(basePath, 'deletedBarcode.txt')
# bcdList = read_deletedBarcode_file(delepath)
# for slist in bcdList:
# getoutFold = slist['SeqDir'].strip()
# getoutPath = os.path.join(basePath, getoutFold)
# '''取出事件文件夹不存在,跳出循环'''
# if not os.path.exists(getoutPath) and not os.path.isdir(getoutPath):
# continue
# ''' 生成取出事件字典 '''
# event = {}
# event['barcode'] = slist['Deleted'].strip()
# event['type'] = 'getout'
# event['basePath'] = getoutPath
return event
def get_std_barcodeDict(bcdpath, bpath):
'''
inputs:
bcdpath: 已清洗的barcode样本图像如果barcode下有'base'文件夹,只选用该文件夹下图像
(default = r'\\192.168.1.28\share\已标注数据备份\对比数据\barcode\barcode_1771')
功能:
生成并保存只有一个key值的字典 {barcode: [imgpath1, imgpath1, ...]}
bpath: 字典存储地址
'''
# bpath = r'\\192.168.1.28\share\测试_202406\contrast\std_barcodes'
'''读取数据集中 barcode 列表'''
stdBlist = []
for filename in os.listdir(bcdpath):
filepath = os.path.join(bcdpath, filename)
if not os.path.isdir(filepath) or not filename.isdigit(): continue
stdBlist.append(filename)
bcdpaths = [(barcode, os.path.join(bcdpath, barcode)) for barcode in stdBlist]
'''遍历数据集针对每一个barcode生成并保存字典{barcode: [imgpath1, imgpath1, ...]}'''
k = 0
for barcode, bpath in bcdpaths:
stdBarcodeDict = {}
stdBarcodeDict[barcode] = []
for root, dirs, files in os.walk(bpath):
imgpaths = []
if "base" in dirs:
broot = os.path.join(root, "base")
for imgname in os.listdir(broot):
imgpath = os.path.join(broot, imgname)
_, ext = os.path.splitext(imgpath)
if ext not in IMG_FORMAT: continue
imgpaths.append(imgpath)
stdBarcodeDict[barcode].extend(imgpaths)
break
else:
for imgname in files:
imgpath = os.path.join(root, imgname)
_, ext = os.path.splitext(imgpath)
if ext not in IMG_FORMAT: continue
imgpaths.append(imgpath)
stdBarcodeDict[barcode].extend(imgpaths)
pickpath = os.path.join(bpath, f"{barcode}.pickle")
with open(pickpath, 'wb') as f:
pickle.dump(stdBarcodeDict, f)
print(f"Barcode: {barcode}")
# k += 1
# if k == 10:
# break
return
def save_event_subimg(event, savepath):
'''
功能: 保存一次购物事件的轨迹子图
9 items: barcode, type, filepath, back_imgpaths, front_imgpaths,
back_boxes, front_boxes, back_feats, front_feats,
feats_compose, feats_select
子图保存次序:先前摄、后后摄,以 k 为编号,和 "feats_compose" 中次序相同
'''
cameras = ('front', 'back')
k = 0
for camera in cameras:
if camera == 'front':
boxes = event['front_boxes']
imgpaths = event['front_imgpaths']
else:
boxes = event['back_boxes']
imgpaths = event['back_imgpaths']
for i, box in enumerate(boxes):
x1, y1, x2, y2, tid, score, cls, fid, bid = box
imgpath = imgpaths[i]
image = cv2.imread(imgpath)
subimg = image[int(y1/2):int(y2/2), int(x1/2):int(x2/2), :]
camerType, timeTamp, _, frameID = os.path.basename(imgpath).split('.')[0].split('_')
subimgName = f"{k}_cam-{camerType}_tid-{int(tid)}_fid-({int(fid)}, {frameID}).png"
spath = os.path.join(savepath, subimgName)
cv2.imwrite(spath, subimg)
k += 1
# basename = os.path.basename(event['filepath'])
print(f"Image saved: {os.path.basename(event['filepath'])}")
def batch_inference(imgpaths, batch):
size = len(imgpaths)
groups = []
for i in range(0, size, batch):
end = min(batch + i, size)
groups.append(imgpaths[i: end])
features = []
for group in groups:
feature = featurize(group, conf.test_transform, model, conf.device)
features.append(feature)
features = np.concatenate(features, axis=0)
return features
def stdfeat_infer(imgPath, featPath, bcdSet=None):
'''
inputs:
imgPath: 该文件夹下的 pickle 文件格式 {barcode: [imgpath1, imgpath1, ...]}
featPath: imgPath图像对应特征的存储地址
功能:
对 imgPath中图像进行特征提取生成只有一个key值的字典
{barcode: features}features.shape=(nsample, 256),并保存至 featPath 中
'''
# imgPath = r"\\192.168.1.28\share\测试_202406\contrast\std_barcodes"
# featPath = r"\\192.168.1.28\share\测试_202406\contrast\std_features"
stdBarcodeDict = {}
k = 0
for filename in os.listdir(imgPath):
bcd, ext = os.path.splitext(filename)
if bcdSet is not None and bcd not in bcdSet:
continue
filepath = os.path.join(imgPath, filename)
stdbDict = {}
t1 = time.time()
try:
with open(filepath, 'rb') as f:
bpDict = pickle.load(f)
for barcode, imgpaths in bpDict.items():
# feature = batch_inference(imgpaths, 8) #from vit distilled model of Lichen
feature = inference_image(imgpaths, conf.test_transform, model, conf.device)
feature /= np.linalg.norm(feature, axis=1)[:, None]
except Exception as e:
print(f"Error accured at: {filename}, with Exception is: {e}")
'''================ 保存单个barcode特征 ================'''
stdbDict["barcode"] = barcode
stdbDict["imgpaths"] = imgpaths
stdbDict["feats"] = feature
pkpath = os.path.join(featPath, f"{barcode}.pickle")
with open(pkpath, 'wb') as f:
pickle.dump(stdbDict, f)
stdBarcodeDict[barcode] = feature
t2 = time.time()
print(f"Barcode: {barcode}, need time: {t2-t1:.1f} secs")
# k += 1
# if k == 10:
# break
pickpath = os.path.join(featPath, f"barcode_features_{k}.pickle")
with open(pickpath, 'wb') as f:
pickle.dump(stdBarcodeDict, f)
def contrast_performance_evaluate():
eventFeatPath = r"\\192.168.1.28\share\测试_202406\contrast\events"
stdBcdPath = r"\\192.168.1.28\share\测试_202406\contrast\std_barcodes"
stdFeatPath = r"\\192.168.1.28\share\测试_202406\contrast\std_features"
subimgPath = r"\\192.168.1.28\share\测试_202406\contrast\subimgs"
# stdBarcode = [p.stem for p in Path(stdFeatPath).iterdir() if p.is_file() and p.suffix=='.pickle']
stdBarcode = [p.stem for p in Path(stdBcdPath).iterdir() if p.is_file() and p.suffix=='.pickle']
'''购物事件列表,该列表中的 Barcode 存在于标准的 stdBarcode 内'''
# evtList = [(p.stem, p.stem.split('_')[1]) for p in Path(eventFeatPath).iterdir()
# if p.is_file()
# and p.suffix=='.pickle'
# and len(p.stem.split('_'))==2
# and p.stem.split('_')[1].isdigit()
# and p.stem.split('_')[1] in stdBarcode
# ]
evtList = [(p.stem, p.stem.split('_')[1]) for p in Path(eventFeatPath).iterdir()
if p.is_file()
and p.suffix=='.pickle'
and len(p.stem.split('_'))==2
and p.stem.split('_')[1].isdigit()
and p.stem.split('_')[1] in stdBarcode
]
barcodes = set([bcd for _, bcd in evtList])
'''标准特征集图像样本经特征提取并保存,运行一次后无需再运行'''
# stdfeat_infer(stdBcdPath, stdFeatPath, barcodes)
'''========= 构建用于比对的标准特征字典 ============='''
stdDict = {}
for barcode in barcodes:
stdpath = os.path.join(stdFeatPath, barcode+'.pickle')
with open(stdpath, 'rb') as f:
stddata = pickle.load(f)
stdDict[barcode] = stddata
'''========= 构建用于比对的操作事件字典 ============='''
evtDict = {}
for event, barcode in evtList:
evtpath = os.path.join(eventFeatPath, event+'.pickle')
with open(evtpath, 'rb') as f:
evtdata = pickle.load(f)
evtDict[event] = evtdata
'''===== 构造 3 个事件对: 扫 A 放 A, 扫 A 放 B, 合并 ===================='''
AA_list = [(event, barcode, "same") for event, barcode in evtList]
AB_list = []
for event, barcode in evtList:
dset = list(barcodes.symmetric_difference(set([barcode])))
idx = random.randint(0, len(dset)-1)
AB_list.append((event, dset[idx], "diff"))
mergePairs = AA_list + AB_list
'''读取事件、标准特征文件中数据,以 AA_list 和 AB_list 中关键字为 key 生成字典'''
results = []
for evt, stdbcd, label in mergePairs:
## 标准特征字典的构造方式不合适需改进不能用具体的barcode做key值
event = evtDict[evt]
## 判断是否存在轨迹图像文件夹,不存在则创建文件夹并保存轨迹图像
pairpath = os.path.join(subimgPath, f"{evt}")
if not os.path.exists(pairpath):
os.makedirs(pairpath)
save_event_subimg(event, pairpath)
## 判断是否存在 barcode 样本集图像文件夹,不存在则创建文件夹并存储 barcode 样本集图像
stdImgpath = stdDict[stdbcd]["imgpaths"]
pstdpath = os.path.join(subimgPath, f"{stdbcd}")
if not os.path.exists(pstdpath):
os.makedirs(pstdpath)
ii = 1
for filepath in stdImgpath:
stdpath = os.path.join(pstdpath, f"{stdbcd}_{ii}.png")
shutil.copy2(filepath, stdpath)
ii += 1
stdfeat = stdDict[stdbcd]["feats"]
evtfeat = event["feats_compose"]
# 比较特征归一化前后的相似度变化情况
matrix = 1 - cdist(stdfeat, evtfeat, 'cosine')
simi_mean = np.mean(matrix)
simi_max = np.max(matrix)
stdfeatm = np.mean(stdfeat, axis=0, keepdims=True)
evtfeatm = np.mean(evtfeat, axis=0, keepdims=True)
simi_mfeat = 1- np.maximum(0.0, cdist(stdfeatm, evtfeatm, 'cosine'))
results.append((label, stdbcd, evt, simi_mean, simi_max, simi_mfeat[0,0]))
print("contrast performance evaluate have done!")
tm = datetime.fromtimestamp(time.time()).strftime('%Y%m%d_%H%M%S')
with open(f'{tm}.pickle', 'wb') as f:
pickle.dump(results, f)
with open(f'{tm}.txt', 'w', encoding='utf-8') as f:
for result in results:
part = [f"{x:.3f}" if isinstance(x, float) else str(x) for x in result]
line = ', '.join(part)
f.write(line + '\n')
def compute_contrast_accuracy(pickpath):
pickfile = os.path.basename(pickpath)
file, _ = os.path.splitext(pickfile)
# tm = datetime.fromtimestamp(time.time()).strftime('%Y%m%d_%H%M%S')
with open(pickpath, 'rb') as f:
results = pickle.load(f)
Same, Cross = [], []
for label, stdbcd, evt, simi_mean, simi_max, simi_mft in results:
if label == "same":
Same.append(simi_max)
if label == "diff":
Cross.append(simi_max)
Same = np.array(Same)
Cross = np.array(Cross)
TPFN = len(Same)
TNFP = len(Cross)
fig, axs = plt.subplots(2, 1)
axs[0].hist(Same, bins=60, edgecolor='black')
axs[0].set_xlim([-0.2, 1])
axs[0].set_title(f'Same Barcode, Num: {TPFN}')
axs[1].hist(Cross, bins=60, edgecolor='black')
axs[1].set_xlim([-0.2, 1])
axs[1].set_title(f'Cross Barcode, Num: {TNFP}')
plt.savefig(f'./result/{file}_hist.png') # svg, png, pdf
Recall_Pos, Recall_Neg = [], []
Precision_Pos, Precision_Neg = [], []
Correct = []
Thresh = np.linspace(-0.2, 1, 100)
for th in Thresh:
TP = np.sum(Same > th)
FN = TPFN - TP
TN = np.sum(Cross < th)
FP = TNFP - TN
Recall_Pos.append(TP/TPFN)
Recall_Neg.append(TN/TNFP)
Precision_Pos.append(TP/(TP+FP+1e-6))
Precision_Neg.append(TN/(TN+FN+1e-6))
Correct.append((TN+TP)/(TPFN+TNFP))
fig, ax = plt.subplots()
ax.plot(Thresh, Correct, 'r', label='Correct: (TN+TP)/(TPFN+TNFP)')
ax.plot(Thresh, Recall_Pos, 'b', label='Recall_Pos: TP/TPFN')
ax.plot(Thresh, Recall_Neg, 'g', label='Recall_Neg: TN/TNFP')
ax.plot(Thresh, Precision_Pos, 'c', label='Precision_Pos: TP/(TP+FP)')
ax.plot(Thresh, Precision_Neg, 'm', label='Precision_Neg: TN/(TN+FN)')
ax.set_xlim([0, 1])
ax.set_ylim([0, 1])
ax.grid(True)
ax.set_title('PrecisePos & PreciseNeg')
ax.set_xlabel(f"Same Num: {TPFN}, Cross Num: {TNFP}")
ax.legend()
plt.show()
plt.savefig(f'./result/{file}_pr.png') # svg, png, pdf
print("Haved done!!!")
def generate_event_and_stdfeatures():
stdSamplePath = r"\\192.168.1.28\share\已标注数据备份\对比数据\barcode\barcode_1771"
stdBarcodePath = r"\\192.168.1.28\share\测试_202406\contrast\std_barcodes"
stdFeaturePath = r"\\192.168.1.28\share\测试_202406\contrast\std_features"
'''=========================== 1. 生成标准特征集 ========================'''
'''1.1 提取并保存标准特征库原始图像文件地址字典'''
# get_std_barcodeDict(stdSamplePath, stdBarcodePath)
# print("standard imgpath have extracted and saved")
'''1.2 特征提取,并保存至文件夹 stdFeaturePath 中也可在运行过程中根据barcodes交集执行'''
# stdfeat_infer(stdBarcodePath, stdFeaturePath, bcdSet=None)
# print("standard features have generated!")
'''=========================== 2. 提取并存储事件特征 ========================'''
eventDatePath = [# r'\\192.168.1.28\share\测试_202406\0723\0723_1',
# r'\\192.168.1.28\share\测试_202406\0723\0723_2',
# r'\\192.168.1.28\share\测试_202406\0723\0723_3',
# r'\\192.168.1.28\share\测试_202406\0722\0722_01',
# r'\\192.168.1.28\share\测试_202406\0722\0722_02'
# r'\\192.168.1.28\share\测试_202406\0719\719_3',
# r'\\192.168.1.28\share\测试_202406\0716\0716_1',
# r'\\192.168.1.28\share\测试_202406\0716\0716_2',
# r'\\192.168.1.28\share\测试_202406\0716\0716_3',
# r'\\192.168.1.28\share\测试_202406\0712\0712_1', # 无帧图像
# r'\\192.168.1.28\share\测试_202406\0712\0712_2', # 无帧图像
]
eventFeatPath = r"\\192.168.1.28\share\测试_202406\contrast\events"
subimgPath = r'\\192.168.1.28\share\测试_202406\contrast\subimgs'
eventList = []
# k = 0
for datePath in eventDatePath:
for eventName in os.listdir(datePath):
pickpath = os.path.join(eventFeatPath, f"{eventName}.pickle")
if os.path.isfile(pickpath):
continue
eventPath = os.path.join(datePath, eventName)
eventDict = creat_shopping_event(eventPath)
if eventDict:
eventList.append(eventDict)
with open(pickpath, 'wb') as f:
pickle.dump(eventDict, f)
print(f"Event: {eventName}, have saved!")
# k += 1
# if k==1:
# break
## 保存轨迹中 boxes 子图
# for event in eventList:
# basename = os.path.basename(event['filepath'])
# savepath = os.path.join(subimgPath, basename)
# if not os.path.exists(savepath):
# os.makedirs(savepath)
# save_event_subimg(event, savepath)
print("eventList have generated and features have saved!")
def main():
# generate_event_and_stdfeatures()
contrast_performance_evaluate()
ppath = r"D:\DetectTracking\contrast"
for filename in os.listdir(ppath):
if filename.find('.pickle') < 0:
continue
pickpath = os.path.join(ppath, filename)
compute_contrast_accuracy(pickpath)
if __name__ == '__main__':
main()