hn3d-server-control.git

			@@ -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