-_-

wu_xinjun

2022-06-10 baee79553738a212d82a12a1966fe57ce4acb937

-_-

 src/Server1.py

New file
@@ -0,0 +1,134 @@
import sys              # NOQA: E402    
import os
from flask import Flask
import pika
import uuid
import queue
from werkzeug.wrappers import request
from threading import Thread
from datetime import datetime



from .engine.DeviceManager import DeviceManager
from .engine import MessageThreads as ths
from .protos import aerial_pb2 as pb



class MyServer(Flask):
    def __init__(self, host, user, password, queue, port, SAVEfolder):
        Flask.__init__(self, __name__)
        self.host = host
        self.user = user
        self.password = password
        self.queue = queue
        self.port = port
        global save_folder
        save_folder = SAVEfolder
        

        self.route('/api', methods=['POST'])(self.api)
        self.route('/', methods=['GET'])(self.hello)
        self._def_routes()
        # threads

        # queue
        self.command_request_queue = queue.Queue()

    def _connect_to_MQ(self):
        credentials = pika.PlainCredentials(self.MQuser, self.MQpassword)
        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()
        


    def _message_receiver_Thread_launcher(self):

        def callback(ch,method,properties,body):
            timestamp = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
            print(f"[x] {timestamp} | [{method.routing_key}]:{body}")

        target_args = (self.parameters,
                        "message_topic",
                        "topic",
                        "message_queue",
                        ["server.server.message", "server.server0.message"],
                        callback)

        self.message_receiver_thread = Thread(target=ths.message_receiver,
                                        args=target_args)
        self.message_receiver_thread.setDaemon(True)
        self.message_receiver_thread.start()


    def _image_receiver_Thread_launcher(self):

        def callback(ch,method,properties,body):
            
            timestamp = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
            print(f"[x] {timestamp} | [{method.routing_key}]:{body}")

        target_args = (self.parameters,
                        "message_topic",
                        "topic",
                        "message_queue",
                        ["server.server.message", "server.server0.message"],
                        callback)

        self.message_receiver_thread = Thread(target=ths.message_receiver,
                                        args=target_args)
        self.message_receiver_thread.setDaemon(True)
        self.message_receiver_thread.start()



    def _def_routes(self):
        self.routes = {
            pb.SysCommand.SYS_COMMAND: self._on_sys_command,
        }
    
    def _on_sys_command(self, req: pb.RequestCommand, resp: pb.ResponseCommand):

        self.command_request_queue.put(ths.Message(None,None,req.command,req.target))
        return

    def _default(self, req: pb.RequestCommand, resp: pb.ResponseCommand):
        print('onknown command')

    def api(self):

        req = pb.RequestCommand()
        req.ParseFromString(request.data)

        # a function: _on_sys_command, refer to self._on_sys_command
        dispatcher = self.routes.get(req.sys_command, self._default)

        resp = pb.ResponseCommand()
        dispatcher(req, resp) # call the function

        resp.checksum = 12345
        response = Flask.response_class(
            resp.SerializeToString(),
            mimetype='application/x-protobuf'
        )
        return response, 200

    def hello(self):
        return 'Hello, This is acquisition demo API!'

    def run_rabbitmq(self):
        self._connect_to_MQ()
        self._start_device_state_up_msg_receiver_thread()
        self._start_alarm_state_up_msg_receiver_thread()
        self._start_command_response_receiver_thread()
        self._start_command_request_sender_thread()

    def run_all(self):
        self.run_rabbitmq()
        self.run(debug=False, host=self.WEBhost, port=self.WEBport)


 src/engine/MessageThreads.py

@@ -1,15 +1,19 @@
import os
import queue
import threading
import pika
import time
import os
from time import sleep
import sys
from datetime import datetime
from typing import List

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



@@ -19,17 +23,27 @@
#         self.sender_key = sender_key
#         self.body = body
#         self.routing_key = routing_key
# 根据新的数据结构改写run

class senderThread(threading.Thread):
    def __init__(self, MQparameters,input_msg_queue,sender_id,exchangeName,exchangeType):
        super(senderThread,self).__init__()
        self.parameters =  MQparameters
        self.queue = input_msg_queue
        self.exchangeName = exchangeName
        self.exchangeType = exchangeType
        self.sender_id = sender_id
        self.is_stop = False
        self.connection = pika.BlockingConnection(self.parameters)

    def  __init__(self,
                    MQparams: pika.ConnectionParameters,
                    exchange_name: str,
                    exchange_type:str,
                    # routing_key:str,
                    message_queue:queue.Queue,
                        ):
            super(senderThread,self).__init__()
            
            self.MQparams = MQparams
            self.exchange_name = exchange_name
            self.exchange_type = exchange_type
            # self.routing_key = routing_key
            self.message_queue = message_queue

            # connect to RabbitMQ
            self.is_stop = False
            self.connection = pika.BlockingConnection(self.MQparams)

    def stop(self):
        """stop the thread"""
@@ -39,42 +53,62 @@
    def run(self):
    
        while not self.is_stop:
            # declare a exchange
            channel = self.connection.channel()
            channel.exchange_declare(exchange=self.exchangeName, exchange_type=self.exchangeType)
            channel.exchange_declare(exchange=self.exchange_name,
                                    exchange_type=self.exchange_type)

            while True:
                try:
                    if not self.queue.empty():
                        data= self.queue.get()
                        if isinstance(data, pb.CapturedData):
                            msg = pb.DataMessage()
                        else:
                            body = msg.body
                            routing_key = msg.routing_key
                            channel.basic_publish(exchange=self.exchangeName, routing_key=routing_key, body=body)
                            
                            response_tag = f"{time.asctime()},{self.sender_id} send {self.exchangeName} message to {routing_key}"
                    if not self.message_queue.empty():
                        
                            print(response_tag)
                        # decode the routing key and message body from tuple list
                        message = self.message_queue.get()
                        msg_routing_key = message[0]
                        msg_body = message[1]

                    elif self.queue.empty():
                        # publish a PERSISTENT message
                        channel.basic_publish(
                        exchange = self.exchange_name,
                        routing_key = msg_routing_key,
                        body = msg_body,
                        properties=pika.BasicProperties(
                            delivery_mode=pika.spec.PERSISTENT_DELIVERY_MODE
                                )
                        )

                        timestamp = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
                        caption = f"[x] {timestamp} | Sending message to[{msg_routing_key}]")
                        print(caption)

                    else:
                        time.sleep(1)

                except Exception as e:
                    print(e)
                    break

# 根据新的数据结构改写run
class receiverThread(threading.Thread):
    def __init__(self, MQparameters,received_msg_queue,output_response_queue,exchangeName,exchangeType,receiver_id):
        super(receiverThread,self).__init__()
        self.parameters =  MQparameters
        self.queue = received_msg_queue
        self.response_queue = output_response_queue
        self.exchangeName = exchangeName
        self.exchangeType = exchangeType
        self.receiver_id = receiver_id
        self.is_stop = False
        self.connection = pika.BlockingConnection(self.parameters)
    def  __init__(self,
                    MQparams: pika.ConnectionParameters,
                    exchange_name: str,
                    exchange_type:str,
                    MQqueue_name:str,
                    binding_keys:List(str),
                    message_queue:queue.Queue,
                        ):
            super(senderThread,self).__init__()
            
            self.MQparams = MQparams
            self.exchange_name = exchange_name
            self.exchange_type = exchange_type
            self.MQqueue_name = MQqueue_name
            self.binding_keys = binding_keys
            self.message_queue = message_queue

            # connect to RabbitMQ
            self.is_stop = False
            self.connection = pika.BlockingConnection(self.MQparams)

    def stop(self):
        """stop the thread"""
@@ -83,25 +117,178 @@


    def run(self):

        while not self.is_stop:

            # declare a exchange
            channel = self.connection.channel()
            channel.exchange_declare(exchange=self.exchangeName, exchange_type=self.exchangeType)
            channel.exchange_declare(exchange=self.exchange_name,
                                    exchange_type=self.exchange_type)

            result = channel.queue_declare(queue='', exclusive=True, durable=True)
            queue_name = result.method.queue
            # declare the queue and bind it with keys
            channel.queue_declare(queue=self.MQqueue_name, durable=True)
            for binding_key in self.binding_keys:
                channel.queue_bind(exchange=self.exchangeName, queue=self.MQqueue_name, routing_key=binding_key)

            channel.queue_bind(exchange=self.exchangeName, queue=queue_name, routing_key=self.receiver_id)

            channel.basic_consume(queue=queue_name, \
                on_message_callback=self.callback)

            # consume the message
            channel.basic_qos(prefetch_count=1)
            channel.basic_consume(
                queue=self.MQqueue_name,
                on_message_callback=self.callback
            )
            channel.start_consuming()

    def callback(ch, method, properties, body):
        raise NotImplementedError(
            "Need to implement the callback function")
    def callback(self,ch, method, properties, body):
        self.message_queue.put(body)
        ch.basic_ack(delivery_tag = method.delivery_tag)

class decodeThread(threading.Thread):
    def __init__(self,
        queue_list:List(queue.Queue),
        queue_tags:List(str),
        save_path:str):
        super().__init__()

        self.queue_list = queue_list
        self.queue_tags = queue_tags
        self.queue_num = 0
        self.save_path = save_path
        self.is_stop = False

        assert len(self.queue_list) == len(self.queue_tags)


    def stop(self):
        """stop the thread"""
        self.is_stop = True

    def run(self) -> None:

        # creating a path for saving data
        time_tag = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
        self.save_root_folder = os.path.join(self.save_path,time_tag)
        os.makedirs(self.save_root_folder)

        self.logsmsg_folder = os.path.join(self.save_root_folder,"LogsMsg")
        self.imagesmsg_folder = os.path.join(self.save_root_folder,"ImagesMsg")
        self.plysmsg_folder = os.path.join(self.save_root_folder,"PlysMsg")
        self.commandsmsg_folder = os.path.join(self.save_root_folder,"CommandsMsg")

        sub_folder_list = [self.logsmsg_folder,self.imagesmsg_folder,
                        self.plysmsg_folder,self.commandsmsg_folder]

        for folder in sub_folder_list:
            os.mkdir(folder)
            print(f"[NOTICE] Creating folder: {folder}")

        # process the queue list
        self.queue_num = len(self.queue_list)
        while self.queue_num:
            i = 0
            while True:
                # select a queue
                i_tag = i % self.queue_num
                queue = self.queue_list[i_tag]

                # make sure the queue is not empty
                while not queue.empty():
                    # decode the msg using cooresponding method
                    queue_tag = self.queue_tags[i_tag]
                    decode_method = self.decode_select(queue_tag)
                    decode_method(queue)
                # next queue
                i += 1

        
    def decode_select(self,queue_tags:str):
        route_dic = {
            "Logs":self.decode_LogsMsg,
            "Images":self.decode_ImagesMsg,
            "Plys":self.decode_PlysMsg,
            "Commands":self.decode_CommandsMsg
        }
        return route_dic[queue_tags]

    def decode_LogsMsg(self,queue):
        nums_limit = 10
        count = 0
        while not queue.empty() and count < nums_limit:

            # extracting the msg from queue
            msg = queue.pop()
            msg_pb = pb.LogsMessage()
            msg_pb.ParseFromString(msg)

            # save the raw pb msg 
            raw_file_name = os.path.join(self.logsmsg_folder,f"{curr_time}.LogsMsg")
            with open(raw_file_name, 'wb') as file:
                file.write(msg)

            # process the msg
            msg_source = msg_pb.source
            msg_timestamp = msg_pb.timestamp
            msg_context = msg_pb.context

            # display the info to the console
            curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
            print(f"[x] {curr_time}|(From {msg_source}): {msg_context}" )

            # next msg
            count += 1


    def decode_ImagesMsg(self,queue):

        while not queue.empty():

            # extracting the msg from queue
            msg = queue.pop()
            msg_pb = pb.ImagesMessage()
            msg_pb.ParseFromString(msg)

            # save the raw pb msg 
            raw_file_name = os.path.join(self.imagesmsg_folder,f"{curr_time}.ImagesMsg")
            with open(raw_file_name, 'wb') as file:
                file.write(msg)

            # process the msg
            msg_source = msg_pb.source
            msg_timestamp = msg_pb.timestamp

            # process the images
            id = 0
            for image in msg_pb.Image:
                imagelegth = image.length
                image_byte = image.data

                # save the images
                if (image_byte is None) | (len(image_byte) == 0):
                    pass
                else:
                    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)

                    # display the info to the console
                    curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
                    print(f"[x] {curr_time}|(From {msg_source}): image{id} saved!" )

                # next image item
                id += 1


    def decode_PlysMsg(self,queue):



        
        pass

    def decode_CommandsMsg(self,queue):

        pass
    


# 根据rabbitmq重新设置消息参数
class device_state_senderThread(senderThread):
@@ -260,6 +447,65 @@
    pass


def message_sender(MQparams: pika.ConnectionParameters,
                exchange_name: str,
                exchange_type:str,
                routing_key:str,
                message:str,
                    ):
    # connect to RabbitMQ
    connection = pika.BlockingConnection(MQparams)

    # declare a exchange
    channel = connection.channel()
    channel.exchange_declare(exchange=exchange_name,
                            exchange_type=exchange_type)
    # publish a PERSISTENT message
    channel.basic_publish(
                exchange=exchange_name,
                routing_key=routing_key,
                body=message,
                properties=pika.BasicProperties(
                    delivery_mode=pika.spec.PERSISTENT_DELIVERY_MODE
                )
        )
    # close the connection
    connection.close()

def message_receiver(MQparams: pika.ConnectionParameters,
                exchange_name: str,
                exchange_type:str,
                queue_name:str,
                binding_keys:List[str],
                callback_fn:callable
                ):
    # connect to RabbitMQ
    connection = pika.BlockingConnection(MQparams)

    # declare a exchange
    channel = connection.channel()
    channel.exchange_declare(exchange=exchange_name,
                            exchange_type=exchange_type)
    
    # declare a durable queue
    channel.queue_declare(queue=queue_name,durable=True)

    # literate the binding keys and bind the queue
    for binding_key in binding_keys:
        channel.queue_bind(queue=queue_name,exchange=exchange_name,
                            routing_key=binding_key)

    # waiting for consuming the messages
    channel.basic_qos(prefetch_count=1)
    channel.basic_consume(queue=queue_name,
                        auto_ack=False,
                        on_message_callback=callback_fn
                        )
    channel.start_consuming()


    





 src/engine/devices.py

@@ -5,12 +5,12 @@
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)
# # 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

class cameraController(object):


 src/protos/aerial.proto

@@ -116,18 +116,38 @@
  CommandData CommandData =4;
}

message LogsMessage{
  string source = 1;
  string timestamp = 2;
  string context = 3;
}


message ImagesMessage
{
  string source = 1;
  string timestamp = 2;

  message ImageItems
  {
    int32 length = 1;                           // 数据长度
    bytes data = 2;                            // 数据内容 
  }

  repeated ImageItems Image = 3;
}


message PlyMessage
{
  string source = 1;
  string timestamp = 2;
  bytes plydata = 3;
}




// message ImagesMessage
// {
//   bytes image1  = 1;
//   bytes image2  = 2;
//   bytes image3  = 3;
//   bytes image4  = 4;
//   bytes image5  = 5;
// }

// message MicroWaveMessage
// {