fix 提取数据不成功

Zmq/zmqServer.py

@@ -339,29 +339,12 @@ class zmqServer(threading.Thread):
             
             # 写入范式缓冲区
             with self.paradigmBufferLock:
+                self.paradigmBuffer.appendBuffer(data_np)
                 if self.interval_inited:
-                    self.epoch_finished = self.detect_event(data_np)
-                    if self.pack_contain_event:
-                        self.paradigmBuffer.resetAllPara()
-                    self.paradigmBuffer.appendBuffer(data_np)
-                    
+                    self.pack_contain_event, self.epoch_finished = self.detect_event(data_np)
                     if self.epoch_finished:
-                        now = datetime.datetime.now()
-                        time_diff_str = ""
-                        # 计算与上一次Epoch完成的时间差
-                        if self.last_epoch_finish_time is not None:
-                            # 时间差 单位：秒，保留3位小数
-                            delta_seconds = (now - self.last_epoch_finish_time).total_seconds()
-                            time_diff_str = f" | 与上一次间隔: {delta_seconds:.3f} s"
-                        
-                        # 拼接日志，增加时间差信息
-                        log_msg = f"Epoch采集完成: {now.strftime('%H:%M:%S.%f')[:-3]}{time_diff_str}"
-                        algo_log(log_msg, level="DEBUG")
-                        
-                        # 更新上一次Epoch完成时间为当前时间
-                        self.last_epoch_finish_time = now
-                else:
-                    self.paradigmBuffer.appendBuffer(data_np)
+                        algo_log(f"Epoch采集完成, 当前数据长度{self.paradigmBuffer.GetDataLenCount()}", level="DEBUG")
+
 
         except Exception as e:
             algo_log(f"数据处理失败: {str(e)}", level="ERROR")
@@ -371,7 +354,8 @@ class zmqServer(threading.Thread):
 
     # -------------------------- 事件检测 --------------------------
     def detect_event(self, samples):
-        self.pack_contain_event = False
+        pack_contain_event = False
+        epoch_finished = False
         # 第65通道为事件通道
         events = np.array(samples[-2], dtype=np.int32).tolist()
         for idx, event in enumerate(events):
@@ -383,14 +367,20 @@ class zmqServer(threading.Thread):
                                                 -%H-%M-%S"),
                     ]
                 )
-                self.currentLabel = event
-                if event == self.predict_event:
-                    self.count_events[new_key] = self.latency + 1
+                if len(self.count_events) > 0:
+                    algo_log(f"当前有事件未采集完成，新事件{new_key}非法，被忽略")
+                    return pack_contain_event, epoch_finished
                 else:
-                    self.count_events[new_key] = self.train_latency + 1
-                self.event_inner_idx = idx
-                algo_log(f"事件检测到: {events}，索引: {idx}", level="DEBUG")
-                self.pack_contain_event = True   
+                    self.currentLabel = event
+                    pack_contain_event = True
+                    if event == self.predict_event:
+                        self.count_events[new_key] = self.latency + 1
+                    else:
+                        self.count_events[new_key] = self.train_latency + 1
+                    self.event_inner_idx = idx
+                    algo_log(f"事件检测到: {events}，索引: {idx}", level="DEBUG")
+            else:
+                pack_contain_event = False
                 
         # 倒计时并清理过期事件
         drop_items = []
@@ -403,9 +393,13 @@ class zmqServer(threading.Thread):
         for key in drop_items:
             del self.count_events[key]
             
-        if drop_items:
-            return True
-        return False
+        if len(drop_items) > 0:
+            epoch_finished = True
+        else:
+            epoch_finished = False
+        return pack_contain_event, epoch_finished
+    
+    
     # -------------------------- 主循环 --------------------------
     def run(self):
         self.running = True