# FlexSim 与 Python Socket 通信完整指南本文档详细介绍如何通过 Socket 通信实现 FlexSim 与 Python 之间的实时数据交互。采用 **Python 作为服务器Server**、**FlexSim 作为客户端Client** 的架构可有效避免 FlexSim 作为服务器时可能出现的 socketinit() 初始化失败问题。## 一、整体架构| 角色 | 功能 ||------|------|| **Python** | TCP 服务器监听指定端口接收并处理 FlexSim 发来的 JSON 数据 || **FlexSim** | TCP 客户端主动连接 Python 服务器定时发送采集的数据 |FlexSim 作为客户端时使用 clientcreate() 创建 Socket使用 clientconnect() 连接服务器。FlexSim 最多可同时创建 50 个客户端 Socket。## 二、Python 服务器端设置### 2.1 环境要求- Python 3.6 或以上版本- 无需安装额外依赖库仅使用标准库 socket、json、csv、datetime### 2.2 创建 Python 服务器脚本创建文件 flexsim_server.py写入以下完整代码python TCP Socket 服务器 - 接收 FlexSim 数据 import socket import json import csv from datetime import datetime # 配置 SERVER_IP 0.0.0.0 # 监听所有网络接口 SERVER_PORT 9000 CSV_FILENAME sim_data.csv # 状态映射表 STATE_MAP { 0: 未定义, 1: 空闲, 2: 处理中, 3: 忙碌, 4: 堵塞, 5: 生成中, 6: 空, 7: 收集中, 8: 释放中, 9: 等待操作员, 10: 等待运输工具, 11: 故障/停机, 12: 计划停机, 13: 暂停中, 14: 装载中, 15: 卸载中, 16: 空载偏移行进, 17: 负载偏移行进, } def get_timestamp(): return datetime.now().strftime(%Y-%m-%d %H:%M:%S.%f)[:-3] def parse_and_display(raw_line): 解析FlexSim发送的JSON数据并显示 try: clean_str raw_line.strip() if not clean_str: return None parsed json.loads(clean_str) t parsed.get(timestamp, 0) r1_code parsed.get(robot1_state, -1) r2_code parsed.get(robot2_state, -1) r1_desc STATE_MAP.get(r1_code, f未知({r1_code})) r2_desc STATE_MAP.get(r2_code, f未知({r2_code})) r1_task parsed.get(robot1_task, 无任务) r2_task parsed.get(robot2_task, 无任务) b1 parsed.get(buff1_content, 0) agv1_load parsed.get(agv1_load, 0) agv2_load parsed.get(agv2_load, 0) agv1_x parsed.get(agv1_x, 0) agv1_y parsed.get(agv1_y, 0) agv1_z parsed.get(agv1_z, 0) agv2_x parsed.get(agv2_x, 0) agv2_y parsed.get(agv2_y, 0) agv2_z parsed.get(agv2_z, 0) print(f[{t:.3f}], flushTrue) print(f robot1: {r1_desc} | 任务: {r1_task}, flushTrue) print(f robot2: {r2_desc} | 任务: {r2_task}, flushTrue) print(f buff1: {b1}, flushTrue) print(f AGV1: 坐标({agv1_x:.2f}, {agv1_y:.2f}, {agv1_z:.2f}) | 载货: {agv1_load}, flushTrue) print(f AGV2: 坐标({agv2_x:.2f}, {agv2_y:.2f}, {agv2_z:.2f}) | 载货: {agv2_load}, flushTrue) print(- * 50, flushTrue) return { timestamp: t, robot1_state: r1_desc, robot2_state: r2_desc, robot1_task: r1_task, robot2_task: r2_task, buff1: b1, agv1_load: agv1_load, agv2_load: agv2_load, agv1_x: agv1_x, agv1_y: agv1_y, agv1_z: agv1_z, agv2_x: agv2_x, agv2_y: agv2_y, agv2_z: agv2_z, } except json.JSONDecodeError: print(f[{get_timestamp()}] JSON解析失败原始数据: {raw_line[:100]}) except Exception as e: print(f[{get_timestamp()}] 解析异常: {e}) return None def save_to_csv(parsed): 将数据保存到CSV文件 if parsed is None: return try: with open(CSV_FILENAME, a, newline, encodingutf-8) as f: fieldnames [ 接收时间, 仿真时间, robot1状态, robot2状态, robot1任务, robot2任务, buff1, AGV1_载货, AGV2_载货, AGV1_X, AGV1_Y, AGV1_Z, AGV2_X, AGV2_Y, AGV2_Z ] writer csv.DictWriter(f, fieldnamesfieldnames) if f.tell() 0: writer.writeheader() writer.writerow({ 接收时间: get_timestamp(), 仿真时间: parsed[timestamp], robot1状态: parsed[robot1_state], robot2状态: parsed[robot2_state], robot1任务: parsed[robot1_task], robot2任务: parsed[robot2_task], buff1: parsed[buff1], AGV1_载货: parsed[agv1_load], AGV2_载货: parsed[agv2_load], AGV1_X: parsed[agv1_x], AGV1_Y: parsed[agv1_y], AGV1_Z: parsed[agv1_z], AGV2_X: parsed[agv2_x], AGV2_Y: parsed[agv2_y], AGV2_Z: parsed[agv2_z], }) except Exception as e: print(f[{get_timestamp()}] CSV写入失败: {e}) def main(): print( * 70) print(FlexSim 数据接收器 (Python 作为服务器)) print(f监听: {SERVER_IP}:{SERVER_PORT}) print(f数据存储: {CSV_FILENAME}) print(按 CtrlC 停止) print( * 70) # 初始化 CSV 文件每次启动覆盖旧数据 try: with open(CSV_FILENAME, w, newline, encodingutf-8) as f: fieldnames [ 接收时间, 仿真时间, robot1状态, robot2状态, robot1任务, robot2任务, buff1, AGV1_载货, AGV2_载货, AGV1_X, AGV1_Y, AGV1_Z, AGV2_X, AGV2_Y, AGV2_Z ] writer csv.DictWriter(f, fieldnamesfieldnames) writer.writeheader() print(f[{get_timestamp()}] CSV 文件已初始化) except Exception as e: print(f[{get_timestamp()}] CSV 初始化失败: {e}) # 创建服务器 Socket server_socket socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server_socket.bind((SERVER_IP, SERVER_PORT)) server_socket.listen(1) print(f[{get_timestamp()}] 服务器已启动等待 FlexSim 连接...) while True: try: conn, addr server_socket.accept() print(f[{get_timestamp()}] FlexSim 已连接: {addr}) conn_file conn.makefile(r, encodingutf-8, buffering1) while True: line conn_file.readline() if not line: print(f[{get_timestamp()}] FlexSim 断开连接) break parsed parse_and_display(line) if parsed: save_to_csv(parsed) conn.close() except KeyboardInterrupt: print(f\n[{get_timestamp()}] 用户中断) break except Exception as e: print(f[{get_timestamp()}] 错误: {e}) continue server_socket.close() print(f[{get_timestamp()}] 服务器已关闭) if __name__ __main__: main() ## 三、FlexSim 端设置### 3.1 模型对象命名在 FlexSim 的 3D 视图中添加需要监控的对象并重命名为**英文名称**避免中文导致 node() 函数找不到对象| 对象类型 | 推荐名称 ||----------|----------|| 机械臂 1 | robot1 || 机械臂 2 | robot2 || 暂存区 1 | buff1 || AGV 1 | AGV1 || AGV 2 | AGV2 | **注意**如果你使用不同的名称请相应修改 DataSender 代码中 node(MODEL:/对象名, model()) 的字符串。### 3.2 设置 OnRunStart 触发器**操作路径**工具箱(Toolbox) → 模型触发器(Model Triggers) → 双击 OnRunStart**作用**模型开始运行时初始化客户端 Socket 并准备连接 Python 服务器。flexscript // // OnRunStart - 初始化客户端连接 // print(OnRunStart 开始执行...); treenode tools node(/Tools, model()); if (!objectexists(tools)) tools node( Tools, model()); // 创建 client 节点并初始化为 0 treenode clientNode assertsubnode(tools, client, DATATYPE_NUMBER); setnodenum(clientNode, 0); print(OnRunStart 执行完毕DataSender 将管理连接); return 1; ### 3.3 创建 DataSender 用户事件**操作路径**工具箱(Toolbox) → 点击 添加(Add) → 选择 用户事件(User Events)**属性设置**| 属性 | 设置值 | 说明 ||------|--------|------|| 名称 | DataSender | 便于识别 || 首次事件时间 | 0 | 模型运行后立即执行第一次 || **重复事件** | ✅ **勾选** | 这是实现定时触发的关键 || **重复事件时间** | 1.0 | 每 1 秒执行一次单位模型时间单位 |**事件代码**粘贴以下完整代码flexscript /** Custom Code */ print( DataSender 开始执行 (时间: numtostring(time(), 3) ) ); // 1. 获取 Tools 和 client 节点 treenode tools node(/Tools, model()); if (!objectexists(tools)) { tools node( Tools, model()); } treenode clientNode node(client, tools); if (!objectexists(clientNode)) { clientNode assertsubnode(tools, client, DATATYPE_NUMBER); setnodenum(clientNode, 0); } int clientID getnodenum(clientNode); // 2. 如果 clientID 无效建立新连接 if (clientID 0) { print(DataSender: 创建新连接...); int newID clientcreate(); // 创建客户端 Socket if (newID 0) { print(DataSender: 创建客户端失败); return 0; } int conn clientconnect(newID, 127.0.0.1, 9000); // 连接 Python 服务器 if (!conn) { print(DataSender: 连接服务器失败); clientclose(newID); setnodenum(clientNode, 0); return 0; } clientID newID; setnodenum(clientNode, clientID); print(DataSender: 连接成功); } else { print(DataSender: 使用已有连接 ID numtostring(clientID, 0)); } // 3. 获取对象节点 treenode robot1 node(MODEL:/robot1, model()); treenode robot2 node(MODEL:/robot2, model()); treenode buff1 node(MODEL:/buff1, model()); treenode agv1 node(MODEL:/AGV1, model()); treenode agv2 node(MODEL:/AGV2, model()); if (!objectexists(robot1) || !objectexists(robot2) || !objectexists(buff1) || !objectexists(agv1) || !objectexists(agv2)) { print(DataSender: 部分对象未找到跳过本次发送); return 0; } // 4. 采集机械臂状态 int state1 getstatenum(robot1, 0); int state2 getstatenum(robot2, 0); // 5. 采集机械臂任务信息 string robot1_task_desc 无任务; string robot2_task_desc 无任务; TaskExecuter te1 robot1.as(TaskExecuter); var task1 te1.activeTask; if (task1 ! 0) { int taskType task1.type; var obj2 task1.involved2; string name2 ; if (objectexists(obj2)) name2 getname(obj2); if (taskType TASKTYPE_LOAD) robot1_task_desc 从 [ name2 ] 装载; else if (taskType TASKTYPE_UNLOAD) robot1_task_desc 在 [ name2 ] 卸载; else if (taskType TASKTYPE_TRAVEL) robot1_task_desc 前往 [ name2 ]; } TaskExecuter te2 robot2.as(TaskExecuter); var task2 te2.activeTask; if (task2 ! 0) { int taskType task2.type; var obj2 task2.involved2; string name2 ; if (objectexists(obj2)) name2 getname(obj2); if (taskType TASKTYPE_LOAD) robot2_task_desc 从 [ name2 ] 装载; else if (taskType TASKTYPE_UNLOAD) robot2_task_desc 在 [ name2 ] 卸载; else if (taskType TASKTYPE_TRAVEL) robot2_task_desc 前往 [ name2 ]; } // 6. 采集 AGV 坐标和载货量 double agv1X xloc(agv1); double agv1Y yloc(agv1); double agv1Z zloc(agv1); double agv2X xloc(agv2); double agv2Y yloc(agv2); double agv2Z zloc(agv2); int agv1_load agv1.subnodes.length; int agv2_load agv2.subnodes.length; // 7. 采集暂存区存货量 int content1 content(buff1); // 8. 构建 JSON 数据包 double t time(); string json {; json json \timestamp\: numtostring(t, 3) ,; json json \robot1_state\: numtostring(state1, 0) ,; json json \robot2_state\: numtostring(state2, 0) ,; json json \robot1_task\:\ robot1_task_desc \,; json json \robot2_task\:\ robot2_task_desc \,; json json \buff1_content\: numtostring(content1, 0) ,; json json \agv1_load\: numtostring(agv1_load, 0) ,; json json \agv2_load\: numtostring(agv2_load, 0) ,; json json \agv1_x\: numtostring(agv1X, 2) ,; json json \agv1_y\: numtostring(agv1Y, 2) ,; json json \agv1_z\: numtostring(agv1Z, 2) ,; json json \agv2_x\: numtostring(agv2X, 2) ,; json json \agv2_y\: numtostring(agv2Y, 2) ,; json json \agv2_z\: numtostring(agv2Z, 2); json json }; // 9. 发送数据 int sendResult clientsend(clientID, json \n); if (sendResult 0) { print(DataSender: 发送失败连接已断开); clientclose(clientID); setnodenum(clientNode, 0); } else { print(DataSender: 【FlexSim已发出】 json); } print( DataSender 执行完毕 ); return 1; **代码说明** - clientcreate() 创建新的客户端 Socket - clientconnect() 连接到指定的 IP 和端口 - clientsend() 发送数据 - clientclose() 关闭连接 - 用户事件的“重复事件”功能驱动定时执行无需在代码中使用 senddelayedmessage 自循环### 3.4 设置 OnReset 触发器**操作路径**工具箱(Toolbox) → 模型触发器(Model Triggers) → 双击 OnReset**作用**模型重置时关闭 Socket 连接并清理状态。flexscript // // OnReset - 清理连接 // print(OnReset 开始执行...); treenode tools node(/Tools, model()); if (!objectexists(tools)) tools node( Tools, model()); treenode clientNode node(client, tools); if (objectexists(clientNode)) { int id getnodenum(clientNode); if (id 0) { clientclose(id); print(已关闭客户端连接 ID: numtostring(id, 0)); } setnodenum(clientNode, 0); } print(OnReset 执行完毕); return 1; ## 四、测试与验证### 4.1 启动顺序严格按顺序1. **启动 Python 服务器**bashpython flexsim_server.py看到 服务器已启动等待 FlexSim 连接... 表示服务器就绪。2. **打开 FlexSim 模型**确保已完成上述所有设置。3. **点击“重置”** → **点击“运行”**。### 4.2 预期输出**FlexSim 控制台**应显示OnRunStart 开始执行...OnRunStart 执行完毕DataSender 将管理连接 DataSender 开始执行 (时间: 0) DataSender: 创建新连接...DataSender: 连接成功DataSender: 【FlexSim已发出】{timestamp:0,...} DataSender 执行完毕 DataSender 开始执行 (时间: 1) DataSender: 使用已有连接 ID1DataSender: 【FlexSim已发出】{timestamp:1,...} DataSender 执行完毕 **Python 服务器**应显示[2026-08-03 14:10:36.495] FlexSim 已连接: (127.0.0.1, 52742)[0.000]robot1: 空闲 | 任务: 无任务robot2: 空闲 | 任务: 无任务buff1: 0AGV1: 坐标(102.00, 18.00, 0.00) | 载货: 0AGV2: 坐标(102.00, 15.00, 0.00) | 载货: 0--------------------------------------------------[1.000]...### 4.3 数据说明FlexSim 发送的 JSON 数据包含以下字段| 字段 | 类型 | 说明 ||------|------|------|| timestamp | float | 当前仿真时间 || robot1_state / robot2_state | int | 机械臂状态码1空闲, 2处理中, 3忙碌, 16空载偏移行进等 || robot1_task / robot2_task | string | 机械臂当前任务描述 || buff1_content | int | 暂存区 1 当前存货量 || agv1_load / agv2_load | int | AGV 当前载货数量 || agv1_x/y/z / agv2_x/y/z | float | AGV 三维坐标 |## 五、常见问题排查| 问题现象 | 可能原因 | 解决方法 ||----------|----------|----------|| Python 显示“FlexSim 已连接”后立即断开 | FlexSim 端 clientconnect 重复调用导致连接重置 | 检查 DataSender 中是否只在新连接时调用 clientconnect已连接时不再重复调用 || FlexSim 只发送了 timestamp0 的数据 | 用户事件的“重复事件”未勾选 | 在 DataSender 属性中**勾选“重复事件”**并设置重复时间为 1.0 || Python 收不到数据但 FlexSim 显示发送成功 | 防火墙拦截或 IP/端口不匹配 | 检查防火墙设置确认 Python 和 FlexSim 使用相同的端口号 || DataSender 显示“连接服务器失败” | Python 服务器未启动 | 先启动 Python 服务器再运行 FlexSim || 对象找不到存在: 0 | 对象名称不匹配 | 检查模型树中的对象名称确保与 node(MODEL:/对象名, model()) 完全一致 || 重置后无法重新连接 | 连接未正确关闭 | 确保 OnReset 中调用了 clientclose(id) 并重置 client 节点为 0 |## 六、架构优势说明本方案采用 **Python 作为服务器、FlexSim 作为客户端** 的架构相比 FlexSim 作为服务器有以下优势1. **避开 socketinit() 问题**FlexSim 作为服务器时需要调用 socketinit() 初始化 Windows Socket 后台进程某些环境下可能因权限或系统配置失败。作为客户端时只需 clientcreate() clientconnect()更加稳定可靠。2. **连接管理更简单**客户端负责主动连接断线后可自动重连。3. **Python 生态优势**Python 作为服务器可以方便地扩展数据存储、可视化、机器学习等高级功能。4. **调试更方便**Python 服务器可以独立运行和测试无需依赖 FlexSim 环境。 FlexSim 与 Python Socket 通信完整指南
# FlexSim 与 Python Socket 通信完整指南本文档详细介绍如何通过 Socket 通信实现 FlexSim 与 Python 之间的实时数据交互。采用 **Python 作为服务器Server**、**FlexSim 作为客户端Client** 的架构可有效避免 FlexSim 作为服务器时可能出现的 socketinit() 初始化失败问题。## 一、整体架构| 角色 | 功能 ||------|------|| **Python** | TCP 服务器监听指定端口接收并处理 FlexSim 发来的 JSON 数据 || **FlexSim** | TCP 客户端主动连接 Python 服务器定时发送采集的数据 |FlexSim 作为客户端时使用 clientcreate() 创建 Socket使用 clientconnect() 连接服务器。FlexSim 最多可同时创建 50 个客户端 Socket。## 二、Python 服务器端设置### 2.1 环境要求- Python 3.6 或以上版本- 无需安装额外依赖库仅使用标准库 socket、json、csv、datetime### 2.2 创建 Python 服务器脚本创建文件 flexsim_server.py写入以下完整代码python TCP Socket 服务器 - 接收 FlexSim 数据 import socket import json import csv from datetime import datetime # 配置 SERVER_IP 0.0.0.0 # 监听所有网络接口 SERVER_PORT 9000 CSV_FILENAME sim_data.csv # 状态映射表 STATE_MAP { 0: 未定义, 1: 空闲, 2: 处理中, 3: 忙碌, 4: 堵塞, 5: 生成中, 6: 空, 7: 收集中, 8: 释放中, 9: 等待操作员, 10: 等待运输工具, 11: 故障/停机, 12: 计划停机, 13: 暂停中, 14: 装载中, 15: 卸载中, 16: 空载偏移行进, 17: 负载偏移行进, } def get_timestamp(): return datetime.now().strftime(%Y-%m-%d %H:%M:%S.%f)[:-3] def parse_and_display(raw_line): 解析FlexSim发送的JSON数据并显示 try: clean_str raw_line.strip() if not clean_str: return None parsed json.loads(clean_str) t parsed.get(timestamp, 0) r1_code parsed.get(robot1_state, -1) r2_code parsed.get(robot2_state, -1) r1_desc STATE_MAP.get(r1_code, f未知({r1_code})) r2_desc STATE_MAP.get(r2_code, f未知({r2_code})) r1_task parsed.get(robot1_task, 无任务) r2_task parsed.get(robot2_task, 无任务) b1 parsed.get(buff1_content, 0) agv1_load parsed.get(agv1_load, 0) agv2_load parsed.get(agv2_load, 0) agv1_x parsed.get(agv1_x, 0) agv1_y parsed.get(agv1_y, 0) agv1_z parsed.get(agv1_z, 0) agv2_x parsed.get(agv2_x, 0) agv2_y parsed.get(agv2_y, 0) agv2_z parsed.get(agv2_z, 0) print(f[{t:.3f}], flushTrue) print(f robot1: {r1_desc} | 任务: {r1_task}, flushTrue) print(f robot2: {r2_desc} | 任务: {r2_task}, flushTrue) print(f buff1: {b1}, flushTrue) print(f AGV1: 坐标({agv1_x:.2f}, {agv1_y:.2f}, {agv1_z:.2f}) | 载货: {agv1_load}, flushTrue) print(f AGV2: 坐标({agv2_x:.2f}, {agv2_y:.2f}, {agv2_z:.2f}) | 载货: {agv2_load}, flushTrue) print(- * 50, flushTrue) return { timestamp: t, robot1_state: r1_desc, robot2_state: r2_desc, robot1_task: r1_task, robot2_task: r2_task, buff1: b1, agv1_load: agv1_load, agv2_load: agv2_load, agv1_x: agv1_x, agv1_y: agv1_y, agv1_z: agv1_z, agv2_x: agv2_x, agv2_y: agv2_y, agv2_z: agv2_z, } except json.JSONDecodeError: print(f[{get_timestamp()}] JSON解析失败原始数据: {raw_line[:100]}) except Exception as e: print(f[{get_timestamp()}] 解析异常: {e}) return None def save_to_csv(parsed): 将数据保存到CSV文件 if parsed is None: return try: with open(CSV_FILENAME, a, newline, encodingutf-8) as f: fieldnames [ 接收时间, 仿真时间, robot1状态, robot2状态, robot1任务, robot2任务, buff1, AGV1_载货, AGV2_载货, AGV1_X, AGV1_Y, AGV1_Z, AGV2_X, AGV2_Y, AGV2_Z ] writer csv.DictWriter(f, fieldnamesfieldnames) if f.tell() 0: writer.writeheader() writer.writerow({ 接收时间: get_timestamp(), 仿真时间: parsed[timestamp], robot1状态: parsed[robot1_state], robot2状态: parsed[robot2_state], robot1任务: parsed[robot1_task], robot2任务: parsed[robot2_task], buff1: parsed[buff1], AGV1_载货: parsed[agv1_load], AGV2_载货: parsed[agv2_load], AGV1_X: parsed[agv1_x], AGV1_Y: parsed[agv1_y], AGV1_Z: parsed[agv1_z], AGV2_X: parsed[agv2_x], AGV2_Y: parsed[agv2_y], AGV2_Z: parsed[agv2_z], }) except Exception as e: print(f[{get_timestamp()}] CSV写入失败: {e}) def main(): print( * 70) print(FlexSim 数据接收器 (Python 作为服务器)) print(f监听: {SERVER_IP}:{SERVER_PORT}) print(f数据存储: {CSV_FILENAME}) print(按 CtrlC 停止) print( * 70) # 初始化 CSV 文件每次启动覆盖旧数据 try: with open(CSV_FILENAME, w, newline, encodingutf-8) as f: fieldnames [ 接收时间, 仿真时间, robot1状态, robot2状态, robot1任务, robot2任务, buff1, AGV1_载货, AGV2_载货, AGV1_X, AGV1_Y, AGV1_Z, AGV2_X, AGV2_Y, AGV2_Z ] writer csv.DictWriter(f, fieldnamesfieldnames) writer.writeheader() print(f[{get_timestamp()}] CSV 文件已初始化) except Exception as e: print(f[{get_timestamp()}] CSV 初始化失败: {e}) # 创建服务器 Socket server_socket socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server_socket.bind((SERVER_IP, SERVER_PORT)) server_socket.listen(1) print(f[{get_timestamp()}] 服务器已启动等待 FlexSim 连接...) while True: try: conn, addr server_socket.accept() print(f[{get_timestamp()}] FlexSim 已连接: {addr}) conn_file conn.makefile(r, encodingutf-8, buffering1) while True: line conn_file.readline() if not line: print(f[{get_timestamp()}] FlexSim 断开连接) break parsed parse_and_display(line) if parsed: save_to_csv(parsed) conn.close() except KeyboardInterrupt: print(f\n[{get_timestamp()}] 用户中断) break except Exception as e: print(f[{get_timestamp()}] 错误: {e}) continue server_socket.close() print(f[{get_timestamp()}] 服务器已关闭) if __name__ __main__: main() ## 三、FlexSim 端设置### 3.1 模型对象命名在 FlexSim 的 3D 视图中添加需要监控的对象并重命名为**英文名称**避免中文导致 node() 函数找不到对象| 对象类型 | 推荐名称 ||----------|----------|| 机械臂 1 | robot1 || 机械臂 2 | robot2 || 暂存区 1 | buff1 || AGV 1 | AGV1 || AGV 2 | AGV2 | **注意**如果你使用不同的名称请相应修改 DataSender 代码中 node(MODEL:/对象名, model()) 的字符串。### 3.2 设置 OnRunStart 触发器**操作路径**工具箱(Toolbox) → 模型触发器(Model Triggers) → 双击 OnRunStart**作用**模型开始运行时初始化客户端 Socket 并准备连接 Python 服务器。flexscript // // OnRunStart - 初始化客户端连接 // print(OnRunStart 开始执行...); treenode tools node(/Tools, model()); if (!objectexists(tools)) tools node( Tools, model()); // 创建 client 节点并初始化为 0 treenode clientNode assertsubnode(tools, client, DATATYPE_NUMBER); setnodenum(clientNode, 0); print(OnRunStart 执行完毕DataSender 将管理连接); return 1; ### 3.3 创建 DataSender 用户事件**操作路径**工具箱(Toolbox) → 点击 添加(Add) → 选择 用户事件(User Events)**属性设置**| 属性 | 设置值 | 说明 ||------|--------|------|| 名称 | DataSender | 便于识别 || 首次事件时间 | 0 | 模型运行后立即执行第一次 || **重复事件** | ✅ **勾选** | 这是实现定时触发的关键 || **重复事件时间** | 1.0 | 每 1 秒执行一次单位模型时间单位 |**事件代码**粘贴以下完整代码flexscript /** Custom Code */ print( DataSender 开始执行 (时间: numtostring(time(), 3) ) ); // 1. 获取 Tools 和 client 节点 treenode tools node(/Tools, model()); if (!objectexists(tools)) { tools node( Tools, model()); } treenode clientNode node(client, tools); if (!objectexists(clientNode)) { clientNode assertsubnode(tools, client, DATATYPE_NUMBER); setnodenum(clientNode, 0); } int clientID getnodenum(clientNode); // 2. 如果 clientID 无效建立新连接 if (clientID 0) { print(DataSender: 创建新连接...); int newID clientcreate(); // 创建客户端 Socket if (newID 0) { print(DataSender: 创建客户端失败); return 0; } int conn clientconnect(newID, 127.0.0.1, 9000); // 连接 Python 服务器 if (!conn) { print(DataSender: 连接服务器失败); clientclose(newID); setnodenum(clientNode, 0); return 0; } clientID newID; setnodenum(clientNode, clientID); print(DataSender: 连接成功); } else { print(DataSender: 使用已有连接 ID numtostring(clientID, 0)); } // 3. 获取对象节点 treenode robot1 node(MODEL:/robot1, model()); treenode robot2 node(MODEL:/robot2, model()); treenode buff1 node(MODEL:/buff1, model()); treenode agv1 node(MODEL:/AGV1, model()); treenode agv2 node(MODEL:/AGV2, model()); if (!objectexists(robot1) || !objectexists(robot2) || !objectexists(buff1) || !objectexists(agv1) || !objectexists(agv2)) { print(DataSender: 部分对象未找到跳过本次发送); return 0; } // 4. 采集机械臂状态 int state1 getstatenum(robot1, 0); int state2 getstatenum(robot2, 0); // 5. 采集机械臂任务信息 string robot1_task_desc 无任务; string robot2_task_desc 无任务; TaskExecuter te1 robot1.as(TaskExecuter); var task1 te1.activeTask; if (task1 ! 0) { int taskType task1.type; var obj2 task1.involved2; string name2 ; if (objectexists(obj2)) name2 getname(obj2); if (taskType TASKTYPE_LOAD) robot1_task_desc 从 [ name2 ] 装载; else if (taskType TASKTYPE_UNLOAD) robot1_task_desc 在 [ name2 ] 卸载; else if (taskType TASKTYPE_TRAVEL) robot1_task_desc 前往 [ name2 ]; } TaskExecuter te2 robot2.as(TaskExecuter); var task2 te2.activeTask; if (task2 ! 0) { int taskType task2.type; var obj2 task2.involved2; string name2 ; if (objectexists(obj2)) name2 getname(obj2); if (taskType TASKTYPE_LOAD) robot2_task_desc 从 [ name2 ] 装载; else if (taskType TASKTYPE_UNLOAD) robot2_task_desc 在 [ name2 ] 卸载; else if (taskType TASKTYPE_TRAVEL) robot2_task_desc 前往 [ name2 ]; } // 6. 采集 AGV 坐标和载货量 double agv1X xloc(agv1); double agv1Y yloc(agv1); double agv1Z zloc(agv1); double agv2X xloc(agv2); double agv2Y yloc(agv2); double agv2Z zloc(agv2); int agv1_load agv1.subnodes.length; int agv2_load agv2.subnodes.length; // 7. 采集暂存区存货量 int content1 content(buff1); // 8. 构建 JSON 数据包 double t time(); string json {; json json \timestamp\: numtostring(t, 3) ,; json json \robot1_state\: numtostring(state1, 0) ,; json json \robot2_state\: numtostring(state2, 0) ,; json json \robot1_task\:\ robot1_task_desc \,; json json \robot2_task\:\ robot2_task_desc \,; json json \buff1_content\: numtostring(content1, 0) ,; json json \agv1_load\: numtostring(agv1_load, 0) ,; json json \agv2_load\: numtostring(agv2_load, 0) ,; json json \agv1_x\: numtostring(agv1X, 2) ,; json json \agv1_y\: numtostring(agv1Y, 2) ,; json json \agv1_z\: numtostring(agv1Z, 2) ,; json json \agv2_x\: numtostring(agv2X, 2) ,; json json \agv2_y\: numtostring(agv2Y, 2) ,; json json \agv2_z\: numtostring(agv2Z, 2); json json }; // 9. 发送数据 int sendResult clientsend(clientID, json \n); if (sendResult 0) { print(DataSender: 发送失败连接已断开); clientclose(clientID); setnodenum(clientNode, 0); } else { print(DataSender: 【FlexSim已发出】 json); } print( DataSender 执行完毕 ); return 1; **代码说明** - clientcreate() 创建新的客户端 Socket - clientconnect() 连接到指定的 IP 和端口 - clientsend() 发送数据 - clientclose() 关闭连接 - 用户事件的“重复事件”功能驱动定时执行无需在代码中使用 senddelayedmessage 自循环### 3.4 设置 OnReset 触发器**操作路径**工具箱(Toolbox) → 模型触发器(Model Triggers) → 双击 OnReset**作用**模型重置时关闭 Socket 连接并清理状态。flexscript // // OnReset - 清理连接 // print(OnReset 开始执行...); treenode tools node(/Tools, model()); if (!objectexists(tools)) tools node( Tools, model()); treenode clientNode node(client, tools); if (objectexists(clientNode)) { int id getnodenum(clientNode); if (id 0) { clientclose(id); print(已关闭客户端连接 ID: numtostring(id, 0)); } setnodenum(clientNode, 0); } print(OnReset 执行完毕); return 1; ## 四、测试与验证### 4.1 启动顺序严格按顺序1. **启动 Python 服务器**bashpython flexsim_server.py看到 服务器已启动等待 FlexSim 连接... 表示服务器就绪。2. **打开 FlexSim 模型**确保已完成上述所有设置。3. **点击“重置”** → **点击“运行”**。### 4.2 预期输出**FlexSim 控制台**应显示OnRunStart 开始执行...OnRunStart 执行完毕DataSender 将管理连接 DataSender 开始执行 (时间: 0) DataSender: 创建新连接...DataSender: 连接成功DataSender: 【FlexSim已发出】{timestamp:0,...} DataSender 执行完毕 DataSender 开始执行 (时间: 1) DataSender: 使用已有连接 ID1DataSender: 【FlexSim已发出】{timestamp:1,...} DataSender 执行完毕 **Python 服务器**应显示[2026-08-03 14:10:36.495] FlexSim 已连接: (127.0.0.1, 52742)[0.000]robot1: 空闲 | 任务: 无任务robot2: 空闲 | 任务: 无任务buff1: 0AGV1: 坐标(102.00, 18.00, 0.00) | 载货: 0AGV2: 坐标(102.00, 15.00, 0.00) | 载货: 0--------------------------------------------------[1.000]...### 4.3 数据说明FlexSim 发送的 JSON 数据包含以下字段| 字段 | 类型 | 说明 ||------|------|------|| timestamp | float | 当前仿真时间 || robot1_state / robot2_state | int | 机械臂状态码1空闲, 2处理中, 3忙碌, 16空载偏移行进等 || robot1_task / robot2_task | string | 机械臂当前任务描述 || buff1_content | int | 暂存区 1 当前存货量 || agv1_load / agv2_load | int | AGV 当前载货数量 || agv1_x/y/z / agv2_x/y/z | float | AGV 三维坐标 |## 五、常见问题排查| 问题现象 | 可能原因 | 解决方法 ||----------|----------|----------|| Python 显示“FlexSim 已连接”后立即断开 | FlexSim 端 clientconnect 重复调用导致连接重置 | 检查 DataSender 中是否只在新连接时调用 clientconnect已连接时不再重复调用 || FlexSim 只发送了 timestamp0 的数据 | 用户事件的“重复事件”未勾选 | 在 DataSender 属性中**勾选“重复事件”**并设置重复时间为 1.0 || Python 收不到数据但 FlexSim 显示发送成功 | 防火墙拦截或 IP/端口不匹配 | 检查防火墙设置确认 Python 和 FlexSim 使用相同的端口号 || DataSender 显示“连接服务器失败” | Python 服务器未启动 | 先启动 Python 服务器再运行 FlexSim || 对象找不到存在: 0 | 对象名称不匹配 | 检查模型树中的对象名称确保与 node(MODEL:/对象名, model()) 完全一致 || 重置后无法重新连接 | 连接未正确关闭 | 确保 OnReset 中调用了 clientclose(id) 并重置 client 节点为 0 |## 六、架构优势说明本方案采用 **Python 作为服务器、FlexSim 作为客户端** 的架构相比 FlexSim 作为服务器有以下优势1. **避开 socketinit() 问题**FlexSim 作为服务器时需要调用 socketinit() 初始化 Windows Socket 后台进程某些环境下可能因权限或系统配置失败。作为客户端时只需 clientcreate() clientconnect()更加稳定可靠。2. **连接管理更简单**客户端负责主动连接断线后可自动重连。3. **Python 生态优势**Python 作为服务器可以方便地扩展数据存储、可视化、机器学习等高级功能。4. **调试更方便**Python 服务器可以独立运行和测试无需依赖 FlexSim 环境。