from concurrent.futures import thread
|
from itertools import count
|
import threading
|
import cv2
|
import sched
|
import time
|
|
IMAGE_WIDTH = 3840
|
IMAGE_HEIGHT = 2880
|
|
class CameraTest:
|
|
def __init__(self, interval: int):
|
self.last_frame = None
|
self.count = 0
|
self.last_time = 0
|
self.interval = interval
|
self.next_time = 0
|
self.cameras = []
|
|
def get_image(self) -> None:
|
# cv2.namedWindow("preview", cv2.WINDOW_NORMAL)
|
# cv2.namedWindow("diff", cv2.WINDOW_NORMAL)
|
while True:
|
this_time = time.time()
|
if this_time > self.next_time:
|
for i, camera in enumerate(self.cameras):
|
ret, frame = camera.read()
|
fn = "./Images/{}-{}.jpg".format(i, self.count)
|
# print(fn, type(frame), frame.shape)
|
encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), 95]
|
result, encimg = cv2.imencode('.jpg', frame, encode_param)
|
print(fn, type(encimg), encimg.shape)
|
# cv2.imshow("preview", frame)
|
# encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), 95]
|
#cv2.imwrite(fn, frame, encode_param)
|
# if not self.last_frame is None:
|
# diff = cv2.absdiff(frame, self.last_frame)
|
# print(self.count, this_time, diff.sum())
|
# cv2.imshow("diff", diff)
|
# self.last_frame = frame1
|
self.count += 1
|
self.next_time += self.interval
|
print("frame ", self.count)
|
# time.sleep(0.1)
|
if cv2.waitKey(1) & 0xFF == ord('q'):
|
break
|
if self.count > 50:
|
break
|
for camera in self.cameras:
|
camera.release()
|
# cv2.destroyAllWindows()
|
|
def open_cameras(self) -> bool:
|
"""
|
Test the ports and returns a tuple with the available ports and the ones that are working.
|
"""
|
for i in range(0,4):
|
camera = cv2.VideoCapture(i, cv2.CAP_DSHOW)
|
#camera = cv2.VideoCapture(i, cv2.CAP_DSHOW)
|
# 设置摄像头设备分辨率
|
# camera.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter.fourcc('M', 'J', 'P', 'G'))
|
camera.set(cv2.CAP_PROP_FRAME_WIDTH, IMAGE_WIDTH)
|
camera.set(cv2.CAP_PROP_FRAME_HEIGHT, IMAGE_HEIGHT)
|
camera.set(cv2.CAP_PROP_FPS, 1)
|
if camera.isOpened():
|
is_reading, img = camera.read()
|
w = camera.get(3)
|
h = camera.get(4)
|
if is_reading:
|
print("Port %s is working and reads images (%s x %s)" % (i, w, h))
|
self.cameras.append(camera)
|
|
def start_capture(self):
|
self.next_time = time.time() + self.interval
|
threading.Thread(target=self.get_image).start()
|
|
if __name__ == '__main__':
|
image = {}
|
camera = CameraTest(1)
|
camera.open_cameras()
|
camera.start_capture()
|
