from lib2to3.pytree import convert
|
import cv2
|
import time
|
import threading
|
import numpy as np
|
import os, sys
|
from typing import List
|
import queue
|
from datetime import datetime
|
|
# # 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
|
from .DustCleaner import DustCleaner
|
from .MessageThreads import send_logs_tool, send_images_tool
|
|
|
######################################### CAMERAS ############################################
|
class cameraController(object):
|
|
def __init__(self,
|
IMAGE_WIDTH:int,
|
IMAGE_HEIGHT:int,
|
CAMERA_COUNT:int,
|
device_id:str,
|
logs_queue:queue.Queue,
|
images_queue:queue.Queue,
|
init_routing_key:str,
|
init_binding_key:str
|
):
|
|
self.IMAGE_WIDTH = IMAGE_WIDTH
|
self.IMAGE_HEIGHT = IMAGE_HEIGHT
|
self.CAMERA_COUNT = CAMERA_COUNT
|
self.device_id = device_id
|
self.logs_queue = logs_queue
|
self.images_queue = images_queue
|
self.routing_key = init_routing_key
|
self.init_routing_key = init_routing_key
|
self.binding_key = init_binding_key
|
self._check_cameras()
|
|
|
def _check_cameras(self):
|
self.cameras = []
|
self.camera_work = []
|
self.working_ports = []
|
test_logs = []
|
test_images = []
|
|
# init the cameras
|
for dev_port in range(self.CAMERA_COUNT):
|
# set the props of the camera by cv2 lib
|
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)
|
self.cameras.append(camera)
|
self.camera_work.append(False)
|
|
# test the cameras
|
(test_logs, test_images) = self._capture_image(mode = "test")
|
|
# generate the msg and put it to queues
|
self._generate_msg(test_logs,test_images)
|
|
|
def _generate_msg(self,logs:List[str],images:List):
|
|
send_logs_tool(logs=logs,
|
logs_queue=self.logs_queue,
|
source=self.device_id,
|
routing_key=self.routing_key+".logs.test0")
|
|
send_images_tool(images=images,
|
images_queue=self.images_queue,
|
source=self.device_id,
|
routing_key=self.routing_key+".images.test0")
|
|
|
def _capture_image(self,mode:str = "capture"):
|
"""
|
mode -> "test" or "capture"
|
"""
|
TEST = False
|
if mode == "test":
|
TEST = True
|
else:
|
pass
|
|
log_list = []
|
image_list = []
|
|
# iterate the cameras
|
for dev_port, camera in enumerate(self.cameras):
|
# init the flags
|
is_reading = False
|
log = None
|
img = None
|
|
if not TEST and self.camera_work[dev_port] is not True:
|
pass
|
else:
|
# capture the image
|
if camera.isOpened():
|
is_reading, img = camera.read()
|
w = camera.get(3)
|
h = camera.get(4)
|
if is_reading:
|
log = f"camera port {dev_port} is working and reads images ({h} x {w})"
|
else:
|
log = f"camera port {dev_port} can not work"
|
else:
|
log = f"camera port {dev_port} can not found"
|
|
# display
|
print(f"{time.asctime()} | {log}")
|
|
# add some tags for camera when test
|
if TEST:
|
self.working_ports.append(dev_port)
|
if is_reading:
|
self.camera_work[dev_port] == True
|
|
|
log_list.append(log)
|
image_list.append(img)
|
|
return log_list, image_list
|
|
|
def capture(self,routing_key:str):
|
|
self.routing_key = routing_key
|
|
(logs,images) = self._capture_image()
|
self._generate_msg = (logs,images)
|
|
self.routing_key = self.init_routing_key
|
|
|
|
######################################## MICROWAVE ###########################################
|
class microwaveController(object):
|
pass
|
|
|
####################################### DUST CLEANER #########################################
|
class dustCleanerController(DustCleaner):
|
def __init__(self, com, bps, timeout, logs_queue, device_id,routing_key):
|
super().__init__(com, bps, timeout)
|
|
self.logs_queue = logs_queue
|
self.device_id = device_id
|
self.routing_key = routing_key
|
self._check_cleaner()
|
|
|
def _check_cleaner(self):
|
|
success = self.Open()
|
|
log = "Unknown log."
|
if success:
|
log = f"The port [{self.port}] refered to the dust cleaner is open."
|
else:
|
log = f"The port [{self.port}] refered to the dust cleaner can not open."
|
|
# display
|
print(f"{time.asctime()} | {log}")
|
|
send_logs_tool(logs=[log],
|
logs_queue=self.logs_queue,
|
source = self.device_id,
|
routing_key=self.routing_key+".logs.test0")
|
|
|
|
# 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
|
|
# # TODO根据新的数据结构修改队消息队列。
|
# 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")
|
