import threading import time import cv2 from AcrossPlatform.get_platform import SYS_PLATFORM, WINDOWS, LINUX from Speaker.speak_base import beep try: from greenlet import getcurrent as get_ident except ImportError: try: from thread import get_ident except ImportError: from _thread import get_ident CAMERA_ID = "camera_id" FRAME_DAT = "frame_data" CATCH_TIME = "catch_time" RESIZE_SIZE = (640, 480) class CameraEvent(object): """An Event-like class that signals all active clients when a new frame is available. """ def __init__(self): self.events = {} def wait(self): """Invoked from each client's thread to wait for the next frame.""" ident = get_ident() if ident not in self.events: # this is a new client # add an entry for it in the self.events dict # each entry has two elements, a threading.Event() and a timestamp self.events[ident] = [threading.Event(), time.time()] return self.events[ident][0].wait() def set(self): """Invoked by the camera thread when a new frame is available.""" now = time.time() remove = None for ident, event in self.events.items(): if not event[0].isSet(): # if this client's event is not set, then set it # also update the last set timestamp to now event[0].set() event[1] = now else: # if the client's event is already set, it means the client # did not process a previous frame # if the event stays set for more than 5 seconds, then assume # the client is gone and remove it if now - event[1] > 1: remove = ident if remove: del self.events[remove] def clear(self): """Invoked from each client's thread after a frame was processed.""" self.events[get_ident()][0].clear() class Camera(object): thread = {} # background thread that reads frames from camera frame = {} # current frame is stored here by background thread queue_frame = [] cache_time = 1 frame_cache = {} last_access = 0 # time of last client access to the camera event = {} # 相机所有源 source_list = set() # 相机记录 record_signal = False # 错误计数 error_counting = 0 max_error = 50 def __init__(self): """Start the background camera thread if it isn't running yet.""" if not Camera.thread: for source in Camera.source_list: Camera.event[source] = CameraEvent() Camera.last_access = time.time() # start background frame thread Camera.thread[source] = threading.Thread(target=self._thread, args=(source,)) Camera.thread[source].start() # wait until frames are available while self.get_frame(video_source=source) is None: time.sleep(0.1) @classmethod def add_source(cls, source): Camera.source_list.add(source) @classmethod def get_cache(cls): if Camera.queue_frame: return Camera.queue_frame.pop(0) @classmethod def get_all_cache(cls): result = Camera.queue_frame.copy() cls.clear_cache() return result @classmethod def clear_cache(cls): Camera.queue_frame.clear() @classmethod def start_record(cls): cls.record_signal = True @classmethod def stop_record(cls): cls.record_signal = False @classmethod def get_frame(cls, video_source=0): """Return the current camera frame.""" Camera.last_access = time.time() # wait for a signal from the camera thread Camera.event[video_source].wait() Camera.event[video_source].clear() return Camera.frame[video_source] @staticmethod def reload_camera(video_source): # 根据平台导入对应的包 if SYS_PLATFORM == WINDOWS: camera = cv2.VideoCapture(video_source, cv2.CAP_DSHOW) elif SYS_PLATFORM == LINUX: camera = cv2.VideoCapture(video_source, cv2.CAP_V4L2) else: camera = None if camera: # 设置摄像头帧率为30 camera.set(cv2.CAP_PROP_FPS, 20) # 设置摄像头图像大小为640x480 camera.set(cv2.CAP_PROP_FRAME_WIDTH, 640) camera.set(cv2.CAP_PROP_FRAME_HEIGHT, 480) camera.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter_fourcc('M', 'J', 'P', 'G')) return camera @staticmethod def reload_camera_normally(video_source): # 根据平台导入对应的包 camera = cv2.VideoCapture(video_source) if camera: # 设置摄像头帧率为30 camera.set(cv2.CAP_PROP_FPS, 20) # 设置摄像头图像大小为640x480 camera.set(cv2.CAP_PROP_FRAME_WIDTH, 640) camera.set(cv2.CAP_PROP_FRAME_HEIGHT, 480) camera.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter_fourcc('M', 'J', 'P', 'G')) return camera @staticmethod def camera_restart(video_source): camera = Camera.reload_camera(video_source) while not camera.isOpened(): print('无法打开摄像头,正在重新初始化,请稍后!') beep() camera = Camera.reload_camera(video_source) return camera @staticmethod def camera_restart_normally(video_source): camera = Camera.reload_camera_normally(video_source) while not camera.isOpened(): print('无法打开摄像头,正在重新初始化,请稍后!') beep() camera = Camera.reload_camera_normally(video_source) return camera @staticmethod def frames(video_source): camera = Camera.camera_restart(video_source) while True: try: ret, img = camera.read() if not ret: Camera.error_counting += 1 if Camera.error_counting > Camera.max_error: camera = Camera.camera_restart_normally(video_source) else: camera = Camera.camera_restart(video_source) # 防止摄像机松动导致帧错误 if img is not None: yield img except cv2.Error: continue @classmethod def _thread(cls, video_source): """Camera background thread.""" print(f'Starting CAMERA@{video_source} thread.') frames_iterator = cls.frames(video_source) for frame in frames_iterator: if cls.record_signal: frame_pkg = { CAMERA_ID: video_source, FRAME_DAT: frame, CATCH_TIME: time.time() } Camera.queue_frame.append(frame_pkg) Camera.frame[video_source] = frame Camera.event[video_source].set() # send signal to clients