在工业自动化领域，高速、实时地从可编程逻辑控制器（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 pyads
import ctypes
import numpy as np
import time
# 假设PLC连接信息
# PLC_AMS_NET_ID = '192.168.1.100.1.1'
# PLC_AMS_PORT = 851
class 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.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值的场景尤其有效。
除了数据采集，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通知机制与高效数据处理教程的详细内容，更多请关注php中文网其它相关文章！
每个人都需要一台速度更快、更稳定的 PC。随着时间的推移，垃圾文件、旧注册表数据和不必要的后台进程会占用资源并降低性能。幸运的是，许多工具可以让 Windows 保持平稳运行。
Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号
597
收藏
