lizhao/bci_algo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v1

Browse Source
This commit is contained in:
lizhao
2026-06-06 15:53:50 +08:00
parent fce7d93d5e
commit 29b6118f11
3 changed files with 31 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
Zmq/zmqServer.py
View File

@@ -217,48 +217,49 @@ class zmqServer(threading.Thread):
处理8100端口原始脑电二进制数据 处理8100端口原始脑电二进制数据
固定格式：上位机发送 (5,66) float32 二维数组字节流（已转换为微伏物理量）→ 转置为 (66,5) 写入双缓冲区 固定格式：上位机发送 (5,66) float32 二维数组字节流（已转换为微伏物理量）→ 转置为 (66,5) 写入双缓冲区
""" """
# 1. 校验ZMQ消息帧完整性 # 1. 校验ZMQ消息帧完整性（ROUTER接收DEALER消息的帧格式：[客户端ID, 发送方ID, 空帧, 数据帧]）
if len(frames) < 3: if len(frames) < 4: # 至少需要4帧
print(f"[ERROR] 无效数据帧：长度不足3帧，实际长度={len(frames)}") algo_log(f"Invalid data frame: 帧数量不足，期望≥4，实际{len(frames)}", level="ERROR")
return return
ident, _, data_bytes = frames[:3] # 2. 正确解析帧（适配DEALER→ROUTER的帧格式）
client_ident, sender_ident, empty_sep, data_bytes = frames[:4]
if empty_sep != b'': # 校验空分隔帧
algo_log(f"Invalid frame separator: 期望空字节，实际{empty_sep}", level="ERROR")
return
# 2. 客户端管理（单客户端场景，自动更新最新身份） # 3. 客户端管理（单客户端场景，自动更新最新身份）
if ident not in self.data_clients: if client_ident not in self.data_clients:
self.data_clients.add(ident) self.data_clients.add(client_ident)
self.current_data_client = ident # 保存唯一客户端身份，用于后续回复滤波结果 self.current_data_client = client_ident # 保存唯一客户端身份，用于后续回复滤波结果
print(f"[INFO] 新数据客户端连接成功：{ident}") print(f"[INFO] 新数据客户端连接成功：{client_ident}")
try: try:
# 3. 精确长度校验（核心：固定(5,66) float32 = 5*66*4=1320字节，与int32字节数相同） # 4. 精确长度校验（核心：固定(5,66) float32 = 5*66*4=1320字节）
EXPECTED_BYTES = self.device_info['frame_points'] * self.device_info['channel_nums'] * 4 # 每个float32占4字节 EXPECTED_BYTES = self.device_info['frame_points'] * self.device_info['channel_nums'] * 4 # 每个float32占4字节
if len(data_bytes) != EXPECTED_BYTES: if len(data_bytes) != EXPECTED_BYTES:
print(f"[ERROR] 数据长度错误：期望{EXPECTED_BYTES}字节，实际{len(data_bytes)}字节") algo_log(f"[ERROR] 数据长度错误：期望{EXPECTED_BYTES}字节，实际{len(data_bytes)}字节", level="ERROR")
return return
# 4. 零拷贝二进制解析 + 维度转换 # 5. 零拷贝二进制解析 + 维度转换
# 步骤：字节流 → (330,) float32数组 → (5,66) 原始格式 → 转置为 (66,5) 缓冲区标准格式
data_np = np.frombuffer(data_bytes, dtype=np.float32) data_np = np.frombuffer(data_bytes, dtype=np.float32)
# 重塑为上位机原始维度
data_np = data_np.reshape(self.device_info['frame_points'], self.device_info['channel_nums']) data_np = data_np.reshape(self.device_info['frame_points'], self.device_info['channel_nums'])
# 转置为(通道数, 采样点数)标准格式，转换为float64保证滤波运算精度
data_np = data_np.T.astype(np.float64) data_np = data_np.T.astype(np.float64)
# 5. 同时写入双环形缓冲区（方法名与现有类保持一致：appendBuffer） # 6. 写入缓冲区
# 注意：上位机已发送微伏物理量，无需再乘以增益系数
self.paradigmBuffer.appendBuffer(data_np) self.paradigmBuffer.appendBuffer(data_np)
self.filterBuffer.appendBuffer(data_np) self.filterBuffer.appendBuffer(data_np)
# 生产环境必须注释！每秒50次打印会导致CPU占用飙升30%以上 algo_log(f"数据写入成功：shape={data_np.shape}, 范围=[{data_np.min():.2f}, {data_np.max():.2f}] μV", level="DEBUG")
algo_log(f"数据写入成功：shape={data_np.shape}, 范围=[{data_np.min():.2f}, {data_np.max():.2f}] μV", level="DEBUG", record_once=True)
except Exception as e: except Exception as e:
algo_log(f"数据处理失败：{str(e)}", level="ERROR") algo_log(f"数据处理失败：{str(e)}", level="ERROR")
# 调试阶段临时打开，生产环境务必注释 if IniRead('system', 'algo_log_level', 'INFO') == 'DEBUG':
import traceback import traceback
traceback.print_exc() traceback.print_exc()
def _process_send_queue(self): def _process_send_queue(self):
"""处理发送队列，向所有命令客户端广播消息""" """处理发送队列，向所有命令客户端广播消息"""
while not self.send_queue.empty(): while not self.send_queue.empty():

11
config.ini
View File

@@ -20,8 +20,9 @@ algo_log_level = DEBUG
console_output = 1 console_output = 1
; 64 导设备配置 ; 64 导设备配置
[device_type] = 1 [device_type_1]
device_sample_rate = 250 sample_rate = 250
device_channel_nums = 66 frame_points = 5
device_channel_names = ['FP1', 'FP2', 'PO6', 'POZ', 'F3', 'F4', 'FPZ', 'AF4', 'FC3', 'PO8', 'CP2', 'CP1', 'FCZ', 'PO5', 'FC2', 'FC1', 'C3', 'C4', 'FC4', 'CP4', 'P3', 'P4', 'F5', 'C5', 'F6', 'PO4', 'CP6', 'CP5', 'PO3', 'CP3', 'FC6', 'FC5', 'CB1', 'CB2', 'P5', 'AF7', 'A1', 'T7', 'FT7', 'TP7', 'FT8', 'AF8', 'F8', 'F7', 'P6', 'C6', 'O2', 'O1', 'T8', 'P7', 'CZ', 'PZ', 'P8', 'FZ', 'OZ', 'PO7', 'TP8', 'AF3', 'C2', 'C1', 'P2', 'P1', 'F2', 'F1', 'label', 'label_tag'] channel_nums = 66
device_channel_index = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65] channel_names = ['FP1', 'FP2', 'PO6', 'POZ', 'F3', 'F4', 'FPZ', 'AF4', 'FC3', 'PO8', 'CP2', 'CP1', 'FCZ', 'PO5', 'FC2', 'FC1', 'C3', 'C4', 'FC4', 'CP4', 'P3', 'P4', 'F5', 'C5', 'F6', 'PO4', 'CP6', 'CP5', 'PO3', 'CP3', 'FC6', 'FC5', 'CB1', 'CB2', 'P5', 'AF7', 'A1', 'T7', 'FT7', 'TP7', 'FT8', 'AF8', 'F8', 'F7', 'P6', 'C6', 'O2', 'O1', 'T8', 'P7', 'CZ', 'PZ', 'P8', 'FZ', 'OZ', 'PO7', 'TP8', 'AF3', 'C2', 'C1', 'P2', 'P1', 'F2', 'F1', 'label', 'label_tag']
channel_index = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65]

2
runDecoder.py
View File

@@ -6,6 +6,7 @@ import time
from Decoder import Decoder_main from Decoder import Decoder_main
from PubLibrary.RunOnce import is_program_running from PubLibrary.RunOnce import is_program_running
from PubLibrary.InifileHelper import IniRead from PubLibrary.InifileHelper import IniRead
from logs.log import algo_log
def get_device_info(device_type): def get_device_info(device_type):
@@ -41,6 +42,7 @@ if __name__ == "__main__":
# ) # )
device_info= get_device_info(1) device_info= get_device_info(1)
algo_log(f"device_info: {device_info}", level="INFO")
decoder = Decoder_main(device_info=device_info) decoder = Decoder_main(device_info=device_info)
try: try: