-
Notifications
You must be signed in to change notification settings - Fork 0
/
threading_utils.py
39 lines (35 loc) · 1.02 KB
/
threading_utils.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
import cv2
import queue
import threading
import time
# bufferless VideoCapture
class VideoCapture:
def __init__(self, name):
self.cap = cv2.VideoCapture(name, cv2.CAP_DSHOW)
self.q = queue.Queue()
self.t = threading.Thread(target=self._reader)
#t.daemon = True
self.t.start()
# read frames as soon as they are available, keeping only most recent one
def _reader(self):
while True:
ret, frame = self.cap.read()
if not ret:
print("break")
break
if not self.q.empty():
try:
self.q.get_nowait() # discard previous (unprocessed) frame
except queue.Empty:
pass
self.q.put(frame)
self.q.task_done()
def read(self):
frame = self.q.get()
return frame
def release(self):
self.cap.release()
self.q.join()
self.q.put(None)
self.t.join()
print("released")