问题描述
我正在尝试执行面部检测功能,并使用Apscheduler仅在特定时间之间运行该功能。我可以正确启动该功能,但是end_time
参数似乎根本不起作用,该功能可以一直运行直到手动关闭。
这是开始时间表的路线:
@app.route("/schedule/start",methods = ['POST'])
def create():
sched = BlockingScheduler()
sched.add_job(detection,'cron',start_date='2020-10-01 15:33:00',end_date='2020-10-01 15:33:05')
sched.start()
return 'Schedule created.'
我的While True
函数中有一个detection
条件,所有检测逻辑都在其中运行。即使我确定了停止时间,也可能是它永远不会停止的原因吗?我该如何解决这个问题?
编辑。从我的detection
函数中-loop时(删除了不必要的部分):
while True:
frame = stream.read()
frame = imutils.resize(frame,width=900)
frame = cv2.cvtColor(frame,cv2.COLOR_BGR2GRAY)
frame = np.dstack([frame,frame,frame])
# frame = cv2.flip(frame,0)
faces = faceCascade.detectMultiScale(
frame,scaleFactor=1.1,minNeighbors=3,# minSize=(10,10),# maxSize=(50,50),# flags=cv2.CASCADE_SCALE_IMAGE
)
for (x,y,w,h) in faces:
name = str(currentframe) + '.jpg'
print('Creating...' + name)
cv2.imwrite(os.path.join(parent_dir,name),frame)
currentframe += 1
if cv2.waitKey(1) & 0xFF == ord('q'):
break
编辑2。我尝试按照以下建议进行操作,并收到以下错误消息:TypeError: func must be a callable or a textual reference to one
。
我还希望集成一个功能,以手动启动和停止我的面部检测功能。我可以这样:
@app.route("/start",methods = ['POST'])
def start():
os.environ['run'] = 'running'
detection()
return 'Detection started'
@app.route("/stop",methods = ['POST'])
def stop():
os.environ['run'] = 'stop'
return 'Detection stopped.'
然后在我的detection.py
中,我只在while循环的beginnig中检查环境变量:
while True:
if os.environ.get('run') == 'stop':
stream.stream.release()
exit()
我想要的是将调度功能集成到此。我不想做单独的功能,因为我希望能够手动停止以计划开始的检测。我有什么技巧可以实现这一目标?
编辑3。日程安排现在正在工作。手动启动也起作用,而停止则意味着停止检测面部。尽管以schedule开始的功能仍然继续,它根本没有迭代到检测部分,因为有一个os.environ['run'] = 'stop'
-flag。知道如何停止函数执行吗?
这是我进行的while -loop检查:
while True:
if self.end_date <= datetime.Now() or os.environ.get('run') == 'stop':
stream.stream.release()
exit()
手动启动时,停止功能按预期工作,但是在停止预定作业时,它会一直循环播放,直到满足end_date时间为止。
解决方法
您需要有一些条件来结束while循环,因为它现在是无限的,传递开始/结束时间并转换为日期时间,然后尝试匹配while end_date_time =< now:
,然后退出任务。如果需要传递开始/结束日期,则可以将detection
函数转换为类,并在初始化时通过c end_date
来停止cron作业。
# detection.py
from datetime import datetime
class Detection:
def __init__(self,end_date):
self.end_date = datetime.strptime(end_date,'%Y-%m-%d %H:%M:%S.%f')
def detection(self):
print(self.end_date)
while self.end_date <= datetime.utcnow():
print('works')
# routes.py
# And then do it like this when starting cron
import Detection
def create():
start_date = '2020-10-02 01:48:00.192386'
end_date = '2020-10-02 05:50:00.192386'
dt = Detection(end_date)
sched = BlockingScheduler()
sched.add_job(dt.detection,'cron',start_date=start_date,end_date=end_date)
sched.start()
return 'Schedule created.'
这应该可以解决问题