中控考勤机demo中提供有C#的示例代码,尝试使用python来实现实时考勤数据的现实,由于刚接触python,很多问题还不明白是为什么,先记录下来。编程工具使用的是vs2017。
在连接考勤机时,比较耗时,界面会有卡顿现象,使用线程QThread,将连接放在run中进行,监听考勤机的事件使用
zk = DispatchWithEvents("zkemkeeper.ZKEM.1",ZKEMEvents)
ZKEMEvents是自定义的类,里面是Event Handlers,可以在C:\Program Files (x86)\Microsoft Visual Studio\Shared\Python36_86\Lib\site-packages\win32com\gen_py\ 目录下找到py文件,里面有Event Handlers的定义。这个文件由win32com\client\makepy.py 生成。
当连接设备成功后再断开设备,再次连接,有人打卡时会出现2条相同的数据,不明白为什么,是zk对象没有销毁,使用QThread创建子线程,线程是在run中开始,初始化时还在主线程,当子线程结束时zk没有被垃圾回收,因此使用pythoncom.CoInitialize()初始化调用线程的COM库,问题解决。
退出子线程,事件监控并没有终止,要终止事件监控使用win32api.PostQuitMessage()
# -*- coding: utf-8 -*-
import sys
import os
import time
from PyQt5 import QtWidgets, QtCore,QtGui
from PyQt5.QtGui import QFont
from PyQt5.QtCore import *
from PyQt5.QtWidgets import QApplication,QMainWindow,QMessageBox
from helloWorld_form import Ui_MainWindow #导入生成的窗口import win32com
from win32com.client import Dispatch, constants,DispatchWithEvents
import win32event
import win32api
import pythoncom#新建一个类来继承生成的窗口,也可以在这里添加关于窗口处理的代码
class MyMainWindow(QMainWindow, Ui_MainWindow):zk = Noneip = ''def __init__(self, parent=None): super(MyMainWindow, self).__init__(parent)# 进度条设置self.pbar = QtWidgets.QProgressBar(self)self.pbar.setGeometry(50,20, 210, 25)self.pbar.setValue(0)self.thread2 = None # 初始化线程self.setupUi(self)pass#连接指纹设备def btnConnect(self):if self.thread2==None:# 创建线程2self.thread2 = Runthread2()# 连接信号self.thread2._signa.connect(self.call_back2) # 进程连接回传到GUI的事件self.thread2._signaConnect.connect(self.SetZKState)self.ip=self.thread2.ip = self.lineEditIp.text()#self.thread2.connect()# 开始线程self.thread2.start()#self.lineEditIp.setFocusPolicy(QtCore.Qt.NoFocus)self.lineEditIp.setEnabled(False)passelse:self.thread2.stop()self.thread2=None#del self.thread2passpass#断开设备def btnDisConnect(self):self.thread2.stop()self.thread2=Noneif self.zk_IsConnected:#self.zk.Disconnect()self.zk_IsConnected = Falseself.textEdit.append(self.ip+':已断开连接')self.lineEditIp.setEnabled(True)pass#线程回调函数def call_back2(self, n,msg):if n>0:QMessageBox.information(self,'提示',msg,QMessageBox.Ok)else:self.textEdit.append(self.ip+':'+ msg)pass#指纹机是否连接回调函数def SetZKState(self,b):self.textEdit.append(str(b))pass
pass#连接指纹机线程
class Runthread2(QtCore.QThread):_signa = pyqtSignal(int,str)_signaConnect = pyqtSignal(bool)bIsConnected = Falseob = NoneiMachineNumber = 1ip = ''is_runnable = Truedef __init__(self):super(Runthread2, self).__init__()pythoncom.CoInitialize()self.zk = DispatchWithEvents("zkemkeeper.ZKEM.1",ZKEMEvents)passdef __del__(self):self.is_runnable = Falseself.quit()self.wait()passdef run(self):self.connect()while self.is_runnable==True:if self.is_runnable==False:breakpasspassthread = win32api.GetCurrentThreadId()self._signa.emit(0,str(thread))pythoncom.CoUninitialize()def connect(self):try:self.bIsConnected = self.zk.Connect_Net(self.ip, 4370) # 连接self._signaConnect.emit(self.bIsConnected)if self.bIsConnected:self._signa.emit(0,'已连接成功')if self.zk.RegEvent(self.iMachineNumber, 65535):thread = win32api.GetCurrentThreadId()self._signa.emit(0,'zk object connect on thread:'+str(thread))sdkVersion = self.zk.GetSDKVersion(None)if sdkVersion[0]:#QMessageBox.information(self,'消息',sdkVersion[1],QMessageBox.Yes | QMessageBox.No)self._signa.emit(0,'控件版本:'+sdkVersion[1])passpass#pythoncom.PumpMessages()else:#QMessageBox.information(self,'错误信息','注册机器失败。',QMessageBox.Ok)self._signa.emit(0,'注册机器失败。')passpasselse:#self.textEdit.append(self.ip+':连接失败')self._signa.emit(0,'连接失败')passexcept BaseException as e:idwErrorCode = 0if self.zk:self.zk.GetLastError(idwErrorCode)self._signa.emit(1,str(idwErrorCode)+':'+e.args[0])else:self._signa.emit(1,e.args[0])passpassdef stop(self):self.is_runnable =Falseself.zk.Disconnect()pass#考勤机实时事件类
class ZKEMEvents:#_signaEvent = pyqtSignal(int,str)myEmpty = pythoncom.Emptydef __init__(self):#self.event = win32event.CreateEvent(None, 0, 0, None)#self.events.append(self.event)#win32event.SetEvent(self.event)passdef OnAttTransactionEx(self,sEnrollNumber=myEmpty,iIsInValid=myEmpty,iAttState=myEmpty,iVerifyMethod=myEmpty,iYear=myEmpty,iMonth=myEmpty,iDay=myEmpty,iHour=myEmpty,iMinute=myEmpty,iSecond=myEmpty,iWorkCode=myEmpty):if myWin.thread2.is_runnable==False:win32api.PostQuitMessage()sTime = str(iYear) + "-" + str(iMonth) + "-" + str(iDay) + " " + str(iHour) + ":" + str(iMinute) + ":" + str(iSecond)txt = "User:" +sEnrollNumber + "[" + sEnrollNumber + "],Method:" + str(iVerifyMethod) + ",Time:" + sTimemyWin.call_back2(0,txt)passpass#主程序,生成一个窗口实例并运行。
if __name__=="__main__": app = QApplication(sys.argv) myWin = MyMainWindow()myWin.show()app.exec_()