从头开始开发基于虹软SDK的人脸识别考勤系统(python+RTSP开源)（四）

终于抽出时间来了，代码继续。

一、摄像头部分

摄像头服务的初始化

# ==================== 摄像头服务 ====================
class CameraWorker(QThread):
    frame_ready = pyqtSignal(np.ndarray)
    error_occurred = pyqtSignal(str)

    def __init__(self, source=0):
        super().__init__()
        self.source = source
        self.running = True
        # 确保 camera_mutex 被正确初始化
        self.camera_mutex = QMutex()

    def run(self):

        self.running = True  # 重置运行状态
        cap = cv2.VideoCapture(self.source)
        if not cap.isOpened():
            self.error_occurred.emit(f"无法打开摄像头：{self.source}")
            print(f"无法打开摄像头：{self.source}")
            return

        try:
            while self.running:
                # 加锁
                self.camera_mutex.lock()
                ret, frame = cap.read()
                # 解锁
                self.camera_mutex.unlock()
                if ret:
                    rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
                    self.frame_ready.emit(rgb_frame)
                else:
                    self.error_occurred.emit("视频流中断")
                    print("视频流中断")
                    break
        finally:
            # 加锁释放资源
            self.camera_mutex.lock()
            cap.release()
            # 解锁
            self.camera_mutex.unlock()

    def stop(self):
        self.running = False
        self.wait(1000)
        if hasattr(self, 'cap'):
            # 加锁释放资源
            self.camera_mutex.lock()
            self.cap.release()
            # 解锁
            self.camera_mutex.unlock()

我这次是加上了RTSP协议的监控摄像头，在配置界面上可以手工添加和启用，其实还没有写完整，应该选择启用RTSP后，自动重启程序。

def _init_camera(self):
        if self.camera_worker:
            self.camera_worker.stop()
            self.camera_worker.quit()
            self.camera_worker.wait()

        cursor = self.db.conn.cursor()
        cursor.execute('SELECT camera_type, rtsp_url, is_enabled, is_connected FROM camera_config WHERE id = 1')
        result = cursor.fetchone()
        if result:
            camera_type, rtsp_url, is_enabled, is_connected = result
            if camera_type == 'rtsp' and is_enabled and is_connected:
                self.camera_worker = CameraWorker(rtsp_url)
            else:
                self.camera_worker = CameraWorker(0)
        else:
            self.camera_worker = CameraWorker(0)

        self.camera_worker.frame_ready.connect(self.update_frame)
        self.camera_worker.error_occurred.connect(self.show_camera_error)
        if not self.camera_worker.isRunning():
            self.camera_worker.start()

def show_camera_error(self, error_msg):
        reply = QMessageBox.question(
            self,
            "摄像头错误",
            f"{error_msg}\n是否尝试重新连接？",
            QMessageBox.Yes | QMessageBox.No
        )

        if reply == QMessageBox.Yes:
            # 停止当前摄像头线程
            if hasattr(self, 'camera_worker'):
                self.camera_worker.stop()
                self.camera_worker.quit()
                self.camera_worker.wait()
            self.init_camera()  # 重新初始化摄像头
        else:
            self.camera_worker.stop()
            self.video_label.setText("摄像头已禁用")

def init_ui(self):
        self.setWindowTitle("摄像头配置")
        layout = QVBoxLayout()

        form = QFormLayout()
        self.camera_type_combo = QComboBox()
        self.camera_type_combo.addItems(['USB', 'RTSP'])
        self.rtsp_url_input = QLineEdit()
        self.enable_rtsp_checkbox = QCheckBox("启用RTSP摄像头")
        self.test_connection_button = QPushButton("测试连接")
        self.test_connection_button.clicked.connect(self.test_rtsp_connection)

        form.addRow("摄像头类型：", self.camera_type_combo)
        form.addRow("RTSP URL：", self.rtsp_url_input)
        form.addRow("", self.enable_rtsp_checkbox)
        form.addRow("", self.test_connection_button)

        btn_box = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel)
        btn_box.accepted.connect(self.save_config)
        btn_box.rejected.connect(self.reject)

        layout.addLayout(form)
        layout.addWidget(btn_box)
        self.setLayout(layout)
        self.load_config()

读取摄像头配置

def load_config(self):
        cursor = self.db.conn.cursor()
        cursor.execute('SELECT camera_type, rtsp_url, is_enabled, is_connected FROM camera_config WHERE id = 1')
        result = cursor.fetchone()
        if result:
            camera_type, rtsp_url, is_enabled, is_connected = result
            self.camera_type_combo.setCurrentText('RTSP' if camera_type == 'rtsp' else 'USB')
            self.rtsp_url_input.setText(rtsp_url)
            self.enable_rtsp_checkbox.setChecked(is_enabled)
            self.test_connection_button.setEnabled(is_connected)

保存摄像头配置并启用

def save_config(self):
        camera_type = 'rtsp' if self.camera_type_combo.currentText() == 'RTSP' else 'usb'
        rtsp_url = self.rtsp_url_input.text() if camera_type == 'rtsp' else None
        is_enabled = self.enable_rtsp_checkbox.isChecked()
        cursor = self.db.conn.cursor()
        cursor.execute('''
            INSERT OR REPLACE INTO camera_config (id, camera_type, rtsp_url, is_enabled, is_connected)
            VALUES (1,?,?,?,?)
        ''', (camera_type, rtsp_url, is_enabled, self.test_connection_button.isEnabled()))
        self.db.conn.commit()
        self.init_camera()  # 调用初始化摄像头的方法
        self.accept()

        # 停止当前摄像头线程
        if self.camera_worker:
            self.camera_worker.stop()
            self.camera_worker.quit()
            self.camera_worker.wait()

        # 重新初始化摄像头
        self.init_camera()

        self.accept()

在RTSP摄像头界面增加测试连接功能

def test_rtsp_connection(self):
        import cv2
        rtsp_url = self.rtsp_url_input.text()
        cap = cv2.VideoCapture(rtsp_url)
        if cap.isOpened():
            cap.release()
            self.test_connection_button.setEnabled(True)
            QMessageBox.information(self, "连接成功", "RTSP摄像头连接可用")
        else:
            self.test_connection_button.setEnabled(False)
            QMessageBox.warning(self, "连接失败", "RTSP摄像头连接不可用")

    def init_camera(self):
        self.main_window._init_camera()

二、界面部分

界面部分上面也都已经介绍了，使用的是PyQt5，直接上代码吧。

1、主界面部分：

def init_ui(self):
        self.setWindowTitle("人脸识别考勤(FaceAttV2)DEV-半熟的皮皮虾")
        self.setWindowIcon(QIcon("icon.png"))
        self.resize(1280, 720)
        central_widget = QWidget()
        self.setCentralWidget(central_widget)

        # 主布局，使用垂直布局
        main_layout = QVBoxLayout(central_widget)

        # 上半部分布局，使用水平布局，包含摄像头显示和考勤列表
        upper_layout = QHBoxLayout()

        # 左侧摄像头显示区域
        left_widget = QWidget()
        left_layout = QVBoxLayout(left_widget)

        self.video_label = QLabel()
        self.video_label.setAlignment(Qt.AlignCenter)
        self.video_label.setScaledContents(True)  # 允许按比例缩放内容
        left_layout.addWidget(self.video_label, 85)

        upper_layout.addWidget(left_widget, 7)  # 左侧占 7 份

        # 右侧已考勤人员列表区域
        right_widget = QWidget()
        right_layout = QVBoxLayout(right_widget)
        right_layout.setAlignment(Qt.AlignCenter)  # 设置右侧布局居中对齐

        # 添加已考勤列表标题
        self.title_label = QLabel("已考勤列表")
        self.title_label.setObjectName("titleLabel")  # 设置对象名称
        self.title_label.setFont(QFont("微软雅黑", 19, QFont.Bold))  # 设置标题字体和加粗
        self.title_label.setAlignment(Qt.AlignCenter)  # 标题居中显示
        right_layout.addWidget(self.title_label)

        # 添加伸缩空间，将考勤列表推到中间
        right_layout.addStretch()

        self.attendance_label = QLabel()
        self.attendance_label.setObjectName("attendanceLabel")  # 设置对象名称
        self.attendance_label.setFont(QFont("宋体", 15, QFont.Bold))
        right_layout.addWidget(self.attendance_label, alignment=Qt.AlignCenter)

        # 添加伸缩空间，将考勤列表推到中间
        right_layout.addStretch()

        upper_layout.addWidget(right_widget, 3)  # 右侧占 3 份

        main_layout.addLayout(upper_layout)

        # 底部布局，使用水平布局，包含日期和打卡提示
        bottom_layout = QHBoxLayout()

        # 日期显示部分，包含时间和农历日期
        date_panel = QWidget()
        date_layout = QHBoxLayout(date_panel)
        date_layout.setContentsMargins(0, 0, 0, 0)

        self.time_label = QLabel()
        self.time_label.setObjectName("timeLabel")  # 设置对象名称
        self.time_label.setFont(QFont("微软雅黑", 19, QFont.Bold))
        date_layout.addWidget(self.time_label)

        self.lunar_date_label = QLabel()  # 新增农历日期显示标签
        self.lunar_date_label.setObjectName("lunarDateLabel")  # 设置对象名称
        self.lunar_date_label.setFont(QFont("微软雅黑", 19, QFont.Bold))
        date_layout.addWidget(self.lunar_date_label)

        bottom_layout.addWidget(date_panel)

        # 打卡提示部分
        self.status_label = QLabel("等待识别...")
        self.status_label.setObjectName("statusLabel")  # 设置对象名称
        # 设置更大的字体大小
        self.status_label.setFont(QFont("微软雅黑", 30, QFont.Bold))
        self.status_label.setAlignment(Qt.AlignCenter)  # 居中显示
        bottom_layout.addStretch()  # 添加弹簧，将打卡提示推到右侧
        bottom_layout.addWidget(self.status_label, alignment=Qt.AlignRight)

        main_layout.addLayout(bottom_layout)

        self.timer = QTimer()
        self.timer.timeout.connect(self.update_display)
        self.timer.start(1000)


        self.create_menu()
        # 修改样式表，使用对象名称选择器调整特定 QLabel 的颜色
        self.setStyleSheet("""
            QMainWindow { background-color: #2D2D2D; }
            QMenuBar { background-color: #3D3D3D; color: white; }
            QMenu { background-color: #3D3D3D; color: white; }
            QTableWidget { background-color: #404040; color: white; }
            QDialog { color: black; }
            QMessageBox { color: black; }
            #titleLabel { color: white; }
            #attendanceLabel { color: white; }
            #timeLabel { color: white; }
            #lunarDateLabel { color: white; }
            #statusLabel { color: white; font-size: 30px; font-family: 微软雅黑; font-weight: bold; }
        """)

2、菜单部分：

def create_menu(self):
        file_menu = self.menuBar().addMenu("文件(&F)")
        backup_action = QAction("数据备份", self)
        backup_action.triggered.connect(self.backup_data)
        file_menu.addAction(backup_action)

        manage_menu = self.menuBar().addMenu("管理(&M)")
        staff_action = QAction("人员管理", self)
        staff_action.triggered.connect(self.show_staff_admin)
        manage_menu.addAction(staff_action)

        view_menu = self.menuBar().addMenu("视图(&V)")
        fullscreen_action = QAction("全屏切换", self, checkable=True)
        fullscreen_action.triggered.connect(self.toggle_fullscreen)
        view_menu.addAction(fullscreen_action)

        config_menu = self.menuBar().addMenu("配置(&C)")
        algo_action = QAction("算法设置", self)
        algo_action.triggered.connect(self.show_algorithm_config)
        config_menu.addAction(algo_action)

        camera_config_action = QAction("摄像头配置", self)
        camera_config_action.triggered.connect(self.show_camera_config)
        config_menu.addAction(camera_config_action)

        # 新增考勤查询菜单
        query_menu = self.menuBar().addMenu("考勤查询(&Q)")
        query_action = QAction("考勤查询", self)
        query_action.triggered.connect(self.show_attendance_query)
        query_menu.addAction(query_action)

3、子菜单部分：

首先是识别算法的选择：

def show_algorithm_config(self):
        dialog = QDialog(self)
        layout = QFormLayout()

        # 算法选择
        algo_combo = QComboBox()
        algo_combo.addItems(['arcsoft'])
        algo_combo.setCurrentText(config.face_algorithm)

        # 虹软配置
        appid_edit = QLineEdit(config.arcsoft_appid.decode())
        sdkkey_edit = QLineEdit(config.arcsoft_sdkkey.decode())
        lib_path_edit = QLineEdit(config.arcsoft_lib_path)

        layout.addRow("识别算法:", algo_combo)
        layout.addRow("虹软APPID:", appid_edit)
        layout.addRow("虹软SDKKEY:", sdkkey_edit)
        layout.addRow("SDK路径:", lib_path_edit)

        btn_box = QDialogButtonBox(QDialogButtonBox.Save | QDialogButtonBox.Cancel)
        btn_box.accepted.connect(lambda: self.save_config(
            algo_combo.currentText(),
            appid_edit.text(),
            sdkkey_edit.text(),
            lib_path_edit.text()
        ))
        btn_box.rejected.connect(dialog.reject)

        layout.addRow(btn_box)
        dialog.setLayout(layout)
        dialog.exec_()

摄像头的在上面说了，这里就不重复了；

考勤结果查询部分，做了个判断，加了个标识，也没啥其他的；

功能界面部分：

def init_ui(self):
        self.setWindowTitle("考勤查询")
        self.setGeometry(100, 100, 400, 300)

        layout = QFormLayout()

        self.query_type_combo = QComboBox()
        self.query_type_combo.addItems(["个人某天考勤", "个人时间区间考勤", "某天所有人员考勤", "某时间段所有人员考勤"])
        self.query_type_combo.currentIndexChanged.connect(self.update_inputs)

        self.staff_name_edit = QLineEdit()
        self.staff_name_edit.setPlaceholderText("输入员工姓名")

        self.date_edit = QDateEdit()
        self.date_edit.setDisplayFormat("yyyy-MM-dd")
        current_date = QDate.currentDate()
        self.date_edit.setDate(current_date)

        self.start_date_edit = QDateEdit()
        self.start_date_edit.setDisplayFormat("yyyy-MM-dd")
        self.start_date_edit.setDate(current_date)

        self.end_date_edit = QDateEdit()
        self.end_date_edit.setDisplayFormat("yyyy-MM-dd")
        self.end_date_edit.setDate(current_date)

        layout.addRow("查询类型:", self.query_type_combo)
        layout.addRow("员工姓名:", self.staff_name_edit)
        layout.addRow("日期:", self.date_edit)
        layout.addRow("开始日期:", self.start_date_edit)
        layout.addRow("结束日期:", self.end_date_edit)

        button_box = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel)
        button_box.accepted.connect(self.query_attendance)
        button_box.rejected.connect(self.reject)
        layout.addRow(button_box)

        self.setLayout(layout)
        self.update_inputs()

def update_inputs(self):
        query_type = self.query_type_combo.currentIndex()
        if query_type in [0, 1]:  # 个人某天考勤和个人时间区间考勤
            self.staff_name_edit.setEnabled(True)
        else:
            self.staff_name_edit.setEnabled(False)

        if query_type == 0:  # 个人某天考勤
            self.date_edit.setEnabled(True)
            self.start_date_edit.setEnabled(False)
            self.end_date_edit.setEnabled(False)
        elif query_type == 1:  # 个人时间区间考勤
            self.date_edit.setEnabled(False)
            self.start_date_edit.setEnabled(True)
            self.end_date_edit.setEnabled(True)
        elif query_type == 2:  # 某天所有人员考勤
            self.date_edit.setEnabled(True)
            self.start_date_edit.setEnabled(False)
            self.end_date_edit.setEnabled(False)
        elif query_type == 3:  # 某时间段所有人员考勤
            self.date_edit.setEnabled(False)
            self.start_date_edit.setEnabled(True)
            self.end_date_edit.setEnabled(True)

考勤的逻辑处理部分：

def show_attendance_results(self, results):
        dialog = QDialog(self)
        dialog.setWindowTitle("考勤查询结果")
        dialog.resize(640, 360)

        layout = QVBoxLayout()

        table = QTableWidget()
        table.setColumnCount(8)
        table.setHorizontalHeaderLabels(["姓名", "日期", "上午打卡时间", "上午状态", "中午打卡时间", "中午状态", "下午打卡时间", "下午状态"])
        table.setRowCount(len(results))

        from datetime import datetime

        for i, (name, date, morning_check_time, noon_check_time, night_check_time) in enumerate(results):
            table.setItem(i, 0, QTableWidgetItem(name))
            table.setItem(i, 1, QTableWidgetItem(date))

            # 处理上午打卡时间和状态
            if morning_check_time:
                try:
                    morning_check_time = datetime.strptime(morning_check_time, '%Y-%m-%d %H:%M:%S.%f')
                    morning_check_time_str = morning_check_time.strftime('%H:%M:%S')
                    morning_status = "正常"
                    if morning_check_time.time() > self.morning_end_time:
                        morning_status = "迟到"
                except ValueError:
                    morning_check_time_str = morning_check_time
                    morning_status = "异常"
            else:
                morning_check_time_str = '未打卡'
                morning_status = "未打卡"
            table.setItem(i, 2, QTableWidgetItem(morning_check_time_str))
            table.setItem(i, 3, QTableWidgetItem(morning_status))

            # 处理中午打卡时间和状态
            if noon_check_time:
                try:
                    noon_check_time = datetime.strptime(noon_check_time, '%Y-%m-%d %H:%M:%S.%f')
                    noon_check_time_str = noon_check_time.strftime('%H:%M:%S')
                    noon_status = "正常"

                except ValueError:
                    noon_check_time_str = noon_check_time
                    noon_status = "异常"
            else:
                noon_check_time_str = '未打卡'
                noon_status = "未打卡"
            table.setItem(i, 4, QTableWidgetItem(noon_check_time_str))
            table.setItem(i, 5, QTableWidgetItem(noon_status))

            # 处理下午打卡时间和状态
            if night_check_time:
                try:
                    night_check_time = datetime.strptime(night_check_time, '%Y-%m-%d %H:%M:%S.%f')
                    night_check_time_str = night_check_time.strftime('%H:%M:%S')
                    night_status = "正常"
                    if night_check_time.time() < self.night_end_time:
                        night_status = "早退"
                except ValueError:
                    night_check_time_str = night_check_time
                    night_status = "异常"
            else:
                night_check_time_str = '未打卡'
                night_status = "未打卡"
            table.setItem(i, 6, QTableWidgetItem(night_check_time_str))
            table.setItem(i, 7, QTableWidgetItem(night_status))

        layout.addWidget(table)

        button_box = QDialogButtonBox(QDialogButtonBox.Save | QDialogButtonBox.Close)
        button_box.accepted.connect(lambda: self.export_attendance_results(table))
        button_box.rejected.connect(dialog.reject)
        layout.addWidget(button_box)

        dialog.setLayout(layout)
        dialog.exec_()

考勤查询条件

def query_attendance(self):
        query_type = self.query_type_combo.currentIndex()
        staff_name = self.staff_name_edit.text() if self.staff_name_edit.isEnabled() else None
        date = self.date_edit.date().toString("yyyy-MM-dd") if self.date_edit.isEnabled() else None
        start_date = self.start_date_edit.date().toString("yyyy-MM-dd") if self.start_date_edit.isEnabled() else None
        end_date = self.end_date_edit.date().toString("yyyy-MM-dd") if self.end_date_edit.isEnabled() else None

        if query_type in [0, 1] and not staff_name:
            QMessageBox.warning(self, "警告", "请输入员工姓名")
            return

        if query_type == 0:  # 个人某天考勤
            if not date:
                QMessageBox.warning(self, "警告", "请输入日期")
                return
            try:
                datetime.strptime(date, '%Y-%m-%d')
            except ValueError:
                QMessageBox.warning(self, "警告", "日期格式不正确")
                return
            results = self.db.query_attendance(staff_name=staff_name, start_date=date, end_date=date)
        elif query_type == 1:  # 个人时间区间考勤
            if not start_date or not end_date:
                QMessageBox.warning(self, "警告", "请输入开始日期和结束日期")
                return
            try:
                datetime.strptime(start_date, '%Y-%m-%d')
                datetime.strptime(end_date, '%Y-%m-%d')
            except ValueError:
                QMessageBox.warning(self, "警告", "日期格式不正确")
                return
            results = self.db.query_attendance(staff_name=staff_name, start_date=start_date, end_date=end_date)
        elif query_type == 2:  # 某天所有人员考勤
            if not date:
                QMessageBox.warning(self, "警告", "请输入日期")
                return
            try:
                datetime.strptime(date, '%Y-%m-%d')
            except ValueError:
                QMessageBox.warning(self, "警告", "日期格式不正确")
                return
            results = self.db.query_attendance(start_date=date, end_date=date)
        elif query_type == 3:  # 某时间段所有人员考勤
            if not start_date or not end_date:
                QMessageBox.warning(self, "警告", "请输入开始日期和结束日期")
                return
            try:
                datetime.strptime(start_date, '%Y-%m-%d')
                datetime.strptime(end_date, '%Y-%m-%d')
            except ValueError:
                QMessageBox.warning(self, "警告", "日期格式不正确")
                return
            results = self.db.query_attendance(start_date=start_date, end_date=end_date)

        self.results = results
        self.accept()

结果导出部分：

def export_attendance_results(self, table):
        # 获取当前时间，格式化为合适的字符串
        current_time = datetime.now().strftime("%Y%m%d%H%M%S")
        # 生成默认文件名
        default_csv_name = f"考勤统计_{current_time}.csv"
        default_excel_name = f"考勤统计_{current_time}.xlsx"

        # 打开文件保存对话框，设置默认文件名
        file_path, _ = QFileDialog.getSaveFileName(
            self, 
            "导出考勤数据", 
            "{default_csv_name}".format(default_csv_name=default_csv_name), 
            "CSV Files (*.csv);;Excel Files (*.xlsx)",
            options=QFileDialog.Options()
        )


        # 如果用户没有输入文件名，根据选择的文件类型使用默认文件名
        if not file_path:
            return
        if not file_path.endswith(('.csv', '.xlsx')):
            if 'CSV' in _:
                file_path = os.path.join(os.path.dirname(file_path), default_csv_name)
            else:
                file_path = os.path.join(os.path.dirname(file_path), default_excel_name)

        data = []
        for row in range(table.rowCount()):
            row_data = []
            for col in range(table.columnCount()):
                item = table.item(row, col)
                row_data.append(item.text() if item else '')
            data.append(row_data)

        df = pd.DataFrame(data, columns=[table.horizontalHeaderItem(col).text() for col in range(table.columnCount())])
        if file_path.endswith('.csv'):
            df.to_csv(file_path, index=False)
        elif file_path.endswith('.xlsx'):
            df.to_excel(file_path, index=False)
        QMessageBox.information(self, "导出成功", f"考勤数据已导出至：{file_path}")

代码有点烟花缭乱了是吧，这篇就先到这里，慢慢来吧，我整理整理看看还有漏掉啥吧，看看再补充。