做协议仿真，最大的挑战不是"能连上"，而是

很多项目的 Mock 只是返回固定 JSON，但工业协议有状态、有会话、有认证、有实时数据流。ProtoForge 的设计核心是：

完整模拟协议交互过程，而不只是数据值。

┌─────────────────────────────────────────┐
│              Vue3 Frontend              │
│   (Naive UI / 场景编辑器 / 调试日志)      │
└──────────────┬──────────────────────────┘
               │ HTTP / WebSocket
┌──────────────▼──────────────────────────┐
│           FastAPI Backend               │
│  ┌─────────┐ ┌─────────┐ ┌──────────┐ │
│  │  REST   │ │ WebSocket│ │  gRPC    │ │
│  │  API    │ │  Logs   │ │  Server  │ │
│  └────┬────┘ └────┬────┘ └────┬─────┘ │
│       └─────────────┴─────────────┘     │
│  ┌─────────┐ ┌─────────┐ ┌──────────┐ │
│  │  Auth   │ │ Engine  │ │ Scenario │ │
│  │(JWT/RBAC)│ │(Device) │ │  Rules   │ │
│  └─────────┘ └────┬────┘ └──────────┘ │
│                   │                     │
│  ┌────────────────┼────────────────┐   │
│  │    Protocol Layer (17 protocols) │   │
│  │  Modbus TCP/RTU · OPC-UA · MQTT  │   │
│  │  S7 · MC · FINS · AB · GB28181   │   │
│  │  BACnet · FOCAS · MTConnect ...  │   │
│  └──────────────────────────────────┘   │
└─────────────────────────────────────────┘

协议层设计：每个协议都是独立服务

# protoforge/protocols/base.py
class ProtocolServer(ABC):
    @abstractmethod
    async def start(self, config: ProtocolConfig): ...

    @abstractmethod
    async def stop(self): ...

    @abstractmethod
    async def create_device(self, device: Device) -> DeviceInstance: ...

    @abstractmethod
    async def read_points(self, device_id: str) -> List[Point]: ...

    @abstractmethod
    async def write_point(self, device_id: str, point: Point, value: Any): ...

 调度。新增协议只需实现 5 个方法。

GB28181 是最复杂的协议之一，涉及 SIP 信令 + RTP 媒体流。

# 简化的 RTP 会话管理
class GB28181Device:
    async def on_invite(self, request: SIPRequest):
        # 解析 SDP，获取目标 IP:Port
        sdp = SDP.parse(request.body)
        # 创建 RTP 会话
        self.rtp_session = RTPSession(
            local_port=0,  # 系统自动分配
            remote_addr=(sdp.connection_ip, sdp.media_port)
        )
        # 启动 PS 流（H.264 over RTP/PS）
        await self.rtp_session.start_stream(self.video_generator)
        # 返回 200 OK + SDP
        return Response(200, body=self.generate_answer_sdp())

：不是读取文件，而是实时生成 H.264 NALU 单元，时间戳连续，符合 RTP 规范。可以用 VLC/FFmpeg 直接播放验证。

class RuleEngine:
    async def evaluate(self, event: Event) -> List[Action]:
        for rule in self.rules:
            if rule.type == "threshold":
                if self.check_threshold(rule, event):
                    return self.execute_actions(rule.actions)
            elif rule.type == "value_change":
                if self.check_delta(rule, event):
                    return self.execute_actions(rule.actions)
            elif rule.type == "timer":
                if self.check_interval(rule):
                    return self.execute_actions(rule.actions)
            elif rule.type == "script":
                # 安全沙箱执行
                if self.safe_eval(rule.expression, event.context):
                    return self.execute_actions(rule.actions)

 模块解析表达式，只允许白名单操作符和函数，防止代码注入。

Device (ProtocolServer)
  ↓ 定时读取/事件触发
Engine (数据调度)
  ↓ WebSocket 广播
Frontend (实时图表)
  ↓ 用户配置规则
Scenario (规则引擎)
  ↓ 条件满足
Action (set/toggle/increment/decrement)
  ↓ 反向写入
Device (更新状态)

整个链路延迟 < 100ms（本地测试），WebSocket 推送保证前端实时性。

# protoforge/core/testing.py
ASSERTIONS = {
    "equals": lambda actual, expected: actual == expected,
    "contains": lambda actual, expected: expected in actual,
    "regex_match": lambda actual, expected: re.match(expected, actual),
    "json_path": lambda actual, expected: jsonpath(actual, expected["json_path"]) == expected["expected"],
    "status_code": lambda response, expected: response.status_code == expected,
    # ... 共 14 种
}

扩展：新增一个协议需要多少代码？

# protoforge/protocols/custom_tcp.py
class CustomTCPServer(ProtocolServer):
    async def start(self, config):
        self.server = await asyncio.start_server(
            self.handle_client, 
            host=config.host, 
            port=config.port
        )

    async def handle_client(self, reader, writer):
        while True:
            data = await reader.read(1024)
            if not data:
                break
            response = self.parse_and_respond(data)
            writer.write(response)
            await writer.drain()

    async def create_device(self, device):
        return CustomDevice(device.config)

    async def read_points(self, device_id):
        return self.devices[device_id].get_points()

即可实现一个基础协议仿真。框架处理了设备生命周期、WebSocket 日志、API 暴露、场景联动等通用逻辑。

如果你在做 IoT 平台开发、工业网关、数据采集系统，需要一个

欢迎 Star、Issue、PR。有协议需求可以提 Feature Request。

很多项目的 Mock 只是返回固定 JSON，但工业协议有状态、有会话、有认证、有实时数据流。ProtoForge 的设计核心是：完整模拟协议交互过程，而不只是数据值。

ProtoForge 架构设计：如何用 Python 模拟 17 种工业协议

后端

物联网

架构设计

ProtoForge是一款工业协议仿真工具，突破传统Mock局限，真实模拟17种协议交互过程。支持Modbus、OPC-UA、MQTT等协议，实现状态会话、认证、实时数据流等完整交互。采用Vue3+FastAPI架构，提供可视化编排、安全沙箱、14种断言测试，单机支持500+并发连接。适用于IoT平台、工业网关等测试场景，80行代码即可扩展新协议。

数据流

Python

GitHub

JSON

MQTT

2026采购季 | AI焕新·智启新局

iotexplorer

iot-class

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云AI代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

功能1上新10个字符

功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符。

功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符。

功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符

功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符

功能4上新

文章&问答评论现已支持表情

全新交互，全新视觉，新增快捷键、悬浮工具栏、高亮块等功能并同时优化现有功能，全面提升创作效率和体验

社区富文本编辑器全新改版！诚邀体验～ 

精选全网热门MCP server，让你的AI更好用 🚀

💥开发者 MCP广场重磅上线！

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

ProtoForge是一款工业协议仿真工具，突破传统Mock局限，真实模拟17种协议交互过程。支持Modbus、OPC-UA、MQTT等协议，实现状态会话、认证、实时数据流等完整交互。采用Vue3+FastAPI架构，提供可视化编排、安全沙箱、14种断言测试，单机支持500+并发连接。适用于IoT平台、工业网关等测试场...

ProtoForge 架构设计：如何用 Python 模拟 17 种工业协议

ProtoForge 架构设计：如何用 Python 模拟 17 种工业协议

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐