-
-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathWebcamStream.py
76 lines (63 loc) · 2.27 KB
/
WebcamStream.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
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
import numpy as np
import argparse
from VideoInterface import VideoInterface
import cv2
from calibrate import calibrate_camera
class WebcamStream(VideoInterface):
'''Gets a stream from the webcam and lets other classes grab information
about the stream.
Attributes:
capture_id: An integer or string indicating the camera device
'''
def __init__(self, capture_id=0):
'''Inits WebcamStream with the capture ID'''
self.vs = cv2.VideoCapture(capture_id, cv2.CAP_DSHOW)
# self.test = cv2.VideoCapture(
# "ball_images/test_throw.mp4", cv2.CAP_DSHOW)
self.vs.set(cv2.CAP_PROP_FPS, 30)
self.undistorted_frame = None
self.camera_matrix, self.dist = calibrate_camera()[:2]
def open_video_stream(self):
'''Grabs a camera stream'''
if not self.vs.isOpened():
print("Video Stream couldn't be created")
def read(self, undistort=True, shape=(640, 360)) -> np.array:
'''Grabs a frame from the video stream.
Waits until a frame can be read
Returns:
Image frame, np.array
'''
while self.vs.isOpened():
ret, frame = self.vs.read()
if ret:
self.frame = frame
# frame = cv2.resize(frame, shape)
undist = cv2.undistort(frame, self.camera_matrix, self.dist)
self.undistorted_frame = undist
if undistort:
return undist
else:
return frame
else:
self.release()
break
def release(self):
self.vs.release()
cv2.destroyAllWindows()
def configure_args() -> dict:
ap = argparse.ArgumentParser()
ap.add_argument("-d", "--debug", action="store_true", help="Show debug information")
ap.add_argument("-v", "--video", help="path to the (optional) video file", default=0)
print(vars(ap.parse_args()))
return vars(ap.parse_args())
if __name__ == "__main__":
print(cv2.__version__)
args = configure_args()
ws = WebcamStream(args)
ws.open_video_stream()
while True:
frame = ws.read()
print(frame)
cv2.imshow("frame", frame)
if (cv2.waitKey(0) & 0xFF == ord('q')):
break