hn3d-server-control.git

			@@ -11,6 +11,8 @@
			from .CaptureThread import CaptureThread
			from .ControllThread import ControllThread
			from .MicroWaveUDPThread import MicroWaveUDPThread
			from . import MessageThreads as ths
			from .CameraDevice import cameraController

			# add python path of src to sys.path
			src_path = os.path.join(__file__, (['..'] 2))
			@@ -60,198 +62,383 @@
			f.close()
			return

			class AcquisitionMain:
			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,
			# 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 _start_udp_thread(self):
			"""start udp thread"""
			if self._udp_thread is None:
			self._udp_thread = MicroWaveUDPThread((self.udp_server_host, 9911), self.msg_queue)
			self._udp_thread.setDaemon(True)
			self._udp_thread.start()

			def _stop_udp_thread(self):
			"""stop udp thread"""
			if not self._udp_thread is None:
			self._udp_thread.stop()
			self._udp_thread = None
			def __init__(self, MQhost, MQuser, MQpassword, deviceID, MICROWAVEhost):

			def _start_capture_thread(self):
			"""start camera capture thread"""
			if self._capture_thread is None:
			self._open_cameras()
			if len(self.cameras) == 0:
			print("No camera available")
			return
			self._capture_thread = CaptureThread(
			self.cameras, self.msg_queue, self.config.interval, self.image_quality)
			self._capture_thread.setDaemon(True)
			self._capture_thread.start()
			# device ID
			self.device_id = deviceID
			# rabbit MQ
			self.MQhost = MQhost
			self.MQuser = MQuser
			self.MQpassword = MQpassword
			self.parameters = None
			# camera
			self.IMAGE_WIDTH = 3840
			self.IMAGE_HEIGHT = 2880
			self.CAMERA_COUNT = 4
			# 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()

			def _stop_capture_thread(self):
			"""stop camera capture thread"""
			if not self._capture_thread is None:
			self._capture_thread.stop()
			self._capture_thread = None
			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
			}

			def _start_controll_thread(self):
			"""start controll thread"""
			if self._controll_thread is None:
			self._controll_thread = ControllThread(
			self.rabbitmq_host, self.rabbitmq_user, self.rabbitmq_password, self.device_id + "_callback",
			self.config)
			self._controll_thread.setDaemon(True)
			self._controll_thread.start()

			def _stop_controll_thread(self):
			"""stop controll thread"""
			if not self._controll_thread is None:
			self._controll_thread.stop()
			self._controll_thread = None

			def _connect_message_queue(self) -> None:
			"""connect message queue of rabbitmq"""
			credentials = pika.PlainCredentials(self.rabbitmq_user, self.rabbitmq_password)
			parameters = pika.ConnectionParameters(
			credentials=credentials, host=self.rabbitmq_host)
			def _connect_to_MQ(self):
			credentials = pika.PlainCredentials(self.MQuser, self.MQpassword)
			self.parameters = pika.ConnectionParameters(
			credentials=credentials, host=self.MQhost)
			# 连接队列服务器
			self._connection = pika.BlockingConnection(parameters)
			self.channel = self._connection.channel()
			self.channel.queue_declare(queue='', exclusive=True)
			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 _close_message_queue(self) -> None:
			"""close message queue of rabbitmq"""
			self._connection.close()
			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 _set_logging(self):
			"""set logging"""
			logging.basicConfig(level=logging.INFO,
			format='%(asctime)s %(levelname)s %(message)s',
			datefmt='%Y-%m-%d %H:%M:%S')
			self.logger = logging.getLogger(__name__)
			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

			def _send_data(self, req: pb.RequestCommand) -> None:
			"""send data to rabbitmq"""
			req.device_id = self.device_id
			body = req.SerializeToString()
			self.channel.basic_publish(
			exchange='',
			routing_key=self.rabbitmq_queue,
			body=body)
			self.message_send += 1
			return
			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()

			def _call(self, req: pb.RequestCommand, timeout: int = 10) -> pb.ResponseCommand:
			"""call server api"""
			req.device_id = self.device_id
			self.response = None
			body = req.SerializeToString()
			response = requests.post(
			self.api_url,
			data=body,
			headers={'Content-Type': 'application/x-protobuf'},
			timeout=timeout)
			resp = pb.ResponseCommand()
			resp.ParseFromString(response.content)
			return resp
			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

			def _send_power_on_message(self) -> pb.ResponseCommand:
			"""send power on message"""
			print("send power on message")
			req = pb.RequestCommand()
			req.sys_command = pb.SysCommand.SYS_COMMAND
			req.sub_command = pb.SubCommand.POWER_ON_SUB_COMMAND
			return self._call(req)

			def _send_power_off_message(self) -> pb.ResponseCommand:
			"""send power off message"""
			print("send power off message")
			req = pb.RequestCommand()
			req.sys_command = pb.SysCommand.SYS_COMMAND
			req.sub_command = pb.SubCommand.POWER_OFF_SUB_COMMAND
			return self._call(req)
			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()

			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(CAMERA_COUNT):
			camera = cv2.VideoCapture(dev_port, cv2.CAP_DSHOW)
			camera.set(cv2.CAP_PROP_FRAME_WIDTH, IMAGE_WIDTH)
			camera.set(cv2.CAP_PROP_FRAME_HEIGHT, 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:
			print("Port %s is working and reads images (%s x %s)" %
			(dev_port, h, w))
			self.working_ports.append(dev_port)
			self.cameras.append(camera)
			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

			def _main_loop(self):
			# main loop of the acquisition main program
			self._connect_message_queue()
			while True:
			try:
			if not self.msg_queue.empty():
			msg = self.msg_queue.get()
			self._send_data(msg)
			print(time.asctime(), "send message")
			time.sleep(0.1)
			except Exception as e:
			break
			self._close_message_queue()
			self._stop_udp_thread()
			self._stop_capture_thread()
			self._stop_controll_thread()
			print("Stopped")

			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 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()

			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()
			pass

			def connecting_microwave(self):
			pass

			def connecting_dustCleaner(self):
			pass



			def run_all(self):
			# connecting to MQ server,launching the message delivery queues
			self.run_rabbitmq()
			# connecting to devices and wrap data to the corressponding queue
			self.run_device()


			def start(self):
			self.run_all()





























			# def _start_udp_thread(self):
			# """start udp thread"""
			# if self._udp_thread is None:
			# self._udp_thread = MicroWaveUDPThread((self.udp_server_host, 9911), self.msg_queue)
			# self._udp_thread.setDaemon(True)
			# self._udp_thread.start()

			# def _stop_udp_thread(self):
			# """stop udp thread"""
			# if not self._udp_thread is None:
			# self._udp_thread.stop()
			# self._udp_thread = None

			# def _start_capture_thread(self):
			# """start camera capture thread"""
			# if self._capture_thread is None:
			# self._open_cameras()
			# if len(self.cameras) == 0:
			# print("No camera available")
			# return
			# self._capture_thread = CaptureThread(
			# self.cameras, self.msg_queue, self.config.interval, self.image_quality)
			# self._capture_thread.setDaemon(True)
			# self._capture_thread.start()

			# def _stop_capture_thread(self):
			# """stop camera capture thread"""
			# if not self._capture_thread is None:
			# self._capture_thread.stop()
			# self._capture_thread = None

			# def _start_controll_thread(self):
			# """start controll thread"""
			# if self._controll_thread is None:
			# self._controll_thread = ControllThread(
			# self.rabbitmq_host, self.rabbitmq_user, self.rabbitmq_password, self.device_id + "_callback",
			# self.config)
			# self._controll_thread.setDaemon(True)
			# self._controll_thread.start()

			# def _stop_controll_thread(self):
			# """stop controll thread"""
			# if not self._controll_thread is None:
			# self._controll_thread.stop()
			# self._controll_thread = None

			# def _connect_message_queue(self) -> None:
			# """connect message queue of rabbitmq"""
			# credentials = pika.PlainCredentials(self.rabbitmq_user, self.rabbitmq_password)
			# parameters = pika.ConnectionParameters(
			# credentials=credentials, host=self.rabbitmq_host)
			# # 连接队列服务器
			# self._connection = pika.BlockingConnection(parameters)
			# self.channel = self._connection.channel()
			# self.channel.queue_declare(queue='', exclusive=True)

			# def _close_message_queue(self) -> None:
			# """close message queue of rabbitmq"""
			# self._connection.close()

			# def _set_logging(self):
			# """set logging"""
			# logging.basicConfig(level=logging.INFO,
			# format='%(asctime)s %(levelname)s %(message)s',
			# datefmt='%Y-%m-%d %H:%M:%S')
			# self.logger = logging.getLogger(__name__)

			# def _send_data(self, req: pb.RequestCommand) -> None:
			# """send data to rabbitmq"""
			# req.device_id = self.device_id
			# body = req.SerializeToString()
			# self.channel.basic_publish(
			# exchange='',
			# routing_key=self.rabbitmq_queue,
			# body=body)
			# self.message_send += 1
			# return

			# def _call(self, req: pb.RequestCommand, timeout: int = 10) -> pb.ResponseCommand:
			# """call server api"""
			# req.device_id = self.device_id
			# self.response = None
			# body = req.SerializeToString()
			# response = requests.post(
			# self.api_url,
			# data=body,
			# headers={'Content-Type': 'application/x-protobuf'},
			# timeout=timeout)
			# resp = pb.ResponseCommand()
			# resp.ParseFromString(response.content)
			# return resp

			# def _send_power_on_message(self) -> pb.ResponseCommand:
			# """send power on message"""
			# print("send power on message")
			# req = pb.RequestCommand()
			# req.sys_command = pb.SysCommand.SYS_COMMAND
			# req.sub_command = pb.SubCommand.POWER_ON_SUB_COMMAND
			# return self._call(req)

			# def _send_power_off_message(self) -> pb.ResponseCommand:
			# """send power off message"""
			# print("send power off message")
			# req = pb.RequestCommand()
			# req.sys_command = pb.SysCommand.SYS_COMMAND
			# req.sub_command = pb.SubCommand.POWER_OFF_SUB_COMMAND
			# return self._call(req)

			# 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(CAMERA_COUNT):
			# camera = cv2.VideoCapture(dev_port, cv2.CAP_DSHOW)
			# camera.set(cv2.CAP_PROP_FRAME_WIDTH, IMAGE_WIDTH)
			# camera.set(cv2.CAP_PROP_FRAME_HEIGHT, 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:
			# print("Port %s is working and reads images (%s x %s)" %
			# (dev_port, h, w))
			# self.working_ports.append(dev_port)
			# self.cameras.append(camera)

			# def _main_loop(self):
			# # main loop of the acquisition main program
			# self._connect_message_queue()
			# while True:
			# try:
			# if not self.msg_queue.empty():
			# msg = self.msg_queue.get()
			# self._send_data(msg)
			# print(time.asctime(), "send message")
			# time.sleep(0.1)
			# except Exception as e:
			# break
			# self._close_message_queue()
			# self._stop_udp_thread()
			# self._stop_capture_thread()
			# self._stop_controll_thread()
			# print("Stopped")

			def start(self) -> None:
			"""start all threads"""
			self._start_controll_thread()
			self._start_capture_thread()
			self._start_udp_thread()
			self._main_loop()
			# def start(self) -> None:
			# """start all threads"""
			# self._start_controll_thread()
			# self._start_capture_thread()
			# self._start_udp_thread()
			# self._main_loop()