AIlib2/utilsK/noParkingUtils.py

464 lines
22 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.

# 耗时最短代码
import numpy as np
import cv2,time,math
import matplotlib.pyplot as plt
def get_ms(time2, time1):
return (time2-time1)*1000.0
# 计算一点到二次函数曲线的距离二次函数的表达式为x = a*(y**2) + b*y + c
def point2QF(a, b, c, y, x): # 坐标点(y, x)
distance = abs(x - a*(y**2) - b*y - c) / math.sqrt(1 + ((2*a*y + b)**2))
return distance
# 存储所有speedRoad的contours
def storageRoad(contours, pars):
allRoadCnt = [] # 存储所有speedRoad的contours
for cnt in contours: # 道路
if len(cnt) >= 6:
rect = cv2.minAreaRect(cnt)
if rect[1][0] * rect[1][1] > pars['RoadArea']: # 过滤掉面积小于阈值的speedRoad
allRoadCnt.append(cnt)
return allRoadCnt
# 返回符合标准的lane的个数及contours
def storageLane(contours, pars):
"""
contourslane分割后的原始contours
newLaneContours符合标准的lane的contours
laneNumber符合标准的lane的个数
符合标准的lane定义如下
1contours中的坐标点个数不小于6
2lane最小外接矩形的面积大于阈值laneArea
3lane最小外接矩形的最短边与最长边的比值小于等于阈值roundness
"""
laneNumber = 0
newLaneContours = ()
for cnt in contours:
if len(cnt) >= 6:
rect = cv2.minAreaRect(cnt)
if rect[1][0] * rect[1][1] > pars['laneArea'] and min(rect[1]) / max(rect[1]) <= pars['roundness']:
laneNumber += 1
newLaneContours = newLaneContours + (cnt, )
return laneNumber, newLaneContours
# 将contours中顶点数大于等于6的车辆信息(合格vehicle)和顶点数小于6的车辆信息(不合格vehicle)分别保存起来
def vehicleDivide(contours, vehicleBD, normVehicle, dets, count, i, unnormVehicle, normVehicleCOOR, centerCOOR):
if len(contours) >= 6:
vehicleBD.append(contours)
normVehicle.append(dets[count])
normVehicleCOOR.append(centerCOOR)
else:
dets[int(i / 2)].append(0)
dets[int(i / 2)].append(0)
unnormVehicle.append(dets[int(i / 2)])
return vehicleBD, normVehicle, unnormVehicle, normVehicleCOOR
# 存储所有vehicle的信息
def storageVehicle(pars, imgVehicle, dets):
"""
输入
pars字典名
imgVehicle分割图只包含vehicle和背景
dets是一个list其中存储检测得到的各vehicle的信息即[[x0, y0, x1, y1, 车辆得分, cls], ...]
输出
dets存储合格vehicle的信息即[x0, y0, x1, y1, 车辆得分, cls]
vehicleBD存储合格vehicle的contours
unnormVehicle存储不合格vehicle的信息即[x0, y0, x1, y1, 车辆得分, cls]
normVehicleCOOR存储合格vehicle的中心点坐标
说明
合格vehiclecontours中的顶点数大于等于6
不合格vehiclecontours中的顶点数小于6
"""
vehicleBD = [] # 存储一副图像中vehicles的contours
normVehicle = [] # 将合格vehicle的信息存储在normVehicle中
unnormVehicle = [] # 将不合格vehicle的信息存储在unnormVehicle中
normVehicleCOOR = [] # 存储合格vehicle的中心点坐标
img = cv2.cvtColor(imgVehicle, cv2.COLOR_BGR2GRAY)
count = 0
for i in range(0, len(pars['vehicleCOOR']), 2):
y1 = int(pars['vehicleCOOR'][i][1] * pars['ZoomFactor']['y'])
y2 = int(pars['vehicleCOOR'][i + 1][1] * pars['ZoomFactor']['y'])
x1 = int(pars['vehicleCOOR'][i][0] * pars['ZoomFactor']['x'])
x2 = int(pars['vehicleCOOR'][i + 1][0] * pars['ZoomFactor']['x'])
if y1 >= 2:
y1 = y1 - 2
if y2 <= (pars['modelSize'][1] - 2):
y2 = y2 + 2
if x1 >= 2:
x1 = x1 - 2
if x2 <= (pars['modelSize'][0] - 2):
x2 = x2 + 2
centerCOOR = (int((x1 + x2) / 2), int((y1 + y2) / 2))
img1 = img[y1:y2, x1:x2]
up = np.zeros((20, (x2 - x1)), dtype='uint8')
left = np.zeros(((40 + y2 - y1), 20), dtype='uint8')
img1 = np.concatenate((up, img1), axis=0)
img1 = np.concatenate((img1, up), axis=0)
img1 = np.concatenate((left, img1), axis=1)
img2 = np.concatenate((img1, left), axis=1)
contours2, hierarchy = cv2.findContours(img2, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_NONE)
if len(contours2) != 0:
if len(contours2) > 1:
vehicleArea = [] # 存储vehicle的最小外接矩形的面积
for j in range(len(contours2)):
rect = cv2.minAreaRect(contours2[j])
vehicleArea.append(rect[1][0] * rect[1][1])
maxAreaIndex = vehicleArea.index(max(vehicleArea))
maxAreaContours = contours2[maxAreaIndex]
vehicleBD, normVehicle, unnormVehicle, normVehicleCOOR = vehicleDivide(maxAreaContours, vehicleBD, normVehicle, dets, count, i, unnormVehicle, normVehicleCOOR, centerCOOR)
elif len(contours2) == 1:
vehicleBD, normVehicle, unnormVehicle, normVehicleCOOR = vehicleDivide(contours2[0], vehicleBD, normVehicle, dets, count, i, unnormVehicle, normVehicleCOOR, centerCOOR)
else:
dets[int(i / 2)].append(0)
dets[int(i / 2)].append(0)
unnormVehicle.append(dets[int(i / 2)])
count += 1
dets = normVehicle
return dets, vehicleBD, unnormVehicle, normVehicleCOOR
# 计算违停得分
def IllegalParkScore1(vehicleBD, allRoadCnt, dets, unnormVehicle, normVehicleCOOR, a_l, b_l, c_l, a_r, b_r, c_r):
"""
对vehicle是否在speedRoad上进行判断并计算违章得分
输出targetList 其格式为:[[cls, x0, y0, x1, y1, score, 违章得分, 违章类别], ...]
"""
if len(vehicleBD) != 0:
for i in range(len(vehicleBD)):
rect = cv2.minAreaRect(vehicleBD[i])
center = normVehicleCOOR[i] # vehicle的中心点坐标
if len(allRoadCnt) != 0: # 当车道线个数至少有两条时,才计算违章得分
for j in range(len(allRoadCnt)):
# 判断车辆矩形框的中心点坐标是否在道路矩形框的范围内
flag = cv2.pointPolygonTest(allRoadCnt[j], center, False)
if flag >= 0:
dets[i].append(0) # 给违章得分占位
dets[i].append(0) # 给违章类别占位
if center[0] < predict(a_l, b_l, c_l, center[1]):
distance = point2QF(a_l, b_l, c_l, center[1], center[0])
if distance >= min(rect[1]) / 2:
dets[i][6], dets[i][7] = 1, 1
else:
dets[i][6], dets[i][7] = distance / (min(rect[1]) / 2), 1
elif center[0] > predict(a_r, b_r, c_r, center[1]):
distance = point2QF(a_r, b_r, c_r, center[1], center[0])
if distance >= min(rect[1]) / 2:
dets[i][6], dets[i][7] = 1, 1
else:
dets[i][6], dets[i][7] = distance / (min(rect[1]) / 2), 1
else:
dets[i][6], dets[i][7] = 0, 0
break
# 如果分割图像中不存在speedRoad则无法进行违章判定将所有车辆的违章类别设为0即没有违章
if len(dets[i]) < 8:
dets[i].append(0) # 违章得分为0
dets[i].append(0) # 0表示没有违章
targetList = dets
if len(unnormVehicle) != 0:
for i in range(len(unnormVehicle)):
targetList.append(unnormVehicle[i]) # 将所有车辆的信息合并到一起
else:
targetList = unnormVehicle
return targetList
# 计算违停得分
def IllegalParkScore2(vehicleBD, dets, unnormVehicle):
"""
计算违章得分
输出targetList 其格式为:[[cls, x0, y0, x1, y1, score, 违章得分, 违章类别], ...]
"""
if len(vehicleBD) != 0:
for i in range(len(vehicleBD)):
if len(dets[i]) < 8:
dets[i].append(0) # 违章得分为0
dets[i].append(0) # 0表示没有违章
targetList = dets
if len(unnormVehicle) != 0:
for i in range(len(unnormVehicle)):
targetList.append(unnormVehicle[i]) # 将所有车辆的信息合并到一起
else:
targetList = unnormVehicle
return targetList
# 找最左侧lane时将要删除的右侧lane的序号存储在delRightLane。找最右侧lane时将要删除的左侧lane的序号存储在delLeftLane。
def devideLane(laneInfo, i, m, delRightLane, delLeftLane, y):
index1 = np.where(laneInfo[i][3] == y)
index1 = index1[0].tolist()
index1.sort()
x_1 = laneInfo[i][5][index1[0]][0]
index2 = np.where(laneInfo[m][3] == y)
index2 = index2[0].tolist()
index2.sort()
x_2 = laneInfo[m][5][index2[0]][0]
if x_1 < x_2:
if i not in delLeftLane:
delLeftLane.append(i) # 保留右侧lane
if m not in delRightLane:
delRightLane.append(m)
else:
if m not in delLeftLane:
delLeftLane.append(m)
if i not in delRightLane:
delRightLane.append(i)
return delRightLane, delLeftLane
# 确定最左侧和最右侧的lane簇
def detLine(contours):
"""
输入
contours各lane的contours
输出
laneInfo存储各lane的信息每条lane的信息为[contours, y坐标范围, lane序号, arr_y, y坐标范围的长度, cnt]
delRightLane在确定最左侧lane时其存储需要删除的lane的序号
delLeftLane在确定最右侧lane时其存储需要删除的lane的序号
"""
mergList = []
for i in range(len(contours)):
cnt = np.squeeze(contours[i], 1)
arr_y = cnt[:, 1]
arrList = list(set(arr_y))
cnt_y = np.sort(np.array(arrList))
mergList.append([contours[i], cnt_y, i, arr_y, len(cnt_y), cnt])
laneInfo = sorted(mergList, key=(lambda x: x[4])) # [[contours[i], cnt_y, i, arr_y, len(cnt_y)],...]
delRightLane = [] # 求最左侧lane
delLeftLane = [] # 求最右侧lane
laneInfoNew = []
for i in range(len(laneInfo)):
laneInfoNew.append([laneInfo[i][1][0], laneInfo[i][1][-1], i]) # [[y_min, y_max, i],...]
laneInfoNew = np.array(laneInfoNew)
new1 = laneInfoNew[:, np.newaxis, :].repeat(laneInfoNew.shape[0], 1)
new2 = laneInfoNew[np.newaxis, ...].repeat(laneInfoNew.shape[0], 0)
new3 = np.concatenate((new1, new2), axis=2)
y_i_min, y_i_max, y_m_min, y_m_max = new3[..., 0], new3[..., 1], new3[..., 3], new3[..., 4]
mask1 = (y_i_min >= y_m_min) & (y_i_min <= y_m_max) & (y_i_max > y_m_max)
mask2 = (y_i_max >= y_m_min) & (y_i_max <= y_m_max) & (y_i_min < y_m_min)
mask3 = (y_i_min >= y_m_min) & (y_i_max <= y_m_max)
mask4 = (y_i_min < y_m_min) & (y_i_max > y_m_max)
if len(np.nonzero(mask1)[0]) != 0:
mask1 = np.triu(mask1, k=1)
serial_i = new3[mask1][..., 2]
serial_m = new3[mask1][..., 5]
for k in range(len(serial_i)):
if (serial_m[k] not in delLeftLane) or (serial_m[k] not in delRightLane) or (serial_i[k] not in delLeftLane) or (serial_i[k] not in delRightLane):
delRightLane, delLeftLane = devideLane(laneInfo, serial_i[k], serial_m[k], delRightLane, delLeftLane, laneInfo[serial_i[k]][1][0])
if len(np.nonzero(mask2)[0]) != 0:
mask2 = np.triu(mask2, k=1)
serial_i = new3[mask2][..., 2]
serial_m = new3[mask2][..., 5]
for k in range(len(serial_i)):
if (serial_m[k] not in delLeftLane) or (serial_m[k] not in delRightLane) or (serial_i[k] not in delLeftLane) or (serial_i[k] not in delRightLane):
delRightLane, delLeftLane = devideLane(laneInfo, serial_i[k], serial_m[k], delRightLane, delLeftLane, laneInfo[serial_i[k]][1][-1])
if len(np.nonzero(mask3)[0]) != 0:
mask3 = np.triu(mask3, k=1)
serial_i = new3[mask3][..., 2]
serial_m = new3[mask3][..., 5]
for k in range(len(serial_i)):
if (serial_m[k] not in delLeftLane) or (serial_m[k] not in delRightLane) or (serial_i[k] not in delLeftLane) or (serial_i[k] not in delRightLane):
delRightLane, delLeftLane = devideLane(laneInfo, serial_i[k], serial_m[k], delRightLane, delLeftLane, laneInfo[serial_i[k]][1][0])
if len(np.nonzero(mask4)[0]) != 0:
mask4 = np.triu(mask4, k=1)
serial_i = new3[mask4][..., 2]
serial_m = new3[mask4][..., 5]
for k in range(len(serial_i)):
if (serial_m[k] not in delLeftLane) or (serial_m[k] not in delRightLane) or (serial_i[k] not in delLeftLane) or (serial_i[k] not in delRightLane):
delRightLane, delLeftLane = devideLane(laneInfo, serial_i[k], serial_m[k], delRightLane, delLeftLane, laneInfo[serial_m[k]][1][0])
return laneInfo, delRightLane, delLeftLane
# 对lane中的y值坐标进行下采样
def downSample(cnt_y):
# number = len(cnt_y) * 0.0125
# cnt_y = np.random.choice(cnt_y, size=number, replace=False)
if len(cnt_y) >= 1000:
cnt_y = cnt_y[1::80]
elif len(cnt_y) >= 900 and len(cnt_y) < 1000:
cnt_y = cnt_y[1::75]
elif len(cnt_y) >= 800 and len(cnt_y) < 900:
cnt_y = cnt_y[1::70]
elif len(cnt_y) >= 700 and len(cnt_y) < 800:
cnt_y = cnt_y[1::65]
elif len(cnt_y) >= 600 and len(cnt_y) < 700:
cnt_y = cnt_y[1::60]
elif len(cnt_y) >= 500 and len(cnt_y) < 600:
cnt_y = cnt_y[1::55]
elif len(cnt_y) >= 400 and len(cnt_y) < 500:
cnt_y = cnt_y[1::40]
elif len(cnt_y) >= 300 and len(cnt_y) < 400:
cnt_y = cnt_y[1::45]
elif len(cnt_y) >= 200 and len(cnt_y) < 300:
cnt_y = cnt_y[1::40]
elif len(cnt_y) >= 100 and len(cnt_y) < 200:
cnt_y = cnt_y[1::35]
elif len(cnt_y) >= 50 and len(cnt_y) < 100:
cnt_y = cnt_y[1::20]
elif len(cnt_y) >= 20 and len(cnt_y) < 50:
cnt_y = cnt_y[1::6]
else:
cnt_y = cnt_y[1::5]
return cnt_y
# 求最左侧lane或最右侧lane中的各点坐标
def targetCOOR(laneInfo, delLane):
"""
输入
laneInfo存储各lane的信息每条lane的信息为[contours, y坐标范围, lane序号, arr_y, y坐标范围的长度, cnt]
delLane在确定最左侧lane或最右侧lane时其存储需要删除的lane的序号。
输出
laneCOOR存储最左侧或最右侧lane簇中各点的坐标
"""
laneCOOR = [] # 存储lane中各点的坐标
centerSort = [] # 存储各lane按照中心点的y坐标排序后的结果
for j in range(len(laneInfo)):
if j not in delLane:
cnt = laneInfo[j][0]
rect = cv2.minAreaRect(cnt)
cnt = np.squeeze(cnt, 1)
cnt_y = laneInfo[j][1]
cnt_y = downSample(cnt_y)
centerSort.append([rect[0][1], cnt_y, laneInfo[j][3], cnt, j])
centerSort = sorted(centerSort, key=(lambda x: x[0]))
for i in range(len(centerSort)):
centerCoordinate = []
for j in range(len(centerSort[i][1])):
index = np.where(centerSort[i][2] == centerSort[i][1][j])
indexList = index[0].tolist()
indexList.sort()
x = (centerSort[i][3][indexList[0]][0] + centerSort[i][3][indexList[-1]][0]) / 2
y = (centerSort[i][3][indexList[0]][1] + centerSort[i][3][indexList[-1]][1]) / 2
centerCoordinate.append([x, y])
laneCOOR = laneCOOR + centerCoordinate
return laneCOOR
# 二次函数曲线表达式x = a*(y**2) + b*y + c根据图像中一点的y坐标求二次曲线中的x坐标
def predict(a, b, c, y):
x = a * (y**2) + b * y + c
return x
def mixNoParking_road_postprocess(dets, mask, pars):
"""
对于字典traffic_dict中的各个键说明如下
RoadAreaspeedRoad的最小外接矩形的面积
vehicleCOOR是一个列表用于存储被检测出的vehicle的坐标vehicle检测模型
roundness圆度 ,lane的长与宽的比率作为判定是否为车道线的标准之一
laneArea车道线的最小外接矩形的面积
ZoomFactor图像在H和W方向上的缩放因子其值小于1
fitOrder多点拟合曲线的阶数
最终输出格式:[[x0, y0, x1, y1, 车辆得分, cls, 违章停车得分, 违章类别], ...]
违章类别0表示正常车辆1表示违章车辆
"""
det_cors = []
for bb in dets:
det_cors.append((int(bb[0]), int(bb[1])))
det_cors.append((int(bb[2]), int(bb[3])))
print('###line341:', det_cors)
pars['vehicleCOOR'] = det_cors
H, W = mask.shape[0:2] # mask的分辨率为360x640
scaleH = pars['modelSize'][1] / H # 自适应调整缩放比例
scaleW = pars['modelSize'][0] / W
pars['ZoomFactor'] = {'x': scaleW, 'y': scaleH}
new_hw = [int(H * scaleH), int(W * scaleW)]
mask = cv2.resize(mask, (new_hw[1], new_hw[0]))
if len(mask.shape) == 3:
mask = mask[:, :, 0]
t1 = time.time()
imgRoad = mask.copy()
imgVehicle = mask.copy()
lane_line = mask.copy()
# 将vehicle和lane过滤掉只包含背景和speedRoad
imgRoad[imgRoad == 2] = 1
imgRoad[imgRoad == 3] = 1
# 将speedRoad和lane过滤掉只保留vehicle和背景
imgVehicle[imgVehicle != 2] = 0
# 将speedRoad和vehicle过滤掉只保留lane和背景
lane_line[lane_line < 3] = 0
imgRoad = cv2.cvtColor(np.uint8(imgRoad), cv2.COLOR_RGB2BGR) # 道路
imgVehicle = cv2.cvtColor(np.uint8(imgVehicle), cv2.COLOR_RGB2BGR) # 车辆
lane_line = cv2.cvtColor(np.uint8(lane_line), cv2.COLOR_RGB2BGR)
# 对车道线进行膨胀操作
# kernel = np.ones((3, 3), np.uint8) # 膨胀范围
# lane_line = cv2.dilate(lane_line, kernel, iterations=2) # 迭代次数为2
t2 = time.time()
img1 = cv2.cvtColor(imgRoad, cv2.COLOR_BGR2GRAY)
roadContours, hierarchy = cv2.findContours(img1, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_NONE)
t3 = time.time()
# 存储所有speedRoad的信息
allRoadCnt = storageRoad(roadContours, pars)
t4 = time.time()
img3 = cv2.cvtColor(lane_line, cv2.COLOR_BGR2GRAY)
laneContours, hierarchy = cv2.findContours(img3, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_NONE)
# 存储所有lane的信息
laneNumber, newLaneContours = storageLane(laneContours, pars)
t5 = time.time()
if laneNumber >= 2:
laneInfo, delRightLane, delLeftLane = detLine(newLaneContours)
t6 = time.time()
# 存储所有vehicle的信息
dets, vehicleBD, unnormVehicle, normVehicleCOOR = storageVehicle(pars, imgVehicle, dets)
t7 = time.time()
leftLaneCOOR = targetCOOR(laneInfo, delRightLane)
rightLaneCOOR = targetCOOR(laneInfo, delLeftLane)
rightLaneCOOR = np.array(rightLaneCOOR)
rightX = rightLaneCOOR[:, 0]
rightY = rightLaneCOOR[:, 1]
leftLaneCOOR = np.array(leftLaneCOOR)
leftX = leftLaneCOOR[:, 0]
leftY = leftLaneCOOR[:, 1]
# a_rb_rc_r分别是最右侧车道线簇拟合的二次函数的二次项系数一次项系数和常数项
a_r, b_r, c_r = np.polyfit(rightY, rightX, pars['fitOrder'])[0], np.polyfit(rightY, rightX, pars['fitOrder'])[1], np.polyfit(rightY, rightX, pars['fitOrder'])[2]
# a_lb_lc_l分别是最左侧车道线簇拟合的二次函数的二次项系数一次项系数和常数项
a_l, b_l, c_l = np.polyfit(leftY, leftX, pars['fitOrder'])[0], np.polyfit(leftY, leftX, pars['fitOrder'])[1], np.polyfit(leftY, leftX, pars['fitOrder'])[2]
# """以下四行代码用于在后处理函数外画图"""
# finalLane = []
# abc = [a_l, b_l, c_l, a_r, b_r, c_r] # abc中存储的是最左侧和最右侧二次函数的各项系数
# finalLane.append(rightLaneCOOR)
# finalLane.append(leftLaneCOOR)
# 计算违停得分
t8 = time.time()
targetList = IllegalParkScore1(vehicleBD, allRoadCnt, dets, unnormVehicle, normVehicleCOOR, a_l, b_l, c_l, a_r, b_r, c_r)
t9 = time.time()
time_infos = 'postTime:%.2f(分割时间:%.2f, findContours:%.2f, ruleJudge:%.2f, storageRoad:%.2f, detLane:%.2f, storageLane:%.2f, storageVehicle:%.2f, fitLine:%.2f, IllegalParkScore1:%.2f)' % (
get_ms(t9, t1), get_ms(t2, t1), get_ms(t3, t2), get_ms(t9, t3), get_ms(t4, t3), get_ms(t6, t5),
get_ms(t5, t4), get_ms(t7, t6), get_ms(t8, t7), get_ms(t9, t8))
# print('####line445:', targetList)
# return targetList, time_infos, finalLane, lane_line, abc
targetList = [ [ *b[0:4],b[6] if b[6]>0 else b[4], b[7] ] for b in targetList ]
return targetList, time_infos
else:
dets, vehicleBD, unnormVehicle, normVehicleCOOR = storageVehicle(pars, imgVehicle, dets)
t6 = time.time()
targetList = IllegalParkScore2(vehicleBD, dets, unnormVehicle)
t7 = time.time()
time_infos = 'postTime:%.2f(分割时间:%.2f, findContours:%.2f, ruleJudge:%.2f, storageRoad:%.2f, storageLane:%.2f, storageVehicle:%.2f, IllegalParkScore2:%.2f)' % (
get_ms(t7, t1), get_ms(t2, t1), get_ms(t3, t2), get_ms(t7, t3), get_ms(t4, t3), get_ms(t5, t4), get_ms(t6, t5), get_ms(t7, t6))
# print('####line456:', targetList)
targetList = [ [ *b[0:4],b[6] if b[6]>0 else b[4], b[7] ] for b in targetList ]
return targetList, time_infos
def mixNoParking_road_postprocess_N(predList, pars):
dets, mask =predList[0:2]
return mixNoParking_road_postprocess(dets, mask, pars)