2026.6.2小更新
This commit is contained in:
@@ -0,0 +1,45 @@
|
||||
# ===================================
|
||||
# 查看视频的帧率等属性
|
||||
# ===================================
|
||||
import cv2
|
||||
import os
|
||||
|
||||
video_path = "../asset/saved_video/output_result.mp4" # 替换成你上传的视频文件名
|
||||
|
||||
if not os.path.exists(video_path):
|
||||
print(f"错误：找不到文件 {video_path}")
|
||||
exit()
|
||||
|
||||
cap = cv2.VideoCapture(video_path)
|
||||
|
||||
if not cap.isOpened():
|
||||
print("错误：无法打开视频文件！")
|
||||
exit()
|
||||
|
||||
# 获取视频自带的物理帧率 (FPS)
|
||||
fps = cap.get(cv2.CAP_PROP_FPS)
|
||||
|
||||
# 获取视频的总帧数
|
||||
total_frames = cap.get(cv2.CAP_PROP_FRAME_COUNT)
|
||||
|
||||
# 获取视频的分辨率宽度和高度
|
||||
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
|
||||
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
|
||||
|
||||
# 计算视频总时长（秒）
|
||||
duration = total_frames / fps if fps > 0 else 0
|
||||
|
||||
print("==========================================")
|
||||
print(f"视频文件: {os.path.basename(video_path)}")
|
||||
print(f"视频分辨率: {width} x {height}")
|
||||
print(f"视频原生帧率 (FPS): {fps:.2f} 帧/秒 <--- [核心原因看这里]")
|
||||
print(f"视频总帧数: {int(total_frames)} 帧")
|
||||
print(f"视频总时长: {duration:.2f} 秒")
|
||||
print("==========================================")
|
||||
|
||||
# if fps <= 25.5:
|
||||
# print("💡 结论：不出所料，你的视频本身就被录制/压制成了 25 帧！")
|
||||
# print("OpenCV 的 cap.read() 会卡在 25fps 的播放速度上等下一帧，导致 NPU 闲置。")
|
||||
# print("你想测试 NPU 的真实上限，需要用前面提到的『纯随机数矩阵压测脚本』。")
|
||||
|
||||
cap.release()
|
||||
@@ -0,0 +1,47 @@
|
||||
import time
|
||||
from pymavlink import mavutil
|
||||
|
||||
# 【核心配置】连接飞控
|
||||
# 这里的路径取决于香橙派和飞控是怎么连的。
|
||||
# 如果是通过 USB 转 TTL 模块插在香橙派上，通常是 '/dev/ttyUSB0'
|
||||
# 如果是直接连在香橙派的板载 GPIO 串口上，可能是 '/dev/ttyAMA0' 或 '/dev/ttyS5'
|
||||
# 波特率通常飞控默认是 115200 或 57600
|
||||
FC_CONNECTION = '/dev/ttyUSB0'
|
||||
BAUD_RATE = 57600
|
||||
|
||||
print(f"正在尝试连接飞控 ({FC_CONNECTION})...")
|
||||
master = mavutil.mavlink_connection(FC_CONNECTION, baud=BAUD_RATE)
|
||||
|
||||
# 等待飞控的心跳信号（确认物理链路已通）
|
||||
print("正在等待飞控心跳信号 (Heartbeat)...")
|
||||
master.wait_heartbeat()
|
||||
print("飞控连接成功！开始获取姿态数据...\n")
|
||||
|
||||
try:
|
||||
while True:
|
||||
# 阻塞或等待接收 MAVLink 消息
|
||||
# ATTITUDE 消息包含了无人机的姿态（滚转、俯仰、偏航）
|
||||
msg = master.recv_match(type='ATTITUDE', blocking=True, timeout=1.0)
|
||||
|
||||
if msg:
|
||||
# 飞控传出来的弧度（radians），需要转换成角度（degrees）
|
||||
import math
|
||||
|
||||
roll = math.degrees(msg.roll)
|
||||
pitch = math.degrees(msg.pitch)
|
||||
yaw = math.degrees(msg.yaw)
|
||||
|
||||
# 如果偏航角是负数（-180到180），可以转换成 0-360 度标准航向
|
||||
if yaw < 0:
|
||||
yaw += 360
|
||||
|
||||
# 刷新打印到你的 PC SSH 终端屏幕上
|
||||
print(f"【飞控姿态】 偏航角(Yaw): {yaw:.2f}° | 俯仰角(Pitch): {pitch:.2f}° | 滚转角(Roll): {roll:.2f}°",
|
||||
end='\r')
|
||||
else:
|
||||
print("未收到 ATTITUDE 消息，正在等待...", end='\r')
|
||||
|
||||
time.sleep(0.01)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\n用户终止，退出监控。")
|
||||
@@ -0,0 +1,198 @@
|
||||
import os
|
||||
import cv2
|
||||
import numpy as np
|
||||
import argparse
|
||||
from ais_bench.infer.interface import InferSession
|
||||
|
||||
class YOLO26Infer:
|
||||
def __init__(self, model_path, device_id=0):
|
||||
"""
|
||||
初始化推理器
|
||||
:param model_path: .om 模型的本地路径
|
||||
:param device_id: NPU 设备ID
|
||||
"""
|
||||
self.device_id = device_id
|
||||
self.model_path = model_path
|
||||
# 创建 ais_bench 推理会话
|
||||
self.session = InferSession(device_id, model_path)
|
||||
# 获取模型输入尺寸，用于后续预处理
|
||||
self.input_width = 640
|
||||
self.input_height = 640
|
||||
|
||||
def letterbox(self, img, new_shape=(640, 640), color=(114, 114, 114)):
|
||||
shape = img.shape[:2]
|
||||
r = min(new_shape[0] / shape[0], new_shape[1] / shape[1])
|
||||
# 计算缩放后的新尺寸（四舍五入）
|
||||
new_unpad = (int(round(shape[1] * r)), int(round(shape[0] * r)))
|
||||
# 计算需要填充的总宽度和高度
|
||||
dw = new_shape[1] - new_unpad[0]
|
||||
dh = new_shape[0] - new_unpad[1]
|
||||
# 分别计算左右/上下填充量（允许奇数不对称）
|
||||
left = dw // 2
|
||||
right = dw - left
|
||||
top = dh // 2
|
||||
bottom = dh - top
|
||||
|
||||
# 缩放图片
|
||||
if shape[::-1] != new_unpad:
|
||||
img = cv2.resize(img, new_unpad, interpolation=cv2.INTER_LINEAR)
|
||||
|
||||
# 添加填充（可能左右或上下不对称，但总尺寸正好是 new_shape）
|
||||
img = cv2.copyMakeBorder(img, top, bottom, left, right, cv2.BORDER_CONSTANT, value=color)
|
||||
return img, r, (left, top) # 返回填充的左上角偏移量
|
||||
|
||||
def preprocess(self, image_path):
|
||||
"""
|
||||
预处理图片，使其符合模型输入要求
|
||||
"""
|
||||
# 使用OpenCV读取图片 (BGR格式)
|
||||
img_bgr = cv2.imread(image_path)
|
||||
if img_bgr is None:
|
||||
raise FileNotFoundError(f"图片未找到: {image_path}")
|
||||
|
||||
# 获取原始图片尺寸，用于后处理时的坐标转换
|
||||
self.original_height, self.original_width = img_bgr.shape[:2]
|
||||
|
||||
# 将图片从BGR转换为RGB
|
||||
img_rgb = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB)
|
||||
|
||||
# 调整图片尺寸并保持宽高比，进行Letterbox填充
|
||||
img_resized, self.ratio, (self.pad_left, self.pad_top) = self.letterbox(
|
||||
img_rgb, (self.input_width, self.input_height)
|
||||
)
|
||||
|
||||
# 归一化：将像素值从 [0, 255] 转为 [0, 1]
|
||||
img_input = (img_resized.astype(np.float32) / 255.0).astype(np.float32)
|
||||
|
||||
# 调整维度：HWC -> NCHW，并转换为连续数组
|
||||
img_input = np.ascontiguousarray(np.transpose(img_input, (2, 0, 1))[np.newaxis, :, :, :])
|
||||
|
||||
return img_input
|
||||
|
||||
def infer(self, img_input):
|
||||
"""
|
||||
执行推理
|
||||
"""
|
||||
# 使用会话的 infer 方法进行推理
|
||||
# infer 方法接收一个列表，包含输入张量，返回一个列表，包含输出张量
|
||||
result = self.session.infer([img_input])
|
||||
print(f"Input tensor shape: {img_input.shape}")
|
||||
print(f"Input tensor dtype: {img_input.dtype}")
|
||||
print(f"Input tensor size: {img_input.nbytes}")
|
||||
return result
|
||||
|
||||
def postprocess(self, infer_result, conf_threshold=0.5):
|
||||
"""
|
||||
对模型输出进行后处理
|
||||
:param infer_result: 模型原始输出，一个列表，通常第一个元素是输出张量
|
||||
:param conf_threshold: 置信度阈值
|
||||
:return: 包含检测结果的列表，每个元素为 [x1, y1, x2, y2, confidence, class_id]
|
||||
"""
|
||||
detections = []
|
||||
# 获取输出张量，假设输出列表的第一个元素是输出数组
|
||||
if isinstance(infer_result, (list, tuple)):
|
||||
output = infer_result[0]
|
||||
else:
|
||||
output = infer_result
|
||||
|
||||
# ========== 调试信息 ==========
|
||||
# print(f"[DEBUG] output shape: {output.shape}")
|
||||
# print(f"[DEBUG] output dtype: {output.dtype}")
|
||||
# print(f"[DEBUG] output min: {output.min()}, max: {output.max()}")
|
||||
# # 打印前5个检测框（如果形状是 [N,6] 或 [1,N,6]）
|
||||
# if output.ndim == 3:
|
||||
# flat_output = output[0]
|
||||
# else:
|
||||
# flat_output = output
|
||||
# print(f"[DEBUG] first 5 rows:\n{flat_output[:5]}")
|
||||
|
||||
# 假设输出维度为 [batch, num_detections, 6]
|
||||
# 移除batch维度，并只取第一个batch的结果
|
||||
if output.ndim == 3:
|
||||
output = output[0]
|
||||
|
||||
# 遍历所有检测框
|
||||
for det in output:
|
||||
confidence = det[4]
|
||||
if confidence < conf_threshold:
|
||||
continue
|
||||
|
||||
# 获取原始输出坐标 (x1, y1, x2, y2) 和 类别ID
|
||||
x1_scale, y1_scale, x2_scale, y2_scale = det[0], det[1], det[2], det[3]
|
||||
cls_id = int(det[5])
|
||||
|
||||
# 将坐标从模型输入尺寸映射回原始图片尺寸
|
||||
x1 = (x1_scale - self.pad_left) / self.ratio
|
||||
y1 = (y1_scale - self.pad_top) / self.ratio
|
||||
x2 = (x2_scale - self.pad_left) / self.ratio
|
||||
y2 = (y2_scale - self.pad_top) / self.ratio
|
||||
|
||||
# 边界限制，防止超出原始图片范围
|
||||
x1 = max(0, min(x1, self.original_width))
|
||||
y1 = max(0, min(y1, self.original_height))
|
||||
x2 = max(0, min(x2, self.original_width))
|
||||
y2 = max(0, min(y2, self.original_height))
|
||||
|
||||
detections.append([x1, y1, x2, y2, confidence, cls_id])
|
||||
|
||||
return detections
|
||||
|
||||
def draw_boxes(image, detections, class_names=None):
|
||||
"""
|
||||
在图片上绘制检测框和标签
|
||||
:param image: 原始图片数组 (BGR格式)
|
||||
:param detections: 后处理得到的结果列表
|
||||
:param class_names: 类别名称列表，默认使用类别ID
|
||||
"""
|
||||
img_copy = image.copy()
|
||||
for det in detections:
|
||||
x1, y1, x2, y2, conf, cls_id = det
|
||||
x1, y1, x2, y2 = int(x1), int(y1), int(x2), int(y2)
|
||||
|
||||
# 绘制矩形框
|
||||
cv2.rectangle(img_copy, (x1, y1), (x2, y2), (0, 255, 0), 2)
|
||||
|
||||
# 构建标签文本
|
||||
label = f"Class: {cls_id}" if class_names is None else class_names[cls_id]
|
||||
label += f" {conf:.2f}"
|
||||
|
||||
# 设置标签背景
|
||||
(label_w, label_h), _ = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, 0.5, 1)
|
||||
cv2.rectangle(img_copy, (x1, y1 - label_h - 5), (x1 + label_w, y1), (0, 255, 0), -1)
|
||||
cv2.putText(img_copy, label, (x1, y1 - 5), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 0), 1)
|
||||
|
||||
return img_copy
|
||||
|
||||
if __name__ == "__main__":
|
||||
parser = argparse.ArgumentParser(description="YOLOv26 目标检测推理")
|
||||
parser.add_argument("--model", type=str, required=True, help="YOLOv26 OM模型文件路径")
|
||||
parser.add_argument("--image", type=str, required=True, help="输入图片路径")
|
||||
parser.add_argument("--conf", type=float, default=0.5, help="置信度阈值")
|
||||
args = parser.parse_args()
|
||||
|
||||
# 初始化推理器
|
||||
yolo_infer = YOLO26Infer(args.model)
|
||||
|
||||
# 图片预处理
|
||||
print("预处理图片...")
|
||||
input_tensor = yolo_infer.preprocess(args.image)
|
||||
|
||||
# 执行推理
|
||||
print("执行推理...")
|
||||
raw_result = yolo_infer.infer(input_tensor)
|
||||
|
||||
# 后处理
|
||||
print("后处理...")
|
||||
detections = yolo_infer.postprocess(raw_result, conf_threshold=args.conf)
|
||||
|
||||
# 打印结果
|
||||
print(f"检测到 {len(detections)} 个目标:")
|
||||
for det in detections:
|
||||
print(f"类别ID: {int(det[5])}, 置信度: {det[4]:.4f}, 边界框: ({det[0]:.2f}, {det[1]:.2f}, {det[2]:.2f}, {det[3]:.2f})")
|
||||
|
||||
# 可选：绘制结果并保存
|
||||
original_img = cv2.imread(args.image)
|
||||
result_img = draw_boxes(original_img, detections)
|
||||
output_path = "output_" + os.path.basename(args.image)
|
||||
cv2.imwrite(output_path, result_img)
|
||||
print(f"结果图片已保存至: {output_path}")
|
||||
@@ -0,0 +1,142 @@
|
||||
import socket
|
||||
import cv2
|
||||
import numpy as np
|
||||
import threading
|
||||
import queue
|
||||
import time
|
||||
import os
|
||||
|
||||
# ==================== 地面站核心配置 ====================
|
||||
PORT = 9999
|
||||
SAVE_DIR = "../asset/saved_video"
|
||||
SAVE_PATH = os.path.join(SAVE_DIR, "ground_station_record.mp4")
|
||||
RECORD_FPS = 24.0 # 录像固定的标准帧率
|
||||
# =======================================================
|
||||
|
||||
# 确保保存视频的文件夹存在
|
||||
if not os.path.exists(SAVE_DIR):
|
||||
os.makedirs(SAVE_DIR)
|
||||
|
||||
# 核心线程共享：最新的一帧画面与互斥锁（防止读写冲突）
|
||||
latest_frame = None
|
||||
frame_lock = threading.Lock()
|
||||
is_running = True
|
||||
|
||||
|
||||
def udp_receiver_thread():
|
||||
""" 📡 后台极速收包线程 """
|
||||
global latest_frame, is_running
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
# 放大系统内核缓冲区到 4MB，全力应对图传电台的无线网络抖动
|
||||
sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 4 * 1024 * 1024)
|
||||
|
||||
try:
|
||||
sock.bind(('0.0.0.0', PORT))
|
||||
print(f"📡 [地面站接收端] 已启动，正在监听 UDP 端口 {PORT}...")
|
||||
except Exception as e:
|
||||
print(f"❌ 端口绑定失败: {e}")
|
||||
is_running = False
|
||||
return
|
||||
|
||||
while is_running:
|
||||
try:
|
||||
packet, addr = sock.recvfrom(65535)
|
||||
data = np.frombuffer(packet, dtype=np.uint8)
|
||||
frame = cv2.imdecode(data, cv2.IMREAD_COLOR)
|
||||
|
||||
if frame is not None:
|
||||
# 安全地更新最新的一帧，瞬间覆盖老帧，杜绝延迟积压
|
||||
with frame_lock:
|
||||
latest_frame = frame
|
||||
except Exception as e:
|
||||
if is_running:
|
||||
print(f"⚠️ 接收解码异常: {e}")
|
||||
time.sleep(0.01)
|
||||
|
||||
sock.close()
|
||||
|
||||
|
||||
def main():
|
||||
global latest_frame, is_running
|
||||
|
||||
# 1. 启动后台无线电收包线程
|
||||
recv_t = threading.Thread(target=udp_receiver_thread, daemon=True)
|
||||
recv_t.start()
|
||||
|
||||
# 2. 初始化地面站专用录像机 (对应香橙派最后一刻 resize 后的 640x480)
|
||||
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
|
||||
video_writer = cv2.VideoWriter(SAVE_PATH, fourcc, RECORD_FPS, (640, 480))
|
||||
print(f"📹 视频录像机就绪，文件将安全保存在: {SAVE_PATH}")
|
||||
|
||||
# 用于计算接收端实时刷新率
|
||||
fps_start_time = time.time()
|
||||
fps_counter = 0
|
||||
fps_display = 0
|
||||
|
||||
# 严格按照发送端的帧率控制循环步长（1 / 24秒 ≈ 0.0416秒）
|
||||
frame_interval = 1.0 / RECORD_FPS
|
||||
last_frame_recorded = None
|
||||
|
||||
print("🖥️ 监视窗口已建立，等待无人机连接推流...")
|
||||
|
||||
try:
|
||||
while is_running:
|
||||
loop_start = time.time()
|
||||
|
||||
# 从共享区取出最新帧
|
||||
current_frame = None
|
||||
with frame_lock:
|
||||
if latest_frame is not None:
|
||||
current_frame = latest_frame.copy()
|
||||
|
||||
# 无线图传时延容错核心逻辑：
|
||||
if current_frame is not None:
|
||||
# 拿到了图传新帧，更新最后记录的有效帧
|
||||
last_frame_recorded = current_frame
|
||||
elif last_frame_recorded is not None:
|
||||
# 【抗干扰核心】当前瞬间电台丢包没收到新图，复制上一帧顶替，防止视频时间轴缩水变快
|
||||
current_frame = last_frame_recorded.copy()
|
||||
|
||||
# 如果已经有画面可渲染（无论是新帧还是补偿帧）
|
||||
if current_frame is not None:
|
||||
# 录像写入（PC 的强悍 CPU 在后台无感硬写磁盘，完全不拖累香橙派的 NPU）
|
||||
video_writer.write(current_frame)
|
||||
|
||||
# 计算并渲染接收端的显示 FPS
|
||||
fps_counter += 1
|
||||
if (time.time() - fps_start_time) > 1.0:
|
||||
fps_display = fps_counter
|
||||
fps_counter = 0
|
||||
fps_start_time = time.time()
|
||||
|
||||
# 在右上角标注实时的图传显示帧率
|
||||
cv2.putText(current_frame, f"GCS FPS: {fps_display}", (current_frame.shape[1] - 140, 30),
|
||||
cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 0), 2)
|
||||
|
||||
# 实时监控显示
|
||||
cv2.imshow("🚀 UAV AI Ground Station Monitor", current_frame)
|
||||
|
||||
# 按键检测（按 'q' 安全退出）
|
||||
if cv2.waitKey(1) & 0xFF == ord('q'):
|
||||
print("👋 接收到关闭指令...")
|
||||
break
|
||||
|
||||
# 严格控制写入频率，使其匹配 24.0 FPS 的时间流速
|
||||
time_used = time.time() - loop_start
|
||||
sleep_time = frame_interval - time_used
|
||||
if sleep_time > 0:
|
||||
time.sleep(sleep_time)
|
||||
|
||||
finally:
|
||||
# 善后处理：确保内存日志和视频安全落盘
|
||||
is_running = False
|
||||
video_writer.release()
|
||||
cv2.destroyAllWindows()
|
||||
print("\n=======================================================")
|
||||
print(f"✅ 地面站已安全关闭。")
|
||||
print(f"🎬 带有精确检测框、对准线和控制误差的视频已成功保存在:\n {SAVE_PATH}")
|
||||
print("=======================================================")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user