前言在第20届全国大学生智能车竞赛智慧医疗机器人创意赛中我们贰贰玖拿下国一。在这里作为队长兼技术主力兼机师兼……我想分享一下在备赛过程中的一些思路。当然为了不把比赛搞成全都是20s以内竞争激烈到前后几名差0.几秒我不会开源我们的惯导和避障思路实在太简单太容易实现了。这是我们两年的备赛日记也有我们第二年区域赛和国赛的全流程。【贰贰玖|从省三到国一从巡线到路径规划到惯导纯视觉避障的贰贰玖智能车日记-哔哩哔哩】 https://b23.tv/IDJyM2P数据集我放在这里了一共2w9张全都是640x480有数据增强的没有旋转https://pan.baidu.com/s/10u4S4fiVATRyEeDpdzpk_A?pwd0229 提取码:022926年回来开源lio定位方案 地瓜机器人智慧医疗——开源里程计定位-CSDN博客下面面我会讲一下我们的网络问题怎么解决上位机的一些辅助处理如何半场扫码如何准确返回 P 点修改stm32以及修改车的ekf.yaml。网络问题我们第一年参赛的时候也和大家一样遇到了非常严重的网络问题——上位机延迟有位湘潭大学的学长帮我们配置了一下路由器完全没有延时现在还是非常想感谢他所以第二年非常重视这个问题。第二年备赛我们一开始用的是华为AX3 Pro在校赛的时候其他队伍连都连不上小车我们跑得飞起来跑出了18s。在校赛拿到第一名之后发了奖金840元立刻就买了华为BE7 Pro。一开始买的是华为企业级路由器但是配了好几天还是没有成功退掉了以后调车就是带着这个路由器去调在实验室没有延时。在赛场上用的是165信道因为这个信道基本没人用。也算是运气好因为区域赛前的调试在165信道只有4个路由器但发现有个ROGxxx跑起来开始延时了图1 刚到手华为BE7 Pro路由器图2 华为BE7 Pro路由器有了路由器那就不要用电脑的板载无线网卡去连路由器用就用网线上位机和开终端的最好都这样。在区域赛调试的时候我们用路由器的中继模式上网但是开了中继模式之后不能改信道刚开始2轮是没有延时的但是第3轮开始就延时卡成ppt了。当时果断放弃上网不用云端API了转用备用方案——本地部署的qwen特别垃圾但逼一下它还是会憋出几个字。在现场有很多队很有实力但是没能进国赛我猜大部分是因为网络问题。我觉得是我觉得哈如果有不对的地方不要骂我今年加的图生文环节不太合理。大家都知道现场网络问题很严重但是联网调用云端API可以有更快更高质量的结果本地部署的特别差除非有钱买卡。要么用调云端API在现场卡死要么铤而走险用本地部署的大模型赌裁判不会特别关注图生文结果。上位机辅助处理如果仔细看我B站发的上位机视角会发现桶和P点的底部会有红线。其实是我在上位机的bridge_client.py 单独又开了一个py脚本来接收YOLO结果用python的tkinter库画出来的。画出障碍物的位置可以帮助机师快速确定障碍物的位置。最初的想法是把障碍物在地图上标出来就像游戏旁边的小地图但是效果不好就放弃了。import tkinter as tk root tk.Tk() root.overrideredirect(True) root.geometry(900x680192215) root.attributes(-topmost, True) root.attributes(-transparentcolor, white) canvas tk.Canvas(root, width900, height680, bgwhite, highlightthickness0) canvas.pack() class LLM2Origincar(): def __init__(self, host, port): self.ros None self.host host self.port port self.roadblock_list [] self.end_list [] self.init_ros() self.init_topic() self.init_thread() self.keep() ... def init_topic(self): ... self.yolo_sub Topic(self.ros, /hobot_dnn_detection, ai_msgs/msg/PerceptionTargets, latch True) self.yolo_sub.subscribe(self.yolo_sub_callback) ... def yolo_sub_callback(self, msg): self.roadblock_list.clear() self.end_list.clear() for target in msg[targets]: if target[type] roadblock: rect target[rois][0][rect] self.roadblock_list.append({ x:rect[x_offset], w:rect[width], b:rect[y_offset] rect[height], }) elif target[type] end:# and target[rois][0][confidence] 0.5: rect target[rois][0][rect] self.end_list.append({ x:rect[x_offset], y:rect[y_offset], w:rect[width], b:rect[y_offset] rect[height], c:target[rois][0][confidence], }) def keep(self): try: while True: canvas.delete(all) # 清空画布 canvas.create_line(141, 0, 141, 680, fillred, width1) canvas.create_line(689, 0, 689, 680, fillred, width1) if self.roadblock_list: for obst in self.roadblock_list: b int(obst[b] * 1.42) # 680 / 480 canvas.create_line( int(obst[x] * 1.41), # 900 / 640 b, int( ( obst[x] obst[w] ) * 1.41 ), # 900 / 640 b, fillred, width2 ) if self.end_list: for end in self.end_list: x1 int(end[x] * 1.41) y1 int(end[y] * 1.41) x2 int( ( end[x] end[w] ) * 1.41 ) # 900 / 640 y2 int(end[b] * 1.42) # 680 / 480 canvas.create_line( x1, y2, x2, y2,fillblue, width1) canvas.create_text(int((x1x2)/2), (y1-20) if (y1-20) 0 else 0, textconf:{:.2f}.format(end[c]), fillcyan) ...除了画出障碍物作为辅助我们还加了几个按键来辅助任务之间的切换还有调用APIdef keyboard_thread(self): # print(help_tips) while True: sleep(0.05) if keyboard.is_pressed(b) or keyboard.is_pressed(B): # 信号连接正常 self.sign4return_pub.publish(self.sign4return_data) sleep(0.5) if keyboard.is_pressed(r) or keyboard.is_pressed(R): # 激活键盘控制 self.sign4return_data[data] 5 # C 区遥测 self.sign4return_pub.publish(self.sign4return_data) self.sign4return_data[data] 0 sleep(0.5) if keyboard.is_pressed(p) or keyboard.is_pressed(P): # 退出键盘控制 self.sign4return_data[data] 6 # Task3 self.sign4return_pub.publish(self.sign4return_data) self.sign4return_data[data] 0 sleep(0.5) if keyboard.is_pressed(j) or keyboard.is_pressed(J): # 车那边处理标志 self.llm_data[data] 1 self.llm_pub.publish(self.llm_data) sleep(1)半场扫码大家应该都知道小粉小橙前面的USB相机拍出来的照片很模糊。但是小橙有个非常大的优势——深度相机。深度相机可以获取深度信息但是除了这个它照出来的照片比USB相机清晰得多得多得多。用深度相机和USB相机同时拍出来的照片被我删掉了大家可以自己来拍二维码对比一下。大家从下面这2个距离应该可以感受到深度相机扫码的优势图3 深度相机扫码图4 USB相机扫码还有一点扫码节点千万不要一直开着特别耗CPU。我们扫码的条件是任务状态是任务一且小车过了半场即小车全局坐标的x超过2m。下面就是我们扫码的节点import rclpy from rclpy.node import Node import cv2 import numpy as np from sensor_msgs.msg import Image from std_msgs.msg import String, Int32 from nav_msgs.msg import Odometry from origincar_msg.msg import Sign from cv_bridge import CvBridge TASK1 1 TASK2_WAITFOR_CMD 2 TASK2 3 TASK3 4 TASK_STOP 5 class QrCodeDetection(Node): def __init__(self): super().__init__(QRcodeSub) self.Sign4ReturnSub self.create_subscription(Int32, sign4return, self.sign4return_callback, 10) self.ImageSub self.create_subscription(Image, /aurora/rgb/image_raw, self.image_callback, 10) self.OdomSub self.create_subscription(Odometry, /odom_combined, self.Odom_callback, 10) self.qrcode_publisher self.create_publisher(String, /qrcode_information, 10) self.info_result String() self.sign_publisher self.create_publisher(Sign, /sign_switch, 10) self.sign_msg Sign() self.detector cv2.wechat_qrcode_WeChatQRCode( /userdata/WorkSpace/codes/src/qrcode/qrcode/model/detect.prototxt, /userdata/WorkSpace/codes/src/qrcode/qrcode/model/detect.caffemodel, /userdata/WorkSpace/codes/src/qrcode/qrcode/model/sr.prototxt, /userdata/WorkSpace/codes/src/qrcode/qrcode/model/sr.caffemodel ) self.bridge CvBridge() self.node_run False self.task TASK1 def image_callback(self, msg): if self.node_run and ( self.task TASK1 or self.task TASK2 ): #cv2_image self.bridge.imgmsg_to_cv2(msg, desired_encodingbgr8)[155:,:] cv2_image self.bridge.imgmsg_to_cv2(msg, desired_encodingmono8)[155:,:] #cv2.imwrite(image.jpg,cv2_image) res self.detector.detectAndDecode(cv2_image)[0] if res: self.node_run False # 停止运行扫码 for r in res: self.info_result.data str(r) self.qrcode_publisher.publish(self.info_result) self.get_logger().info(\033[94m{}\033[0m.format(self.info_result.data)) if self.info_result.data AntiClockWise: self.sign_msg.sign_data 4 elif self.info_result.data ClockWise: self.sign_msg.sign_data 3 else: # 数字表示 try: data int(r) if data % 2: # 奇数顺时针 self.sign_msg.sign_data 3 else: # 偶数逆时针 self.sign_msg.sign_data 4 except: pass self.sign_publisher.publish(self.sign_msg) self.info_result.data None self.sign_msg.sign_data 0 else: return def sign4return_callback(self, msg): if msg.data 0 or msg.data -1: self.task TASK1 self.node_run False if msg.data 5: # 遥操作 self.task TASK2 elif msg.data 6: # 退出遥操作 self.task TASK3 def Odom_callback(self, msg): if self.task TASK1 and msg.pose.pose.position.x 2: self.node_run True # 若车过了半场且是task1则开启扫码 def main(argsNone): rclpy.init(argsargs) qrCodeDetection QrCodeDetection() while rclpy.ok(): rclpy.spin(qrCodeDetection) qrCodeDetection.destroy_node() rclpy.shutdown() if __name__ __main__: main()我们没有把整张照片丢给wechat的扫码模型而是裁掉了一部分比如下面这张图片我先取最近的扫码距离把红线上面的给去掉这样图像就小了很多。图5 裁掉一部分图片准确返回P点我们准确返回 P 点的思路有 3 个有一个是另一队的学弟跟我讲的。他们是任务二停车对准 P 点然后退出遥操作这样 P 点就在车的正前方加上YOLO识别 P 点也可以回去但是这很看机师的操作他们最后有一次成功返回有一次失败了。我的思路是校准 P 点的坐标。思路1——使用地图的固定元素来校准。这个思路要重置里程计每次都在通道重置那么相当于把原点设在了这里这里计算出来的相对坐标可以直接用作全局坐标用作后面的导航。如果小车每次从任务二出来都停在同一个位置然后重置里程计那么一定会有一个终点可以让小车回去的。比如我每次都停在这个位置那么一定有个坐标是可以回去的以车头方向为x车左边为y。图6 固定小橙的位置终点是(1.9m, -1.5m)但是正式比赛的时候机师是最紧张的很难做到每次都停的这么准确所以得想办法校准用地图的固定元素。大家可以先想想通道出来有什么是一定可以识别到的图7 大家找找地图固定元素是不是很明显前面有线。线在地图的位置绝对是固定的那么接下来的操作就是在线上了。有了线那就先求线的相对位置相对于小车的因为重置了里程计。求线的相对位置可以参考这位大佬发的逆透视变换的文章。基于透视变换的单目摄像头单一平面障碍物深度信息获取算法 - 火花绝赞摸鱼中✨️https://hibanaw.com/archives/171/我们先把车固定在一个位置比如图6的位置用小橙的USB相机拍一张照片然后给终点让小橙跑过去多试几次每次都要放回原来的位置跑过去小橙正好回去的点就是(1.9m, -1.5m)记得记得记得是清空里程计之后再跑。图8 固定位置的视角这个 (1.9m, -1.5m) 是关键。有了一个点接下来就是用逆透视变换求前面 2 根线的相对位置必须是最近的 2 根为什么是 2 根线后面会解释这样我们就有了 3 个点的坐标分别是 ABP。有了这 3 个点我们再任意摆放车比如图7的位置再用逆透视变换计算图7视角下最近的 2 根线的相对位置。这样就又有了 2 个点分别是AB一共有了 5 个点现在就是用这 5 个点来计算 P。问题简化一下在原来坐标系下我知道 ABP 3个点。原坐标系经过变换之后我可以知道AB接下来我想求P’。要求 P’那就得先知道前后 2 个坐标系是怎么变换的也就是前后 2 个坐标系之间的旋转矩阵R和平移变量 t。利用已知的两个对应点对 (A, A) 和 (B, B) 来求解 R 和 t 。对于点 A 和 AA′ RA t (1)对于点 B 和 BB′ RB t (2)将方程 (1) 和 (2) 相减A′ − B′R( A − B )令ΔAB A − BΔA′B′ A′ − B′则ΔA′B′ RΔAB因此旋转矩阵 R 可以这样求解R ΔA′B′ ΔAB^(−1)一旦得到 R可以通过任一对点求解 t。比如用 A 和 At A′ − RA最后对于 P其在新坐标系中的坐标 P′ 为P′ RP t这就是校准的所有步骤了用 2 根线的坐标去校准 P 点的坐标。实际情况是二维平面操作可以更简单大家把 C 换成 P 来看就可以了。最后计算出来这个 P 点是非常准确的用 python 写的代码有些参数是固定的大家可以先离线计算def end_point( x1, y1, x2, y2, x3, y3, x1_, y1_, x2_, y2_ ): delta_x x1 - x2 delta_y y1 - y2 delta_x_ x1_ - x2_ delta_y_ y1_ - y2_ den delta_x ** 2 delta_y ** 2 a ( delta_x * delta_x_ delta_y * delta_y_ ) / den # 旋转参数 a b ( delta_x * delta_y_ - delta_y * delta_x_ ) / den # 旋转参数 b tx x1_ - a * x1 b * y1 # 平移 tx ty y1_ - b * x1 - a * y1 # 平移 ty x3_ a * x3 - b * y3 tx # 变换后的 x3 y3_ b * x3 a * y3 ty # 变换后的 y3 print(f(x1, y1): ({x1, y1}), (x2, y2): ({x2, y2}), (x3, y3): ({x3, y3}) delta x: {delta_x}, delta y: {delta_y}, den: {den}) return x3_, y3_ print( fend: {end_point( ptx1, pty1, ptx2, pty2, 1.9, -1.5, ptx3, pty3, ptx4, pty4 )} )思路2——不重置里程计使用YOLO识别 P 点结果来校正终点我们比赛时候使用的就是这个思路上一个思路要在通道处停一下这个是不需要停下来直接冲出去就行了。这个思路比上一个简单很多能回去的概率非常大。这个思路是使用YOLO识别 P 点然后还是用逆透视变换计算 P 点相对坐标再通过小车的坐标计算这个P点的全局坐标。# 单应性矩阵计算全局坐标 H np.array([ [-4.66389128e-04, -2.26288030e-04, -4.92300831e-02], [ 7.59821540e-04, 5.20569143e-05, -2.33074608e-01], [-6.59643252e-04, -7.15022786e-03, 1.00000000e00], ]) def pixel2global(self, pixel_x, pixel_y): # 计算全局坐标从像素到局部再到全局 pixel np.array([pixel_x, pixel_y, 1], dtypenp.float32) # 应用逆透视变换矩阵 local np.dot(H, pixel) # 归一化坐标 local / local[2] local[0] 0.25 # 摄像头坐标转车底盘坐标 car_cos np.cos(self.current_pos[2]) car_sin np.sin(self.current_pos[2]) global_x self.current_pos[0] car_cos * local[0] - car_sin * local[1] global_y self.current_pos[1] car_sin * local[0] car_cos * local[1] return global_x, global_y将 P 点中心的像素坐标丢到这个函数里面输出的就是校正之后的终点了车头朝向为x车左边为y。因为全程不重置里程计所以未校正的终点设为出发的原点(0.5m, 0.2m)。校正之前小车的终点就是这个。这种思路对YOLO的要求比较高所以必须采集非常多的数据我们在实验室采了 7k 数据增强之后就有了差不多 2w2 训练出来的效果杠杠的。最后在比赛现场采了1k4增强到7k一共2w9张加急训练了22小时。其中数据集有超过一半的是P点的。采集数据的时候这种只有一个小角或被桶挡住了或比较远看起来很扁的也贴上凡是小车可以看到 P 点哪怕是一点点也要标上而且尽量采集多一些这样训练出来也是可以识别到的回去更轻松。不过要注意一下任务三出来的位置可能会误识成 P 点所以在任务三出来包括白色网格那里也要采集一点不需要太多。修改STM32源码stm32源码我基本上都看过了也找到了很多可以修改的地方。最重要的是舵机转角。我找到了舵机转角的限制但是转角对不对称跟有没有限制是没关系的如果去掉限制反而有可能会把前轮给跑坏。所以问题应该是出在了计算上面计算舵机转角的地方也就只有这个多项式了。它的输入是右前轮的转角输出是舵机角度。右前轮的转向角度的限幅是-0.490.320.32是左转最大的角度-0.49是右转最大角度说明右前轮的角度在 -0.49 和 0.32 的时候小车的舵机量应该是相同的。但是直接把这 2 个值带入原来的多项式中得到的舵机量是这样的这就说明小车左转转角小右转转角大。用上位机跑的时候确实是左转小右转大。调整多项式的二次项系数之后大概让左右转的舵机量相同。经过这样的调整之后小车左转和右转是差不多的角度了。在上位机跑起来左右转是对称的。为了提高惯导的计算频率我把串口发送的频率提高到了 50Hz串口波特率提高到了 921600而且也把其他没用到的外设比如CAN蓝牙给关了只留下串口1和串口3。这里为什么用串口1呢因为我们小橙底板的串口3有问题会出现断连干脆直接换成串口1了也就是烧录口。不知道为什么把串口3关掉之后oled会卡住只能打开它了在X5上面找到/root/dev_ws/origincar/origincar_base/launch/base_serial.launch.py波特率改成921600再编译一下就可以正常通信了不用管clear_flag。因为提高串口发送的频率到50Hz所以小车上EKF的计算频率也要设为50Hz。我们实际用下来发现小车的odom_combined挺准的一圈下来x和y偏得都不算太大。我们也改了一下ekf.yaml大家可以参考一下我们改的地方。原来的/imu/data_raw是原始数据/imu/data是经过滤波之后的。除了删掉一些外设我们还把电位器决定车型号的部分也删掉了固定车型号为Ackermanoled刷屏显示也把跟Ackerman无关的给删掉了。补充忘记把处理数据的代码放上来了就在这里补充一下吧。我们贴标签不是傻乎乎地全部自己来贴。我们先用以前训练过的模型贴一遍然后再人工检查一遍。这样子操作一个人不用一天时间就可以贴差不多6k因为当时我一个人花了半天来贴。除此之外我们还写了删除无效图片和无效标签的脚本图片没有对应同名的标签或者标签没有对应同名的图片、数据增强的脚本没有旋转和将数据分批次让队友来帮忙的脚本。最好在检查完模型贴的标签之后再进行数据增强。代码我都放在后面。后记希望这份分享可以帮助大家接下来的比赛。附这份是让模型贴标签的import argparse import os import shutil import time from pathlib import Path import torch import torch.backends.cudnn as cudnn import cv2 from models.experimental import attempt_load from utils.datasets import LoadImages from utils.utils import non_max_suppression, scale_coords, xyxy2xywh from utils.torch_utils import select_device, time_synchronized 用训练过的模型贴标签 def auto_annotate(source, weights, output, img_size640, conf_thres0.25, iou_thres0.45, device, view_imgFalse): 使用YOLOv5模型自动标注图像 参数: source (str): 输入图像文件夹路径 weights (str): 模型权重路径 output (str): 输出文件夹路径 img_size (int): 推理尺寸 conf_thres (float): 置信度阈值 iou_thres (float): IOU阈值 device (str): 使用的设备 (cpu, 0, 1, ...) view_img (bool): 是否显示结果图像 # 初始化 device select_device(device) half device.type ! cpu # 半精度仅在CUDA上支持 # # 创建输出文件夹 # if os.path.exists(output): # shutil.rmtree(output) # 删除输出文件夹 # os.makedirs(output) # 创建新的输出文件夹 # os.makedirs(os.path.join(output, labels)) # 创建标签文件夹 # 加载模型 model attempt_load(weights, map_locationdevice) # 加载FP32模型 imgsz img_size if half: model.half() # 转换为FP16 # 获取类别名称 names model.module.names if hasattr(model, module) else model.names # 设置数据加载器 dataset LoadImages(source, img_sizeimgsz) # 运行推理 t0 time.time() img torch.zeros((1, 3, imgsz, imgsz), devicedevice) # 初始化图像 _ model(img.half() if half else img) if device.type ! cpu else None # 运行一次 for path, img, im0s, _ in dataset: img torch.from_numpy(img).to(device) img img.half() if half else img.float() # uint8 to fp16/32 img / 255.0 # 0 - 255 to 0.0 - 1.0 if img.ndimension() 3: img img.unsqueeze(0) # 推理 t1 time_synchronized() pred model(img, augmentFalse)[0] # 应用NMS pred non_max_suppression(pred, conf_thres, iou_thres, classesNone, agnosticFalse) t2 time_synchronized() # 处理检测结果 p, im0 path, im0s.copy() txt_path str(Path(output) / Path(p).stem) (.txt) # 标签保存路径 # 确保标签文件存在即使为空 open(txt_path, w).close() # 创建空文件或清空现有文件 # 归一化增益 whwh gn torch.tensor(im0.shape)[[1, 0, 1, 0]] # 处理检测结果如果有 if pred is not None: for i, det in enumerate(pred): # 每张图像的检测结果 if det is not None and len(det): # 将边界框从img_size调整到im0大小 det[:, :4] scale_coords(img.shape[2:], det[:, :4], im0.shape).round() # 写入结果 # 修改写入标签的部分 with open(txt_path, w) as f: if det is not None and len(det): for *xyxy, conf, cls in reversed(det): xywh (xyxy2xywh(torch.tensor(xyxy).view(1, 4)) / gn).view(-1).tolist() # 格式化输出6位小数无行末空格 line %d %.6f %.6f %.6f %.6f % (cls, *xywh) f.write(line \n) # 注意换行符前无空格 else: f.write() # 空文件或按需写入占位符 # 打印时间 (推理 NMS) print(f{Path(p).name} done. ({t2 - t1:.3f}s)) # 显示结果(可选) if view_img: cv2.imshow(Path(p).name, im0) if cv2.waitKey(1) ord(q): # 按q退出 raise StopIteration print(fDone. ({time.time() - t0:.3f}s)) if __name__ __main__: parser argparse.ArgumentParser() parser.add_argument(--source, typestr, defaultdataset_process/new1/images, help输入图像文件夹路径) parser.add_argument(--weights, typestr, defaultruns/2025.7.28/weights/last.pt, help模型权重路径) parser.add_argument(--output, typestr, defaultdataset_process/new1/labels, help输出标签路径) parser.add_argument(--img-size, typeint, default640, help推理尺寸 (像素)) parser.add_argument(--conf-thres, typefloat, default0.25, help目标置信度阈值) parser.add_argument(--iou-thres, typefloat, default0.45, helpNMS的IOU阈值) parser.add_argument(--device, default, helpcuda设备, 如 0 或 0,1,2,3 或 cpu) parser.add_argument(--view-img, actionstore_true, help显示结果) opt parser.parse_args() print(opt) with torch.no_grad(): auto_annotate( sourceopt.source, weightsopt.weights, outputopt.output, img_sizeopt.img_size, conf_thresopt.conf_thres, iou_thresopt.iou_thres, deviceopt.device, view_imgopt.view_img )这份是删除无效数据的import os from pathlib import Path def remove_invalid_images_labels(image_dir, label_dir): 删除无效的图片和标签文件标签为空或不存在 参数: image_dir (str): 图片文件夹路径 label_dir (str): 标签文件夹路径 deleted_images 0 deleted_labels 0 # 遍历图片文件夹 for image_file in os.listdir(image_dir): if image_file.lower().endswith((.jpg, .png, .jpeg)): image_path os.path.join(image_dir, image_file) label_path os.path.join(label_dir, Path(image_file).stem .txt) # 检查标签文件是否存在或为空 if not os.path.exists(label_path): os.remove(image_path) deleted_images 1 print(f删除图片无标签: {image_file}) else: with open(label_path, r) as f: content f.read().strip() if not content: # 标签文件为空 os.remove(image_path) os.remove(label_path) deleted_images 1 deleted_labels 1 print(f删除无效数据: {image_file} 和对应标签) print(f\n操作完成共删除: {deleted_images} 张图片, {deleted_labels} 个标签) if __name__ __main__: # 设置路径修改为你的实际路径 image_dir os.path.join(os.path.dirname(__file__), new1/images/) label_dir os.path.join(os.path.dirname(__file__), new1/labels/) # 确认操作 print(f即将检查:\n图片目录: {image_dir}\n标签目录: {label_dir}) confirm input(是否继续(y/n): ).lower() if confirm y: remove_invalid_images_labels(image_dir, label_dir) else: print(操作已取消)这份是数据增强的import torch import torchvision.transforms as T import torchvision.transforms.functional as TF from pathlib import Path import shutil from PIL import Image import random from multiprocessing import Pool import os 数据增强 class YOLOAugment: def __init__(self, output_dir): self.output_dir output_dir Path(f{output_dir}/images).mkdir(parentsTrue, exist_okTrue) Path(f{output_dir}/labels).mkdir(parentsTrue, exist_okTrue) # 定义基础增强仅影响图像 self.img_augment T.Compose([ T.ColorJitter(brightness0.3, contrast0.3, saturation0.2), T.GaussianBlur(kernel_size(3, 7)) ]) def apply_augment(self, img_path, label_path, aug_id): 处理单张图像和对应标签 # 读取原始数据 img Image.open(img_path).convert(RGB) with open(label_path) as f: bboxes [list(map(float, line.strip().split())) for line in f] # 转换为Tensor格式 img_tensor TF.to_tensor(img) bboxes_tensor torch.tensor(bboxes) # 应用图像增强不影响框 img_tensor self.img_augment(img_tensor) # 保存增强结果 stem Path(img_path).stem self._save_results(img_tensor, bboxes_tensor, stem, aug_id) return img, bboxes def _save_results(self, img_tensor, bboxes, stem, aug_id): 保存增强图像和标签 # 保存图像 aug_img TF.to_pil_image(img_tensor) aug_img.save(f{self.output_dir}/images/{stem}_aug{aug_id}.jpg) # 保存标签YOLO格式 with open(f{self.output_dir}/labels/{stem}_aug{aug_id}.txt, w) as f: for bbox in bboxes.numpy(): line .join(map(str, bbox)) f.write(line \n) def process_file(args): 多进程处理函数 img_path, label_path, output_dir, aug_per_image args augmenter YOLOAugment(output_dir) for i in range(1, aug_per_image 1): augmenter.apply_augment(img_path, label_path, i) # 保留原始文件 shutil.copy(img_path, f{output_dir}/images/{Path(img_path).name}) shutil.copy(label_path, f{output_dir}/labels/{Path(label_path).name}) if __name__ __main__: root_path os.path.dirname(__file__) # 配置参数 TUDO input_dir os.path.join(root_path, new1) # 原始数据集路径 output_dir os.path.join(root_path, new1_aug) # 输出路径 aug_per_image 3 # 每张图片生成4个增强版本 num_workers 4 # 并行进程数 # 准备列表 tasks [] for img_file in Path(f{input_dir}/images).glob(*.*): if img_file.suffix.lower() in (.jpg, .png, .jpeg): label_file Path(f{input_dir}/labels/{img_file.stem}.txt) if label_file.exists(): tasks.append((str(img_file), str(label_file), output_dir, aug_per_image)) # 多进程处理 print(f开始增强 {len(tasks)} 张图像...) with Pool(processesnum_workers) as pool: pool.map(process_file, tasks) # 统计结果 orig_count len(tasks) aug_count orig_count * aug_per_image print(f处理完成\n f- 原始图像保留: {orig_count} 张\n f- 增强图像生成: {aug_count} 张\n f- 总数据量: {orig_count aug_count} 张)这份是让队友打工的import os import zipfile import math from pathlib import Path 将数据集分好份打包好 def create_task_packs(images_dir, labels_dir, output_dir, tasks3, label_txtFalse): 创建包含匹配images和labels的task压缩包 :param images_dir: 图片文件夹路径 :param labels_dir: 标注文件夹路径 :param output_dir: 输出目录 :param tasks: 需要划分的任务数 # 获取匹配的文件对确保严格对应 image_files sorted([f for f in os.listdir(images_dir) if f.endswith((.jpg, .png))]) label_files sorted([f for f in os.listdir(labels_dir) if f.endswith(.txt) ]) # 验证一致性 image_stems {Path(f).stem for f in image_files} label_stems {Path(f).stem for f in label_files} unmatched image_stems.symmetric_difference(label_stems) if unmatched: print(f⚠️ 警告: 发现 {len(unmatched)} 个不匹配文件示例: {list(unmatched)[:3]}) print(建议先运行数据校验脚本修复不一致问题) return # 计算每个task应包含的文件数 total_pairs len(image_files) pairs_per_task math.ceil(total_pairs / tasks) print(f数据集统计:) print(f- 图片数量: {len(image_files)}) print(f- 标注数量: {len(label_files)}) print(f- 将分成 {tasks} 个任务包每个约 {pairs_per_task} 对数据\n) # 创建输出目录 os.makedirs(output_dir, exist_okTrue) for task_num in range(1, tasks 1): start_idx ( task_num - 1 ) * pairs_per_task end_idx min( start_idx pairs_per_task, total_pairs ) task_images image_files[start_idx:end_idx] task_labels [Path(f).stem .txt for f in task_images ] # 自动匹配对应的labels zip_path os.path.join(output_dir, ftask_{task_num}.zip) print(f创建任务包 {task_num}:) print(f- 包含图片: {len(task_images)} 张) print(f- 包含标注: {len(task_labels)} 个) print(f- 保存到: {zip_path}) with zipfile.ZipFile(zip_path, w, zipfile.ZIP_DEFLATED) as zipf: # 添加图片 for img in task_images: img_path os.path.join(images_dir, img) zipf.write(img_path, fimages/{img}) # 添加对应的标注 for label in task_labels: label_path os.path.join(labels_dir, label) if os.path.exists(label_path): # 双重验证 zipf.write(label_path, flabels/{label}) else: print(f⚠️ 缺失标注文件: {label}) if label_txt is not False: label_info Path(label_txt).open(r).read() zipf.writestr(flabels/labels.txt, label_info) # 每个任务里面都放进一个 labels.txt print(- * 50) print(f\n 任务包创建完成共生成 {tasks} 个压缩包保存在: {output_dir}) if __name__ __main__: root_path os.path.dirname(__file__) # 配置参数 dataset_dir os.path.join(root_path, new1) # 数据集根目录 output_dir os.path.join(root_path, package) # 输出目录 label_txt os.path.join(root_path, labels.txt) # 标签文件 num_tasks 4 # 需要划分的任务数量 # 运行打包 create_task_packs( images_diros.path.join(dataset_dir, images), labels_diros.path.join(dataset_dir, labels), output_diroutput_dir, tasksnum_tasks, # label_txtlabel_txt, )