from lib2to3.pytree import convert
|
import cv2
|
import time
|
import threading
|
import numpy as np
|
import os, sys
|
|
# add python path of src to sys.path
|
src_path = os.path.join(__file__, *(['..'] * 2))
|
src_path = os.path.abspath(src_path)
|
sys.path.insert(0, src_path)
|
|
from protos import aerial_pb2 as pb
|
|
class cameraController(object):
|
|
def __init__(self,IMAGE_WIDTH,IMAGE_HEIGHT,CAMERA_COUNT,device_id,queue_dict):
|
self.IMAGE_WIDTH = IMAGE_WIDTH
|
self.IMAGE_HEIGHT = IMAGE_HEIGHT
|
self.CAMERA_COUNT = CAMERA_COUNT
|
self.device_id = device_id
|
self.queue_dict = queue_dict
|
self.device_state_queue = self.queue_dict["device_state"]
|
self.image_data_queue = self.queue_dict["image_data"]
|
self._capture_thread = None
|
|
def open_cameras(self):
|
"""
|
Test the ports and returns a tuple with the available ports and the ones that are working.
|
"""
|
self.cameras = []
|
self.working_ports = []
|
for dev_port in range(self.CAMERA_COUNT):
|
camera = cv2.VideoCapture(dev_port, cv2.CAP_DSHOW)
|
camera.set(cv2.CAP_PROP_FRAME_WIDTH, self.IMAGE_WIDTH)
|
camera.set(cv2.CAP_PROP_FRAME_HEIGHT, self.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:
|
self.working_ports.append(dev_port)
|
self.cameras.append(camera)
|
msg = f"{time.asctime()} {self.device_id} camera port {dev_port} is working and reads images ({h} x {w})"
|
|
else:
|
msg = f"{time.asctime()} {self.device_id} camera port {dev_port} can not work"
|
else:
|
msg = f"{time.asctime()} {self.device_id} camera port {dev_port} can not found"
|
|
# add msg to queue
|
self.device_state_queue.put(msg)
|
print(msg)
|
|
|
def _start_captureThread(self):
|
"""start capture thread"""
|
if self._capture_thread is None:
|
self._capture_thread = \
|
CaptureThread(self.cameras,self.image_data_queue,self.device_id,interval=10)
|
self._capture_thread.setDaemon(True)
|
self._capture_thread.start()
|
|
def _stop_captureThread(self):
|
"""stop capture thread"""
|
if self._capture_thread is not None:
|
self._capture_thread.stop()
|
self._capture_thread = None
|
|
|
class microwaveController(object):
|
pass
|
|
class dustCleanerController(object):
|
pass
|
|
|
class CaptureThread(threading.Thread):
|
"""
|
camera capture thread
|
"""
|
def __init__(self, cameras, msg_queue, interval, device_id,image_quality = 95):
|
"""init the thread"""
|
threading.Thread.__init__(self)
|
self.cameras = cameras
|
self.msg_queue = msg_queue
|
self.interval = interval
|
self.image_quality = image_quality
|
self.is_stop = False
|
self.device_id = device_id
|
|
def stop(self):
|
"""stop the thread"""
|
self.is_stop = True
|
|
def run(self):
|
print("CaptureThread started")
|
encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), self.image_quality]
|
self.next_time = time.time() + self.interval
|
while not self.is_stop:
|
this_time = time.time()
|
if this_time > self.next_time:
|
for i in range(len(self.cameras)):
|
_, frame = self.cameras[i].read()
|
capture_time = time.time()
|
_, encimg = cv2.imencode('.jpg', frame, encode_param)
|
if frame is not None:
|
np_array = np.asarray(encimg) # frame 转换为numpy数组
|
byte_array = np_array.tobytes() # numpy数组转换为byte数组
|
# convert the image to pb.DataField
|
imagedata = pb.DataField()
|
imagedata.data = byte_array
|
imagedata.length = len(byte_array)
|
|
data_str = imagedata.SerializeToString()
|
|
# create protobuf message
|
data_msg = pb.CapturedData()
|
data_msg.generate_time = capture_time
|
data_msg.generate_device = self.device_id
|
data_msg.camera_port = i
|
data_msg.image_data = data_str
|
|
|
|
# add message to queue
|
self.msg_queue.put(data_msg)
|
|
self.next_time += self.interval
|
time.sleep(0.1)
|
for camera in self.cameras:
|
camera.release()
|
print("CaptureThread stopped")
|
