本教程详细探讨了如何利用 PyADS 库的通知机制，高效、Pythonic地处理来自PLC的大量实时数据。文章介绍了通过类封装回调函数来管理内部状态和累积数据的方法，有效避免了全局变量的使用。同时，教程深入讲解了优化数据解析性能的策略，包括使用 return_ctypes=True 结合 NumPy 进行批量数据转换，并提供了监控PLC状态变化的实用示例，旨在帮助开发者构建健壮、高性能的工业数据采集应用。
PyADS通知机制与数据处理挑战
在工业自动化领域，高速、实时地从可编程逻辑控制器（plc）采集数据是一项常见需求。pyads库提供了强大的ads（automation device specification）接口，允许python应用程序与twincat等plc系统进行通信。其中，ads通知（notification）机制尤为关键，它使得plc变量值发生变化时能够主动推送数据到python应用程序，而非通过频繁轮询。这对于200毫秒甚至更快的采集周期，以及每周期传输大量数据（例如，单个信号1000个值，多个信号累积达到100,000个值）的场景至关重要。
然而，在使用PyADS通知时，开发者常面临一个挑战：如何优雅且高效地从异步回调函数中获取数据，并将其整合到主应用程序逻辑中，特别是当需要累积大量数据时。PyADS的回调函数通常独立运行，直接在回调函数内部处理所有数据可能不切实际，而使用全局变量来传递数据则违背了良好的编程实践。PyADS的 @plc.notification 装饰器虽然方便，但不能直接应用于类方法，这使得将回调逻辑封装到类中变得不那么直观。
采用类封装回调逻辑
解决上述挑战的最佳实践是采用面向对象的方法，将PyADS连接、通知设置以及数据处理逻辑封装到一个类中。通过这种方式，我们可以利用类的实例变量来存储和管理回调函数接收到的数据，从而避免使用全局变量，提高代码的可维护性和可读性。
虽然 @plc.notification 装饰器不能直接用于类方法，但我们可以使用 plc.add_device_notification() 方法，并将类实例的方法作为回调函数传递。
以下是一个示例，展示了如何构建一个类来处理PyADS通知，并累积数据：
import pyadsimport ctypesimport numpy as npimport time# 假设PLC连接信息# PLC_AMS_NET_ID = '192.168.1.100.1.1'# PLC_AMS_PORT = 851class PlcDataManager: def __init__(self, plc_ams_net_id, plc_ams_port): self.plc = pyads.Connection(plc_ams_net_id, plc_ams_port) self.plc.open() self.data_buffer = {} # 用于存储各个变量的数据 self.notification_handles = {} # 存储通知句柄，以便后续删除 print(f"Connected to PLC: {self.plc.ams_net_id}:{self.plc.ams_port}") def setup_notifications(self, variables_info): """ 设置多个变量的通知。 variables_info 示例: [ {'name': 'global.sample_structure', 'type': pyads.PLCTYPE_DINT, 'count': 1000, 'struct_def': [('nVar', pyads.PLCTYPE_DINT, 1000)]}, {'name': 'global.heartbeat_var', 'type': pyads.PLCTYPE_BOOL, 'count': 1} ] """ for var_info in variables_info: var_name = var_info['name'] var_type = var_info.get('type') var_count = var_info.get('count', 1) struct_def = var_info.get('struct_def') # 初始化数据缓冲区 self.data_buffer[var_name] = [] if struct_def: # 对于结构体，需要计算结构体大小 size_of_struct = pyads.size_of_structure(struct_def) attr = pyads.NotificationAttrib(size_of_struct) # 使用lambda表达式或functools.partial来传递额外参数 # 这里为了简化，直接在回调中处理结构体解析 handle = self.plc.add_device_notification( var_name, attr, lambda handle, name, timestamp, value, s_def=struct_def: self._on_structured_data_callback(handle, name, timestamp, value, s_def) ) print(f"Added notification for structured variable: {var_name}") else: # 对于单个变量或数组 size = ctypes.sizeof(var_type) * var_count attr = pyads.NotificationAttrib(size) handle = self.plc.add_device_notification( var_name, attr, lambda handle, name, timestamp, value, v_type=var_type, v_count=var_count: self._on_simple_data_callback(handle, name, timestamp, value, v_type, v_count) ) print(f"Added notification for simple variable/array: {var_name}") self.notification_handles[var_name] = handle def _on_simple_data_callback(self, handle, name, timestamp, value, var_type, var_count): """处理简单类型变量（如BOOL, INT, DINT）的通知回调""" # pyads.parse_notification 可以解析原始字节数据 # 对于简单类型，可以直接从value字节串解析 # 或者使用 pyads.read_by_name 并指定 return_ctypes=True 来获取原始数据 # 这里为了演示，我们假设value已经是可解析的字节串 # 实际应用中，如果value是原始字节，需要根据var_type和var_count进行解析 # 示例：假设我们想将DINT数组解析为Python列表 if var_type == pyads.PLCTYPE_DINT and var_count > 1: # 假设value是一个 ctypes.c_int32 * var_count 的实例 # 这里需要根据实际情况解析value # 一个更通用的方法是使用 pyads.read_by_name(var_name, var_type, var_count, return_ctypes=True) # 但这里我们是在回调中，value已经是传入的字节数据 # 简单演示，直接将字节转换为DINT列表（如果value是原始字节） # 实际转换会更复杂，可能需要struct.unpack或numpy # 例如，对于DINT，每4个字节一个值 try: # 假设value是bytes对象，直接转换为numpy数组 # np.frombuffer 效率很高 data = np.frombuffer(value, dtype=np.int32).tolist() self.data_buffer[name].extend(data) # print(f"Received {len(data)} DINTs for {name}. Total: {len(self.data_buffer[name])}") except Exception as e: print(f"Error parsing simple data for {name}: {e}") # Fallback to direct print if parsing fails # print(f"Received simple data for {name}: {value}") else: # 对于单个值，可以直接打印或存储 # 对于单个PLCTYPE_BOOL，value可能就是bytes(b'x00')或bytes(b'x01') parsed_value = pyads.AmsAddr.parse_notification(pyads.AmsAddr(), notification={'name': name, 'value': value}, plc_type=var_type)[-1] self.data_buffer[name].append(parsed_value) # print(f"Received {name}: {parsed_value}") def _on_structured_data_callback(self, handle, name, timestamp, value, structure_def): """处理结构体类型变量的通知回调""" try: # 使用 pyads.dict_from_bytes 解析结构体数据 parsed_values = pyads.dict_from_bytes(value, structure_def) # 将解析后的数据添加到缓冲区 for key, val_list in parsed_values.items(): full_key = f"{name}.{key}" # 例如: global.sample_structure.nVar if full_key not in self.data_buffer: self.data_buffer[full_key] = [] self.data_buffer[full_key].extend(val_list) # print(f"Received structured data for {name}: {parsed_values.keys()}") # print(f"Total data points for {name}.nVar: {len(self.data_buffer.get(f'{name}.nVar', []))}") except Exception as e: print(f"Error parsing structured data for {name}: {e}") def get_data(self, var_name): """获取指定变量的当前累积数据并清空缓冲区""" data = self.data_buffer.get(var_name, []) self.data_buffer[var_name] = [] # 清空已获取的数据 return data def close(self): """关闭PLC连接并移除所有通知""" for var_name, handle in self.notification_handles.items(): try: self.plc.del_device_notification(handle) print(f"Removed notification for {var_name}") except Exception as e: print(f"Error removing notification for {var_name}: {e}") self.plc.close() print("PLC connection closed.")# --- 使用示例 ---if __name__ == "__main__": # 请根据您的PLC配置修改以下参数 # 警告：此示例需要一个运行中的TwinCAT PLC，并包含以下变量： # GLOBAL VARS: # sample_structure : STRUCT # nVar : ARRAY[0..999] OF DINT; # nVar2 : ARRAY[0..999] OF DINT; # ... (最多nVar5) # END_STRUCT # heartbeat_var : BOOL; PLC_AMS_NET_ID = '192.168.1.100.1.1' # 替换为您的PLC AmsNetId PLC_AMS_PORT = 851 # 替换为您的PLC ADS端口 # 定义结构体（与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) ) # 定义需要监听的变量信息 variables_to_monitor = [ {'name': 'global.sample_structure', 'struct_def': structure_def}, {'name': 'global.heartbeat_var', 'type': pyads.PLCTYPE_BOOL, 'count': 1} ] manager = None try: manager = PlcDataManager(PLC_AMS_NET_ID, PLC_AMS_PORT) manager.setup_notifications(variables_to_monitor) # 模拟主程序循环，等待数据并处理 print("nListening for PLC notifications... Press Ctrl+C to exit.") start_time = time.time() while True: time.sleep(1) # 每秒检查一次数据 # 获取并处理结构体数据 for i in range(1, 6): var_name_in_buffer = f'global.sample_structure.nVar{i}' if i > 1 else 'global.sample_structure.nVar' struct_data = manager.get_data(var_name_in_buffer) if struct_data: print(f"[{time.time()-start_time:.2f}s] Processed {len(struct_data)} new data points for {var_name_in_buffer}.") # 这里可以对 struct_data 进行进一步处理，例如存储到数据库、绘图等 # 例如：计算平均值 # print(f" Average: {np.mean(struct_data)}") # 获取并处理心跳变量数据 heartbeat_data = manager.get_data('global.heartbeat_var') if heartbeat_data: print(f"[{time.time()-start_time:.2f}s] Received {len(heartbeat_data)} heartbeat updates. Last value: {heartbeat_data[-1]}") # 示例：达到一定数据量后退出 if len(manager.data_buffer.get('global.sample_structure.nVar', [])) > 10000: print("Collected enough data, exiting...") break except pyads.ADSError as e: print(f"ADS Error: {e}") except Exception as e: print(f"An unexpected error occurred: {e}") finally: if manager: manager.close()
代码解析：
PlcDataManager 类： 封装了PyADS连接 (self.plc)、数据缓冲区 (self.data_buffer) 和通知句柄 (self.notification_handles)。__init__： 初始化PyADS连接并打开。setup_notifications： 遍历需要监听的变量列表。对于结构体，它会根据 structure_def 计算大小；对于简单类型或数组，则根据 pyads.PLCTYPE 和 count 计算大小。然后，调用 self.plc.add_device_notification() 来注册通知。注意：为了将 structure_def 或 var_type/var_count 传递给回调函数，这里使用了 lambda 表达式，通过默认参数捕获这些值，确保回调函数能访问到正确的上下文信息。_on_simple_data_callback 和 _on_structured_data_callback： 这些是实际的回调方法。当PLC变量变化时，PyADS会调用它们。_on_structured_data_callback 使用 pyads.dict_from_bytes(value, structure_def) 将接收到的原始字节数据解析为Python字典，然后将数据追加到 self.data_buffer 中。_on_simple_data_callback 演示了如何处理简单变量或数组。对于数组，它展示了如何使用 numpy.frombuffer 高效地将字节数据转换为数值数组。get_data： 提供一个公共方法，允许外部代码获取某个变量的累积数据，并在获取后清空缓冲区，避免重复处理。close： 在程序退出时，负责移除所有注册的通知并关闭PLC连接，释放资源。
通过这种类封装的方式，数据累积和处理逻辑被清晰地组织在 PlcDataManager 实例内部，避免了全局变量的弊端，并且易于扩展和维护。
优化数据解析与性能
对于高速、大批量的数据采集，回调函数内部的数据解析性能至关重要。pyads.dict_from_bytes 对于结构化数据非常方便，但对于包含大量相同类型元素的数组，它可能不是最高效的选择。
read_by_name(return_ctypes=True) 与手动解析：
当PLC变量是一个大型同类型数组时，可以考虑在通知回调中接收原始字节数据，然后使用更高效的方法进行解析。PyADS允许在读取变量时设置 return_ctypes=True，这将返回C语言类型（ctypes）的原始数据结构，而不是PyADS自动转换的Python对象。虽然在回调函数中 value 参数本身就是原始字节，但理解其背后的原理有助于优化。
对于原始字节数据，尤其是数值数组，Python的 numpy 库提供了极高的性能优势。你可以直接将 value 字节串转换为 numpy 数组：
import numpy as np# ... 在回调函数中 ...def _on_simple_data_callback(self, handle, name, timestamp, value, var_type, var_count): if var_type == pyads.PLCTYPE_DINT and var_count > 1: # 假设 value 是一个 bytes 对象，代表 var_count 个 DINT 值 # DINT 对应 ctypes.c_int32 或 numpy.int32 # 注意：这里假设字节序正确，通常PyADS会处理好 data_array = np.frombuffer(value, dtype=np.int32) # 将 numpy 数组转换为列表或直接存储 numpy 数组 self.data_buffer[name].extend(data_array.tolist()) # 或者直接存储 numpy 数组，如果后续处理也使用 numpy # self.data_buffer[name].append(data_array) # 如果每次回调都收到一个完整的数组
使用 numpy.frombuffer 可以避免Python层面的循环和类型转换开销，实现数量级上的性能提升。这对于每周期1000个甚至更多DINT值的场景尤其有效。
监控PLC状态变化的实践
除了数据采集，PyADS通知机制还可以用于监控PLC的运行状态，这对于构建健壮的工业通信应用至关重要。例如，可以监听PLC的ADS状态（ADSSTATE），当PLC从运行模式切换到配置模式或发生错误时，及时得到通知。
以下是一个监控PLC状态变化的示例：
# ... 在 PlcDataManager 类中添加以下方法 ... def setup_plc_status_notification(self): """设置PLC状态变化的通知""" # ADSIGRP_DEVICE_DATA 和 ADSIOFFS_DEVDATA_ADSSTATE 是预定义的常量，用于获取PLC的ADS状态 # 0xF100 (ADSIGRP_DEVICE_DATA) 和 0x0000 (ADSIOFFS_DEVDATA_ADSSTATE) self.plc.add_device_notification( (int("0xF100", 16), int("0x0000", 16)), # ADS组和偏移量 pyads.NotificationAttrib(ctypes.sizeof(pyads.PLCTYPE_INT)), # ADSSTATE是INT类型 self._on_plc_status_change, ) print("Added notification for PLC status changes.") def _on_plc_status_change(self, notification, _): """PLC状态变化通知回调""" # pyads.parse_notification 可以解析通知字典中的原始值 # notification 参数是一个字典，包含句柄、名称、时间戳和原始值 # 这里我们只关心值，所以使用 _plc.parse_notification 来解析原始字节 *_, value = self.plc.parse_notification(notification, pyads.PLCTYPE_INT) if value != pyads.ADSSTATE_RUN:
以上就是PyADS通知机制与高效数据处理教程的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1366994.html
微信扫一扫 
支付宝扫一扫 
