From 74c96c72625aff6613d69c7168e1a7d76da22870 Mon Sep 17 00:00:00 2001
From: wu_xinjun <bent20140204@outlook.com>
Date: 星期五, 24 六月 2022 17:18:31 +0800
Subject: [PATCH] add taskuuid
---
src/Server.py | 71 ++++++++++++++++++++++++++++++++---
1 files changed, 65 insertions(+), 6 deletions(-)
diff --git a/src/Server.py b/src/Server.py
index 9194321..5818ae8 100644
--- a/src/Server.py
+++ b/src/Server.py
@@ -1,10 +1,9 @@
import sys # NOQA: E402
import os
-from flask import Flask
+from flask import Flask, request
import pika
import uuid
import queue
-from werkzeug.wrappers import request
from threading import Thread
from datetime import datetime
@@ -13,6 +12,7 @@
from .engine.DeviceManager import DeviceManager
from .engine import MessageThreads as ths
from .protos import aerial_pb2 as pb
+from .utils.tools import update_folder_for_today
@@ -35,9 +35,22 @@
self.WEBport = WEBport
self.save_folder = save_folder
self.server_name = server_name
-
- # self.route('/api', methods=['POST'])(self.api)
+ today = datetime.now().strftime('%Y-%m-%d')
+ today_folder = os.path.join(self.save_folder,today)
+
+ self.today = {"today":today,
+ "today_folder":today_folder,
+ "sup_folder":self.save_folder,
+ "sub_folder": ""}
+
+ self.task_uuid_folder = {"init":os.path.join(self.today["today_folder"],"init")}
+
+ if not os.path.exists(self.task_uuid_folder["init"]):
+ os.makedirs(self.task_uuid_folder["init"])
+
+
+ self.route('/api', methods=['POST'])(self.api)
self.route('/', methods=['GET'])(self.hello)
# self._def_routes()
@@ -51,7 +64,7 @@
def _connect_to_MQ(self):
credentials = pika.PlainCredentials(self.MQuser, self.MQpassword)
self.parameters = pika.ConnectionParameters(
- credentials=credentials, host=self.MQhost)
+ credentials=credentials, host=self.MQhost,heartbeat=0)
# 杩炴帴闃熷垪鏈嶅姟鍣�
try:
self.connection = pika.BlockingConnection(self.parameters)
@@ -110,7 +123,7 @@
self.MsgDecoder_Thread = ths.decodeThread(
queue_list= queue_list,
queue_tags= queue_tags,
- save_path= self.save_folder
+ task_folder= self.task_uuid_folder
)
self.MsgDecoder_Thread.setDaemon(True)
@@ -135,6 +148,52 @@
def hello(self):
return 'Hello, This is acquisition demo API!'
+ def api(self):
+
+ try:
+ success, response_bytes = self.verify_commands(request.data)
+ except:
+ response_bytes = b'command error!'
+
+ response = Flask.response_class(
+ response_bytes,
+ mimetype='application/x-bytes'
+ )
+ return response,200
+
+
+ def verify_commands(self,input_command:bytes):
+
+ # parse the command from bytes to the pb object
+ req_command = pb.CommandMessage()
+ req_command.ParseFromString(input_command)
+
+ # extracting some info for use
+ source = req_command.source
+ # generatetime = req_command.generatetime
+ # syscommand = req_command.syscommand
+ # subcommand = req_command.subcommand
+ targetclient = req_command.targetclient
+ # returnserver = req_command.returnserver
+ taskuuid = req_command.taskuuid
+
+ update_folder_for_today(self.today)
+ task_path = os.path.join(self.today["today_folder"],taskuuid)
+
+ self.task_uuid_folder[taskuuid] = task_path
+
+ if not os.path.exists(task_path):
+ os.makedirs(task_path)
+
+ curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
+ print(f"[x] {curr_time}|(From {source}): task [{taskuuid}] received and sending it to [{targetclient}]." )
+
+ self.CommandsSender_queue.put([targetclient,input_command])
+ response_bytes = bytes(f"Receive command success! The data will be save in [{os.path.abspath(task_path)}]",\
+ encoding="utf8")
+
+ return True, response_bytes
+
def run_rabbitmq(self):
self._connect_to_MQ()
self._receiver_Thread_Launcher()
--
Gitblit v1.9.1