基于二哈识图2的电动车入梯检测与联动控制系统（持续完善中） 简单

import os
import shutil
import xml.etree.ElementTree as ET

# ===================== 1. 配置参数（无需修改，适配你的数据集结构） =====================
XML_DIR = "Annotations"               # VOC标注目录
IMG_DIR = "JPEGImages"                # 图片目录
LABEL_LIST_PATH = "label_list.txt"    # 类别文件
TRAIN_LIST_PATH = "train.txt"    # 训练集列表
VAL_LIST_PATH = "val.txt"        # 验证集列表
# YOLO输出目录（自动创建）
YOLO_ROOT = "yolo_dataset"
YOLO_IMG_TRAIN = os.path.join(YOLO_ROOT, "images", "train")
YOLO_IMG_VAL = os.path.join(YOLO_ROOT, "images", "val")
YOLO_LBL_TRAIN = os.path.join(YOLO_ROOT, "labels", "train")
YOLO_LBL_VAL = os.path.join(YOLO_ROOT, "labels", "val")

# ===================== 2. 初始化（创建目录+读取类别映射） =====================
# 创建YOLO目录结构
for dir_path in [YOLO_IMG_TRAIN, YOLO_IMG_VAL, YOLO_LBL_TRAIN, YOLO_LBL_VAL]:
    os.makedirs(dir_path, exist_ok=True)

# 读取类别并建立ID映射（person=0, bicycle=1, motorcycle=2）
with open(LABEL_LIST_PATH, "r", encoding="utf-8") as f:
    class_names = [line.strip() for line in f if line.strip()]
class_id_map = {name: idx for idx, name in enumerate(class_names)}
print(f"类别ID映射：{class_id_map}")

# ===================== 3. 核心函数：XML转YOLO TXT =====================
def xml_to_yolo(xml_path, img_width, img_height):
    """解析XML，返回YOLO格式的标注内容（归一化坐标）"""
    tree = ET.parse(xml_path)
    root = tree.getroot()
    yolo_lines = []
    
    for obj in root.findall("object"):
        # 获取类别ID（跳过未定义类别）
        obj_name = obj.find("name").text.strip()
        if obj_name not in class_id_map:
            print(f"警告：跳过未定义类别 {obj_name}（XML文件：{xml_path}）")
            continue
        class_id = class_id_map[obj_name]
        
        # 解析VOC边界框（xmin, ymin, xmax, ymax）
        bndbox = obj.find("bndbox")
        xmin = int(bndbox.find("xmin").text)
        ymin = int(bndbox.find("ymin").text)
        xmax = int(bndbox.find("xmax").text)
        ymax = int(bndbox.find("ymax").text)
        
        # 转换为YOLO格式（归一化中心坐标+宽高）
        center_x = (xmin + xmax) / (2 * img_width)
        center_y = (ymin + ymax) / (2 * img_height)
        width = (xmax - xmin) / img_width
        height = (ymax - ymin) / img_height
        
        # 保留6位小数，拼接成一行
        yolo_lines.append(f"{class_id} {center_x:.6f} {center_y:.6f} {width:.6f} {height:.6f}\n")
    return yolo_lines

# ===================== 4. 批量处理训练集/验证集 =====================
def process_split(split_list_path, yolo_img_dir, yolo_lbl_dir):
    """根据数据集列表，批量复制图片+生成YOLO标注"""
    with open(split_list_path, "r", encoding="utf-8") as f:
        # 读取所有文件名（无后缀）
        base_names = [line.strip() for line in f if line.strip()]
    
    for base_name in base_names:
        # 1. 复制图片到YOLO images目录
        img_src = os.path.join(IMG_DIR, f"{base_name}.jpg")
        img_dst = os.path.join(yolo_img_dir, f"{base_name}.jpg")
        if not os.path.exists(img_src):
            print(f"警告：图片不存在 {img_src}，跳过")
            continue
        shutil.copyfile(img_src, img_dst)
        
        # 2. 解析XML生成YOLO TXT标注
        xml_src = os.path.join(XML_DIR, f"{base_name}.xml")
        lbl_dst = os.path.join(yolo_lbl_dir, f"{base_name}.txt")
        if not os.path.exists(xml_src):
            print(f"警告：标注不存在 {xml_src}，跳过")
            continue
        
        # 获取图片宽高（用于坐标归一化）
        img = ET.parse(xml_src).find("size")
        img_w = int(img.find("width").text)
        img_h = int(img.find("height").text)
        
        # 转换并写入YOLO标注
        yolo_content = xml_to_yolo(xml_src, img_w, img_h)
        with open(lbl_dst, "w", encoding="utf-8") as f:
            f.writelines(yolo_content)

# 执行训练集处理
print("正在处理训练集...")
process_split(TRAIN_LIST_PATH, YOLO_IMG_TRAIN, YOLO_LBL_TRAIN)
# 执行验证集处理
print("正在处理验证集...")
process_split(VAL_LIST_PATH, YOLO_IMG_VAL, YOLO_LBL_VAL)

print(f"\n转换完成！YOLO数据集保存至：{YOLO_ROOT}")
print("YOLO目录结构：")
print(f"- {YOLO_ROOT}/images/train：训练集图片")
print(f"- {YOLO_ROOT}/images/val：验证集图片")
print(f"- {YOLO_ROOT}/labels/train：训练集YOLO标注（.txt）")
print(f"- {YOLO_ROOT}/labels/val：验证集YOLO标注（.txt）")

from unihiker import GUI, Audio
import time
import json
import cv2
import base64
import random
from datetime import datetime
from dfrobot_huskylensv2 import *
from pinpong.board import Board, Pin
import siot

# ---------- 1. 初始化硬件 ----------
Board().begin()
u_gui = GUI()
u_audio = Audio()  # 初始化音频播放模块

# 继电器控制引脚初始化
RELAY_PIN = Pin(Pin.P22, Pin.OUT)
RELAY_PIN.write_digital(0)
print("[INFO] 继电器初始化完成，状态：断开")

# ---------- 2. 初始化二哈识图2 (I2C) ----------
huskylens = HuskylensV2_I2C()
huskylens.knock()
ALGORITHM_CUSTOM_MODEL = 129  # 自训练模型算法ID
huskylens.switchAlgorithm(ALGORITHM_CUSTOM_MODEL)
print("[INFO] 二哈识图2初始化完成，自训练模型已加载。")

# ---------- 3. 初始化USB摄像头 ----------
cap = cv2.VideoCapture(0)
if cap.isOpened():
    cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640)
    cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)
    print("[INFO] USB摄像头初始化成功。")
else:
    print("[ERROR] 摄像头初始化失败！")
    cap = None

# ---------- 4. MQTT与标签配置 ----------
SIOT_SERVER = "127.0.0.1"  # 改为实际SIoT服务器IP
SIOT_PORT = 1883
SIOT_USER = "siot"
SIOT_PASSWORD = "dfrobot"
DEVICE_ID = "siot"
PROJECT_NAME = "elevator_project"
TOPIC = f"{DEVICE_ID}/{PROJECT_NAME}"

# 标签中英文映射（匹配你的新标签）
LABEL_MAP = {
    'person': '人',
    'bicycle': '自行车',
    'motorcycle': '电动车',
}

# ---------- 5. 屏幕显示初始化 ----------
title = u_gui.draw_text(text="电梯安全联动监控系统", x=10, y=10, font_size=16, color="#0000FF")
status_label = u_gui.draw_text(text="状态：初始化...", x=10, y=50, font_size=12)
relay_label = u_gui.draw_text(text="继电器：断开", x=10, y=80, font_size=12)
count_label = u_gui.draw_text(text="检测目标：0", x=10, y=110, font_size=12)
last_event_label = u_gui.draw_text(text="最后事件：无", x=10, y=140, font_size=10, color="#666666")

# ---------- 6. 连接SIoT服务器 ----------
def connect_siot():
    try:
        client_id = f"m10_{random.randint(10000, 99999)}"
        siot.init(client_id=client_id, server=SIOT_SERVER, port=SIOT_PORT, 
                  user=SIOT_USER, password=SIOT_PASSWORD)
        siot.connect()
        siot.loop()
        print(f"[INFO] SIoT连接成功: {SIOT_SERVER}")
        status_label.config(text="状态：已连接，监控中")
        status_label.config(color="#008000")
        return True
    except Exception as e:
        print(f"[ERROR] SIoT连接失败: {e}")
        status_label.config(text="状态：网络连接失败")
        status_label.config(color="red")
        return False

siot_connected = connect_siot()

# ---------- 7. 核心功能函数 ----------
def set_relay(state):
    """控制继电器状态 (0:断开/正常, 1:闭合/开门)"""
    RELAY_PIN.write_digital(state)
    state_text = "闭合" if state else "断开"
    color = "red" if state else "green"
    relay_label.config(text=f"继电器：{state_text}")
    relay_label.config(color=color)
    print(f"[RELAY] 状态 -> {state_text}")

def play_alarm():
    """播放告警语音"""
    try:
        u_audio.start_play("/home/pi/record.wav")  # 请确保语音文件路径正确
        print("[AUDIO] 播放告警语音")
    except Exception as e:
        print(f"[ERROR] 播放语音失败: {e}")

def capture_photo():
    """捕获单张现场照片并返回Base64字符串"""
    if cap and cap.isOpened():
        ret, frame = cap.read()
        if ret:
            _, buffer = cv2.imencode('.jpg', frame, [cv2.IMWRITE_JPEG_QUALITY, 85])
            return base64.b64encode(buffer).decode('utf-8')
    return ""

def send_target_data(target_obj, photo_base64, timestamp):
    """为单个目标生成并上报数据到SIoT"""
    if not siot_connected:
        return False
    try:
        label_en = target_obj.name
        label_cn = LABEL_MAP.get(label_en, label_en)
        payload = {
            "time": timestamp,
            "lng": 116.397129,  # 模拟数据，实际项目可接GPS模块
            "lat": 39.916527,
            "photo": photo_base64,
            "deviceId": "1号电梯",
            "damageCategory": label_cn
        }
        siot.publish_save(topic=TOPIC, data=json.dumps(payload, ensure_ascii=False))
        print(f"[SIoT] 上报成功: {label_cn}")
        return True
    except Exception as e:
        print(f"[ERROR] 上报失败: {e}")
        return False

# ---------- 8. 主状态循环 ----------
print("[INFO] 系统启动，进入监控状态...")
electric_vehicle_present = False
departure_timer = 0
DEPARTURE_CONFIRM_TIME = 3  # 电动车离开确认时间（秒）
last_trigger_time = 0
EVENT_COOLDOWN = 10  # 事件上报冷却时间（秒）

while True:
    # 维护MQTT连接
    if siot_connected:
        siot.loop()
    
    # 获取二哈识图2识别结果
    huskylens.getResult(ALGORITHM_CUSTOM_MODEL)
    
    if huskylens.available(ALGORITHM_CUSTOM_MODEL):
        target_count = huskylens.getCachedResultNum(ALGORITHM_CUSTOM_MODEL)
        count_label.config(text=f"检测目标：{target_count}")
        current_ev_detected = False
        detected_targets = []
        
        # 遍历所有识别到的目标
        for i in range(target_count):
            target = huskylens.getCachedResultByIndex(ALGORITHM_CUSTOM_MODEL, i)
            if target:
                if target.name == 'motorcycle':
                    current_ev_detected = True
                if target.name in LABEL_MAP:
                    detected_targets.append(target)
        
        current_time = time.time()
        
        # 状态1：检测到电动车进入
        if current_ev_detected and not electric_vehicle_present:
            print("[ALERT] 电动车进入！触发联动。")
            electric_vehicle_present = True
            departure_timer = 0
            
            # 执行联动动作
            play_alarm()
            set_relay(1)
            
            # 拍照并上报
            if current_time - last_trigger_time > EVENT_COOLDOWN:
                snapshot = capture_photo()
                event_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                for target in detected_targets:
                    if target.name == 'motorcycle':
                        send_target_data(target, snapshot, event_time)
                        last_trigger_time = current_time
                        break
            
            status_label.config(text="状态：电动车入梯，电梯已锁定")
            status_label.config(color="red")
        
        # 状态2：电动车持续存在
        elif current_ev_detected and electric_vehicle_present:
            departure_timer = 0  # 重置离开计时器
            status_label.config(text="状态：电动车仍在电梯内")
        
        # 状态3：电动车离开（检测中）
        elif not current_ev_detected and electric_vehicle_present:
            departure_timer += 0.1
            if departure_timer >= DEPARTURE_CONFIRM_TIME:
                print("[INFO] 电动车已离开，恢复电梯。")
                electric_vehicle_present = False
                set_relay(0)
                status_label.config(text="状态：已恢复，监控中")
                status_label.config(color="green")
                time.sleep(2)
                status_label.config(text="状态：监控中")
                status_label.config(color="black")
    
    time.sleep(0.1)

# 资源释放
if cap:
    cap.release()
print("[INFO] 系统停止。")