update

Author: Likecolor

cal_metrics.py

@@ -72,7 +72,7 @@ def cal_mAP(self):
         sum_delay = 0.0
         filtered_out = 0
         i = 0
-
+        last_result = None
         while i < len(result):
             index, start_time, end_time, res, log = result[i]
             gap = end_time-start_time
@@ -84,6 +84,7 @@ def cal_mAP(self):
                 pred = eval(res)
             else:
                 pred = eval(res)
+                last_result = pred
             ground_truth = ground_truths['{}'.format(index)]
             map = calculate_map(ground_truth, pred)
             sum_map += map

cloud_server.py

@@ -34,7 +34,7 @@ def __init__(self, config):
 
         # start the thread for local process
         self.local_queue = Queue(config.local_queue_maxsize)
-        self.local_processor = threading.Thread(target=self.cloud_local, )
+        self.local_processor = threading.Thread(target=self.cloud_local, daemon=True)
         self.local_processor.start()
 
 

config/config.yaml

@@ -47,20 +47,21 @@ client:
   #The task from another edge node is offloaded to the cloud after local inference, default 85.
   another_cloud: 90
  #server_ip
- server_ip: '127.0.0.1:50051'
+ server_ip: '43.137.51.180:50051'
  #edge nodes
  edge_id: 1
  destinations: {'id': [2], 'ip':['127.0.0.1:50050']}
  #database config
  database:
-  connection: { 'user': 'root', 'password': 'root', 'host': '127.0.0.1', 'raise_on_warnings': True }
+  connection: { 'user': 'root', 'password': 'root', 'host': '43.137.51.180', 'raise_on_warnings': True }
   database_name: 'mydatabase'
  # retrain
  retrain:
-  num_epoch: 4
+  num_epoch: 2
   cache_path: './cache'
-  collect_num: 20
-  select_num: 20
+  collect_num: 10
+  select_num: 5
+  interval: 2
 
 server:
  server_id: 0

edge/edge_worker.py

@@ -15,13 +15,14 @@
 from database.database import DataBase
 from difference.diff import DiffProcessor
 from edge.info import TASK_STATE
+from edge.resample import history_sample, annotion_process
 from edge.task import Task
 from edge.transmit import get_cloud_target
 from grpc_server import message_transmission_pb2_grpc, message_transmission_pb2
 from grpc_server.rpc_server import MessageTransmissionServicer
 
 from tools.convert_tool import cv2_to_base64
-from tools.file_op import clear_folder, creat_folder
+from tools.file_op import clear_folder, creat_folder, sample_files
 from tools.preprocess import frame_resize
 from model_management.object_detection import Object_Detection
 from apscheduler.schedulers.background import BackgroundScheduler
@@ -53,53 +54,37 @@ def __init__(self, config):
         self.avg_scores = []
         self.select_index = []
         self.annotations = []
-        # read video
-        self.video_reader = threading.Thread(target=self.video_read,)
-        self.video_reader.start()
+
         # start the thread for diff
         self.diff = 0
         self.key_task = None
-        self.diff_processor = threading.Thread(target=self.diff_worker,)
+        self.diff_processor = threading.Thread(target=self.diff_worker,daemon=True)
         self.diff_processor.start()
 
         # start the thread for edge server
-        #self.edge_server = threading.Thread(target=self.start_edge_server, )
+        #self.edge_server = threading.Thread(target=self.start_edge_server, daemon=True)
         #self.edge_server.start()
 
         # start the thread for local process
-        self.local_processor = threading.Thread(target=self.local_worker,)
+        self.local_processor = threading.Thread(target=self.local_worker,daemon=True)
         self.local_processor.start()
 
         # start the thread for retrain process
         self.retrain_flag = False
         self.collect_flag = True
         self.cache_count = 0
-        self.retrain_processor = threading.Thread(target=self.retrain_worker,)
+
+        self.use_history = True
+        self.test_only = False
+        self.retrain_no = 0
+
+        self.retrain_processor = threading.Thread(target=self.retrain_worker,daemon=True)
         self.retrain_processor.start()
 
         # start the thread pool for offload
         self.offloading_executor = futures.ThreadPoolExecutor(max_workers=config.offloading_max_worker,)
 
-    def video_read(self):
-        with VideoProcessor(self.config.source) as video:
-            video_fps = video.fps
-            logger.info("the video fps is {}".format(video_fps))
-            index = 0
-            if self.config.interval == 0:
-                logger.error("the interval error")
-                sys.exit(1)
-            logger.info("Take the frame interval is {}".format(self.config.interval))
-            while True:
-                frame = next(video)
-                if frame is None:
-                    logger.info("the video finished")
-                    break
-                index += 1
-                if index % self.config.interval == 0:
-                    start_time = time.time()
-                    task = Task(self.edge_id, index, frame, start_time, frame.shape)
-                    self.frame_cache.put(task, block=True)
-                    time.sleep((self.config.interval * 1.0) / video_fps)
+
 
 
     def diff_worker(self):
@@ -110,7 +95,6 @@ def diff_worker(self):
             self.pre_frame_feature = self.edge_processor.get_frame_feature(frame)
             self.key_task = task
             # Create an entry for the task in the database table
-            logger.debug("start time {}".format(task.start_time))
             data = (
                 task.frame_index,
                 task.start_time,
@@ -151,9 +135,6 @@ def diff_worker(self):
                     task.state = TASK_STATE.FINISHED
                     self.update_table(task)
 
-        else:
-            pass
-
     def update_table(self, task):
         state = "Finished" if task.state == TASK_STATE.FINISHED else ""
         if task.ref is not None:
@@ -207,7 +188,7 @@ def local_worker(self):
                     task.frame_cloud = offloading_image
                     self.offloading_executor.submit(self.offload_worker, task)
                 end_time = time.time()
-                task.set_end_time(end_time)
+                task.end_time = end_time
                 task.state = TASK_STATE.FINISHED
                 # upload the result to database
                 self.update_table(task)
@@ -298,55 +279,74 @@ def offload_worker(self, task, destination_edge_id=None):
             else:
                 logger.info(str(res))
 
-    #
+    # collect data for retrain
     def collect_data(self, task, frame ,detection_boxes, detection_class, detection_score):
-        self.select_index = []
-        creat_folder(self.config.retrain.cache_path)
-        cv2.imwrite(os.path.join(self.config.retrain.cache_path,'frames', str(task.frame_index) + '.jpg'), frame)
-        self.avg_scores.append({task.frame_index:np.mean(detection_score)})
-        self.cache_count += 1
-        logger.debug("count {}".format(self.cache_count))
-        for score, label, box in zip(detection_score, detection_class, detection_boxes):
-            self.pred_res.append((task.frame_index, label, box[0], box[1], box[2], box[3], score))
+        if detection_score is not None:
+            creat_folder(self.config.retrain.cache_path)
+            cv2.imwrite(os.path.join(self.config.retrain.cache_path,'frames', str(task.frame_index) + '.jpg'), frame)
+            self.avg_scores.append({task.frame_index: np.mean(detection_score)})
+            self.cache_count += 1
+            logger.debug("count {}".format(self.cache_count))
             if self.cache_count >= self.config.retrain.collect_num:
+                self.retrain_no += 1
                 logger.debug("enough")
                 smallest_elements = sorted(self.avg_scores, key=lambda d: list(d.values())[0])[:self.config.retrain.select_num]
                 self.select_index = [list(d.keys())[0] for d in smallest_elements]
-                print(self.select_index)
-
-                np.savetxt(os.path.join(self.config.retrain.cache_path,'pred_res.txt'), self.pred_res,
-                           fmt=['%d', '%d', '%f', '%f', '%f', '%f', '%f'], delimiter=',')
-
+                logger.debug(self.select_index)
                 self.pred_res = []
-                self.retrain_flag = True
                 self.collect_flag = False
                 self.cache_count = 0
+                if self.retrain_no % self.config.retrain.interval == 0:
+                    self.test_only = True
+                else:
+                    self.test_only = False
+                self.retrain_flag = True
 
 
     # retrain
     def retrain_worker(self):
+        self.annotations = []
         while True:
             if self.retrain_flag:
                 logger.debug("retrain")
-                self.annotations = []
+
                 for index in self.select_index:
                     path = os.path.join(self.config.retrain.cache_path, 'frames', '{}.jpg'.format(index))
+                    logger.debug(path)
                     frame = cv2.imread(path)
-                    logger.debug("get index {}".format(index))
+                    logger.debug("get index {} {}".format(index, time.time()))
                     target_res = get_cloud_target(self.config.server_ip, frame)
-                    logger.debug("get target {}".format(target_res))
+                    logger.debug("get target {} {}".format(target_res, time.time()))
                     for score, label, box in zip(target_res['scores'], target_res['labels'], target_res['boxes']):
                         self.annotations.append((index, label, box[0], box[1], box[2], box[3], score))
                 if len(self.annotations):
                     np.savetxt(os.path.join(self.config.retrain.cache_path,'annotation.txt'), self.annotations,
                                fmt=['%d', '%d', '%f', '%f', '%f', '%f', '%f'], delimiter=',')
 
                 logger.debug("select num {}".format(int(self.config.retrain.select_num*0.8)))
-                self.small_object_detection.retrain(self.config.retrain.cache_path, self.select_index[:int(self.config.retrain.select_num*0.8)])
-                self.small_object_detection.model_evaluation(self.config.retrain.cache_path, self.select_index[int(self.config.retrain.select_num*0.8):])
-                self.collect_flag = True
+                if self.test_only:
+                    logger.debug("test only")
+                    self.small_object_detection.model_evaluation(
+                        self.config.retrain.cache_path, self.select_index)
+                else:
+                    self.small_object_detection.model_evaluation(
+                        self.config.retrain.cache_path, self.select_index[int(self.config.retrain.select_num * 0.8):])
+                    self.small_object_detection.retrain(
+                        self.config.retrain.cache_path, self.select_index[:int(self.config.retrain.select_num*0.8)])
+                    self.small_object_detection.model_evaluation(
+                        self.config.retrain.cache_path, self.select_index[int(self.config.retrain.select_num*0.8):])
                 self.retrain_flag = False
-                clear_folder(self.config.retrain.cache_path)
+                if self.use_history:
+                    self.select_index,self.avg_scores = history_sample(self.select_index,self.avg_scores)
+                    self.annotations = annotion_process(self.annotations, self.select_index)
+                    sample_files(os.path.join(self.config.retrain.cache_path, 'frames') ,self.select_index)
+                    self.cache_count = len(self.select_index)
+                else:
+                    clear_folder(self.config.retrain.cache_path)
+                    self.select_index = []
+                    self.avg_scores = []
+                    self.annotations = []
+                self.collect_flag = True
             time.sleep(1)
 
 

edge/resample.py

@@ -0,0 +1,17 @@
+import random
+
+def history_sample(index_list, score_list):
+    num_elements = min(len(index_list), len(score_list)) // 2
+    index_list_c = random.sample(index_list, num_elements)
+    score_list_c = []
+    for item in score_list:
+        if list(item.keys())[0] in index_list_c:
+            score_list_c.append(item)
+    return index_list_c, score_list_c
+
+def annotion_process(annotations, index_list):
+    new_annotations = []
+    for sublist in annotations:
+        if sublist[0] in index_list:
+            new_annotations.append(sublist)
+    return new_annotations

edge_client.py

@@ -1,12 +1,23 @@
 import argparse
+import signal
+import sys
+import threading
+import time
+
 import munch
 import yaml
 
 from edge.edge_worker import EdgeWorker
 from loguru import logger
 
+from edge.task import Task
+from tools.file_op import clear_folder
+from tools.video_processor import VideoProcessor
 
-
+def signal_handler(signal, frame):
+    logger.debug("Received Ctrl+C. Cleaning up...")
+    clear_folder(config.retrain.cache_path)
+    sys.exit(0)
 
 
 if __name__ == '__main__':
@@ -18,10 +29,31 @@
     #provide class-like access for dict
     config = munch.munchify(config)
     config = config.client
+    signal.signal(signal.SIGINT, signal_handler)
+    event = threading.Event()
     edge = EdgeWorker(config)
     logger.add("log/client/client_{time}.log", level="INFO", rotation="500 MB")
-
-
-
-
+    try:
+        with VideoProcessor(config.source) as video:
+            video_fps = video.fps
+            logger.info("the video fps is {}".format(video_fps))
+            index = 0
+            if config.interval == 0:
+                logger.error("the interval error")
+                sys.exit(1)
+            logger.info("Take the frame interval is {}".format(config.interval))
+            while True:
+                frame = next(video)
+                if frame is None:
+                    logger.debug("The video finished")
+                    break
+                index += 1
+                if index % config.interval == 0:
+                    start_time = time.time()
+                    task = Task(config.edge_id, index, frame, start_time, frame.shape)
+                    edge.frame_cache.put(task, block=True)
+                    time.sleep((config.interval * 1.0) / video_fps)
+        event.wait()
+    except KeyboardInterrupt:
+        pass
 

grpc_server/rpc_server.py

@@ -23,12 +23,9 @@ def task_processor(self, request, context):
         task.other = True
 
         if request.part_result != "":
-            logger.debug("put1")
             part_result = eval(request.part_result)
             if len(request.part_result['boxes']) != 0:
-                logger.debug("put2")
                 task.add_result(part_result['boxes'], part_result['labels'], part_result['scores'])
-            logger.debug("put3")
 
         if request.note == "edge process":
             task.edge_process = True
@@ -49,13 +46,13 @@ def frame_processor(self, request, context):
         frame_shape = tuple(int(s) for s in request.frame_shape[1:-1].split(","))
         frame = base64_to_cv2(base64_frame).reshape(frame_shape)
         pred_boxes, pred_class, pred_score = self.object_detection.large_inference(frame)
-        res_dict = {
+        res = {
             'boxes': pred_boxes,
             'labels': pred_class,
             'scores': pred_score
         }
         reply = message_transmission_pb2.FrameReply(
-            response=str(res_dict),
+            response=str(res),
             frame_shape=str(frame_shape),
         )
         return reply