From cf60d9606677250c06c86f40d90873a14634883d Mon Sep 17 00:00:00 2001 From: Viajero <2737079298@qq.com> Date: Sat, 18 Oct 2025 18:21:30 +0800 Subject: [PATCH] =?UTF-8?q?=E6=9B=B4=E6=96=B0=E6=8E=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- LPRNET_part/lpr_interface.py | 328 -------------------------------- gate_control.py | 251 ++++++++++++++++++++++++ main.py | 356 ++++++++++++++++++++++++++++++++++- simple_client.py | 58 ++++++ 4 files changed, 663 insertions(+), 330 deletions(-) delete mode 100644 LPRNET_part/lpr_interface.py create mode 100644 gate_control.py create mode 100644 simple_client.py diff --git a/LPRNET_part/lpr_interface.py b/LPRNET_part/lpr_interface.py deleted file mode 100644 index 2b688ba..0000000 --- a/LPRNET_part/lpr_interface.py +++ /dev/null @@ -1,328 +0,0 @@ -import torch -import torch.nn as nn -import cv2 -import numpy as np -import os -import sys -from torch.autograd import Variable -from PIL import Image - -# 添加父目录到路径，以便导入模型和数据加载器 -sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - -# LPRNet字符集定义（与训练时保持一致） -CHARS = ['京', '沪', '津', '渝', '冀', '晋', '蒙', '辽', '吉', '黑', - '苏', '浙', '皖', '闽', '赣', '鲁', '豫', '鄂', '湘', '粤', - '桂', '琼', '川', '贵', '云', '藏', '陕', '甘', '青', '宁', '新', - '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', - 'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'J', 'K', - 'L', 'M', 'N', 'P', 'Q', 'R', 'S', 'T', 'U', 'V', - 'W', 'X', 'Y', 'Z', 'I', 'O', '-'] - -CHARS_DICT = {char: i for i, char in enumerate(CHARS)} - -# 简化的LPRNet模型定义 -class small_basic_block(nn.Module): - def __init__(self, ch_in, ch_out): - super(small_basic_block, self).__init__() - self.block = nn.Sequential( - nn.Conv2d(ch_in, ch_out // 4, kernel_size=1), - nn.ReLU(), - nn.Conv2d(ch_out // 4, ch_out // 4, kernel_size=(3, 1), padding=(1, 0)), - nn.ReLU(), - nn.Conv2d(ch_out // 4, ch_out // 4, kernel_size=(1, 3), padding=(0, 1)), - nn.ReLU(), - nn.Conv2d(ch_out // 4, ch_out, kernel_size=1), - ) - - def forward(self, x): - return self.block(x) - -class LPRNet(nn.Module): - def __init__(self, lpr_max_len, phase, class_num, dropout_rate): - super(LPRNet, self).__init__() - self.phase = phase - self.lpr_max_len = lpr_max_len - self.class_num = class_num - self.backbone = nn.Sequential( - nn.Conv2d(in_channels=3, out_channels=64, kernel_size=3, stride=1), # 0 - nn.BatchNorm2d(num_features=64), - nn.ReLU(), # 2 - nn.MaxPool3d(kernel_size=(1, 3, 3), stride=(1, 1, 1)), - small_basic_block(ch_in=64, ch_out=128), # *** 4 *** - nn.BatchNorm2d(num_features=128), - nn.ReLU(), # 6 - nn.MaxPool3d(kernel_size=(1, 3, 3), stride=(2, 1, 2)), - small_basic_block(ch_in=64, ch_out=256), # 8 - nn.BatchNorm2d(num_features=256), - nn.ReLU(), # 10 - small_basic_block(ch_in=256, ch_out=256), # *** 11 *** - nn.BatchNorm2d(num_features=256), - nn.ReLU(), # 13 - nn.MaxPool3d(kernel_size=(1, 3, 3), stride=(4, 1, 2)), # 14 - nn.Dropout(dropout_rate), - nn.Conv2d(in_channels=64, out_channels=256, kernel_size=(1, 4), stride=1), # 16 - nn.BatchNorm2d(num_features=256), - nn.ReLU(), # 18 - nn.Dropout(dropout_rate), - nn.Conv2d(in_channels=256, out_channels=class_num, kernel_size=(13, 1), stride=1), # 20 - nn.BatchNorm2d(num_features=class_num), - nn.ReLU(), # 22 - ) - self.container = nn.Sequential( - nn.Conv2d(in_channels=448+self.class_num, out_channels=self.class_num, kernel_size=(1,1), stride=(1,1)), - ) - - def forward(self, x): - keep_features = list() - for i, layer in enumerate(self.backbone.children()): - x = layer(x) - if i in [2, 6, 13, 22]: # [2, 4, 8, 11, 22] - keep_features.append(x) - - global_context = list() - for i, f in enumerate(keep_features): - if i in [0, 1]: - f = nn.AvgPool2d(kernel_size=5, stride=5)(f) - if i in [2]: - f = nn.AvgPool2d(kernel_size=(4, 10), stride=(4, 2))(f) - f_pow = torch.pow(f, 2) - f_mean = torch.mean(f_pow) - f = torch.div(f, f_mean) - global_context.append(f) - - x = torch.cat(global_context, 1) - x = self.container(x) - logits = torch.mean(x, dim=2) - - return logits - -class LPRNetInference: - def __init__(self, model_path=None, img_size=[94, 24], lpr_max_len=8, dropout_rate=0.5): - """ - 初始化LPRNet推理类 - Args: - model_path: 训练好的模型权重文件路径 - img_size: 输入图像尺寸 [width, height] - lpr_max_len: 车牌最大长度 - dropout_rate: dropout率 - """ - self.img_size = img_size - self.lpr_max_len = lpr_max_len - self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') - - # 设置默认模型路径 - if model_path is None: - current_dir = os.path.dirname(os.path.abspath(__file__)) - model_path = os.path.join(current_dir, 'LPRNet__iteration_74000.pth') - - # 初始化模型 - self.model = LPRNet(lpr_max_len=lpr_max_len, phase=False, class_num=len(CHARS), dropout_rate=dropout_rate) - - # 加载模型权重 - if model_path and os.path.exists(model_path): - print(f"Loading LPRNet model from {model_path}") - try: - self.model.load_state_dict(torch.load(model_path, map_location=self.device)) - print("LPRNet模型权重加载成功") - except Exception as e: - print(f"Warning: 加载模型权重失败: {e}. 使用随机权重.") - else: - print(f"Warning: 模型文件不存在或未指定: {model_path}. 使用随机权重.") - - self.model.to(self.device) - self.model.eval() - - print(f"LPRNet模型加载完成，设备: {self.device}") - print(f"模型参数数量: {sum(p.numel() for p in self.model.parameters()):,}") - - def preprocess_image(self, image_array): - """ - 预处理图像数组 - 使用与训练时相同的预处理方式 - Args: - image_array: numpy数组格式的图像 (H, W, C) - Returns: - preprocessed_image: 预处理后的图像tensor - """ - if image_array is None: - raise ValueError("Input image is None") - - # 确保图像是numpy数组 - if not isinstance(image_array, np.ndarray): - raise ValueError("Input must be numpy array") - - # 检查图像维度 - if len(image_array.shape) != 3: - raise ValueError(f"Expected 3D image array, got {len(image_array.shape)}D") - - height, width, channels = image_array.shape - if channels != 3: - raise ValueError(f"Expected 3 channels, got {channels}") - - # 调整图像尺寸到模型要求的尺寸 - if height != self.img_size[1] or width != self.img_size[0]: - image_array = cv2.resize(image_array, tuple(self.img_size)) - - # 使用与训练时相同的预处理方式 - image_array = image_array.astype('float32') - image_array -= 127.5 - image_array *= 0.0078125 - image_array = np.transpose(image_array, (2, 0, 1)) # HWC -> CHW - - # 转换为tensor并添加batch维度 - image_tensor = torch.from_numpy(image_array).unsqueeze(0) - - return image_tensor - - def decode_prediction(self, logits): - """ - 解码模型预测结果 - 使用正确的CTC贪婪解码 - Args: - logits: 模型输出的logits [batch_size, num_classes, sequence_length] - Returns: - predicted_text: 预测的车牌号码 - """ - # 转换为numpy进行处理 - prebs = logits.cpu().detach().numpy() - preb = prebs[0, :, :] # 取第一个batch [num_classes, sequence_length] - - # 贪婪解码：对每个时间步选择最大概率的字符 - preb_label = [] - for j in range(preb.shape[1]): # 遍历每个时间步 - preb_label.append(np.argmax(preb[:, j], axis=0)) - - # CTC解码：去除重复字符和空白字符 - no_repeat_blank_label = [] - pre_c = preb_label[0] - - # 处理第一个字符 - if pre_c != len(CHARS) - 1: # 不是空白字符 - no_repeat_blank_label.append(pre_c) - - # 处理后续字符 - for c in preb_label: - if (pre_c == c) or (c == len(CHARS) - 1): # 重复字符或空白字符 - if c == len(CHARS) - 1: - pre_c = c - continue - no_repeat_blank_label.append(c) - pre_c = c - - # 转换为字符 - decoded_chars = [CHARS[idx] for idx in no_repeat_blank_label] - return ''.join(decoded_chars) - - def predict(self, image_array): - """ - 预测单张图像的车牌号码 - Args: - image_array: numpy数组格式的图像 - Returns: - prediction: 预测的车牌号码 - confidence: 预测置信度 - """ - try: - # 预处理图像 - image = self.preprocess_image(image_array) - if image is None: - return None, 0.0 - - image = image.to(self.device) - - # 模型推理 - with torch.no_grad(): - logits = self.model(image) - # logits shape: [batch_size, class_num, sequence_length] - - # 计算置信度（使用softmax后的最大概率平均值） - probs = torch.softmax(logits, dim=1) - max_probs = torch.max(probs, dim=1)[0] - confidence = torch.mean(max_probs).item() - - # 解码预测结果 - prediction = self.decode_prediction(logits) - - return prediction, confidence - - except Exception as e: - print(f"预测图像失败: {e}") - return None, 0.0 - -# 全局变量 -lpr_model = None - -def LPRNinitialize_model(): - """ - 初始化LPRNet模型 - - 返回: - bool: 初始化是否成功 - """ - global lpr_model - - try: - # 模型权重文件路径 - model_path = os.path.join(os.path.dirname(__file__), 'LPRNet__iteration_74000.pth') - - # 创建推理对象 - lpr_model = LPRNetInference(model_path) - - print("LPRNet模型初始化完成") - return True - - except Exception as e: - print(f"LPRNet模型初始化失败: {e}") - import traceback - traceback.print_exc() - return False - -def LPRNmodel_predict(image_array): - """ - LPRNet车牌号识别接口函数 - - 参数: - image_array: numpy数组格式的车牌图像，已经过矫正处理 - - 返回: - list: 包含最多8个字符的列表，代表车牌号的每个字符 - 例如: ['京', 'A', '1', '2', '3', '4', '5'] (蓝牌7位) - ['京', 'A', 'D', '1', '2', '3', '4', '5'] (绿牌8位) - """ - global lpr_model - - if lpr_model is None: - print("LPRNet模型未初始化，请先调用LPRNinitialize_model()") - return ['待', '识', '别', '0', '0', '0', '0', '0'] - - try: - # 预测车牌号 - predicted_text, confidence = lpr_model.predict(image_array) - - if predicted_text is None: - print("LPRNet识别失败") - return ['识', '别', '失', '败', '0', '0', '0', '0'] - - print(f"LPRNet识别结果: {predicted_text}, 置信度: {confidence:.3f}") - - # 将字符串转换为字符列表 - char_list = list(predicted_text) - - # 确保返回至少7个字符，最多8个字符 - if len(char_list) < 7: - # 如果识别结果少于7个字符，用'0'补齐到7位 - char_list.extend(['0'] * (7 - len(char_list))) - elif len(char_list) > 8: - # 如果识别结果多于8个字符，截取前8个 - char_list = char_list[:8] - - # 如果是7位，补齐到8位以保持接口一致性（第8位用空字符或占位符） - if len(char_list) == 7: - char_list.append('') # 添加空字符作为第8位占位符 - - return char_list - - except Exception as e: - print(f"LPRNet识别失败: {e}") - import traceback - traceback.print_exc() - return ['识', '别', '失', '败', '0', '0', '0', '0'] \ No newline at end of file diff --git a/gate_control.py b/gate_control.py new file mode 100644 index 0000000..f9a4582 --- /dev/null +++ b/gate_control.py @@ -0,0 +1,251 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +道闸控制模块 +负责与Hi3861设备通信，控制道闸开关 +""" + +import socket +import json +import time +from datetime import datetime, timedelta +from PyQt5.QtCore import QObject, pyqtSignal, QThread + + +class GateControlThread(QThread): + """道闸控制线程，用于异步发送命令""" + command_sent = pyqtSignal(str, bool) # 信号：命令内容，是否成功 + + def __init__(self, ip, port, command): + super().__init__() + self.ip = ip + self.port = port + self.command = command + + def run(self): + """发送命令到Hi3861设备""" + try: + # 创建UDP socket + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + + # 发送命令 + json_command = json.dumps(self.command, ensure_ascii=False) + sock.sendto(json_command.encode('utf-8'), (self.ip, self.port)) + + # 发出成功信号 + self.command_sent.emit(json_command, True) + + except Exception as e: + # 发出失败信号 + self.command_sent.emit(f"发送失败: {e}", False) + finally: + sock.close() + + +class GateController(QObject): + """道闸控制器""" + + # 信号 + log_message = pyqtSignal(str) # 日志消息 + gate_opened = pyqtSignal(str) # 道闸打开信号，附带车牌号 + + def __init__(self, ip="192.168.43.12", port=8081): + super().__init__() + self.ip = ip + self.port = port + self.last_pass_times = {} # 记录车牌上次通过时间 + self.thread_pool = [] # 线程池 + + def send_command(self, cmd, text=""): + """ + 发送命令到道闸 + + 参数: + cmd: 命令类型 (1-4) + text: 显示文本 + + 返回: + bool: 是否发送成功 + """ + # 创建JSON命令 + command = { + "cmd": cmd, + "text": text + } + + # 创建并启动线程发送命令 + thread = GateControlThread(self.ip, self.port, command) + thread.command_sent.connect(self.on_command_sent) + thread.start() + self.thread_pool.append(thread) + + # 记录日志 + cmd_desc = { + 1: "自动开闸(10秒后关闭)", + 2: "手动开闸", + 3: "手动关闸", + 4: "仅显示信息" + } + self.log_message.emit(f"发送命令: {cmd_desc.get(cmd, '未知命令')} - {text}") + + return True + + def on_command_sent(self, message, success): + """命令发送结果处理""" + if success: + self.log_message.emit(f"命令发送成功: {message}") + else: + self.log_message.emit(f"命令发送失败: {message}") + + def auto_open_gate(self, plate_number): + """ + 自动开闸（检测到白名单车牌时调用） + + 参数: + plate_number: 车牌号 + """ + # 获取当前时间 + current_time = datetime.now() + time_diff_str = "" + + # 检查是否是第一次通行 + if plate_number in self.last_pass_times: + # 第二次或更多次通行，计算时间差 + last_time = self.last_pass_times[plate_number] + time_diff = current_time - last_time + + # 格式化时间差 + total_seconds = int(time_diff.total_seconds()) + minutes = total_seconds // 60 + seconds = total_seconds % 60 + + if minutes > 0: + time_diff_str = f" {minutes}min{seconds}sec" + else: + time_diff_str = f" {seconds}sec" + + # 计算时间差后清空之前记录的时间点 + del self.last_pass_times[plate_number] + log_msg = f"检测到白名单车牌: {plate_number}，自动开闸{time_diff_str}，已清空时间记录" + else: + # 第一次通行，只记录时间，不计算时间差 + self.last_pass_times[plate_number] = current_time + log_msg = f"检测到白名单车牌: {plate_number}，首次通行，已记录时间" + + # 发送开闸命令 + display_text = f"{plate_number} 通行{time_diff_str}" + self.send_command(1, display_text) + + # 发出信号 + self.gate_opened.emit(plate_number) + + # 记录日志 + self.log_message.emit(log_msg) + + def manual_open_gate(self): + """手动开闸""" + self.send_command(2, "") + self.log_message.emit("手动开闸") + + def manual_close_gate(self): + """手动关闸""" + self.send_command(3, "") + self.log_message.emit("手动关闸") + + def display_message(self, text): + """仅显示信息，不控制道闸""" + self.send_command(4, text) + self.log_message.emit(f"显示信息: {text}") + + def deny_access(self, plate_number): + """ + 拒绝通行（检测到非白名单车牌时调用） + + 参数: + plate_number: 车牌号 + """ + self.send_command(4, f"{plate_number} 禁止通行") + self.log_message.emit(f"检测到非白名单车牌: {plate_number}，拒绝通行") + + +class WhitelistManager(QObject): + """白名单管理器""" + + # 信号 + whitelist_changed = pyqtSignal(list) # 白名单变更信号 + + def __init__(self): + super().__init__() + self.whitelist = [] # 白名单车牌列表 + + def add_plate(self, plate_number): + """ + 添加车牌到白名单 + + 参数: + plate_number: 车牌号 + + 返回: + bool: 是否添加成功 + """ + if not plate_number or plate_number in self.whitelist: + return False + + self.whitelist.append(plate_number) + self.whitelist_changed.emit(self.whitelist.copy()) + return True + + def remove_plate(self, plate_number): + """ + 从白名单移除车牌 + + 参数: + plate_number: 车牌号 + + 返回: + bool: 是否移除成功 + """ + if plate_number in self.whitelist: + self.whitelist.remove(plate_number) + self.whitelist_changed.emit(self.whitelist.copy()) + return True + return False + + def edit_plate(self, old_plate, new_plate): + """ + 编辑白名单中的车牌 + + 参数: + old_plate: 原车牌号 + new_plate: 新车牌号 + + 返回: + bool: 是否编辑成功 + """ + if old_plate in self.whitelist and new_plate not in self.whitelist: + index = self.whitelist.index(old_plate) + self.whitelist[index] = new_plate + self.whitelist_changed.emit(self.whitelist.copy()) + return True + return False + + def is_whitelisted(self, plate_number): + """ + 检查车牌是否在白名单中 + + 参数: + plate_number: 车牌号 + + 返回: + bool: 是否在白名单中 + """ + return plate_number in self.whitelist + + def get_whitelist(self): + """获取白名单副本""" + return self.whitelist.copy() + + def clear_whitelist(self): + """清空白名单""" + self.whitelist.clear() + self.whitelist_changed.emit(self.whitelist.copy()) \ No newline at end of file diff --git a/main.py b/main.py index 9b26bf7..1616a99 100644 --- a/main.py +++ b/main.py @@ -1,13 +1,16 @@ import sys import os import cv2 +import time import numpy as np from collections import defaultdict, deque from PyQt5.QtWidgets import QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, \ - QFileDialog, QFrame, QScrollArea, QComboBox -from PyQt5.QtCore import QTimer, Qt, pyqtSignal, QThread + QFileDialog, QFrame, QScrollArea, QComboBox, QListWidget, QListWidgetItem, QLineEdit, QMessageBox, QDialog, \ + QDialogButtonBox, QFormLayout, QTextEdit +from PyQt5.QtCore import QTimer, Qt, pyqtSignal, QThread, QDateTime from PyQt5.QtGui import QImage, QPixmap, QFont, QPainter, QPen, QColor from yolopart.detector import LicensePlateYOLO +from gate_control import GateController, WhitelistManager #选择使用哪个模块 # from LPRNET_part.lpr_interface import LPRNmodel_predict @@ -19,6 +22,38 @@ from yolopart.detector import LicensePlateYOLO # from CRNN_part.crnn_interface import LPRNmodel_predict # from CRNN_part.crnn_interface import LPRNinitialize_model +class PlateInputDialog(QDialog): + """车牌输入对话框""" + def __init__(self, title, default_text=""): + super().__init__() + self.setWindowTitle(title) + self.setFixedSize(300, 100) + self.setWindowModality(Qt.ApplicationModal) + + layout = QVBoxLayout() + + # 车牌输入框 + self.plate_input = QLineEdit() + self.plate_input.setPlaceholderText("请输入车牌号") + self.plate_input.setText(default_text) + self.plate_input.setMaxLength(10) + + # 按钮 + buttons = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel) + buttons.accepted.connect(self.accept) + buttons.rejected.connect(self.reject) + + layout.addWidget(QLabel("车牌号:")) + layout.addWidget(self.plate_input) + layout.addWidget(buttons) + + self.setLayout(layout) + + def get_plate_number(self): + """获取输入的车牌号""" + return self.plate_input.text().strip() + + class PlateStabilizer: """车牌识别结果稳定器""" @@ -449,13 +484,27 @@ class MainWindow(QMainWindow): stability_frames=5 # 需要5帧稳定 ) + # 初始化道闸控制器和白名单管理器 + self.gate_controller = GateController() + self.whitelist_manager = WhitelistManager() + + # 记录车牌首次检测时间和上次发送指令时间 + self.plate_first_detected = {} # 记录车牌首次检测时间 + self.plate_last_command_time = {} # 记录车牌上次发送指令时间 + self.init_ui() self.init_detector() self.init_camera() self.init_video() + self.init_gate_control() # 初始化默认识别方法（CRNN）的模型 self.change_recognition_method(self.current_recognition_method) + + # 设置定时器，每30秒清理一次过期的车牌记录 + self.cleanup_timer = QTimer(self) + self.cleanup_timer.timeout.connect(self.cleanup_plate_records) + self.cleanup_timer.start(30000) # 30秒 def init_ui(self): @@ -524,6 +573,156 @@ class MainWindow(QMainWindow): right_frame.setStyleSheet("QFrame { background-color: #fafafa; border: 2px solid #ddd; }") right_layout = QVBoxLayout(right_frame) + # 道闸控制区域 + gate_frame = QFrame() + gate_frame.setFrameStyle(QFrame.StyledPanel) + gate_frame.setStyleSheet("QFrame { background-color: #f0f8ff; border: 1px solid #b0d4f1; border-radius: 5px; }") + gate_layout = QVBoxLayout(gate_frame) + + # 道闸控制标题 + gate_title = QLabel("道闸控制") + gate_title.setAlignment(Qt.AlignCenter) + gate_title.setFont(QFont("Arial", 14, QFont.Bold)) + gate_title.setStyleSheet("QLabel { color: #1976d2; padding: 5px; }") + + # 道闸控制按钮 + gate_button_layout = QHBoxLayout() + self.open_gate_button = QPushButton("手动开闸") + self.close_gate_button = QPushButton("手动关闸") + self.open_gate_button.clicked.connect(self.manual_open_gate) + self.close_gate_button.clicked.connect(self.manual_close_gate) + + # 设置道闸按钮样式 + gate_button_style = """ + QPushButton { + background-color: #4CAF50; + color: white; + border: none; + padding: 8px 16px; + border-radius: 4px; + font-weight: bold; + } + QPushButton:hover { + background-color: #45a049; + } + QPushButton:pressed { + background-color: #3d8b40; + } + """ + self.open_gate_button.setStyleSheet(gate_button_style) + + close_button_style = """ + QPushButton { + background-color: #f44336; + color: white; + border: none; + padding: 8px 16px; + border-radius: 4px; + font-weight: bold; + } + QPushButton:hover { + background-color: #d32f2f; + } + QPushButton:pressed { + background-color: #b71c1c; + } + """ + self.close_gate_button.setStyleSheet(close_button_style) + + gate_button_layout.addWidget(self.open_gate_button) + gate_button_layout.addWidget(self.close_gate_button) + + # 白名单管理区域 + whitelist_layout = QVBoxLayout() + whitelist_label = QLabel("车牌白名单") + whitelist_label.setFont(QFont("Arial", 12, QFont.Bold)) + whitelist_label.setStyleSheet("QLabel { color: #333; padding: 5px; }") + + # 白名单按钮 + whitelist_button_layout = QHBoxLayout() + self.add_plate_button = QPushButton("添加车牌") + self.edit_plate_button = QPushButton("编辑车牌") + self.delete_plate_button = QPushButton("删除车牌") + self.add_plate_button.clicked.connect(self.add_plate_to_whitelist) + self.edit_plate_button.clicked.connect(self.edit_plate_in_whitelist) + self.delete_plate_button.clicked.connect(self.delete_plate_from_whitelist) + + # 设置白名单按钮样式 + whitelist_button_style = """ + QPushButton { + background-color: #2196F3; + color: white; + border: none; + padding: 6px 12px; + border-radius: 4px; + font-weight: bold; + font-size: 11px; + } + QPushButton:hover { + background-color: #1976D2; + } + QPushButton:pressed { + background-color: #0D47A1; + } + """ + self.add_plate_button.setStyleSheet(whitelist_button_style) + self.edit_plate_button.setStyleSheet(whitelist_button_style) + self.delete_plate_button.setStyleSheet(whitelist_button_style) + + whitelist_button_layout.addWidget(self.add_plate_button) + whitelist_button_layout.addWidget(self.edit_plate_button) + whitelist_button_layout.addWidget(self.delete_plate_button) + + # 白名单列表 + self.whitelist_list = QListWidget() + self.whitelist_list.setMaximumHeight(120) + self.whitelist_list.setStyleSheet(""" + QListWidget { + border: 1px solid #ddd; + background-color: white; + border-radius: 4px; + padding: 5px; + } + QListWidget::item { + padding: 5px; + border-bottom: 1px solid #eee; + } + QListWidget::item:selected { + background-color: #e3f2fd; + color: #1976d2; + } + """) + + # 调试日志区域 + log_label = QLabel("调试日志") + log_label.setFont(QFont("Arial", 10, QFont.Bold)) + log_label.setStyleSheet("QLabel { color: #333; padding: 5px; }") + + self.log_text = QTextEdit() + self.log_text.setMaximumHeight(100) + self.log_text.setReadOnly(True) + self.log_text.setStyleSheet(""" + QTextEdit { + border: 1px solid #ddd; + background-color: #f9f9f9; + border-radius: 4px; + padding: 5px; + font-family: 'Consolas', 'Courier New', monospace; + font-size: 10px; + } + """) + + # 添加到道闸控制布局 + whitelist_layout.addWidget(whitelist_label) + whitelist_layout.addLayout(whitelist_button_layout) + whitelist_layout.addWidget(self.whitelist_list) + + gate_layout.addWidget(gate_title) + gate_layout.addLayout(gate_button_layout) + gate_layout.addLayout(whitelist_layout) + gate_layout.addWidget(log_label) + gate_layout.addWidget(self.log_text) + # 标题 title_label = QLabel("检测结果") title_label.setAlignment(Qt.AlignCenter) @@ -576,6 +775,7 @@ class MainWindow(QMainWindow): self.current_method_label.setFont(QFont("Arial", 9)) self.current_method_label.setStyleSheet("QLabel { color: #666; padding: 5px; }") + right_layout.addWidget(gate_frame) right_layout.addWidget(title_label) right_layout.addLayout(method_layout) right_layout.addWidget(self.count_label) @@ -1014,6 +1214,13 @@ class MainWindow(QMainWindow): # 更新存储的结果 self.last_plate_results = stable_results + + # 处理道闸控制逻辑 + for result in stable_results: + plate_number = result.get('plate_number', '') + if plate_number and plate_number != "识别失败": + # 调用道闸控制逻辑 + self.process_gate_control(plate_number) # 清理旧的车牌记录 current_plate_ids = [result['id'] for result in stable_results] @@ -1111,6 +1318,151 @@ class MainWindow(QMainWindow): if self.current_frame is not None: self.process_frame(self.current_frame) + def init_gate_control(self): + """初始化道闸控制功能""" + # 更新白名单列表显示 + self.update_whitelist_display() + + # 添加初始日志 + self.add_log("道闸控制系统已初始化") + + # GateController的IP地址在初始化时已设置，默认为192.168.43.12 + + def manual_open_gate(self): + """手动开闸""" + self.gate_controller.manual_open_gate() + self.add_log("手动开闸指令已发送") + + def manual_close_gate(self): + """手动关闸""" + self.gate_controller.manual_close_gate() + self.add_log("手动关闸指令已发送") + + def cleanup_plate_records(self): + """清理过期的车牌记录""" + current_time = time.time() + + # 清理超过30秒的首次检测记录 + expired_plates = [] + for plate, first_time in self.plate_first_detected.items(): + if current_time - first_time > 30: + expired_plates.append(plate) + + for plate in expired_plates: + del self.plate_first_detected[plate] + self.add_log(f"清理过期的首次检测记录: {plate}") + + # 清理超过1小时的指令发送记录 + expired_commands = [] + for plate, last_time in self.plate_last_command_time.items(): + if current_time - last_time > 3600: + expired_commands.append(plate) + + for plate in expired_commands: + del self.plate_last_command_time[plate] + self.add_log(f"清理过期的指令记录: {plate}") + + def add_plate_to_whitelist(self): + """添加车牌到白名单""" + dialog = PlateInputDialog("添加车牌", "") + if dialog.exec_() == QDialog.Accepted: + plate_number = dialog.get_plate_number() + if plate_number: + self.whitelist_manager.add_plate(plate_number) + self.update_whitelist_display() + self.add_log(f"已添加车牌到白名单: {plate_number}") + + def edit_plate_in_whitelist(self): + """编辑白名单中的车牌""" + current_item = self.whitelist_list.currentItem() + if not current_item: + QMessageBox.warning(self, "提示", "请先选择要编辑的车牌") + return + + old_plate = current_item.text() + dialog = PlateInputDialog("编辑车牌", old_plate) + if dialog.exec_() == QDialog.Accepted: + new_plate = dialog.get_plate_number() + if new_plate and new_plate != old_plate: + self.whitelist_manager.remove_plate(old_plate) + self.whitelist_manager.add_plate(new_plate) + self.update_whitelist_display() + self.add_log(f"已修改车牌: {old_plate} -> {new_plate}") + + def delete_plate_from_whitelist(self): + """从白名单中删除车牌""" + current_item = self.whitelist_list.currentItem() + if not current_item: + QMessageBox.warning(self, "提示", "请先选择要删除的车牌") + return + + plate = current_item.text() + reply = QMessageBox.question(self, "确认", f"确定要删除车牌 {plate} 吗？", + QMessageBox.Yes | QMessageBox.No) + + if reply == QMessageBox.Yes: + self.whitelist_manager.remove_plate(plate) + self.update_whitelist_display() + self.add_log(f"已从白名单删除车牌: {plate}") + + def update_whitelist_display(self): + """更新白名单列表显示""" + self.whitelist_list.clear() + for plate in self.whitelist_manager.get_whitelist(): + self.whitelist_list.addItem(plate) + + def add_log(self, message): + """添加日志消息""" + current_time = QDateTime.currentDateTime().toString("hh:mm:ss") + log_message = f"[{current_time}] {message}" + self.log_text.append(log_message) + # 限制日志行数，避免内存占用过多 + if self.log_text.document().blockCount() > 100: + cursor = self.log_text.textCursor() + cursor.movePosition(cursor.Start) + cursor.select(cursor.BlockUnderCursor) + cursor.removeSelectedText() + cursor.deleteChar() # 删除换行符 + + def process_gate_control(self, plate_number): + """处理道闸控制逻辑""" + # 检查车牌是否在白名单中 + if self.whitelist_manager.is_whitelisted(plate_number): + current_time = time.time() + + # 检查是否在10秒内已发送过指令 + if plate_number in self.plate_last_command_time: + time_since_last_command = current_time - self.plate_last_command_time[plate_number] + if time_since_last_command < 10: # 10秒内不再发送指令 + self.add_log(f"车牌 {plate_number} 在10秒内已发送过指令，跳过") + return + + # 记录车牌首次检测时间 + if plate_number not in self.plate_first_detected: + self.plate_first_detected[plate_number] = current_time + self.add_log(f"车牌 {plate_number} 首次检测，等待2秒稳定确认") + return + + # 检查是否已稳定2秒 + time_since_first_detected = current_time - self.plate_first_detected[plate_number] + if time_since_first_detected >= 2: # 稳定2秒后发送指令 + # 使用GateController的auto_open_gate方法，它会自动处理时间差 + self.gate_controller.auto_open_gate(plate_number) + self.add_log(f"车牌 {plate_number} 验证通过，已发送开闸指令") + + # 更新上次发送指令时间 + self.plate_last_command_time[plate_number] = current_time + # 清除首次检测时间，以便下次重新检测 + if plate_number in self.plate_first_detected: + del self.plate_first_detected[plate_number] + else: + # 还未稳定2秒，继续等待 + self.add_log(f"车牌 {plate_number} 检测中，已等待 {time_since_first_detected:.1f} 秒") + else: + # 不在白名单中，发送禁行指令 + self.gate_controller.deny_access(plate_number) + self.add_log(f"车牌 {plate_number} 不在白名单中，已发送禁行指令") + def closeEvent(self, event): """窗口关闭事件""" if self.camera_thread and self.camera_thread.running: diff --git a/simple_client.py b/simple_client.py new file mode 100644 index 0000000..65f85d5 --- /dev/null +++ b/simple_client.py @@ -0,0 +1,58 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +简单的UDP客户端程序 +向Hi3861设备发送JSON命令 +""" + +import socket +import json +import time + +def send_command(): + """发送命令到Hi3861设备""" + # 目标设备信息 + target_ip = "192.168.43.12" + target_port = 8081 + + #cmd为1，道闸打开十秒后关闭,oled显示字符串信息（默认使用及cmd为4） + #cmd为2，道闸舵机向打开方向旋转90度，oled上不显示（仅在qt界面手动开闸时调用） + #cmd为3，道闸舵机向关闭方向旋转90度，oled上不显示（仅在qt界面手动关闸时调用） + #cmd为4，oled显示字符串信息，道闸舵机不旋转 + + # 创建JSON命令 + command = { + "cmd": 1, + "text": "沪AAAAAA 通行" + } + + json_command = json.dumps(command, ensure_ascii=False) + + try: + # 创建UDP socket + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + + # 发送命令 + print(f"正在向 {target_ip}:{target_port} 发送命令...") + print(f"命令内容: {json_command}") + + sock.sendto(json_command.encode('utf-8'), (target_ip, target_port)) + + print("命令发送成功！") + print("设备将执行以下操作：") + print("1. 顺时针旋转舵机90度") + print("2. 在OLED屏幕上显示：沪AAAAAA") + print("3. 等待10秒") + print("4. 逆时针旋转舵机90度") + print("5. 清空OLED屏幕") + + except Exception as e: + print(f"发送命令失败: {e}") + finally: + sock.close() + +if __name__ == "__main__": + print("Hi3861 简单客户端程序") + print("=" * 30) + send_command() + print("程序结束") \ No newline at end of file