1. opencv從攝像頭抽幀
camera = cv2.VideoCapture(0)
if camera.isOpened():
success, frame = camera.read()
if success:
print('capture success')
2. RGB轉YUV編碼，JPG格式壓縮
# 直接壓縮到最小
result, img_code = cv2.imencode('.jpg', frame)
# 可以指定壓縮后的圖像質量
img_quality = 15
result, img_code = cv2.imencode('.jpg', frame, [int(cv2.IMWRITE_JPEG_QUALITY), img_quality])
# 這里，img_code經過編碼后，可以直接tobytes寫入文件了。
3. numpy ndarray 轉bytes
buffer = frame.tobytes()
4. numpy ndarray 從buffer讀取圖像矩陣
# buffer中讀取矩陣需要手動指定dtype，之后reshape調整shape，因此如果通過網絡傳輸矩陣，需要同時傳輸其dtype和shape。
buffer = numpy.frombuffer(frame.tobytes(), frame.dtype)
buffer.reshape(frame.shape)
5. 圖像轉為矩陣
frame = cv2.imdecode(numpy.frombuffer(img_code.tobytes(), img_code.dtype), -1)
img = Image.fromarray(frame)
# 一般來說，圖像矩陣元素類型為 uint8 ， 解碼時可直接指定dtype為 numpy.uint8
6. BGR 轉 RGB 的幾種方式
# 如果發現圖片顯示的時候顏色不對勁，紅色變成了藍色，說明顏色信息放反了，需要轉換一下
# opencv默認使用BGR格式保存圖像
frame = frame[:, :, [2, 1, 0]]
frame = frame[:, :, ::-1]
frame = frame[..., ::-1]
frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
7. opencv 、socket 實現簡單的視頻處理（不帶音頻）
視頻源 —> 生產者 —> 中間處理 —> 消費者
視頻源：可以是靜態視頻文件，也可以是一些實時視頻流。
生產者：opencv的VideoCapture，從視頻源抽幀。
中間處理：圖像處理程序等。
消費者：播放器、視頻封裝程序等。
server.py
class FrameProducer(threading.Thread):
def __init__(self, frame_stack):
super(FrameProducer, self).__init__()
self.camera = cv2.VideoCapture(0)
self.running = True
self.frame_stack: Stack = frame_stack
def run(self) -> None:
while self.running:
if self.camera.isOpened():
res, frame = self.camera.read()
if res:
print('clip a frame from camera')
self.frame_stack.push(frame)
class Sender(threading.Thread):
def __init__(self, sock: socket.socket, frame_stack: Stack):
super(Sender, self).__init__()
self.client: socket = sock
self.resolution = (640, 480)
self.img_quality = 15
self.running = True
self.frame_stack = frame_stack
def run(self) -> None:
while self.running:
frame = self.frame_stack.pop()
if frame is not None:
print('got a frame from stack')
frame = cv2.resize(frame, self.resolution)
result, img_encode = cv2.imencode('.jpg', frame, [int(cv2.IMWRITE_JPEG_QUALITY), self.img_quality])
data = img_encode.tostring()
msg = Message(data, *self.resolution)
try:
self.client.send(msg.get_head())
self.client.send(msg.get_body())
print(msg.length)
except Exception as ex:
self.running = False
print(ex)
client.py
class Receiver(threading.Thread):
def __init__(self, _sock: socket.socket, frame_stack: Stack):
super(Receiver, self).__init__()
self.sock = _sock
self.running = True
self.frame_stack = frame_stack
def run(self) -> None:
try:
while self.running:
head = self.sock.recv(Message.head_length)
msg = Message()
msg.parse_head(head)
data = self.sock.recv(msg.length)
msg.parse_body(data)
print(msg.length, len(msg.data))
frame = cv2.imdecode(np.frombuffer(msg.data, np.uint8), -1)
self.frame_stack.push(frame)
except Exception as ex:
print(ex)
self.running = False
class Consumer(threading.Thread):
def __init__(self, frame_stack: Stack):
super(Consumer, self).__init__()
self.frame_stack = frame_stack
self.running = True
def run(self) -> None:
while self.running:
frame = self.frame_stack.pop()
if frame is not None:
cv2.imshow('image', frame)
if cv2.waitKey(100) & 0xFF == ord('q'):
break
可能遇到的問題：
- 如果消費者這邊處理速度低於opencv抽幀的速度，由於opencv自帶幀緩沖區，每一幀圖像都不會被丟棄，會使幀數據在緩沖區堆積，結果處理后的視頻延時越來越高。消費者端，准備一個棧，將接收到的幀存放到棧里，用於丟幀，防止實時視頻的延時累加。每當棧內積壓的數據超過一個閥值，就將棧內數據清空，防止內存溢出。
- 直接使用socket時需要注意socket的粘包問題。粘包問題可以通過多種方式解決，如定界符加轉義、固定報文長度、固定首部長度並在首部指明數據部分長度。