update

Author: Likecolor

config/config.yaml

@@ -2,7 +2,7 @@ client:
  # the source video
  source:
   video_path: ./video_data/dayroad.mp4
-  max_count: 5
+  max_count: 1000
   rtsp:
    flag: False
    account: your account
@@ -21,13 +21,13 @@ client:
  # the queue threshold for offloading
  queue_thresh: 5
  # the max wait time
- wait_thresh: 10
+ wait_thresh: 100
  # Number of worker threads to offloading thread pool
  offloading_max_worker: 1
  frame_cache_maxsize: 100
  small_model_name: fasterrcnn_mobilenet_v3_large_fpn
  # select the offloading policy
- policy: Edge-Shortest
+ policy: Edge-Local
  # change frame resolution using frame new height
  new_height:
   # offload to another edge node [1080, 720], default 720
@@ -52,24 +52,24 @@ client:
  server_ip: '127.0.0.1:50051'
  # edge nodes
  edge_id: 1
- edge_num: 2
- destinations: {'id': [2], 'ip':['192.168.0.185:50050']}
+ edge_num: 1
+ destinations: {'id': [], 'ip':[]}
  # database config
  database:
   connection: { 'user': 'root', 'password': 'root', 'host': '127.0.0.1', 'raise_on_warnings': True }
   database_name: 'mydatabase'
  # retrain
  retrain:
-  flag: False
+  flag: True
   num_epoch: 2
   cache_path: './cache'
-  collect_num: 10
-  select_num: 5
-  interval: 2
+  collect_num: 20
+  select_num: 15
+  window: 90
 
 server:
  server_id: 0
- edge_ids: [1,2]
+ edge_ids: [1]
  large_model_name: fasterrcnn_resnet50_fpn
  #the queue maxsize
  local_queue_maxsize: 10

edge/edge_worker.py

@@ -63,20 +63,21 @@ def __init__(self, config):
         self.diff_processor.start()
 
         # start the thread for edge server
-        self.edge_server = threading.Thread(target=self.start_edge_server, daemon=True)
-        self.edge_server.start()
+        #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,daemon=True)
         self.local_processor.start()
 
         # start the thread for retrain process
         self.collect_flag = self.config.retrain.flag
+        self.collect_time = None
+        self.collect_time_flag = True
         self.retrain_flag = False
         self.cache_count = 0
 
         self.use_history = True
-        self.test_only = False
         self.retrain_no = 0
 
         self.retrain_processor = threading.Thread(target=self.retrain_worker,daemon=True)
@@ -180,7 +181,12 @@ def local_worker(self):
 
             # collect data for retrain
             if self.collect_flag:
-                self.collect_data(task, current_frame ,detection_boxes, detection_class, detection_score)
+                if self.collect_time_flag:
+                    self.collect_time = time.time()
+                    self.collect_time_flag = False
+                duration = time.time() - self.collect_time
+                if duration > self.config.retrain.window:
+                    self.collect_data(task, current_frame ,detection_boxes, detection_class, detection_score)
 
             if detection_boxes is not None:
                 task.add_result(detection_boxes, detection_class, detection_score)
@@ -308,10 +314,6 @@ def collect_data(self, task, frame ,detection_boxes, detection_class, detection_
                 self.pred_res = []
                 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
 
 
@@ -332,17 +334,10 @@ def retrain_worker(self):
                     np.savetxt(os.path.join(self.config.retrain.cache_path,'annotation.txt'), self.annotations,
                                fmt=['%d', '%d', '%f', '%f', '%f', '%f', '%f'], delimiter=',')
 
-                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.small_object_detection.model_evaluation(
+                    self.config.retrain.cache_path, self.select_index)
+                self.small_object_detection.retrain(
+                    self.config.retrain.cache_path, self.select_index[:int(self.config.retrain.select_num*0.8)])
                 self.retrain_flag = False
                 if self.use_history:
                     self.select_index,self.avg_scores = history_sample(self.select_index,self.avg_scores)
@@ -354,8 +349,10 @@ def retrain_worker(self):
                     self.select_index = []
                     self.avg_scores = []
                     self.annotations = []
+                self.collect_time_flag = True
                 self.collect_flag = True
-            time.sleep(1)
+            time.sleep(0.2)
+
 
 
     def start_edge_server(self):

edge_client.py

@@ -4,6 +4,8 @@
 import threading
 import time
 
+
+import wandb
 import munch
 import yaml
 
@@ -28,6 +30,7 @@ def signal_handler(signal, frame):
         config = yaml.load(f, Loader=yaml.SafeLoader)
     #provide class-like access for dict
     config = munch.munchify(config)
+    wandb.init(project="filter", config=config)
     config = config.client
     signal.signal(signal.SIGINT, signal_handler)
     event = threading.Event()

model_management/object_detection.py

@@ -107,10 +107,13 @@ def retrain(self, path, select_index):
             # Update the learning rate
             lr_scheduler.step()
         torch.save(tmp_model.state_dict(), "./model_management/tmp_model.pth")
+
+        if torch.cuda.is_available():
+            torch.cuda.empty_cache()
         state_dict = torch.load("./model_management/tmp_model.pth", map_location=device)
         with self.model_lock:
             self.model.load_state_dict(state_dict)
-        self.model.eval()
+            self.model.eval()
 
     def model_evaluation(self,cache_path, select_index):
         map = []
@@ -222,6 +225,8 @@ def large_inference(self, img):
         return pred_boxes, pred_class, pred_score
 
     def get_model_prediction(self, img, threshold, model=None):
+        if torch.cuda.is_available():
+            torch.cuda.empty_cache()
         #process the image
         img = Image.fromarray(cv2.cvtColor(img, cv2.COLOR_BGR2RGB))
         transform = transforms.Compose([transforms.ToTensor()])