基于pyads的PLC ADS回调数据整合：非全局变量实现方案咨询

针对你用pyads通过ADS通知机制高速拉取PLC数据并整合为大数组的需求，推荐用类封装状态+线程安全锁的方案替代全局变量，既符合Python风格，又能安全高效地处理多线程回调数据。

核心思路 #

用类封装数据收集的所有状态（目标样本数、已收集数据、完成标志），通过线程锁保护多线程环境下的数据操作（pyads回调在独立线程执行），最后在循环中等待收集完成。

完整代码实现 #

import pyads
import threading
import time
import numpy as np

class PLCDataCollector:
    def __init__(self, plc_conn, struct_def, target_samples=100000):
        self.plc = plc_conn
        self.struct_def = struct_def
        self.target_samples = target_samples
        # 初始化每个信号的收集列表
        self.signal_data = {name: [] for name, _, _ in struct_def}
        # 线程锁：保护多线程下的数据读写
        self.data_lock = threading.Lock()
        self.collection_done = False
        self.struct_size = pyads.size_of_structure(struct_def)
        # 自动注册ADS通知
        self._setup_notification()

    def _setup_notification(self):
        attr = pyads.NotificationAttrib(self.struct_size)
        # 注册类方法作为回调
        self.plc.add_device_notification(
            'global.sample_structure',
            attr,
            self._handle_data,
            ctypes.c_ubyte * self.struct_size
        )

    def _handle_data(self, handle, name, timestamp, raw_value):
        # 解析PLC传来的结构化数据
        parsed_data = pyads.dict_from_bytes(raw_value, self.struct_def)
        
        with self.data_lock:
            for signal_name, batch_values in parsed_data.items():
                # 追加当前批次的1000个值
                self.signal_data[signal_name].extend(batch_values)
                # 超过目标长度时截断，保证最终数组刚好10万个值
                if len(self.signal_data[signal_name]) >= self.target_samples:
                    self.signal_data[signal_name] = self.signal_data[signal_name][:self.target_samples]
            
            # 检查所有信号是否都收集完成
            if all(len(arr) >= self.target_samples for arr in self.signal_data.values()):
                self.collection_done = True

    def wait_and_get_data(self, timeout=None):
        start_time = time.time()
        while not self.collection_done:
            if timeout and (time.time() - start_time) > timeout:
                raise TimeoutError("数据收集超时")
            # 短暂休眠，避免CPU空转
            time.sleep(0.01)
        
        # 转换为numpy数组（可选，根据需求调整）
        return {name: np.array(arr) for name, arr in self.signal_data.items()}

# -------------------------- 使用示例 --------------------------
if __name__ == "__main__":
    # 建立PLC连接
    plc = pyads.Connection('127.0.0.1.1.1', pyads.PORT_TC3PLC1)
    plc.open()

    # 定义PLC侧的结构（需与PLC内定义完全匹配）
    structure_def = (
        ('nVar', pyads.PLCTYPE_DINT, 1000),
        ('nVar2', pyads.PLCTYPE_DINT, 1000),
        ('nVar3', pyads.PLCTYPE_DINT, 1000),
        ('nVar4', pyads.PLCTYPE_DINT, 1000),
        ('nVar5', pyads.PLCTYPE_DINT, 1000))

    # 初始化收集器，目标每个信号收集10万个值
    collector = PLCDataCollector(plc, structure_def, target_samples=100000)
    
    print("开始收集数据...")
    try:
        # 等待收集完成，可设置超时（例如300秒）
        final_data = collector.wait_and_get_data(timeout=300)
        print("数据收集完成！")
        
        # 输出结果信息
        for sig_name, arr in final_data.items():
            print(f"{sig_name}: 有效数据长度 {len(arr)}，前5个值: {arr[:5]}")
    except TimeoutError as e:
        print(e)
    finally:
        plc.close()

方案优势 #

1. 无全局变量：所有状态封装在类实例中，代码更整洁、可复用
2. 线程安全：用threading.Lock避免回调线程与主线程的数据竞争
3. 自动截断：确保每个信号数组严格达到10万个值，不会超额
4. 可扩展性：轻松调整目标样本数、结构定义，或添加超时、日志等功能
5. 清晰职责：类内部处理ADS通知注册、数据解析、状态管理，主线程只需等待结果

注意事项 #

• 确保Python中的structure_def与PLC内的结构体定义完全一致（变量名、类型、长度），否则解析会出错
• 如果不需要numpy数组，可直接返回原始列表，去掉np.array转换步骤
• 若PLC通知频率不稳定，可在wait_and_get_data中添加进度打印逻辑

内容的提问来源于stack exchange，提问作者Maik_FLS