当前位置： 首页 > news >正文

FlexSim 与 Python Socket 通信完整指南

news 2026/8/4 15:57:11

# 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}]", flush=True) print(f" robot1: {r1_desc} | 任务: {r1_task}", flush=True) print(f" robot2: {r2_desc} | 任务: {r2_task}", flush=True) print(f" buff1: {b1}", flush=True) print(f" AGV1: 坐标({agv1_x:.2f}, {agv1_y:.2f}, {agv1_z:.2f}) | 载货: {agv1_load}", flush=True) print(f" AGV2: 坐标({agv2_x:.2f}, {agv2_y:.2f}, {agv2_z:.2f}) | 载货: {agv2_load}", flush=True) print("-" * 50, flush=True) 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='', encoding='utf-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, fieldnames=fieldnames) 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("按 Ctrl+C 停止") print("=" * 70) # 初始化 CSV 文件（每次启动覆盖旧数据） try: with open(CSV_FILENAME, 'w', newline='', encoding='utf-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, fieldnames=fieldnames) 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', encoding='utf-8', buffering=1) 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 服务器**：
```bash
python 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: 使用已有连接 ID=1
DataSender: 【FlexSim已发出】{"timestamp":1,...}
========== DataSender 执行完毕 ==========
```

**Python 服务器**应显示：
```
[2026-08-03 14:10:36.495] FlexSim 已连接: ('127.0.0.1', 52742)
[0.000]
robot1: 空闲 | 任务: 无任务
robot2: 空闲 | 任务: 无任务
buff1: 0
AGV1: 坐标(102.00, 18.00, 0.00) | 载货: 0
AGV2: 坐标(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 只发送了 timestamp=0 的数据 | 用户事件的“重复事件”未勾选 | 在 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 环境。

http://www.jsqmd.com/news/1327409/

相关文章：

  • 零基础AI应用编程开发入门 | 吴恩达Prompt工程极简通关指南：新手从零学会工业级提示词开发（可直接复用代码）
  • 输电线路缺陷数据集 电缆缺陷识别 输电线破损数据集
  • 武汉城铁技工学校电话号码多少？ - 升学择校早知道
  • UnityWebRequest SSL证书验证：自定义CertificateHandler实现安全绕过与指纹钉扎
  • Windows系统CertPolEng.dll丢失的解决方案与预防措施
  • 宫老师助力廊坊十八酒坊客户私享会 赋能酒水商户数字化经营升级
  • 5分钟实战：从OFD到PDF的高效转换全解
  • 新手申请香港优才不想费心，深圳一站式服务商从哪些维度筛选？ - 滚动商讯
  • 2026年Q3膳食营养分析与配餐实训系统厂家观察：从商科实训到垂直细分，谁在引领院校数智化教学新浪潮——西安青软领衔五大实训厂商盘点，膳食营养配餐系统成新商科建设热门赛道 - 深度智识库
  • 从初稿到终稿一步到位！Gradpaper不愧是论文顶配工具
  • 企业级数据中台开源软件：数智云的数据治理与服务化架构
  • 山西GEO优化哪家服务好？优选山西中航云创科技有限公司 - GrowUME
  • Clockwork for Dynamo：450+节点打造BIM参数化设计的终极工具箱
  • 2026年8月蛋品企业全自动鸡蛋品质哈夫值测定仪实测：四大品牌选购指南 - 云唐专业仪器测评
  • 终极指南：如何用Hotkey Detective快速检测Windows热键冲突
  • 2026德阳卫生间、外墙、楼顶、地下室、阳台阳光房渗漏不用愁！3家正规靠谱防水服务商甄选：选对专业团队，告别反复渗水，长效售后有保障 - 吉林同城获客
  • 2026届毕业生推荐的六大AI辅助论文工具横评
  • 家政管理APP开发分析（功能、难点与源码自研差异）
  • Django构建汽车检测站管理系统的技术实践
  • TV Bro：为智能电视设计的终极遥控器浏览器解决方案
  • GPU越多，算力就越强吗？很多企业忽略了真正的瓶颈
  • 厂房夏季高温怎么解决？一套工业屋面隔热降温技术方案的完整拆解
  • 如何构建高性能物联网边缘网关：Linux多线程架构的完整实现
  • 武汉南华光电职业技术学校招生老师电话微信_在线报名咨询入口 - 武汉中职最新信息发布
  • 货车路线规划合规接入：限高限重限行全场景处理 - 资讯综合
  • SpringBoot+Vue旅游预算线路推荐系统开发实践
  • 从零构建高可用IM系统：核心技术拆解与工程实践指南
  • 贵州工业地板怎么选？从产品性能、供应链稳定性到交付标准，看科洛弗如何解决工程项目的地材痛点 - 中国华商产业观察网
  • FNF QT重制版“Blissful Erect”更新：从开源游戏框架到模组开发实战
  • ABPP探针合成与活性验证：从结构设计到质控全攻略（二）