131 lines
4.4 KiB
Python
Executable File
131 lines
4.4 KiB
Python
Executable File
import multiprocessing
|
||
import cv2 as cv
|
||
from multiprocessing import Process, Queue
|
||
from app.lib.face.facepp import API, File
|
||
from app.util.system_api import ge_device_info, lock_screen, unlock, sudo
|
||
from app.config import Config
|
||
from os import path, remove
|
||
|
||
class Application:
|
||
|
||
def __init__(self):
|
||
|
||
# 保存当前锁屏状态
|
||
Config.is_locked = ge_device_info().get('CGSSessionScreenIsLocked', False)
|
||
|
||
def run(self):
|
||
try:
|
||
|
||
mydict = multiprocessing.Manager().dict()
|
||
mydict['get_pic'] = 'true'
|
||
|
||
frame = Queue()
|
||
|
||
# 启动摄像头
|
||
camera_frame = Process(
|
||
target=self.OpenCamera,
|
||
args=("摄像头", frame, mydict)
|
||
)
|
||
|
||
|
||
# 实时图片处理
|
||
img_frame = Process(
|
||
target=self.ProcessingPictures,
|
||
args=("处理图片", frame, mydict)
|
||
)
|
||
|
||
camera_frame.start()
|
||
img_frame.start()
|
||
|
||
camera_frame.join()
|
||
img_frame.join()
|
||
|
||
except KeyboardInterrupt:
|
||
print("任务被终止了")
|
||
|
||
|
||
# 打开摄像头捕捉图片
|
||
def OpenCamera(self, task_name, queue, mydict):
|
||
|
||
# 初始化状态，不进行图像传输
|
||
mydict["get_pic"] = "true"
|
||
|
||
capture = cv.VideoCapture(0)
|
||
print(task_name + "任务启动..")
|
||
|
||
try:
|
||
while True:
|
||
# 一帧一帧读取视频
|
||
ret, frame = capture.read()
|
||
if mydict["get_pic"] == "true":
|
||
mydict["get_pic"] = "false"
|
||
queue.put(frame)
|
||
|
||
# cv.imshow('capture', frame)
|
||
# cv.namedWindow("摄像头", 0)
|
||
# cv.resizeWindow("capture", 0, 0)
|
||
# cv.imshow('capture', frame)
|
||
cv.waitKey(1)
|
||
|
||
except KeyboardInterrupt:
|
||
capture.release()
|
||
print(task_name + "任务关闭...")
|
||
|
||
# 处理图片，face++人脸对比
|
||
def ProcessingPictures(self, task_name, queue, mydict):
|
||
print(task_name + "任务启动")
|
||
try:
|
||
while True:
|
||
|
||
# 从队列中获取图片，显示图像质量
|
||
mydict["get_pic"] = "true"
|
||
|
||
frame = queue.get()
|
||
|
||
cv.imwrite(self.PicPath("current_user.jpg"), frame)
|
||
|
||
compare_res = Config.faceApi.compare(
|
||
image_file1=File(self.PicPath("administrators.jpg")),
|
||
image_file2=File(self.PicPath("current_user.jpg"))
|
||
)
|
||
|
||
if ('confidence' in compare_res):
|
||
print("捕捉的摄像头人物相似度：", compare_res['confidence'])
|
||
# 如果是锁屏状态
|
||
if Config.is_locked:
|
||
# 如果是管理员
|
||
if compare_res['confidence'] > 50.0:
|
||
print("锁屏状态下，是管理员，所以需要进行解锁....")
|
||
Config.is_locked = False
|
||
sudo("say '欢迎，正在自动解锁中！'")
|
||
unlock()
|
||
else:
|
||
sudo("say '不是管理员，无法解锁！'")
|
||
else:
|
||
if compare_res['confidence'] > 50.0:
|
||
print("非锁屏状态下，是管理员，无需操作....")
|
||
else:
|
||
sudo("say '不是管理员，即将锁屏！'")
|
||
Config.is_locked = True
|
||
lock_screen()
|
||
|
||
|
||
# 摄像头前没有检测到人的时候，进行锁屏
|
||
else:
|
||
if Config.is_first_count == 1:
|
||
print("是第一次运行，什么都不操作")
|
||
Config.is_first_count+=1
|
||
else:
|
||
print("不是第一次运行，屏幕前还没有人，进行锁屏")
|
||
# remove(self.PicPath("current_user.jpg"))
|
||
Config.is_locked = True
|
||
lock_screen()
|
||
|
||
except KeyboardInterrupt:
|
||
# remove(self.PicPath("current_user.jpg"))
|
||
print(task_name + "任务被终止")
|
||
|
||
|
||
def PicPath(self, fileName):
|
||
parent = path.abspath(path.dirname(path.realpath(__file__)) + path.sep + 'res/img/'+fileName);
|
||
return parent |