v1.4实现机器人一起踢乒乓球功能，使用maixcam，离线运行。
This commit is contained in:
myperworld committed
1 parent
650797fe28
commit
9d3aa2684b
13 files changed
+19044
-1041
No files matched your search
@@ -0,0 +1,48 @@
|
||||
# ------------------------------
|
||||
# LED配置
|
||||
# ------------------------------
|
||||
|
||||
from maix import camera, display, image, nn, app, sys
|
||||
from maix import uart, pinmap, time, err, gpio
|
||||
import traceback
|
||||
|
||||
LED_PIN_NAME = "A14"
|
||||
LED_GPIO_NAME = "GPIOA14"
|
||||
BLINK_INTERVAL = 500 # LED闪烁间隔(ms)
|
||||
|
||||
|
||||
# ------------------------------
|
||||
# 全局业务状态变量
|
||||
# ------------------------------
|
||||
led = None # LED实例
|
||||
serial_dev = None # 舵机串口全局变量
|
||||
previous_millis = 0 # LED闪烁计时
|
||||
|
||||
# ------------------------------
|
||||
# LED控制
|
||||
# ------------------------------
|
||||
def init_led():
|
||||
"""初始化LED引脚"""
|
||||
global led
|
||||
try:
|
||||
err.check_raise(
|
||||
pinmap.set_pin_function(LED_PIN_NAME, LED_GPIO_NAME),
|
||||
f"设置LED引脚{LED_PIN_NAME}功能失败"
|
||||
)
|
||||
led = gpio.GPIO(LED_GPIO_NAME, gpio.Mode.OUT)
|
||||
led.value(0) # 初始熄灭
|
||||
print(f"LED初始化完成：引脚{LED_PIN_NAME}（{LED_GPIO_NAME}）")
|
||||
except Exception as e:
|
||||
print(f"LED初始化失败: {e}")
|
||||
led = None
|
||||
|
||||
def update_led():
|
||||
"""更新LED状态：改为固定闪烁模式"""
|
||||
global previous_millis
|
||||
if led is None:
|
||||
return
|
||||
|
||||
current_millis = time.time_ms()
|
||||
if current_millis - previous_millis >= BLINK_INTERVAL:
|
||||
previous_millis = current_millis
|
||||
led.toggle()
|
||||
+33
-424
@@ -1,288 +1,19 @@
|
||||
'''
|
||||
主题：MAIXCAM 云台
|
||||
作者：Tommy
|
||||
日期：2025年-12月-21日
|
||||
'''
|
||||
# main.py - 主业务逻辑（人脸跟踪、舵机控制、LED）
|
||||
from maix import camera, display, image, nn, app, sys
|
||||
from maix import uart, pinmap, time, err, gpio, network
|
||||
import paho.mqtt.client as mqtt
|
||||
import asyncio
|
||||
import gc
|
||||
from maix import uart, pinmap, time, err, gpio
|
||||
from maix import app, nn
|
||||
import traceback
|
||||
|
||||
|
||||
DEVICE_ID = "XXX"
|
||||
|
||||
# 两套WiFi/MQTT配置
|
||||
WIFI_CONFIGS = [
|
||||
{
|
||||
"SSID": "X",
|
||||
"PASSWORD": "Y",
|
||||
"MQTT_BROKER": "Z",
|
||||
},
|
||||
]
|
||||
|
||||
MQTT_PORT = 1883
|
||||
|
||||
# 发送的主题
|
||||
MQTT_PUB_TOPIC_IMAGE = "hangtianyuan/head_video" # 图像发布主题
|
||||
|
||||
# 订阅的主题
|
||||
MQTT_SUB_TOPIC_TEXT = "hangtianyuan/text" # 文本指令订阅主题
|
||||
MQTT_SUB_TOPIC_MOVE = "hangtianyuan/set_move" # 舵机移动指令订阅主题
|
||||
# 批量订阅
|
||||
MQTT_SUB_TOPICS = [
|
||||
(MQTT_SUB_TOPIC_TEXT, 0),
|
||||
(MQTT_SUB_TOPIC_MOVE, 0),
|
||||
]
|
||||
|
||||
# LED配置
|
||||
LED_PIN_NAME = "A14"
|
||||
LED_GPIO_NAME = "GPIOA14"
|
||||
BLINK_INTERVAL = 500 # LED闪烁间隔(ms)
|
||||
|
||||
# 人脸追踪参数
|
||||
kp_x = 0.05 # 水平方向 0.04
|
||||
kp_y = 0.07 # 垂直方向
|
||||
angle_threshold = 1
|
||||
s1_range = [0, 180]
|
||||
s2_range = [0, 180]
|
||||
no_face_threshold = 30
|
||||
|
||||
# 滤波缓存
|
||||
last_dev_x = 0.0
|
||||
last_dev_y = 0.0
|
||||
filter_alpha = 0.85
|
||||
|
||||
# ------------------------------
|
||||
# 全局状态变量（新增track_mode控制跟踪开关）
|
||||
# ------------------------------
|
||||
mqtt_connected = False
|
||||
wifi_connected = False
|
||||
current_wifi_idx = 0
|
||||
led = None
|
||||
mqtt_client = None
|
||||
previous_millis = 0
|
||||
send_frame_interval = 8
|
||||
frame_count = 0
|
||||
track_mode = 1 # 0:关闭追踪，1:人脸跟踪模式
|
||||
serial_dev = None # 新增：舵机串口全局变量
|
||||
|
||||
# ------------------------------
|
||||
# 工具函数
|
||||
# ------------------------------
|
||||
def constrain(val, min_val, max_val):
|
||||
return min(max_val, max(min_val, val))
|
||||
|
||||
# ------------------------------
|
||||
# LED控制
|
||||
# ------------------------------
|
||||
def init_led():
|
||||
"""初始化LED引脚"""
|
||||
global led
|
||||
try:
|
||||
err.check_raise(
|
||||
pinmap.set_pin_function(LED_PIN_NAME, LED_GPIO_NAME),
|
||||
f"设置LED引脚{LED_PIN_NAME}功能失败"
|
||||
)
|
||||
led = gpio.GPIO(LED_GPIO_NAME, gpio.Mode.OUT)
|
||||
led.value(0) # 初始熄灭
|
||||
print(f"LED初始化完成：引脚{LED_PIN_NAME}（{LED_GPIO_NAME}）")
|
||||
except Exception as e:
|
||||
print(f"LED初始化失败: {e}")
|
||||
led = None
|
||||
|
||||
def update_led():
|
||||
"""更新LED状态：MQTT未连接闪烁，连接成功常亮"""
|
||||
global previous_millis, mqtt_connected
|
||||
if led is None:
|
||||
return
|
||||
|
||||
current_millis = time.time_ms()
|
||||
if not mqtt_connected:
|
||||
if current_millis - previous_millis >= BLINK_INTERVAL:
|
||||
previous_millis = current_millis
|
||||
led.toggle()
|
||||
else:
|
||||
if led.value() != 1:
|
||||
led.value(1)
|
||||
|
||||
# ------------------------------
|
||||
# WiFi连接（异步）
|
||||
# ------------------------------
|
||||
async def connect_wifi_async():
|
||||
"""异步尝试连接WiFi（自动切换两套配置）"""
|
||||
global wifi_connected, current_wifi_idx
|
||||
w = network.wifi.Wifi()
|
||||
|
||||
wifi_connected = False
|
||||
current_wifi_idx = 0
|
||||
|
||||
while current_wifi_idx < len(WIFI_CONFIGS) and not wifi_connected:
|
||||
config = WIFI_CONFIGS[current_wifi_idx]
|
||||
ssid = config["SSID"]
|
||||
password = config["PASSWORD"]
|
||||
|
||||
print(f"[WiFi] 尝试连接 {ssid} (第{current_wifi_idx+1}套)")
|
||||
try:
|
||||
e = w.connect(ssid, password, wait=True, timeout=10)
|
||||
err.check_raise(e, f"WiFi连接失败")
|
||||
wifi_connected = True
|
||||
ip = w.get_ip()
|
||||
print(f"[WiFi] 连接成功！IP: {ip}")
|
||||
except Exception as e:
|
||||
print(f"[WiFi] {ssid} 连接失败: {e}")
|
||||
current_wifi_idx += 1
|
||||
await asyncio.sleep(1)
|
||||
|
||||
if not wifi_connected:
|
||||
print("[WiFi] 所有配置连接失败，无联网状态")
|
||||
|
||||
# ------------------------------
|
||||
# MQTT回调函数（新增track_mode控制逻辑）
|
||||
# ------------------------------
|
||||
def on_mqtt_connect(client, userdata, flags, rc):
|
||||
"""MQTT连接回调"""
|
||||
global mqtt_connected
|
||||
if rc == 0:
|
||||
mqtt_connected = True
|
||||
print("[MQTT] 连接成功")
|
||||
# 订阅主题（如有）
|
||||
if MQTT_SUB_TOPICS:
|
||||
client.subscribe(MQTT_SUB_TOPICS)
|
||||
print(f"[MQTT] 已订阅主题: {MQTT_SUB_TOPICS}")
|
||||
else:
|
||||
mqtt_connected = False
|
||||
print(f"[MQTT] 连接失败，错误码: {rc}")
|
||||
|
||||
def on_mqtt_message(client, userdata, msg):
|
||||
global track_mode, serial_dev
|
||||
if msg.topic == MQTT_SUB_TOPIC_TEXT:
|
||||
payload = msg.payload.decode().strip() # 去除首尾空格，提高容错
|
||||
print(f"文本指令: {payload}")
|
||||
# 解析跟踪模式指令
|
||||
if payload.lower() == "track_off":
|
||||
track_mode = 0
|
||||
print("[跟踪模式] 已关闭人脸跟踪，停止绘制人脸框和追踪")
|
||||
elif payload.lower() == "track_face":
|
||||
track_mode = 1
|
||||
print("[跟踪模式] 已开启人脸跟踪，开始绘制人脸框和追踪")
|
||||
elif payload.lower() == "track_hand":
|
||||
track_mode = 2
|
||||
print("[跟踪模式] 已开启手部跟踪，开始绘制手部框和追踪")
|
||||
elif msg.topic == MQTT_SUB_TOPIC_MOVE:
|
||||
payload = msg.payload.decode().strip()
|
||||
# 修复：判空保护 + 使用全局serial_dev
|
||||
if serial_dev is not None:
|
||||
try:
|
||||
serial_dev.write_str(payload)
|
||||
print(f"运动指令已发送: {payload}")
|
||||
except Exception as e:
|
||||
print(f"发送运动指令失败: {e} | 指令内容: {payload}")
|
||||
else:
|
||||
print(f"运动指令: {payload}，但串口未初始化")
|
||||
|
||||
def on_mqtt_disconnect(client, userdata, rc):
|
||||
"""MQTT断开回调"""
|
||||
global mqtt_connected
|
||||
mqtt_connected = False
|
||||
if rc != 0:
|
||||
print(f"[MQTT] 意外断开，错误码: {rc}")
|
||||
|
||||
# ------------------------------
|
||||
# MQTT初始化（异步）
|
||||
# ------------------------------
|
||||
async def init_mqtt_async():
|
||||
"""异步初始化MQTT（依赖WiFi，自动重试）"""
|
||||
global mqtt_client, mqtt_connected, current_wifi_idx
|
||||
mqtt_connected = False
|
||||
mqtt_client = None
|
||||
|
||||
# 等待WiFi连接（最多20秒）
|
||||
wait_time = 0
|
||||
while not app.need_exit() and not wifi_connected and wait_time < 20:
|
||||
await asyncio.sleep(1)
|
||||
wait_time += 1
|
||||
|
||||
if app.need_exit() or not wifi_connected:
|
||||
print("[MQTT] WiFi未连接，跳过MQTT初始化")
|
||||
return
|
||||
|
||||
# 兜底索引
|
||||
if current_wifi_idx >= len(WIFI_CONFIGS):
|
||||
current_wifi_idx = 0
|
||||
current_broker = WIFI_CONFIGS[current_wifi_idx]["MQTT_BROKER"]
|
||||
|
||||
try:
|
||||
mqtt_client = mqtt.Client(DEVICE_ID)
|
||||
mqtt_client.on_connect = on_mqtt_connect
|
||||
mqtt_client.on_message = on_mqtt_message
|
||||
mqtt_client.on_disconnect = on_mqtt_disconnect
|
||||
|
||||
# 重试连接
|
||||
retry_count = 0
|
||||
while not app.need_exit() and not mqtt_connected and retry_count < 5:
|
||||
try:
|
||||
print(f"[MQTT] 尝试连接 {current_broker}:{MQTT_PORT} (重试{retry_count+1}/5)")
|
||||
mqtt_client.connect(current_broker, MQTT_PORT, 60)
|
||||
mqtt_client.loop_start()
|
||||
|
||||
# 等待连接结果
|
||||
t = time.time()
|
||||
while not mqtt_connected and time.time() - t < 5:
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
if mqtt_connected:
|
||||
break
|
||||
else:
|
||||
mqtt_client.disconnect()
|
||||
retry_count += 1
|
||||
await asyncio.sleep(3)
|
||||
except Exception as e:
|
||||
print(f"[MQTT] 连接异常: {e}")
|
||||
retry_count += 1
|
||||
await asyncio.sleep(3)
|
||||
|
||||
if not mqtt_connected:
|
||||
print("[MQTT] 多次重试后仍无法连接")
|
||||
except Exception as e:
|
||||
print(f"[MQTT] 初始化失败: {e}")
|
||||
mqtt_client = None
|
||||
mqtt_connected = False
|
||||
|
||||
# ------------------------------
|
||||
# 图像MQTT发送
|
||||
# ------------------------------
|
||||
def send_image_via_mqtt(img):
|
||||
"""发送图像到MQTT（JPEG压缩）"""
|
||||
global mqtt_connected, mqtt_client
|
||||
if not mqtt_connected or mqtt_client is None:
|
||||
return False
|
||||
|
||||
try:
|
||||
jpeg_data = img.to_jpeg(quality=85).to_bytes()
|
||||
# 二进制发送（高效）
|
||||
mqtt_client.publish(MQTT_PUB_TOPIC_IMAGE, jpeg_data)
|
||||
gc.collect()
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"[MQTT] 图像发送失败: {e}")
|
||||
try:
|
||||
if mqtt_client:
|
||||
mqtt_client.reconnect()
|
||||
except:
|
||||
mqtt_connected = False
|
||||
return False
|
||||
from config import *
|
||||
from test_face import *
|
||||
from test_ball import *
|
||||
|
||||
# ------------------------------
|
||||
# 系统初始化
|
||||
# ------------------------------
|
||||
def init_system():
|
||||
|
||||
"""初始化相机、舵机、串口、LED"""
|
||||
global serial_dev # 声明使用全局serial_dev
|
||||
|
||||
"""初始化相机、舵机、串口、LED"""
|
||||
init_led()
|
||||
|
||||
# 人脸检测器
|
||||
@@ -304,167 +35,45 @@ def init_system():
|
||||
f"Failed set pin{pin} function to {func}")
|
||||
serial_dev = uart.UART(device, 115200)
|
||||
|
||||
# 舵机初始角度
|
||||
current_s1, current_s2 = 90, 90
|
||||
initial_cmd = f"S1:{current_s1},S2:{current_s2};"
|
||||
# 初始化值 距离中间点像素距离
|
||||
initial_cmd = f"DX:0,DY:0;"
|
||||
serial_dev.write_str(initial_cmd)
|
||||
print(f"[舵机] 初始角度: {initial_cmd}")
|
||||
time.sleep(1)
|
||||
|
||||
return {
|
||||
"detector": detector, "cam": cam, "disp": disp, "serial_dev": serial_dev,
|
||||
"img_width": img_width, "img_height": img_height, "image_center": image_center,
|
||||
"current_s1": current_s1, "current_s2": current_s2
|
||||
"img_width": img_width, "img_height": img_height, "image_center": image_center
|
||||
}
|
||||
|
||||
# ------------------------------
|
||||
# 人脸检测处理（新增track_mode判断，关闭时不绘制人脸框）
|
||||
# ------------------------------
|
||||
def process_faces(img, objs):
|
||||
def callback(data: list[float], len: int):
|
||||
|
||||
global track_mode
|
||||
# 跟踪模式关闭时，直接返回无脸坐标，不绘制任何框
|
||||
if not track_mode :
|
||||
return (-1, -1, -1)
|
||||
|
||||
max_face = None
|
||||
max_width = 0
|
||||
|
||||
if objs:
|
||||
for obj in objs:
|
||||
if obj.w > max_width:
|
||||
max_width = obj.w
|
||||
max_face = obj
|
||||
|
||||
# 绘制人脸框（仅跟踪模式开启时执行）
|
||||
for obj in objs:
|
||||
color = image.COLOR_GREEN if obj == max_face else image.COLOR_RED
|
||||
img.draw_rect(obj.x, obj.y, obj.w, obj.h, color=color, thickness=2)
|
||||
|
||||
if max_face:
|
||||
center_x = int(max_face.x + max_face.w / 2)
|
||||
center_y = int(max_face.y + max_face.h / 2)
|
||||
img.draw_keypoints([center_x, center_y], image.COLOR_GREEN, size=3)
|
||||
return (center_x, center_y, int(max_face.w))
|
||||
else:
|
||||
return (-1, -1, -1)
|
||||
|
||||
# 2. 找到概率最大的关键词索引
|
||||
max_prob = max(data) # 获取最大概率值
|
||||
max_index = data.index(max_prob) # 获取最大概率对应的kw索引
|
||||
|
||||
# 3. 判断：仅当最大概率>0时，输出对应value
|
||||
if max_prob > 0:
|
||||
# 根据索引映射到kw_list的value（kw0→0，kw1→1，kw2→2）
|
||||
output_value = [0, 1, 2][max_index]
|
||||
print(f"识别结果：{output_value}") # 输出最终的value
|
||||
track_mode = output_value
|
||||
|
||||
# ------------------------------
|
||||
# 人脸追踪
|
||||
# 系统执行
|
||||
# ------------------------------
|
||||
def track_face(center, image_center, current_angles, serial_dev):
|
||||
global last_dev_x, last_dev_y
|
||||
center_x, center_y = center
|
||||
current_s1, current_s2 = current_angles
|
||||
|
||||
# 偏差计算+滤波
|
||||
deviation_x = center_x - image_center[0]
|
||||
deviation_y = center_y - image_center[1]
|
||||
smoothed_x = filter_alpha * last_dev_x + (1 - filter_alpha) * deviation_x
|
||||
smoothed_y = filter_alpha * last_dev_y + (1 - filter_alpha) * deviation_y
|
||||
last_dev_x, last_dev_y = smoothed_x, smoothed_y
|
||||
|
||||
# 角度计算+限制
|
||||
# 金属舵机
|
||||
# new_s1 = constrain(int(current_s1 + smoothed_x * kp_x), s1_range[0], s1_range[1])
|
||||
# new_s2 = constrain(int(current_s2 + smoothed_y * kp_y), s2_range[0], s2_range[1])
|
||||
# 塑料舵机
|
||||
new_s1 = constrain(int(current_s1 - smoothed_x * kp_x), s1_range[0], s1_range[1])
|
||||
new_s2 = constrain(int(current_s2 - smoothed_y * kp_y), s2_range[0], s2_range[1])
|
||||
|
||||
# 发送舵机指令
|
||||
if abs(new_s1 - current_s1) > angle_threshold or abs(new_s2 - current_s2) > angle_threshold:
|
||||
cmd = f"S1:{new_s1},S2:{new_s2};"
|
||||
serial_dev.write_str(cmd)
|
||||
return (new_s1, new_s2, cmd)
|
||||
else:
|
||||
return (current_s1, current_s2, None)
|
||||
try:
|
||||
# 初始化系统资源
|
||||
system_resources = init_system()
|
||||
# 进入主循环
|
||||
# trace_face_main(system_resources)
|
||||
trace_ball_main(system_resources)
|
||||
|
||||
except Exception as e:
|
||||
print(f"[异常] 程序异常: {e}")
|
||||
traceback.print_exc()
|
||||
|
||||
# ------------------------------
|
||||
# 主循环（新增track_mode判断和状态显示）
|
||||
# ------------------------------
|
||||
async def main_loop_async(resources):
|
||||
"""异步主循环"""
|
||||
detector = resources["detector"]
|
||||
cam = resources["cam"]
|
||||
disp = resources["disp"]
|
||||
serial_dev = resources["serial_dev"]
|
||||
img_width = resources["img_width"]
|
||||
img_height = resources["img_height"]
|
||||
image_center = resources["image_center"]
|
||||
current_s1, current_s2 = resources["current_s1"], resources["current_s2"]
|
||||
|
||||
# 启动网络异步任务
|
||||
asyncio.create_task(connect_wifi_async())
|
||||
asyncio.create_task(init_mqtt_async())
|
||||
|
||||
# 主循环
|
||||
global frame_count, mqtt_connected, wifi_connected, track_mode
|
||||
while not app.need_exit():
|
||||
# 图像读取+人脸检测
|
||||
img = cam.read().flip(image.FlipDir.Y)
|
||||
objs = detector.detect(img, conf_th=0.4, iou_th=0.45)
|
||||
center_x, center_y, face_width = process_faces(img, objs)
|
||||
|
||||
# 绘制基础状态
|
||||
img.draw_string(20, 20, f"Center: ({center_x}, {center_y})", image.COLOR_GREEN, 1.0)
|
||||
img.draw_string(20, 40, f"Max Width: {face_width}", image.COLOR_YELLOW, 1.0)
|
||||
|
||||
# 绘制跟踪模式状态（直观显示当前模式）
|
||||
track_status = "ON" if track_mode else "OFF"
|
||||
track_color = image.COLOR_GREEN if track_mode else image.COLOR_RED
|
||||
img.draw_string(20, 100, f"Track Mode: {track_status}", track_color, 1.0)
|
||||
|
||||
# 人脸追踪（仅跟踪模式开启时执行）
|
||||
if track_mode and center_x != -1 and center_y != -1:
|
||||
new_s1, new_s2, cmd = track_face((center_x, center_y), image_center, (current_s1, current_s2), serial_dev)
|
||||
current_s1, current_s2 = new_s1, new_s2
|
||||
if cmd:
|
||||
img.draw_string(20, 60, f"Tracking: {cmd}", image.COLOR_BLUE, 1.0)
|
||||
elif track_mode and center_x == -1:
|
||||
# 跟踪模式开启但无脸
|
||||
img.draw_string(20, 60, "No face detected", image.COLOR_RED, 1.0)
|
||||
else:
|
||||
# 跟踪模式关闭
|
||||
img.draw_string(20, 60, "Track Mode: OFF", image.COLOR_ORANGE, 1.0)
|
||||
|
||||
# 网络状态显示
|
||||
net_status = f"WiFi: {'OK' if wifi_connected else 'NO'} | MQTT: {'OK' if mqtt_connected else 'NO'}"
|
||||
img.draw_string(20, 80, net_status, image.COLOR_YELLOW, 1.0)
|
||||
|
||||
# LED更新+图像发送
|
||||
update_led()
|
||||
frame_count += 1
|
||||
if frame_count >= send_frame_interval:
|
||||
send_image_via_mqtt(img)
|
||||
frame_count = 0
|
||||
|
||||
disp.show(img)
|
||||
await asyncio.sleep(0.001)
|
||||
|
||||
# ------------------------------
|
||||
# 程序入口
|
||||
# ------------------------------
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
system_resources = init_system()
|
||||
asyncio.run(main_loop_async(system_resources))
|
||||
except Exception as e:
|
||||
print(f"[异常] 程序异常: {e}")
|
||||
traceback.print_exc()
|
||||
finally:
|
||||
# 清理MQTT
|
||||
if mqtt_client:
|
||||
try:
|
||||
mqtt_client.loop_stop()
|
||||
mqtt_client.disconnect()
|
||||
print("[MQTT] 客户端已断开")
|
||||
except Exception as e:
|
||||
print(f"[MQTT] 断开失败: {e}")
|
||||
# 清理LED
|
||||
if led:
|
||||
try:
|
||||
led.value(0)
|
||||
print("[LED] 已关闭")
|
||||
except Exception as e:
|
||||
print(f"[LED] 关闭失败: {e}")
|
||||
print("[系统] 程序已退出")
|
||||
@@ -0,0 +1,134 @@
|
||||
from maix import image, app, time
|
||||
|
||||
# 橘黄色乒乓球的颜色阈值配置（HSV色域）
|
||||
thresholds = [
|
||||
[73, 93, 7, 27, 13, 33],
|
||||
[40, 65, 30, 60, 40, 70],
|
||||
[39, 59, 41, 61, 48, 68],
|
||||
]
|
||||
|
||||
# 新增：丢失目标的阈值配置（可根据实际帧率调整，建议5-10帧）
|
||||
LOST_FRAME_THRESHOLD = 10 # 连续x帧没识别到才判定为真正丢失
|
||||
SEND_INTERVAL = 5 # 控制指令发送间隔（避免高频发送，单位：ms）
|
||||
|
||||
def select_largest_blob(blobs):
|
||||
"""
|
||||
筛选并选取最大的有效blob（替换原融合逻辑）
|
||||
1. 筛除宽/高小于10px的小blob（过滤噪声）
|
||||
2. 从剩余blob中选取面积最大的单个blob
|
||||
:param blobs: 识别到的所有blob列表
|
||||
:return: 最大blob的矩形参数 (x, y, width, height) 和中心点 (cx, cy)；未识别到则返回None, None
|
||||
"""
|
||||
if not blobs:
|
||||
return None, (-1,-1)
|
||||
|
||||
# 步骤1：过滤掉宽/高小于10px的无效blob
|
||||
valid_blobs = []
|
||||
for b in blobs:
|
||||
x, y, w, h = b[0], b[1], b[2], b[3]
|
||||
if w >= 10 and h >= 10: # 筛除边距（宽/高）小于10px的blob
|
||||
valid_blobs.append(b)
|
||||
|
||||
# 过滤后无有效blob，返回None
|
||||
if not valid_blobs:
|
||||
return None, (-1,-1)
|
||||
|
||||
# 步骤2：选取面积（w*h）最大的blob
|
||||
largest_blob = max(valid_blobs, key=lambda b: b[2] * b[3])
|
||||
|
||||
# 提取最大blob的矩形参数和中心点
|
||||
x, y, w, h = largest_blob[0], largest_blob[1], largest_blob[2], largest_blob[3]
|
||||
cx = x + w // 2
|
||||
cy = y + h // 2
|
||||
|
||||
return (x, y, w, h), (cx, cy)
|
||||
|
||||
def calculate_deviation_ball(center, image_center):
|
||||
"""
|
||||
优化：仅计算偏移量，不直接发送指令（将指令发送逻辑移到主循环统一处理）
|
||||
:return: (deviation_x, deviation_y) 或 None（无有效目标）
|
||||
"""
|
||||
center_x, center_y = center
|
||||
|
||||
if center_x == -1 and center_y == -1 :
|
||||
return None # 无有效目标
|
||||
|
||||
# 计算x/y方向的偏移偏差（目标中心 - 图像中心）
|
||||
deviation_x = center_x - image_center[0]
|
||||
deviation_y = center_y - image_center[1]
|
||||
|
||||
# 仅当偏移超过阈值时返回有效偏移，否则返回None（避免微小抖动）
|
||||
if abs(deviation_x) > 10 or abs(deviation_y) > 10 :
|
||||
return (deviation_x, deviation_y)
|
||||
else:
|
||||
return None
|
||||
|
||||
def trace_ball_main(resources):
|
||||
detector = resources["detector"]
|
||||
cam = resources["cam"]
|
||||
disp = resources["disp"]
|
||||
serial_dev = resources["serial_dev"]
|
||||
img_width = resources["img_width"]
|
||||
img_height = resources["img_height"]
|
||||
image_center = resources["image_center"]
|
||||
|
||||
# 新增：状态变量初始化
|
||||
last_valid_deviation = None # 记录最后一次有效偏移量
|
||||
lost_frames = 0 # 丢失帧数计数
|
||||
last_send_time = 0 # 最后一次发送指令的时间（防高频）
|
||||
|
||||
while not app.need_exit():
|
||||
current_time = time.ticks_ms()
|
||||
img = cam.read().flip(image.FlipDir.Y)
|
||||
|
||||
# 识别色块（像素阈值设为10，过滤小噪声）
|
||||
blobs = img.find_blobs(thresholds, pixels_threshold=10)
|
||||
|
||||
# 筛选并选取最大的有效blob
|
||||
largest_rect, target_center = select_largest_blob(blobs)
|
||||
|
||||
# 计算当前帧的偏移量
|
||||
current_deviation = calculate_deviation_ball(target_center, image_center)
|
||||
|
||||
# -------------------------- 核心优化逻辑 --------------------------
|
||||
if current_deviation is not None:
|
||||
# 1. 识别到有效目标：重置丢失计数，更新最后一次有效偏移
|
||||
lost_frames = 0
|
||||
last_valid_deviation = current_deviation
|
||||
# 控制发送频率，避免串口堵塞
|
||||
if current_time - last_send_time >= SEND_INTERVAL:
|
||||
dx, dy = current_deviation
|
||||
cmd = f"DX:{dx},DY:{dy};"
|
||||
serial_dev.write_str(cmd)
|
||||
print(f"识别到目标: {cmd}")
|
||||
last_send_time = current_time
|
||||
else:
|
||||
# 2. 未识别到目标：累计丢失帧数
|
||||
lost_frames += 1
|
||||
if lost_frames < LOST_FRAME_THRESHOLD and last_valid_deviation is not None:
|
||||
# 2.1 丢失帧数未达阈值：继续发送最后一次有效偏移
|
||||
if current_time - last_send_time >= SEND_INTERVAL:
|
||||
dx, dy = last_valid_deviation
|
||||
cmd = f"DX:{dx},DY:{dy};"
|
||||
serial_dev.write_str(cmd)
|
||||
print(f"短暂丢失，沿用最后偏移: {cmd} (丢失帧数: {lost_frames})")
|
||||
last_send_time = current_time
|
||||
else:
|
||||
# 2.2 丢失帧数达阈值：发送9999，重置最后有效偏移
|
||||
if current_time - last_send_time >= SEND_INTERVAL:
|
||||
cmd = f"DX:9999,DY:9999;"
|
||||
serial_dev.write_str(cmd)
|
||||
print(f"目标丢失超过阈值: {cmd} (丢失帧数: {lost_frames})")
|
||||
last_valid_deviation = None
|
||||
last_send_time = current_time
|
||||
# -----------------------------------------------------------------
|
||||
|
||||
# 绘制识别结果
|
||||
if largest_rect:
|
||||
x, y, w, h = largest_rect
|
||||
img.draw_rect(x, y, w, h, image.COLOR_GREEN, 2)
|
||||
cx, cy = target_center
|
||||
img.draw_circle(cx, cy, 3, image.COLOR_RED, -1)
|
||||
|
||||
disp.show(img)
|
||||
time.sleep_ms(1)
|
||||
@@ -0,0 +1,88 @@
|
||||
# ------------------------------
|
||||
# 人脸检测处理
|
||||
# ------------------------------
|
||||
from maix import image, app, time
|
||||
from config import *
|
||||
|
||||
def process_faces(img, objs):
|
||||
|
||||
max_face = None
|
||||
max_width = 0
|
||||
|
||||
if objs:
|
||||
for obj in objs:
|
||||
if obj.w > max_width:
|
||||
max_width = obj.w
|
||||
max_face = obj
|
||||
|
||||
# 绘制人脸框（仅跟踪模式开启时执行）
|
||||
for obj in objs:
|
||||
color = image.COLOR_GREEN if obj == max_face else image.COLOR_RED
|
||||
img.draw_rect(obj.x, obj.y, obj.w, obj.h, color=color, thickness=2)
|
||||
|
||||
if max_face:
|
||||
center_x = int(max_face.x + max_face.w / 2)
|
||||
center_y = int(max_face.y + max_face.h / 2)
|
||||
img.draw_keypoints([center_x, center_y], image.COLOR_GREEN, size=3)
|
||||
return (center_x, center_y, int(max_face.w))
|
||||
else:
|
||||
return (-1, -1, -1)
|
||||
|
||||
# ------------------------------
|
||||
# 人脸追踪
|
||||
# ------------------------------
|
||||
def calculate_deviation_face(center, image_center, serial_dev):
|
||||
|
||||
center_x, center_y = center
|
||||
|
||||
# 偏差计算
|
||||
deviation_x = center_x - image_center[0]
|
||||
deviation_y = center_y - image_center[1]
|
||||
|
||||
# 输出像素偏差
|
||||
if abs(deviation_x) > 10 or abs(deviation_y) > 10 :
|
||||
cmd = f"DX:{deviation_x},DY:{deviation_y};" # 图像尺寸: w320xh224
|
||||
serial_dev.write_str(cmd)
|
||||
print(cmd)
|
||||
return (cmd)
|
||||
else:
|
||||
return (None)
|
||||
|
||||
# ------------------------------
|
||||
# 主循环（同步）
|
||||
# ------------------------------
|
||||
def trace_face_main(resources):
|
||||
|
||||
detector = resources["detector"]
|
||||
cam = resources["cam"]
|
||||
disp = resources["disp"]
|
||||
serial_dev = resources["serial_dev"]
|
||||
img_width = resources["img_width"]
|
||||
img_height = resources["img_height"]
|
||||
image_center = resources["image_center"]
|
||||
|
||||
while not app.need_exit():
|
||||
|
||||
img = cam.read().flip(image.FlipDir.Y)
|
||||
objs = detector.detect(img, conf_th=0.4, iou_th=0.45)
|
||||
center_x, center_y, face_width = process_faces(img, objs)
|
||||
|
||||
# 绘制基础状态
|
||||
img.draw_string(20, 20, f"Center: ({center_x}, {center_y})", image.COLOR_GREEN, 1.0)
|
||||
img.draw_string(20, 40, f"Max Width: {face_width}", image.COLOR_YELLOW, 1.0)
|
||||
|
||||
# 人脸追踪（仅跟踪模式开启时执行）
|
||||
if center_x != -1 and center_y != -1:
|
||||
cmd = calculate_deviation_face((center_x, center_y), image_center, serial_dev)
|
||||
if cmd:
|
||||
img.draw_string(20, 60, f"Tracking: {cmd}", image.COLOR_BLUE, 1.0)
|
||||
elif center_x == -1:
|
||||
cmd = f"DX:9999,DY:9999;" # 未找到
|
||||
serial_dev.write_str(cmd)
|
||||
print(cmd)
|
||||
img.draw_string(20, 60, "No face detected", image.COLOR_RED, 1.0)
|
||||
else:
|
||||
img.draw_string(20, 60, "Track Mode: OFF", image.COLOR_ORANGE, 1.0)
|
||||
|
||||
disp.show(img)
|
||||
time.sleep_ms(1)
|
||||
Reference in new issue
Block a user