FactoryIO仿真入门：手把手教你用Python Modbus库实现‘Sort by Weight’分拣控制

FactoryIO仿真与Python ModbusTCP实战：重量分拣系统从零构建指南

工业自动化仿真技术正逐渐成为工程师和开发者验证控制逻辑的高效工具。FactoryIO作为一款功能强大的仿真软件，配合Python的灵活性，能够快速搭建出接近真实场景的自动化测试环境。本文将完整演示如何利用Python的modbus_tk库与FactoryIO协同工作，实现一个基于重量检测的智能分拣系统。

1. 环境准备与基础配置

1.1 FactoryIO场景搭建

首先需要从FactoryIO官方场景库中加载"Sort by Weight"示例场景。这个场景包含以下核心组件：

• 称重皮带输送机：用于检测物品重量
• 三个分流输送机（左、前、右）：根据重量进行分拣
• 多个光电传感器：检测物品位置
• 转向装置：控制物品流向

关键配置步骤：

1. 启动FactoryIO后，点击"Scenes"→"Samples"→"Conveyors"→选择"Sort by Weight"
2. 进入"Settings"→"Drivers"，选择ModbusTCP作为通信协议
3. 记录下自动分配的Modbus地址映射表（后续Python编程会用到）

提示：FactoryIO默认ModbusTCP端口为502，IP地址为本机地址，可在网络设置中查看

1.2 Python开发环境配置

推荐使用Python 3.8+版本，并安装以下关键库：

pip install modbus-tk # Modbus协议实现库 pip install pymodbus # 可选替代库

验证安装是否成功：

import modbus_tk print(modbus_tk.__version__) # 应显示版本号如1.1.0

2. ModbusTCP通信原理与地址映射

2.1 Modbus数据模型解析

FactoryIO通过ModbusTCP暴露了四种基本数据类型：

2.2 重量分拣场景的地址分配

在"Sort by Weight"场景中，关键设备的Modbus地址映射如下：

离散输入(DI)地址：

• 10000：称重入口传感器
• 10001：称重位置传感器
• 10002：称重出口传感器
• 10003：左分拣入口传感器
• ...（其他传感器依次排列）

线圈(DO)地址：

• 00000：主输送机控制
• 00002：左转向控制
• 00004：右转向控制
• 00006：前向控制

输入寄存器(IR)：

• 30000：称重值（0-1000模拟量）

3. Python控制程序开发

3.1 建立ModbusTCP连接

创建基本的通信框架：

import modbus_tk.modbus_tcp as mt import modbus_tk.defines as md class ModbusController: def __init__(self, host='127.0.0.1', port=502): self.master = mt.TcpMaster(host=host, port=port) self.master.set_timeout(5.0) # 设置超时时间 def read_sensors(self): """读取所有传感器状态""" di_values = self.master.execute(1, md.READ_DISCRETE_INPUTS, 0, 14) weight = self.master.execute(1, md.READ_INPUT_REGISTERS, 0, 1)[0] return di_values, weight def write_actuators(self, do_values): """控制执行器输出""" self.master.execute(1, md.WRITE_MULTIPLE_COILS, 0, output_value=do_values)

3.2 重量分拣逻辑实现

基于状态机的控制逻辑实现：

class WeightSorter: def __init__(self): self.state = "IDLE" self.weight_threshold_light = 350 self.weight_threshold_heavy = 700 def update(self, sensors, weight): """根据传感器和重量更新状态""" at_scale = sensors[1] # 称重位置传感器 if self.state == "IDLE" and at_scale: self.state = "WEIGHING" self.current_weight = weight elif self.state == "WEIGHING" and not at_scale: self.determine_direction() self.state = "SORTING" elif self.state == "SORTING": if not any(sensors[3:9]): # 所有分拣区无物品 self.state = "IDLE" def determine_direction(self): """根据重量确定分拣方向""" if self.current_weight >= self.weight_threshold_heavy: self.direction = "LEFT" elif self.current_weight >= self.weight_threshold_light: self.direction = "RIGHT" else: self.direction = "FRONT"

3.3 主控制循环集成

将各个模块整合为完整系统：

def main_control_loop(): controller = ModbusController() sorter = WeightSorter() try: while True: # 1. 读取现场状态 di_values, weight = controller.read_sensors() # 2. 更新分拣逻辑 sorter.update(di_values, weight) # 3. 生成控制输出 do_values = [0] * 12 do_values[0] = 1 # 主输送机常开 if sorter.state == "SORTING": if sorter.direction == "LEFT": do_values[2] = 1 # 左转向 elif sorter.direction == "RIGHT": do_values[4] = 1 # 右转向 else: do_values[6] = 1 # 前向 # 4. 输出控制信号 controller.write_actuators(do_values) time.sleep(0.1) # 控制周期 except KeyboardInterrupt: print("系统安全停止")

4. 高级功能扩展与调试技巧

4.1 可视化监控界面

使用Tkinter添加简单的监控UI：

import tkinter as tk from tkinter import ttk class MonitoringUI: def __init__(self, controller): self.root = tk.Tk() self.controller = controller self.setup_ui() def setup_ui(self): self.root.title("重量分拣监控") # 传感器状态显示 ttk.Label(self.root, text="传感器状态").grid(row=0, column=0) self.di_labels = [] for i in range(14): lbl = ttk.Label(self.root, text=f"DI{i}: 0", width=8) lbl.grid(row=1+i//3, column=i%3) self.di_labels.append(lbl) # 重量显示 ttk.Label(self.root, text="当前重量:").grid(row=6, column=0) self.weight_var = tk.StringVar(value="0") ttk.Label(self.root, textvariable=self.weight_var).grid(row=6, column=1) # 更新按钮 ttk.Button(self.root, text="刷新", command=self.update).grid(row=7, column=0) def update(self): di_values, weight = self.controller.read_sensors() for i, val in enumerate(di_values): self.di_labels[i].config(text=f"DI{i}: {val}") self.weight_var.set(str(weight)) def run(self): self.root.mainloop()

4.2 常见问题排查

连接失败排查步骤：

1. 确认FactoryIO的ModbusTCP服务已启用
2. 检查防火墙是否阻止了502端口
3. 验证IP地址是否正确（可使用ping测试）
4. 在FactoryIO中查看Modbus日志是否有错误

数据不同步解决方案：

• 添加重试机制：

def safe_read(controller, retries=3): for _ in range(retries): try: return controller.read_sensors() except Exception as e: print(f"读取失败: {e}") time.sleep(1) raise Exception("超过最大重试次数")

• 实现数据校验：

def validate_weight(weight): return 0 <= weight <= 1000 # 根据场景调整范围

5. 性能优化与生产部署

5.1 控制周期优化

通过时间统计优化循环性能：

import time class TimingController: def __init__(self): self.cycle_time = 0.1 # 初始100ms self.avg_cycle = 0 self.cycle_count = 0 def run_cycle(self): start = time.perf_counter() # 执行控制逻辑 self.control_logic() elapsed = time.perf_counter() - start self.update_timing(elapsed) # 动态调整周期 sleep_time = max(0, self.cycle_time - elapsed) time.sleep(sleep_time) def update_timing(self, elapsed): self.avg_cycle = (self.avg_cycle * self.cycle_count + elapsed) / (self.cycle_count + 1) self.cycle_count += 1 # 每100次循环调整一次周期 if self.cycle_count % 100 == 0: if self.avg_cycle < self.cycle_time * 0.8: self.cycle_time *= 0.9 # 加快10% elif self.avg_cycle > self.cycle_time * 1.2: self.cycle_time *= 1.1 # 减慢10%

5.2 日志记录与分析

添加详细的运行日志：

import logging from datetime import datetime def setup_logging(): logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(f"sorting_{datetime.now().strftime('%Y%m%d_%H%M')}.log"), logging.StreamHandler() ] ) class LoggedController(ModbusController): def read_sensors(self): try: di, weight = super().read_sensors() logging.debug(f"读取传感器: DI={di}, 重量={weight}") return di, weight except Exception as e: logging.error(f"传感器读取失败: {e}") raise

6. 项目扩展方向

6.1 多维度分拣策略

除了重量外，可以扩展其他分拣维度：

1. 颜色识别分拣：

   • 通过摄像头获取物品颜色
   • 在Python中使用OpenCV处理图像
   • 新增颜色分拣逻辑分支

2. 尺寸分拣：

   • 使用多个传感器检测物品长度
   • 在Modbus中添加尺寸寄存器

6.2 与MES/ERP系统集成

将分拣数据上传至上层系统：

import requests class ERPIntegration: def __init__(self, api_url): self.api_url = api_url def report_sorting(self, item_id, weight, direction, timestamp): data = { "item_id": item_id, "weight": weight, "direction": direction, "timestamp": timestamp.isoformat() } try: response = requests.post( f"{self.api_url}/sorting_records", json=data, timeout=3 ) response.raise_for_status() return True except Exception as e: print(f"上报失败: {e}") return False

6.3 机器学习优化分拣

收集历史数据训练分拣模型：

import pandas as pd from sklearn.ensemble import RandomForestClassifier class SmartSorter: def __init__(self, model_path=None): if model_path: self.load_model(model_path) else: self.model = RandomForestClassifier() def train(self, X, y): """X: 特征(重量、尺寸等), y: 分拣方向""" self.model.fit(X, y) def predict_direction(self, features): return self.model.predict([features])[0]

实际部署中发现，将控制周期稳定在80-120ms之间可以获得最佳响应速度，同时避免过高的CPU占用。对于更复杂的控制逻辑，建议将状态管理部分单独抽象为状态机类，这样既便于维护也方便后续扩展。