1 from __future__ import annotations
  2 import cv2
  3 import numpy
  4 import time
  5 
  6 '''
  7 1、允许同一文件下不同类之间的类型提示，py3.7以上特性
  8 2、cv2——获取摄像头和展示
  9 3、numpy——对展示画面进行左右翻转（fliplr）
 10 4、time——精确获取时间间隔，然后计算帧率估值
 11 '''
 12 
 13 
 14 class CaptureManager:
 15     def __init__(self, capture: cv2.VideoCapture,
 16                  previewWindowManager: WindowManager = None,
 17                  shouldMirrorPreview: bool = False):
 18         self.previewWindowManager = previewWindowManager
 19         self.shouldMirrorPreview = shouldMirrorPreview
 20 
 21         self._capture = capture
 22         self._channel = 0
 23         self._enteredFrame = False
 24         self._frame = None
 25 
 26         self._imageFilename = None
 27         self._videoFilename = None
 28         self._videoEncoding = None
 29         self._videoWriter = None
 30 
 31         self._startTime = None
 32         self._framesElapsed = 0
 33         self._fpsEstimate = None
 34 
 35     @property
 36     def channel(self):
 37         return self._channel
 38 
 39     @channel.setter
 40     def channel(self, value):
 41         if self._channel != value:
 42             self._channel = value
 43             self._frame = None  # 对通道赋值时需要将当前帧置空，否则可能出现通道为未更改现象
 44 
 45     # 如果进入帧，且当前帧为None则对已捕获的帧进行解码和获取实际图像帧
 46     @property
 47     def frame(self):
 48         if self._enteredFrame and self._frame is None:
 49             _, self._frame = self._capture.retrieve(self._frame, self.channel)
 50         return self._frame
 51 
 52     @property
 53     def isWritingImage(self):
 54         return self._imageFilename is not None
 55 
 56     @property
 57     def isWritingVideo(self):
 58         return self._videoFilename is not None
 59 
 60     def enterFrame(self):
 61         # 检查上一帧是否被处理完，如未处理完，则会被下一帧覆盖，最终导致实时画面或者保存的视频不连续
 62         assert not self._enteredFrame, 'previous enterFrame() had no matching exitFrame()'
 63 
 64         if self._capture is not None:
 65             self._enteredFrame = self._capture.grab()
 66 
 67     def exitFrame(self):
 68         # 如果当前帧为None，表示未成功捕获有效帧或视频流已处理完成，此时则直接结束此方法
 69         if self.frame is None:
 70             self._enteredFrame = False
 71             return
 72 
 73         # 更新FPS估值
 74         if self._framesElapsed == 0:
 75             self._startTime = time.perf_counter()
 76         else:
 77             timeElapsed = time.perf_counter() - self._startTime
 78             self._fpsEstimate = self._framesElapsed / timeElapsed
 79         self._framesElapsed += 1
 80 
 81         # 是否水平反转画面并展示
 82         if self.previewWindowManager is not None:
 83             if self.shouldMirrorPreview:
 84                 mirroredFrame = numpy.fliplr(self._frame)
 85                 self.previewWindowManager.show(mirroredFrame)
 86             else:
 87                 self.previewWindowManager.show(self._frame)
 88 
 89         # 将当前帧保存为图片
 90         if self.isWritingImage:
 91             cv2.imwrite(self._imageFilename, self._frame)
 92             self._imageFilename = None
 93 
 94         # 将当前帧写入视频
 95         self._writeVideoFrame()
 96 
 97         # 释放并退出当前帧
 98         self._frame = None
 99         self._enteredFrame = False
100 
101     def writeImage(self, filename):
102         self._imageFilename = filename
103 
104     def startWritingVideo(self, filename, encoding=cv2.VideoWriter_fourcc('M', 'J', 'P', 'G')):
105         self._videoFilename = filename
106         self._videoEncoding = encoding
107 
108     def stopWritingVideo(self):
109         self._videoFilename = None
110         self._videoEncoding = None
111         self._videoWriter = None
112 
113     def _writeVideoFrame(self):
114         if not self.isWritingVideo:
115             return
116 
117         # 检查是否创建了VideoWriter对象
118         if self._videoWriter is None:
119             # 获取摄像头帧率
120             fps = self._capture.get(cv2.CAP_PROP_FPS)
121             # 如果帧率获取失败则使用估计值
122             if numpy.isnan(fps) or fps <= 0.0:
123                 # 获取更多的以处理帧数，以求得更稳定的帧率
124                 if self._framesElapsed < 20:
125                     return
126                 else:
127                     fps = self._fpsEstimate
128             size = (int(self._capture.get(
129                 cv2.CAP_PROP_FRAME_WIDTH)),
130                     int(self._capture.get(
131                         cv2.CAP_PROP_FRAME_HEIGHT)))
132             self._videoWriter = cv2.VideoWriter(
133                 self._videoFilename, self._videoEncoding,
134                 fps, size)
135 
136         self._videoWriter.write(self._frame)
137 
138 
139 class WindowManager(object):
140 
141     def __init__(self, windowName: str
142                  , keypressCallback=None):
143         self.keypressCallback = keypressCallback
144 
145         self._windowName = windowName
146         self._isWindowCreated = False
147 
148     @property
149     def isWindowCreated(self):
150         return self._isWindowCreated
151 
152     def createWindow(self):
153         cv2.namedWindow(self._windowName)
154         self._isWindowCreated = True
155 
156     def show(self, frame):
157         cv2.imshow(self._windowName, frame)
158 
159     def destroyWindow(self):
160         cv2.destroyWindow(self._windowName)
161         self._isWindowCreated = False
162 
163     def processEvents(self):
164         keycode = cv2.waitKey(1)
165         if self.keypressCallback is not None and keycode != -1:
166             self.keypressCallback(keycode)

 1 import cv2
 2 from managers import WindowManager, CaptureManager
 3 
 4 
 5 class Cameo:
 6     def __init__(self):
 7         self._windowManager = WindowManager('Cameo',
 8                                             self.onKeypress)
 9         self._captureManager = CaptureManager(
10             cv2.VideoCapture(0), self._windowManager, True)
11 
12     def run(self):
13         # 创建窗口
14         self._windowManager.createWindow()
15         while self._windowManager.isWindowCreated:
16             # 开始捕获摄像头帧
17             self._captureManager.enterFrame()
18             # 解码并获取实际帧
19             frame = self._captureManager.frame
20             # 预留后续新功能
21             if frame is not None:
22                 pass
23             # 写入图片文件或者视频文件并释放帧资源
24             self._captureManager.exitFrame()
25             # 检查并返回键盘状态
26             self._windowManager.processEvents()
27 
28     def onKeypress(self, keycode):
29         if keycode == 32:  # space
30             self._captureManager.writeImage(r'screenshot.png')
31         elif keycode == 9:  # tab
32             if not self._captureManager.isWritingVideo:
33                 self._captureManager.startWritingVideo('screencast.avi')
34             else:
35                 self._captureManager.stopWritingVideo()
36         elif keycode == 27:  # escape
37             self._windowManager.destroyWindow()
38 
39 
40 if __name__ == "__main__":
41     Cameo().run()