From 0dc337197311d0fec373bf479f8ee0ac26b4cad6 Mon Sep 17 00:00:00 2001
From: wu_xinjun <bent20140204@outlook.com>
Date: 星期三, 15 六月 2022 16:33:47 +0800
Subject: [PATCH] client enhanced&camera contorller
---
src/protos/aerial.proto | 14
scripts/mqClient.py | 20 -
scripts_launch/TEST_startClient.bat | 4
src/Server.py | 21 +
src/engine/MessageThreads.py | 20 +
src/engine/AcquisitionMain.py | 317 ++++++++++++-------------------
src/engine/devices.py | 170 +++++++++++++---
7 files changed, 304 insertions(+), 262 deletions(-)
diff --git a/scripts/mqClient.py b/scripts/mqClient.py
index 56f546c..cca6c24 100644
--- a/scripts/mqClient.py
+++ b/scripts/mqClient.py
@@ -8,7 +8,7 @@
from src.engine.AcquisitionMain import AcquisitionMain
-from src.engine.AcquisitionMain import Config
+# from src.engine.AcquisitionMain import Config
def quit(signum, frame):
print('You choose to stop me.')
@@ -20,16 +20,14 @@
print("EXAMPLE: Client.py 172.17.17.206 chai password123 aerial_rpc pod1 http://172.17.17.206/api 192.168.2.65")
exit(1)
- if len(sys.argv) != 8:
+ if len(sys.argv) != 6:
help()
try:
MQhost = str(sys.argv[1])
MQuser = str(sys.argv[2])
MQpassword = str(sys.argv[3])
- QUEUEname = str(sys.argv[4])
- DEVICEname = str(sys.argv[5])
- SERVERhost = str(sys.argv[6])
- MICROWAVEhost = str(sys.argv[7])
+ DEVICEname = str(sys.argv[4])
+ MICROWAVEhost = str(sys.argv[5])
except Exception as e:
print(e)
@@ -40,15 +38,13 @@
signal.signal(signal.SIGTERM, quit)
# client = AcquisitionMain('192.168.3.54', 'chai',
# 'password123', 'aerial_rpc', 'pod1', 'http://192.168.3.54:5000/api', 10)
- config_dir = os.path.join(__file__, "..", "../src/config.json" )
- config_path = os.path.abspath(config_dir)
- config = Config(config_path)
+ # config_dir = os.path.join(__file__, "..", "../src/config.json" )
+ # config_path = os.path.abspath(config_dir)
+ # config = Config(config_path)
client = AcquisitionMain(MQhost,
MQuser,
MQpassword,
- QUEUEname,
DEVICEname,
- SERVERhost,
MICROWAVEhost, # udp server host, 192.168.2.65 for production
- config)
+ )
client.start()
diff --git a/scripts_launch/TEST_startClient.bat b/scripts_launch/TEST_startClient.bat
index f399e0d..fcbf203 100644
--- a/scripts_launch/TEST_startClient.bat
+++ b/scripts_launch/TEST_startClient.bat
@@ -20,7 +20,7 @@
set MQuser=chai
set MQpassword=password123
set QUEUEname=aerial_rpc
-set DEVICEname=pod1
+set DEVICEname=client.client.client
set SERVERhost=http://192.168.1.224/api
set MICROWAVEhost=localhost
set LOGfile=ClientTEST.log
@@ -30,6 +30,6 @@
echo Client.py is running ...
@REM -------- Usage: Client.py [MQhost] [MQuser] [MQpassword] [QUEUEname] [DEVICEname] [SERVERhost] [MICROWAVEhost]
@REM %aerialpython% ./src/Client.py %MQhost% %MQuser% %MQpassword% %QUEUEname% %DEVICEname% %SERVERhost% %MICROWAVEhost% >> %LOGfile%
-%aerialpython% ./scripts/mqClient.py %MQhost% %MQuser% %MQpassword% %QUEUEname% %DEVICEname% %SERVERhost% %MICROWAVEhost%
+%aerialpython% ./scripts/mqClient.py %MQhost% %MQuser% %MQpassword% %DEVICEname% %MICROWAVEhost%
echo Client.py has been terminited.
pause
diff --git a/src/Server.py b/src/Server.py
index 3b50887..af20966 100644
--- a/src/Server.py
+++ b/src/Server.py
@@ -34,7 +34,7 @@
self.WEBhost = WEBhost
self.WEBport = WEBport
self.save_folder = save_folder
- self.server_name = "server0"
+ self.server_name = server_name
# self.route('/api', methods=['POST'])(self.api)
@@ -53,9 +53,14 @@
self.parameters = pika.ConnectionParameters(
credentials=credentials, host=self.MQhost)
# 杩炴帴闃熷垪鏈嶅姟鍣�
- connection = pika.BlockingConnection(self.parameters)
- print(f"connected to MQ server: {self.MQhost} with user {self.MQuser}")
- connection.close()
+ try:
+ self.connection = pika.BlockingConnection(self.parameters)
+ print(f"connected to MQ server: {self.MQhost} with user {self.MQuser}")
+ self.connection.close()
+ except Exception :
+ print("Can't connecting to the MQ server")
+ exit(1)
+
def _receiver_Thread_Launcher(self):
@@ -66,7 +71,7 @@
exchange_name='logs_topic',
exchange_type='topic',
MQqueue_name=f'logs_queue',
- binding_keys='server.logs.test0',
+ binding_keys=['server.logs.test0'],
message_queue=self.LogsReceiver_queue
)
self.LogsReceiver_Thread.setDaemon(True)
@@ -79,7 +84,7 @@
exchange_name='images_topic',
exchange_type='topic',
MQqueue_name=f'images_queue',
- binding_keys="server.images.test0",
+ binding_keys=["server.images.test0"],
message_queue=self.ImagesReceiver_queue
)
self.ImagesReceiver_Thread.setDaemon(True)
@@ -92,7 +97,7 @@
exchange_name='plys_topic',
exchange_type='topic',
MQqueue_name=f'plys_queue',
- binding_keys='server.plys.test0',
+ binding_keys=['server.plys.test0'],
message_queue=self.PlysReceiver_queue
)
self.PlysReceiver_Thread.setDaemon(True)
@@ -115,7 +120,7 @@
def _sender_Thread_Launcher(self):
- # launcher command sender thread
+ # launcher commands sender thread
self.CommandsSender_Thread = ths.senderThread(
MQparams=self.parameters,
exchange_name='commands_topic',
diff --git a/src/engine/AcquisitionMain.py b/src/engine/AcquisitionMain.py
index bb2c8be..17d80ed 100644
--- a/src/engine/AcquisitionMain.py
+++ b/src/engine/AcquisitionMain.py
@@ -1,112 +1,71 @@
import sys # NOQA: E402
import os
-import cv2
-import logging
-import json
import pika
import queue
-import requests
-import time
-import threading
-from .CaptureThread import CaptureThread
-from .ControllThread import ControllThread
-from .MicroWaveUDPThread import MicroWaveUDPThread
+# from .CaptureThread import CaptureThread
+# from .ControllThread import ControllThread
+# from .MicroWaveUDPThread import MicroWaveUDPThread
from . import MessageThreads as ths
-from .devices import cameraController
+from .devices import cameraController, dustCleanerController
# # 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 ..protos import aerial_pb2 as pb
IMAGE_WIDTH = 3840
IMAGE_HEIGHT = 2880
CAMERA_COUNT = 4
-class Config:
- """
- config class
- """
- def __init__(self, config_file: str):
- self.interval: float = 10
- self.resolution = '4k'
+# class Config:
+# """
+# config class
+# """
+# def __init__(self, config_file: str):
+# self.interval: float = 10
+# self.resolution = '4k'
- self.clean_lens_start_angle = 10
- self.clean_lens_end_angle = 90
- self.clean_lens_num = 2
- self.clean_lens_period = 1200
- self.clean_self_start_angle = 10
- self.clean_self_end_angle = 55
- self.clean_self_num = 2
- self.clean_self_period = 240
+# self.clean_lens_start_angle = 10
+# self.clean_lens_end_angle = 90
+# self.clean_lens_num = 2
+# self.clean_lens_period = 1200
+# self.clean_self_start_angle = 10
+# self.clean_self_end_angle = 55
+# self.clean_self_num = 2
+# self.clean_self_period = 240
- # load config file
- with open(config_file, 'r') as f:
- self.config = json.load(f)
- self.device_id = self.config['Device_ID']
- self.interval = self.config['Captue_Image_para']['capture_interval']
- self.resolution = self.config['Captue_Image_para']['resolution']
+# # load config file
+# with open(config_file, 'r') as f:
+# self.config = json.load(f)
+# self.device_id = self.config['Device_ID']
+# self.interval = self.config['Captue_Image_para']['capture_interval']
+# self.resolution = self.config['Captue_Image_para']['resolution']
- Clean_Brush_para = self.config['Clean_Brush_para']
- self.clean_lens_start_angle = Clean_Brush_para['clean_lens_start_angle']
- self.clean_lens_end_angle = Clean_Brush_para['clean_lens_end_angle']
- self.clean_lens_num = Clean_Brush_para['clean_lens_num']
- self.clean_lens_period = Clean_Brush_para['clean_lens_period']
- self.clean_self_start_angle = Clean_Brush_para['clean_self_start_angle']
- self.clean_self_end_angle = Clean_Brush_para['clean_self_end_angle']
- self.clean_self_num = Clean_Brush_para['clean_self_num']
- self.clean_self_period = Clean_Brush_para['clean_self_period']
- f.close()
- return
+# Clean_Brush_para = self.config['Clean_Brush_para']
+# self.clean_lens_start_angle = Clean_Brush_para['clean_lens_start_angle']
+# self.clean_lens_end_angle = Clean_Brush_para['clean_lens_end_angle']
+# self.clean_lens_num = Clean_Brush_para['clean_lens_num']
+# self.clean_lens_period = Clean_Brush_para['clean_lens_period']
+# self.clean_self_start_angle = Clean_Brush_para['clean_self_start_angle']
+# self.clean_self_end_angle = Clean_Brush_para['clean_self_end_angle']
+# self.clean_self_num = Clean_Brush_para['clean_self_num']
+# self.clean_self_period = Clean_Brush_para['clean_self_period']
+# f.close()
+# return
class AcquisitionMain (object):
- """
- acquisition main class
- """
- # def __init__(self,
- # rabbitmq_host: str,
- # rabbitmq_user: str,
- # rabbitmq_password: str,
- # rabbitmq_queue: str,
- # device_id: str,
- # api_url: str,
- # udp_server_host: str,
- # config: Config) -> None:
- # """
- # init AcquisitionMain class
- # Args:
- # rabbitmq_host: rabbitmq server host
- # rabbitmq_user: rabbitmq user
- # rabbitmq_password: rabbitmq password
- # rabbitmq_queue: rabbitmq upload queue
- # device_id: device id
- # api_url: server api url
- # udp_server_host: udp server host for micro wave, localhost for testing or 192.168.2.65 for production
- # config: config object
- # """
- # self.rabbitmq_host: str = rabbitmq_host
- # self.udp_server_host: str = udp_server_host
- # self.rabbitmq_user: str = rabbitmq_user
- # self.rabbitmq_password: str = rabbitmq_password
- # self.rabbitmq_queue: str = rabbitmq_queue
- # self.device_id: str = device_id
- # self.message_send: int = 0
- # self.api_url: str = api_url
- # self.config = config
- # self.image_quality: float = 95
- # self.cameras: list = []
- # self.msg_queue = queue.Queue()
- # # thread members
- # self._controll_thread = None
- # self._capture_thread = None
- # self._udp_thread = None
- def __init__(self, MQhost, MQuser, MQpassword, deviceID, MICROWAVEhost):
+ def __init__(self,
+ MQhost:str,
+ MQuser:str,
+ MQpassword:str,
+ deviceID:str,
+ MICROWAVEhost:str):
# device ID
self.device_id = deviceID
@@ -122,28 +81,14 @@
# microwave
self.MICROWAVEhost = MICROWAVEhost
# threads
- self._device_state_sender_thread = None
- self._alarm_state_sender_thread = None
- self._command_request_receiver_thread = None
- self._command_response_sender_thread = None
- # queue
- self.device_state_queue = queue.Queue()
- self.alarm_state_queue = queue.Queue()
- self.command_request_queue = queue.Queue()
- self.command_response_queue = queue.Queue()
- self.image_data_queue = queue.Queue()
- self.udp_data_queue = queue.Queue()
- self.ply_data_queue = queue.Queue()
+ self.threads = []
- self.queue_dict = {
- "device_state":self.device_state_queue,
- "alarm_state":self.alarm_state_queue,
- "command_request":self.command_request_queue,
- "command_response":self.command_response_queue,
- "image_data":self.image_data_queue,
- "udp_data":self.udp_data_queue,
- "ply_data":self.ply_data_queue
- }
+ # queue
+ self.CommandsReceiver_queue = queue.Queue()
+
+ self.LogsSender_queue = queue.Queue()
+ self.ImagesSender_queue = queue.Queue()
+ self.PlysSender_queue = queue.Queue()
# device
self.cc = None # camera controller
@@ -159,6 +104,7 @@
}
+
def _connect_to_MQ(self):
credentials = pika.PlainCredentials(self.MQuser, self.MQpassword)
self.parameters = pika.ConnectionParameters(
@@ -169,100 +115,92 @@
print(f"connected to MQ server: {self.MQhost} with user {self.MQuser}")
self.connection.close()
except Exception :
- print("can't connecting to the MQ server")
+ print("Can't connecting to the MQ server")
exit(1)
- def _start_device_state_up_msg_sender_thread(self):
- """start device state thread"""
- if self._device_state_sender_thread is None:
- self._device_state_sender_thread = \
- ths.device_state_senderThread(self.parameters,\
- self.device_state_queue,self.device_id)
- self._device_state_sender_thread.setDaemon(True)
- self._device_state_sender_thread.start()
+ def _receiver_Thread_Launcher(self):
- def _stop_device_state_up_msg_sender_thread(self):
- """stop device state thread"""
- if self._device_state_sender_thread is not None:
- self._device_state_sender_thread.stop()
- self._device_state_sender_thread = None
+ # launch a commands receiver thread
+ self.CommandsReceiver_Thread = ths.receiverThread(
+ MQparams=self.parameters,
+ exchange_name='commands_topic',
+ exchange_type='topic',
+ MQqueue_name="commands_queue",
+ binding_keys=[f"{self.device_id}"],
+ message_queue=self.CommandsReceiver_queue
+ )
+ self.CommandsReceiver_Thread.setDaemon(True)
+ self.CommandsReceiver_Thread.setName("commands_thread")
+ self.CommandsReceiver_Thread.start()
- def _start_alarm_state_up_msg_sender_thread(self):
- """start alarm state thread"""
- if self._alarm_state_sender_thread is None:
- self._alarm_state_sender_thread = \
- ths.alarm_state_senderThread(self.parameters,\
- self.alarm_state_queue,self.device_id)
- self._alarm_state_sender_thread.setDaemon(True)
- self._alarm_state_sender_thread.start()
+ self.threads.append(self.CommandsReceiver_Thread)
+
+ def _sender_Thread_Launcher(self):
- def _stop_alarm_state_up_msg_sender_thread(self):
- """stop alarm state thread"""
- if self._alarm_state_sender_thread is not None:
- self._alarm_state_sender_thread.stop()
- self._alarm_state_sender_thread = None
+ # launch a logs sender thread
+ self.LogsSender_Thread = ths.senderThread(
+ MQparams=self.parameters,
+ exchange_name="logs_topic",
+ exchange_type="topic",
+ message_queue=self.LogsSender_queue
+ )
+ self.LogsSender_Thread.setDaemon(True)
+ self.LogsSender_Thread.setName("logs_thread")
+ self.LogsSender_Thread.start()
+
+ self.threads.append(self.LogsSender_Thread)
- def _start_command_response_sender_thread(self):
- """start command response thread"""
- if self._command_response_sender_thread is None:
- self._command_response_sender_thread = \
- ths.command_response_senderThread(self.parameters, \
- self.command_response_queue,self.device_id)
- self._command_response_sender_thread.setDaemon(True)
- self._command_response_sender_thread.start()
+ # launch a images sender thread
+ self.ImagesSender_Thread = ths.senderThread(
+ MQparams=self.parameters,
+ exchange_name="images_topic",
+ exchange_type="topic",
+ message_queue=self.ImagesSender_queue
+ )
+ self.ImagesSender_Thread.setDaemon(True)
+ self.ImagesSender_Thread.setName("image_thread")
+ self.ImagesSender_Thread.start()
- def _stop_command_response_sender_thread(self):
- """stop command response thread"""
- if self._command_response_sender_thread is not None:
- self._command_response_sender_thread.stop()
- self._command_response_sender_thread = None
+ self.threads.append(self.ImagesSender_Thread)
+
+ # launch a plys sender thread
+ self.PlysSender_Thread = ths.senderThread(
+ MQparams=self.parameters,
+ exchange_name="plys_topic",
+ exchange_type="topic",
+ message_queue=self.PlysSender_queue
+ )
+ self.PlysSender_Thread.setDaemon(True)
+ self.PlysSender_Thread.setName("plys_thread")
+ self.PlysSender_Thread.start()
+
+ self.threads.append(self.PlysSender_Thread)
- def _start_command_request_receiver_thread(self):
- """start command request thread"""
- if self._command_request_receiver_thread is None:
- self._command_request_receiver_thread = \
- ths.command_request_receiverThread(self.parameters,\
- self.command_request_queue,self.command_response_queue,self.device_id)
- self._command_request_receiver_thread.setDaemon(True)
- self._command_request_receiver_thread.start()
- def _stop_command_request_receiver_thread(self):
- """stop command request thread"""
- if self._command_request_receiver_thread is not None:
- self._command_request_receiver_thread.stop()
- self._command_request_receiver_thread = None
+ def _connect_device(self):
+ # connect to cameras
+ self.cc = cameraController(
+ IMAGE_WIDTH=IMAGE_WIDTH,
+ IMAGE_HEIGHT=IMAGE_HEIGHT,
+ CAMERA_COUNT=CAMERA_COUNT,
+ device_id=self.device_id,
+ logs_queue=self.LogsSender_queue,
+ images_queue=self.ImagesSender_queue,
+ init_binding_key="client.client.client",
+ init_routing_key="server"
+ )
+ # self.dc = dustCleanerController()
def run_rabbitmq(self):
self._connect_to_MQ()
- self._start_device_state_up_msg_sender_thread()
- self._start_alarm_state_up_msg_sender_thread()
- self._start_command_response_sender_thread()
- self._start_command_request_receiver_thread()
+ self._receiver_Thread_Launcher()
+ self._sender_Thread_Launcher()
def run_device(self):
- self.connecting_camera()
- self.connecting_microwave()
- self.connecting_dustCleaner()
-
- def connecting_camera(self):
- # init the camera controller
- cc = cameraController(self.IMAGE_WIDTH,self.IMAGE_HEIGHT,\
- self.CAMERA_COUNT,self.device_id,self.queue_dict)
- # connecting to camera via opencv
- cc.open_cameras()
- # capture the data, and wrap it to data queue
- cc._start_captureThread()
-
-
- def connecting_microwave(self):
- pass
-
- def connecting_dustCleaner(self):
- pass
-
+ self._connect_device()
def run_all(self):
@@ -273,21 +211,20 @@
self.run_device()
def stop_all(self):
- self._stop_alarm_state_up_msg_sender_thread()
- self._stop_device_state_up_msg_sender_thread()
- self._stop_command_request_receiver_thread()
- self._start_command_response_sender_thread()
+ for t in self.threads:
+ t.stop()
+ print(f"stop {t.name}")
def start(self):
try:
self.run_all()
- self._device_state_sender_thread.join()
+ print("[*] Client is runing ...")
+ self.CommandsReceiver_Thread.join()
except Exception as e:
+ print(e)
print("running error, stop all threads")
-
- self._stop_alarm_state_up_msg_sender_thread()
- self.stop
+ self.stop_all()
print("main thread end.")
return 0
diff --git a/src/engine/MessageThreads.py b/src/engine/MessageThreads.py
index 04d86e2..98e7cc0 100644
--- a/src/engine/MessageThreads.py
+++ b/src/engine/MessageThreads.py
@@ -50,7 +50,7 @@
def stop(self):
"""stop the thread"""
self.is_stop = True
- self.connection.close()
+ # self.connection.close()
def run(self):
@@ -62,7 +62,7 @@
while True:
try:
- if not self.message_queue.empty():
+ if not self.message_queue.empty() and not self.is_stop:
# decode the routing key and message body from tuple list
message = self.message_queue.get()
@@ -115,7 +115,7 @@
def stop(self):
"""stop the thread"""
self.is_stop = True
- self.connection.close()
+ # self.connection.close()
def run(self):
@@ -130,6 +130,7 @@
channel.queue_declare(queue=self.MQqueue_name, durable=True)
for binding_key in self.binding_keys:
channel.queue_bind(exchange=self.exchange_name, queue=self.MQqueue_name, routing_key=binding_key)
+ print(f"[*] Using [{binding_key}] bind [{self.MQqueue_name}] on [{self.exchange_name}]")
# consume the message
channel.basic_qos(prefetch_count=1)
@@ -140,6 +141,9 @@
channel.start_consuming()
def callback(self,ch, method, properties, body):
+
+ if self.is_stop:
+ ch.close()
self.message_queue.put(body)
ch.basic_ack(delivery_tag = method.delivery_tag)
@@ -191,7 +195,7 @@
queue = self.queue_list[i_tag]
# make sure the queue is not empty
- while not queue.empty():
+ while not queue.empty() and not self.is_stop:
# decode the msg using cooresponding method
queue_tag = self.queue_tags[i_tag]
decode_method = self.decode_select(queue_tag)
@@ -214,7 +218,7 @@
while not queue.empty() and count < nums_limit:
# extracting the msg from queue
- msg = queue.pop()
+ msg = queue.get()
msg_pb = pb.LogsMessage()
msg_pb.ParseFromString(msg)
@@ -241,7 +245,7 @@
while not queue.empty():
# extracting the msg from queue
- msg = queue.pop()
+ msg = queue.get()
msg_pb = pb.ImagesMessage()
msg_pb.ParseFromString(msg)
@@ -268,7 +272,7 @@
filename = f"{msg_source}_{msg_timestamp}_image{id}.jpg"
filepath = os.path.join(self.imagesmsg_folder,filename)
with open(filepath, 'wb') as f:
- f.write(image)
+ f.write(image_byte)
# display the info to the console
curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
@@ -282,7 +286,7 @@
while not queue.empty():
# extracting the msg from queue
- msg = queue.pop()
+ msg = queue.get()
msg_pb = pb.PlyMessage()
msg_pb.ParseFromString(msg)
diff --git a/src/engine/devices.py b/src/engine/devices.py
index aacc2ac..29f4340 100644
--- a/src/engine/devices.py
+++ b/src/engine/devices.py
@@ -4,6 +4,9 @@
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))
@@ -14,65 +17,162 @@
class cameraController(object):
- def __init__(self,IMAGE_WIDTH,IMAGE_HEIGHT,CAMERA_COUNT,device_id,queue_dict):
+ 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.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
+ 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 open_cameras(self):
- """
- Test the ports and returns a tuple with the available ports and the ones that are working.
- """
+
+ 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)
- if camera.isOpened():
- is_reading, img = camera.read()
- w = camera.get(3)
- h = camera.get(4)
+ 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):
+
+ # constract a images message
+ imagesMsg = pb.ImagesMessage()
+ imagesMsg.source = self.device_id
+ curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
+ imagesMsg.timestamp = curr_time
+ log_rk = self.routing_key+".logs.test0"
+ image_rk = self.routing_key+".images.test0"
+
+ for log, img in zip(logs,images):
+
+ # encoding the images to the bytes
+ # encode_params = [int(cv2.IMWRITE_JPEG_QUALITY), self.image_quality]
+ # image_encoded = cv2.imencode(".jpg",single_img,encode_params)
+ if img is not None:
+ success,image_encoded = cv2.imencode(".jpg",img)
+ if success:
+ image_bytes = image_encoded.tobytes()
+ image_length = len(image_bytes)
+ # add the encoded image to msg
+ image_item = imagesMsg.Image.add()
+ image_item.length = image_length
+ image_item.data = image_bytes
+
+ if log is not None:
+ # constract a logs message
+ logsMsg = pb.LogsMessage()
+ logsMsg.source = self.device_id
+ logsMsg.timestamp = f"{time.asctime()}"
+ logsMsg.context = log
+ # convert the proto object to string
+ logsMsg_bytes = logsMsg.SerializeToString()
+ # add logsmsg to queue
+ self.logs_queue.put([log_rk,logsMsg_bytes])
+
+
+ # converting the pb image msg to bytes
+ imageMsg_bytes = imagesMsg.SerializeToString()
+ # add the imageMsg to the queue
+ self.images_queue.put([image_rk,imageMsg_bytes])
+
+
+ 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.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})"
+ self.camera_work[dev_port] == True
+
- 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"
+ log_list.append(log)
+ image_list.append(img)
- # add msg to queue
- self.device_state_queue.put(msg)
- print(msg)
+ return log_list, image_list
- 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 capture(self,routing_key:str):
- def _stop_captureThread(self):
- """stop capture thread"""
- if self._capture_thread is not None:
- self._capture_thread.stop()
- self._capture_thread = None
+ self.routing_key = routing_key
+
+ (logs,images) = self._capture_image()
+ self._generate_msg = (logs,images)
+
+ self.routing_key = self.init_routing_key
class microwaveController(object):
pass
class dustCleanerController(object):
+
pass
diff --git a/src/protos/aerial.proto b/src/protos/aerial.proto
index 04e7b2f..bbab4ff 100644
--- a/src/protos/aerial.proto
+++ b/src/protos/aerial.proto
@@ -140,12 +140,12 @@
enum DataType
{
- c_int16 = 1;
- c_int32 = 2;
- c_int64 = 3;
- c_float16 = 4;
- c_float32 = 5;
- c_float = 6;
+ c_int16 = 0;
+ c_int32 = 1;
+ c_int64 = 2;
+ c_float16 = 3;
+ c_float32 = 4;
+ c_float = 5;
}
@@ -161,7 +161,7 @@
int32 cols = 3;
string order = 4;
DataType d_types = 5;
- bytes data = 6
+ bytes data = 6;
}
--
Gitblit v1.9.1