以下是Python基础学习第三十二课的完整内容，聚焦Python在机器人技术与自动化控制中的创新应用，涵盖ROS机器人操作系统开发、计算机视觉引导与工业自动化控制，带你进入物理世界的数字化交互领域：

 

Python基础学习第三十二课：机器人技术与自动化控制

 

一、课程目标

 

1. 掌握ROS（机器人操作系统）基础开发

2. 学习计算机视觉引导机器人运动（OpenCV+ROS）

3. 构建工业自动化控制系统（Modbus+PLC通信）

4. 开发自主移动机器人路径规划算法

 

二、ROS机器人操作系统开发

 

1. ROS基础架构与节点通信

 

(1) 安装ROS Noetic（Ubuntu环境）

 

# 设置软件源

sudo sh -c 'echo "deb http://packages.ros.org/ros/ubuntu $(lsb_release -sc) main" > /etc/apt/sources.list.d/ros-latest.list'

sudo apt-key adv --keyserver 'hkp://keyserver.ubuntu.com:80' --recv-key C1CF6E31E6BADE8868B172B4F42ED6FBAB17C654

sudo apt update

 

# 安装ROS核心包

sudo apt install ros-noetic-desktop-full

 

# 初始化rosdep

sudo rosdep init

rosdep update

 

# 环境配置

echo "source /opt/ros/noetic/setup.bash" >> ~/.bashrc

source ~/.bashrc

 

(2) 创建ROS工作空间与功能包

 

# 创建工作空间

mkdir -p ~/catkin_ws/src

cd ~/catkin_ws/

catkin_make

 

# 创建功能包（依赖roscpp和std_msgs）

catkin_create_pkg my_robot_pkg roscpp std_msgs

 

# 编译工作空间

cd ~/catkin_ws/

catkin_make

source devel/setup.bash

 

3. 发布与订阅话题（Python节点）

 

(1) 创建发布者节点（发布速度指令）

 

#!/usr/bin/env python

import rospy

from geometry_msgs.msg import Twist

 

def velocity_publisher():

    # 初始化节点

    rospy.init_node('velocity_publisher', anonymous=True)

    

    # 创建发布者（话题名：/cmd_vel，消息类型：Twist）

    pub = rospy.Publisher('/cmd_vel', Twist, queue_size=10)

    

    # 设置发布频率（10Hz）

    rate = rospy.Rate(10)

    

    while not rospy.is_shutdown():

        # 创建Twist消息

        cmd_vel = Twist()

        cmd_vel.linear.x = 0.2 # 线速度 (m/s)

        cmd_vel.angular.z = 0.1 # 角速度 (rad/s)

        

        # 发布消息

        pub.publish(cmd_vel)

        rate.sleep()

 

if __name__ == '__main__':

    try:

        velocity_publisher()

    except rospy.ROSInterruptException:

        pass

 

(2) 创建订阅者节点（接收激光雷达数据）

 

#!/usr/bin/env python

import rospy

from sensor_msgs.msg import LaserScan

 

def laser_callback(data):

    # 处理激光雷达数据（范围数组）

    min_distance = min(data.ranges)

    print(f"最近障碍物距离: {min_distance:.2f} 米")

 

def laser_listener():

    rospy.init_node('laser_listener', anonymous=True)

    rospy.Subscriber('/scan', LaserScan, laser_callback)

    rospy.spin() # 保持节点运行

 

if __name__ == '__main__':

    laser_listener()

 

三、计算机视觉引导机器人运动

 

1. OpenCV图像处理与目标检测

 

(1) 实时摄像头图像采集

 

import cv2

 

# 初始化摄像头

cap = cv2.VideoCapture(0)

 

while True:

    ret, frame = cap.read()

    if not ret:

        break

    

    # 转换为灰度图

    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)

    

    # 边缘检测

    edges = cv2.Canny(gray, 100, 200)

    

    # 显示结果

    cv2.imshow('Original', frame)

    cv2.imshow('Edges', edges)

    

    if cv2.waitKey(1) & 0xFF == ord('q'):

        break

 

cap.release()

cv2.destroyAllWindows()

 

(2) ArUco标记检测（机器人定位）

 

import cv2

import cv2.aruco as aruco

 

# 加载预定义字典

aruco_dict = aruco.Dictionary_get(aruco.DICT_6X6_250)

parameters = aruco.DetectorParameters_create()

 

# 检测标记

cap = cv2.VideoCapture(0)

while True:

    ret, frame = cap.read()

    if not ret:

        break

    

    # 检测ArUco标记

    corners, ids, _ = aruco.detectMarkers(frame, aruco_dict, parameters=parameters)

    

    # 绘制检测结果

    if ids is not None:

        aruco.drawDetectedMarkers(frame, corners, ids)

    

    cv2.imshow('ArUco Detection', frame)

    if cv2.waitKey(1) & 0xFF == ord('q'):

        break

 

cap.release()

cv2.destroyAllWindows()

 

2. ROS与OpenCV集成（图像话题订阅）

 

#!/usr/bin/env python

import rospy

from sensor_msgs.msg import Image

from cv_bridge import CvBridge

import cv2

 

# 初始化ROS节点和转换器

rospy.init_node('image_converter', anonymous=True)

bridge = CvBridge()

 

def image_callback(msg):

    try:

        # 将ROS图像消息转换为OpenCV格式

        cv_image = bridge.imgmsg_to_cv2(msg, desired_encoding='bgr8')

        

        # 图像处理（示例：转换为灰度图）

        gray = cv2.cvtColor(cv_image, cv2.COLOR_BGR2GRAY)

        

        # 显示结果（可选）

        cv2.imshow('ROS Image', gray)

        cv2.waitKey(1)

        

    except Exception as e:

        print(e)

 

# 订阅摄像头话题

rospy.Subscriber('/camera/image_raw', Image, image_callback)

rospy.spin()

 

四、工业自动化控制（Modbus通信）

 

1. Modbus TCP协议基础

 

(1) 安装Modbus库

 

pip install pymodbus

 

(2) Modbus TCP客户端（读写寄存器）

 

from pymodbus.client import ModbusTcpClient

 

# 连接Modbus TCP设备

client = ModbusTcpClient('192.168.1.10', port=502)

client.connect()

 

# 读取保持寄存器（地址0-9，共10个寄存器）

result = client.read_holding_registers(address=0, count=10, slave=1)

print("寄存器值:", result.registers)

 

# 写入单个寄存器（地址0，值1234）

client.write_register(address=0, value=1234, slave=1)

 

client.close()

 

2. PLC控制集成（与西门子S7通信）

 

from snap7.client import Client

 

# 连接西门子PLC

plc = Client()

plc.connect('192.168.1.20', rack=0, slot=1)

 

# 读取DB块数据（DB1.DBW0开始的2个字）

data = plc.db_read(1, 0, 4) # DB1, 起始偏移0, 4字节

value = int.from_bytes(data[0:2], byteorder='big')

print("DB1.DBW0值:", value)

 

# 写入数据

new_value = 1000

plc.db_write(1, 0, new_value.to_bytes(2, byteorder='big'))

 

plc.disconnect()

 

五、自主移动机器人路径规划

 

1. A算法实现（网格地图寻路）*

 

import heapq

 

def a_star_search(grid, start, goal):

    """A*路径规划算法"""

    neighbors = [(0,1), (1,0), (0,-1), (-1,0)] # 4方向移动

    close_set = set()

    came_from = {}

    gscore = {start: 0}

    fscore = {start: heuristic(start, goal)}

    open_heap = [(fscore[start], start)]

    

    while open_heap:

        current = heapq.heappop(open_heap)[1]

        

        if current == goal:

            return reconstruct_path(came_from, current)

        

        close_set.add(current)

        for i, j in neighbors:

            neighbor = current[0] + i, current[1] + j

            if 0 <= neighbor[0] < len(grid) and 0 <= neighbor[1] < len(grid[0]):

                if grid[neighbor[0]][neighbor[1]] == 1: # 障碍物

                    continue

                tentative_g = gscore[current] + 1

                if neighbor in close_set and tentative_g >= gscore.get(neighbor, float('inf')):

                    continue

                if tentative_g < gscore.get(neighbor, float('inf')) or neighbor not in [i[1] for i in open_heap]:

                    came_from[neighbor] = current

                    gscore[neighbor] = tentative_g

                    fscore[neighbor] = tentative_g + heuristic(neighbor, goal)

                    heapq.heappush(open_heap, (fscore[neighbor], neighbor))

    return []

 

def heuristic(a, b):

    """曼哈顿距离启发式函数"""

    return abs(a[0] - b[0]) + abs(a[1] - b[1])

 

def reconstruct_path(came_from, current):

    path = [current]

    while current in came_from:

        current = came_from[current]

        path.append(current)

    return path[::-1]

 

# 示例：5x5网格地图（0=自由空间，1=障碍物）

grid = [

    [0, 0, 0, 0, 0],

    [0, 1, 1, 1, 0],

    [0, 0, 0, 1, 0],

    [0, 1, 0, 0, 0],

    [0, 0, 0, 0, 0]

]

start = (0, 0)

goal = (4, 4)

path = a_star_search(grid, start, goal)

print("规划路径:", path)

 

2. ROS导航栈集成

 

#!/usr/bin/env python

import rospy

from geometry_msgs.msg import PoseStamped

 

def send_goal(x, y, theta):

    # 初始化ROS节点

    rospy.init_node('navigation_goal_sender', anonymous=True)

    

    # 创建目标位姿发布者

    pub = rospy.Publisher('/move_base_simple/goal', PoseStamped, queue_size=10)

    

    # 创建目标消息

    goal = PoseStamped()

    goal.header.frame_id = "map"

    goal.pose.position.x = x

    goal.pose.position.y = y

    goal.pose.orientation.z = theta # 简化示例

    

    # 发布目标

    pub.publish(goal)

    print(f"已发送目标位姿: x={x}, y={y}, theta={theta}")

 

if __name__ == '__main__':

    try:

        send_goal(2.0, 3.0, 0.5)

    except rospy.ROSInterruptException:

        pass

 

六、课后练习

 

1. 基础题：

   - 在ROS中创建一个节点，订阅激光雷达数据并检测圆形障碍物

   - 使用OpenCV实现颜色跟踪（HSV色彩空间过滤）

2. 进阶题：

   - 开发基于深度强化学习的机器人路径规划系统（使用Stable-Baselines3）

   - 构建工业机器人视觉引导抓取系统（结合ROS与OpenCV）

 

七、常见问题解答

 

1. Q：ROS节点通信延迟如何优化？A：

   - 使用

"tcp_nodelay"选项减少网络延迟

   - 优化消息频率（避免过高发布频率）

2. Q：Modbus通信不稳定怎么办？A：

   - 增加重试机制（

"pymodbus"的

"retries"参数）

   - 使用心跳包检测连接状态

3. Q：A算法性能瓶颈？*A：

   - 使用更高效的启发式函数（如对角线距离）

   - 实现跳点搜索(JPS)优化

 

通过本课，你已掌握机器人技术的数字化核心！从ROS开发到工业自动化控制，具备构建智能机器人系统的能力。下一步可探索无人机集群控制或人机协作机器人(HRC)，持续拓展物理世界的数字化交互边界。 🤖🚀