把以上编码写成详细的实验步骤(并进行逐句分析)
时间: 2024-12-04 08:28:36 浏览: 23
### 实验步骤及详细分析
#### 步骤1:导入必要的库和模块
```python
from sklearn.cluster import KMeans
import numpy as np
import cv2
from enum import Enum
import base64
import math
# import serial_driver
import struct
import time
```
**分析**:
- `sklearn.cluster.KMeans`:用于聚类算法,帮助确定车道线的位置。
- `numpy`:用于数值计算,特别是数组操作。
- `cv2`:OpenCV库,用于图像处理。
- `Enum`:枚举类,定义状态常量。
- `base64`:用于图像数据的Base64编码,方便在网络传输中使用。
- `math`:数学运算。
- `struct`:用于二进制数据的打包和解包。
- `time`:时间相关操作。
- `serial_driver`:注释掉了,可能是因为不需要或暂时不用。
#### 步骤2:定义状态枚举类
```python
class State(Enum):
left = 1
right = 2
center = 3
left_to_center = 4 # 左线到中线的过渡段
right_to_center = 5 # 右线到中线的过渡段
```
**分析**:
- 定义了一个枚举类 `State`,表示车辆的状态,包括左线、右线、中心线以及从侧线到中心线的过渡状态。
#### 步骤3:定义自行车参数类
```python
class Bike_args:
def __init__(self, _camera):
self.camera = _camera # 0是openni摄像头 1是免驱摄像头
# 图像阈值参数
self.blue_low = np.array([65, 66, 60])
self.blue_upper = np.array([125, 254, 255])
self.yellow_low = np.array([23, 0, 144])
self.yellow_upper = np.array([180, 255, 255])
# 寻线参数
self.find_current_x_delta_x = 130 # 寻找左右线的距离
self.shifted_x = 22 # 线偏移量
# 避障参数
self.block_detect_y = 290
self.block_detect_delta_y = 80
self.block_h_upper = 30 # 障碍物在窗内的高度的最大值
# 停车线参数
self.stop_line1_detect_y = 400
# 通信数据
self.wait_back_center_data = struct.pack("!BB", 0xA5, 0X01)
self.first_stop_line = struct.pack("!BB", 0xA5, 0X02)
```
**分析**:
- 初始化了摄像头类型、颜色阈值、寻线参数、避障参数、停车线参数和通信数据。
- 使用 `np.array` 定义了颜色范围,便于后续的颜色检测。
- `struct.pack` 用于打包通信数据。
#### 步骤4:定义自行车主类
```python
class Bike_class:
def __init__(self, _socketio, _camera, _task=0, _block_nums=3, _skip_times=400, _block_dirc=(3, 3, 3)):
self.socketio = _socketio
self.init_flag = 0
self.kmeans = KMeans(n_clusters=2)
self.bike_args = Bike_args(_camera)
self.M = None
self.M_inverse = None
self.M_inverse_list = None
self.get_warp_M()
self.color_stream = None
self.cap = None
if self.bike_args.camera == 0:
from openni import openni2
openni2.initialize()
dev = openni2.Device.open_any()
self.color_stream = dev.create_color_stream()
self.color_stream.start()
elif self.bike_args.camera == 1:
self.cap = cv2.VideoCapture(0)
else:
pass
# 串口初始化
# self.serial_status = serial_driver.start_serial("/dev/upper_serial")
# 寻线
self.state = State.center
self.dynamic_center_x = 320
self.left_fit = None
self.right_fit = None
# 过度线
self.skip_frame_times = 0
self.initial_skip_frame_times = 80
self.shift_to_center_x = 0
self.to_center_x = 0
self.dynamic_shift_x = 0
# 避障
self.task = _task
self.skip_block_nums = 0
self.last_block_state = State.center
self.catch_block_times = 0
self.wait_back_center_flag = 0
self.block_nums = _block_nums
self.stop_skip_times = _skip_times
self.block_direction_list = []
for _ in _block_dirc:
if _ == 1:
self.block_direction_list.append(State.right)
elif _ == 2:
self.block_direction_list.append(State.left)
else:
self.block_direction_list.append(State.center)
# 周期控制
self.log_times = 0
self.block_detect_times = 0
self.stop_line_times = 0
# 全局变量
self.img = None
self.warp_img = None
self.edges = None
self.line_points_x = []
self.left_line_x = []
self.right_line_x = []
self.leftx_mean = None
self.left_point_source = None
self.rightx_mean = None
self.right_point_source = None
self.catch_point_source = None
self.rightx_mean_list = []
self.leftx_mean_list = []
self.x_mean_list = []
self.x_mean = None
self.x_error = None
self.last_x_error = None
self.error_flag = 0
self.img_size = (640, 480)
self.y = 180
# 调试变量
self.wait_back_center_flag_debug = 0
self.record_flag = 0
self.error_x_save_list = []
```
**分析**:
- 初始化了各种参数和变量,包括摄像头、KMeans模型、透视变换矩阵、状态、避障参数等。
- 根据摄像头类型选择不同的初始化方式。
- 初始化了多个列表和变量,用于存储中间结果和状态信息。
#### 步骤5:显示图像
```python
def show_image(self, _img, flag):
if flag == 0:
frame = cv2.resize(_img, (160, 120))
encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), 85]
_, compressed_frame = cv2.imencode('.jpg', frame, encode_param)
frame_data = base64.b64encode(compressed_frame).decode('utf-8')
self.socketio.emit('image', {'image_data': frame_data})
else:
encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), 85]
_, compressed_frame = cv2.imencode('.jpg', _img, encode_param)
frame_data = base64.b64encode(compressed_frame).decode('utf-8')
self.socketio.emit('event_image', {'image_data': frame_data})
```
**分析**:
- 将图像压缩并转换为Base64格式,通过WebSocket发送给前端。
- `flag` 参数决定发送的事件类型。
#### 步骤6:显示日志
```python
def show_log(self, _text):
self.socketio.emit('log', {'log': _text})
```
**分析**:
- 发送日志信息到前端,用于调试和监控。
#### 步骤7:获取透视变换矩阵
```python
def get_warp_M(self):
objdx = 200
objdy = 230
imgdx = 220
imgdy = 250
list_pst = [[172, 330], [461, 330], [75, 475], [546, 475]]
pts1 = np.float32(list_pst)
pts2 = np.float32([[imgdx, imgdy], [imgdx + objdx, imgdy], [imgdx, imgdy + objdy], [imgdx + objdx, imgdy + objdy]])
self.M = cv2.getPerspectiveTransform(pts1, pts2)
self.M_inverse = cv2.getPerspectiveTransform(pts2, pts1)
self.M_inverse_list = self.M_inverse.flatten()
```
**分析**:
- 计算透视变换矩阵 `M` 和其逆矩阵 `M_inverse`,用于将图像从原始视角转换为鸟瞰图。
#### 步骤8:反向透视变换
```python
def point_reverse_perspective(self, point):
x, y = point
denom = self.M_inverse_list[6] * x + self.M_inverse_list[7] * y + 1
x_transformed = (self.M_inverse_list[0] * x + self.M_inverse_list[1] * y + self.M_inverse_list[2]) / denom
y_transformed = (self.M_inverse_list[3] * x + self.M_inverse_list[4] * y + self.M_inverse_list[5]) / denom
return (int(x_transformed), int(y_transformed))
```
**分析**:
- 将鸟瞰图中的点转换回原始图像中的位置。
#### 步骤9:插值函数
```python
def interpolate_value(self, start_value, end_value, initial_times, current_times):
step = (end_value - start_value) / initial_times
current_value = start_value + step * current_times
return int(current_value)
```
**分析**:
- 计算从起始值到结束值的插值,用于平滑过渡。
#### 步骤10:图像预处理
```python
def img_preprocess(self):
self.img = cv2.medianBlur(self.img, 9)
self.warp_img = cv2.warpPerspective(self.img, self.M, self.img_size)
warp_img_hsv = cv2.cvtColor(self.warp_img, cv2.COLOR_BGR2HSV)
blue_mask_white = cv2.inRange(warp_img_hsv, self.bike_args.blue_low, self.bike_args.blue_upper)
blue_mask_white = cv2.bitwise_not(blue_mask_white)
kernel = np.ones((5, 5), np.uint8)
blue_mask_white = cv2.erode(blue_mask_white, kernel, iterations=1)
edges = cv2.Canny(self.warp_img, 50, 40, apertureSize=3)
edges = cv2.bitwise_and(edges, edges, mask=blue_mask_white)
kernel = np.ones((3, 3), np.uint8)
edges = cv2.dilate(edges, kernel, iterations=2)
edges_mask = np.zeros((self.img_size[1], self.img_size[0]), dtype=np.uint8)
cv2.rectangle(edges_mask, (160, 0), (480, 480), 255, thickness=cv2.FILLED)
self.edges = cv2.bitwise_and(edges, edges, mask=edges_mask)
```
**分析**:
- 对图像进行中值滤波、透视变换、颜色阈值处理、Canny边缘检测、遮罩处理等预处理操作。
- 最终得到处理后的边缘图像 `self.edges`。
#### 步骤11:霍夫直线检测
```python
def img_HoughLines(self):
self.line_points_x.clear()
lines = cv2.HoughLines(self.edges, 1, np.pi / 180, threshold=260)
if lines is not None:
for line in lines:
rho, theta = line[0]
theta_degree = np.degrees(theta)
if theta_degree > 90:
theta_degree = 180 - theta_degree
if np.abs(theta_degree) > 35:
continue
elif np.abs(theta) == 0:
b = rho
self.line_points_x.append(int(b))
else:
m = -1 / np.tan(theta)
b = rho / np.sin(theta)
self.line_points_x.append(int((self.y - b) / m))
```
**分析**:
- 使用霍夫变换检测图像中的直线,并将检测到的直线点存储在 `self.line_points_x` 列表中。
#### 步骤12:霍夫直线过滤
```python
def img_HoughLines_filter(self):
self.left_line_x.clear()
self.right_line_x.clear()
if len(self.line_points_x) != 0:
for point_x in self.line_points_x:
if point_x < self.dynamic_center_x and point_x > (self.dynamic_center_x - self.bike_args.find_current_x_delta_x):
self.left_line_x.append(point_x)
cv2.circle(self.warp_img, (point_x, self.y), radius=5, color=(255, 255, 255), thickness=-1)
elif point_x > self.dynamic_center_x and point_x < (self.dynamic_center_x + self.bike_args.find_current_x_delta_x):
self.right_line_x.append(point_x)
cv2.circle(self.warp_img, (point_x, self.y), radius=5, color=(255, 255, 255), thickness=-1)
if self.state == State.left or self.state == State.left_to_center or self.state == State.center:
if len(self.left_line_x) != 0:
self.leftx_mean = int(np.mean(self.left_line_x))
cv2.line(self.warp_img, (self.leftx_mean, 0), (self.leftx_mean, 480), (255, 0, 0), 3)
self.error_flag = 0
else:
self.error_flag = 1
if self.state == State.right or self.state == State.right_to_center or self.state == State.center:
if len(self.right_line_x) != 0:
self.rightx_mean = int(np.mean(self.right_line_x))
cv2.line(self.warp_img, (self.rightx_mean, 0), (self.rightx_mean, 480), (255, 0, 0), 3)
self.error_flag = 0
else:
self.error_flag = 1
else:
self.error_flag = 1
```
**分析**:
- 过滤霍夫检测到的直线点,将其分为左侧线和右侧线。
- 计算左侧线和右侧线的平均位置,并绘制线条。
#### 步骤13:滑动窗口法
```python
def img_swap_windows(self):
margin = 35
minpix = 25
try:
if self.error_flag != 1:
left_lane_inds = []
last_good_left_inds_len = 0
right_lane_inds = []
last_good_right_inds_len = 0
nwindows = 8
window_height = np.int32(self.img_size[1] / nwindows)
nonzero = self.edges.nonzero()
nonzeroy = np.array(nonzero[0])
nonzerox = np.array(nonzero[1])
for window in range(nwindows):
win_y_low = self.img_size[1] - (window + 1) * window_height
win_y_high = self.img_size[1] - window * window_height
if self.state == State.left or self.state == State.center or self.state == State.left_to_center:
win_xleft_low = self.leftx_mean - margin
win_xleft_high = self.leftx_mean + margin
if self.state == State.right or self.state == State.center or self.state == State.right_to_center:
win_xright_low = self.rightx_mean - margin
win_xright_high = self.rightx_mean + margin
if self.state == State.left or self.state == State.center or self.state == State.left_to_center:
good_left_inds = ((nonzeroy >= win_y_low) & (nonzeroy < win_y_high) & (nonzerox >= win_xleft_low) & (nonzerox < win_xleft_high)).nonzero()[0]
left_lane_inds.append(good_left_inds)
last_good_left_inds_len = len(good_left_inds)
if self.state == State.right or self.state == State.center or self.state == State.right_to_center:
good_right_inds = ((nonzeroy >= win_y_low) & (nonzeroy < win_y_high) & (nonzerox >= win_xright_low) & (nonzerox < win_xright_high)).nonzero()[0]
right_lane_inds.append(good_right_inds)
last_good_right_inds_len = len(good_right_inds)
if self.state == State.left or self.state == State.center or self.state == State.left_to_center:
if last_good_left_inds_len > minpix:
self.leftx_mean = np.int32(np.mean(nonzerox[good_left_inds]))
if self.state == State.right or self.state == State.center or self.state == State.right_to_center:
if last_good_right_inds_len > minpix:
self.rightx_mean = np.int32(np.mean(nonzerox[good_right_inds]))
if self.state == State.left or self.state == State.center or self.state == State.left_to_center:
left_lane_inds = np.concatenate(left_lane_inds)
leftx = nonzerox[left_lane_inds]
lefty = nonzeroy[left_lane_inds]
if self.state == State.right or self.state == State.center or self.state == State.right_to_center:
right_lane_inds = np.concatenate(right_lane_inds)
rightx = nonzerox[right_lane_inds]
righty = nonzeroy[right_lane_inds]
if self.state == State.left or self.
阅读全文