工业数据自动化采集实战PythonADS高效处理TwinCAT3 PLC数据每次月底报表截止日前产线工程师小李总要加班到深夜——他需要从12台设备的PLC中手动记录温度、压力和产量数据再逐个录入Excel制作日报。这种重复劳动不仅耗时耗力还容易因人为失误导致数据偏差。直到他发现了Python与TwinCAT3的ADS通讯方案原本3天的工作现在只需15分钟自动完成。1. 为什么需要自动化数据采集在智能制造和工业4.0背景下设备产生的数据量呈指数级增长。传统的手动记录方式存在三大致命缺陷效率低下每条产线每天产生500数据点人工记录耗时约2小时准确性差某汽车零部件厂的统计显示手动记录的错误率高达3.2%追溯困难纸质记录或分散的Excel文件难以进行历史数据对比分析以某注塑车间的实际案例为例改用自动化采集后| 指标 | 手动采集 | 自动采集 | 提升幅度 | |--------------|---------|---------|---------| | 单次耗时 | 120分钟 | 3分钟 | 97.5% | | 数据准确率 | 96.8% | 100% | 3.2% | | 月度分析效率 | 40小时 | 1小时 | 97.5% |2. 环境搭建与基础配置2.1 TwinCAT3工程准备首先在TwinCAT3开发环境中创建测试项目新建PLC项目并添加全局变量PROGRAM MAIN VAR {attribute hide} Temperature : ARRAY[1..24] OF LREAL : [20.5, 21.1, ..., 19.8]; Pressure : ARRAY[1..24] OF REAL : [101.3, 102.1, ..., 100.9]; OutputCount : ARRAY[1..7] OF INT : [1500, 1600, ..., 1800]; END_VAR获取目标系统的AMS NetId在SYSTEM → General → Choose Target System查看典型格式192.168.1.100.1.1IP.端口.实例注意生产环境中建议为数据采集创建专用账号而非使用默认管理员权限2.2 Python环境配置安装必要的Python包pip install pyads openpyxl numpy pandas验证安装import pyads print(pyads.__version__) # 应输出3.x.x版本3. 核心数据采集逻辑实现3.1 建立ADS连接创建安全连接的最佳实践def create_plc_connection(net_id, port851, timeout5000): plc pyads.Connection(net_id, port) try: plc.open() plc.set_timeout(timeout) print(f成功连接到 {net_id}) return plc except pyads.ADSError as e: print(f连接失败: {e}) return None3.2 处理不规则数组数据PLC中的数组常存在长度不一致问题这是数据清洗的关键步骤def read_and_align_arrays(plc, var_names, max_length100): data [] for name in var_names: try: # 动态确定数组类型 if LREAL in name: arr plc.read_by_name(name, pyads.PLCTYPE_ARR_LREAL(max_length)) elif REAL in name: arr plc.read_by_name(name, pyads.PLCTYPE_ARR_REAL(max_length)) else: arr plc.read_by_name(name, pyads.PLCTYPE_ARR_INT(max_length)) data.append(arr) except pyads.ADSError as e: print(f读取{name}失败: {e}) data.append([]) # 数组长度对齐 aligned_data [] max_len max(len(arr) for arr in data) for arr in data: aligned_data.append(arr [None]*(max_len - len(arr))) return aligned_data4. 生成专业级Excel报表4.1 基础数据导出使用openpyxl创建带格式的工作表from openpyxl.styles import Font, Alignment from openpyxl.utils import get_column_letter def create_excel_report(data, headers, filename): wb Workbook() ws wb.active ws.title 生产数据 # 设置表头样式 header_font Font(boldTrue, colorFFFFFF) header_fill PatternFill(start_color4472C4, end_color4472C4, fill_typesolid) # 写入表头 ws.append(headers) for col in range(1, len(headers)1): cell ws.cell(row1, columncol) cell.font header_font cell.fill header_fill ws.column_dimensions[get_column_letter(col)].width 15 # 写入数据 for row in zip(*data): ws.append(row) # 添加自动筛选 ws.auto_filter.ref ws.dimensions wb.save(filename)4.2 高级功能扩展为报表添加可视化元素from openpyxl.chart import LineChart, Reference def add_trend_chart(ws, data_range, title): chart LineChart() chart.title title chart.style 13 chart.y_axis.title 数值 chart.x_axis.title 时间序列 data Reference(ws, min_col2, min_row2, max_col4, max_row25) chart.add_data(data, titles_from_dataTrue) ws.add_chart(chart, F2)5. 生产环境部署方案5.1 定时任务配置使用Windows任务计划程序实现每日自动运行创建批处理文件run_collect.batecho off C:\Python39\python.exe D:\scripts\plc_data_collect.py在任务计划程序中设置触发器每天凌晨2点操作启动程序run_collect.bat条件只在网络连接时运行5.2 异常处理机制增强脚本的健壮性def main(): plc create_plc_connection(192.168.1.100.1.1) if not plc: send_alert_email(PLC连接失败) return try: data read_and_align_arrays(plc, [MAIN.Temperature, MAIN.Pressure]) create_excel_report(data, [时间, 温度, 压力], daily_report.xlsx) upload_to_cloud(daily_report.xlsx) # 自动上传到企业云盘 except Exception as e: log_error(e) send_alert_email(f数据采集失败: {str(e)}) finally: plc.close()6. 性能优化技巧处理大型数据集时的建议内存优化# 使用生成器分批处理数据 def batch_process(plc, var_name, batch_size1000): total plc.read_by_name(f{var_name}.LENGTH) for i in range(0, total, batch_size): yield plc.read_by_name( var_name, pyads.PLCTYPE_ARR_LREAL(batch_size), index_offseti )并行采集from concurrent.futures import ThreadPoolExecutor def parallel_read(plc, variables): with ThreadPoolExecutor() as executor: results list(executor.map( lambda v: plc.read_by_name(v[0], v[1]), variables )) return results数据压缩存储# 使用Parquet格式替代Excel import pyarrow.parquet as pq df pd.DataFrame(data) pq.write_table(pa.Table.from_pandas(df), data.parquet)在实际项目中这套方案将PLC数据采集效率提升了40倍。一个典型的应用场景是凌晨2点自动采集全厂设备数据3:00生成日报并发送给管理层5:00触发MES系统更新生产计划——全程无需人工干预。