简介这份资源面向欧标充电桩开发、测试与协议对接人员整理了OCPP1.6标准中全部消息事件的JSON格式定义可用于模拟桩端与后台之间的完整业务流程。包内共78个json文件压缩包约28KB每个文件对应一条消息的request或response结构并标注必要字段与可选字段涵盖BootNotification、Authorize、StartTransaction、MeterValues、RemoteStartTransaction、SetChargingProfile、FirmwareStatusNotification等核心交互。OCPP采用一问一答模式读者可参照这些JSON包构造请求与响应快速验证充电桩业务逻辑、排查字段缺失或格式错误。目前已有5462人学习下载适合需要落地OCPP1.6通信、搭建模拟测试环境或进行协议联调的开发者参考。1. 欧标充电桩 OCPP1.6 消息事件 JSON 格式从一份报文样本到能跑通的桩端联调第一次拿到欧标充电桩的联调日志时我盯着满屏的 JSON 有点懵——BootNotification、Heartbeat、StatusNotification、StartTransaction这些消息像流水一样刷过去但真要把它们和桩端代码里的状态机对上才发现每条报文的字段含义、触发时机、uniqueId怎么回填全是坑。这份资源就是围绕 OCPP1.6 的消息事件 JSON 格式展开的把充电桩和中央系统CSMS之间那套基于 WebSocket 的 RPC 交互拆成了可读、可改、可复现的报文样本和解析逻辑。它适合正在做欧标桩端固件、后台 CSMS 对接或者想用 Linux 环境跑一个 OCPP 模拟器的开发者。不管你是用 Python 写测试桩还是用 C 写嵌入式端只要涉及 OCPP1.6 的 JSON 报文这份东西都能当字典翻。2. 协议底座拆解OCPP1.6 的 RPC 帧结构与 JSON 字段语义OCPP1.6 本质上是「WebSocket 之上跑 JSON-RPC」但它的 RPC 不是标准 JSON-RPC 2.0而是自己定义了一套四元组帧格式。很多人第一次抓包会疑惑为什么消息数组里第一个元素是数字 2、3、4这就是 OCPP 的消息类型标识搞不清这个后面所有解析都无从谈起。2.1 四种消息类型CALL、CALLRESULT、CALLERROR、CALLRESULTERROROCPP1.6 的每一条 WebSocket 文本帧都是一个 JSON 数组数组第一个元素是消息类型 ID类型 ID名称方向结构2CALL双向[2, uniqueId, action, payload]3CALLRESULT双向[3, uniqueId, payload]4CALLERROR双向[4, uniqueId, errorCode, errorDescription, errorDetails]5CALLRESULTERROR双向保留实际很少用这里最容易翻车的是uniqueId。CALL 发出时由发起方生成CALLRESULT 和 CALLERROR 必须原样带回同一个uniqueId否则对方无法匹配请求。我见过有开发者用自增整数做uniqueId结果桩端重启后从 0 重新计数和 CSMS 那边还没超时的旧请求撞了 ID导致响应被错误匹配。常见做法是用 UUID 或者「时间戳随机数」来保证全局唯一。// CALL 示例桩端向 CSMS 发起 BootNotification [2, a1b2c3d4-0001, BootNotification, { chargePointModel: AC7K-01, chargePointVendor: VendorX, chargePointSerialNumber: SN20240101, firmwareVersion: 1.2.3, iccid: 89860012345678901234, imsi: 460001234567890, meterSerialNumber: MTR-0001, meterType: AC }]上面这段是桩端上电后发的第一条业务消息。chargePointModel和chargePointVendor是必填字段CSMS 通常用这两个字段做设备白名单校验。iccid和imsi是可选字段但在欧标场景下如果桩端带 4G 模块建议填上方便后台做 SIM 卡绑定和流量统计。meterType填AC或DC对应交流桩和直流桩后台的电价模型和计费逻辑会依赖这个字段。// CALLRESULT 示例CSMS 对 BootNotification 的响应 [3, a1b2c3d4-0001, { status: Accepted, currentTime: 2024-01-01T12:00:00Z, interval: 300 }]status有三个取值Accepted、Pending、Rejected。如果返回Pending桩端需要等待 CSMS 后续下发BootNotification的确认或者按interval秒后重试。interval是心跳间隔单位秒桩端收到后要把本地的Heartbeat定时器改成这个值。注意currentTime是 ISO8601 格式的 UTC 时间桩端应该用它来校准本地时钟而不是自己随便设。2.2 核心事件消息的 JSON 字段与触发时机OCPP1.6 的消息按功能分几大类引导与注册、心跳与状态、事务与计量、远程控制、固件与诊断。这份资源里覆盖了最常用的十几条我挑几条联调时出现频率最高的说。Heartbeat是最简单的payload 是空对象{}CSMS 返回currentTime。但别小看它心跳间隔设太短会浪费流量设太长会导致 CSMS 判定桩端离线。一般交流桩设 300 秒直流桩设 60 到 120 秒具体看后台配置。StatusNotification是桩端上报连接器状态的消息字段比较多[2, a1b2c3d4-0002, StatusNotification, { connectorId: 1, errorCode: NoError, status: Available, timestamp: 2024-01-01T12:00:05Z, info: , vendorId: , vendorErrorCode: }]connectorId为 0 表示桩体本身1 及以上表示具体的充电枪。status的取值有Available、Preparing、Charging、SuspendedEVSE、SuspendedEV、Finishing、Reserved、Unavailable、Faulted。联调时最常见的坑是桩端插枪后先发Preparing然后等 CSMS 下发RemoteStartTransaction才发Charging但有些后台逻辑是收到Preparing就自动授权两边状态机对不上就会卡住。解决方法是先和后台确认状态流转图再在桩端代码里把状态迁移条件写死。StartTransaction是事务开始的标志[2, a1b2c3d4-0003, StartTransaction, { connectorId: 1, idTag: RFID-0001, meterStart: 0, reservationId: 0, timestamp: 2024-01-01T12:01:00Z }]meterStart是电表起始读数单位 Wh。reservationId如果没有预约就填 0。CSMS 返回transactionId和idTagInfo桩端拿到transactionId后要存下来后续StopTransaction和MeterValues都要带上。[3, a1b2c3d4-0003, { transactionId: 10001, idTagInfo: { status: Accepted, expiryDate: 2024-12-31T23:59:59Z, parentIdTag: } }]idTagInfo.status如果是Blocked或Expired桩端应该拒绝启动充电并给出提示。expiryDate是卡片的有效期桩端可以缓存这个信息减少对后台的查询。2.3 WebSocket 子协议与连接保活OCPP1.6 规定 WebSocket 连接必须使用子协议ocpp1.6URL 路径通常是ws://csms-host:port/chargePointIdentity。chargePointIdentity就是桩的唯一标识一般用序列号。如果子协议不对CSMS 会直接拒绝握手返回 400 或 426。连接建立后除了 OCPP 层面的HeartbeatWebSocket 层面也有 Ping/Pong 帧。有些 CSMS 实现依赖 WebSocket Ping 来判断链路存活有些则只看 OCPPHeartbeat。我一般会在桩端同时处理两者收到 Ping 自动回 Pong大多数 WebSocket 库内置同时按interval发Heartbeat。如果连续两次Heartbeat没收到响应就主动断开重连避免「假在线」——链路还在但后台已经认为你离线了。3. 桩端跑通实战用 Python 模拟器验证 JSON 报文收发光看报文格式不够得跑起来才能验证字段对不对、时序对不对。这一章用一个 Python 写的 OCPP1.6 桩端模拟器把上面那些消息串成完整流程。选 Python 是因为websockets库足够简单改起来快适合联调阶段快速验证。3.1 环境准备与依赖安装Linux 环境下需要 Python 3.8 以上依赖只有websockets和asyncio标准库自带。我一般用虚拟环境隔离python3 -m venv ocpp-env source ocpp-env/bin/activate pip install websockets如果是在嵌入式 Linux 上跑比如某款 ARM 开发板websockets也能装但要注意 Python 版本别太低。有些老系统自带 Python 3.6websockets新版本不支持得降级到 9.x。我踩过一次坑在板子上装了最新版websockets结果ssl模块编译不过折腾了半天才发现是 OpenSSL 版本太老。后来统一用pip install websockets9.1就稳了。3.2 建立 WebSocket 连接并发送 BootNotification先写连接和注册部分。核心是构造符合 OCPP1.6 的 CALL 帧发出去后等 CALLRESULT解析status和interval。import asyncio import json import uuid import websockets class OCPPChargePoint: def __init__(self, csms_url, cp_id): self.csms_url csms_url # 例如 ws://127.0.0.1:9000/CP001 self.cp_id cp_id self.ws None self.heartbeat_interval 300 # 默认心跳间隔秒 self.pending {} # uniqueId - Future用于匹配响应 def _make_call(self, action, payload): 构造 CALL 帧返回 uniqueId 和 JSON 字符串 uid str(uuid.uuid4()) frame [2, uid, action, payload] return uid, json.dumps(frame) async def _send_call(self, action, payload): 发送 CALL 并等待 CALLRESULT uid, text self._make_call(action, payload) future asyncio.get_event_loop().create_future() self.pending[uid] future await self.ws.send(text) # 设置超时避免永久阻塞 return await asyncio.wait_for(future, timeout30) async def boot(self): payload { chargePointModel: AC7K-01, chargePointVendor: VendorX, chargePointSerialNumber: self.cp_id, firmwareVersion: 1.2.3 } result await self._send_call(BootNotification, payload) if result[status] Accepted: self.heartbeat_interval result.get(interval, 300) print(f[BOOT] Accepted, heartbeat{self.heartbeat_interval}s) else: print(f[BOOT] Rejected: {result}) return result_make_call用uuid4生成uniqueId保证全局唯一。_send_call把uniqueId和一个Future绑定发出去后等响应。asyncio.wait_for设 30 秒超时防止 CSMS 不响应导致程序卡死。boot方法里如果status是Accepted就把interval存下来后续心跳定时器用这个值。3.3 心跳与状态上报的定时器实现心跳和状态上报需要定时触发用asyncio.create_task起两个协程一个发Heartbeat一个发StatusNotification。async def heartbeat_loop(self): 按 heartbeat_interval 周期发送 Heartbeat while True: await asyncio.sleep(self.heartbeat_interval) try: result await self._send_call(Heartbeat, {}) # 用 CSMS 返回的时间校准本地时钟这里只打印 print(f[HEARTBEAT] server time: {result[currentTime]}) except Exception as e: print(f[HEARTBEAT] failed: {e}) break # 心跳失败退出循环触发重连 async def status_notification(self, connector_id, status, error_codeNoError): 上报连接器状态 payload { connectorId: connector_id, errorCode: error_code, status: status, timestamp: self._now_iso() } return await self._send_call(StatusNotification, payload) def _now_iso(self): 返回 ISO8601 UTC 时间字符串 from datetime import datetime, timezone return datetime.now(timezone.utc).strftime(%Y-%m-%dT%H:%M:%SZ)heartbeat_loop里如果_send_call抛异常说明连接可能断了直接break退出循环外层重连逻辑会重新建立连接。status_notification封装了状态上报connectorId为 0 表示桩体1 表示枪口。_now_iso生成的时间格式必须带Z后缀表示 UTC有些 CSMS 对时间格式校验很严少个Z就报FormationViolation。3.4 接收 CSMS 下发的 CALL 并回 CALLRESULT桩端不只是发消息还要接收 CSMS 下发的指令比如RemoteStartTransaction、Reset、UnlockConnector。接收循环里要解析 CALL 帧根据action分发处理然后回 CALLRESULT。async def recv_loop(self): 接收 CSMS 消息并处理 async for message in self.ws: try: frame json.loads(message) except json.JSONDecodeError: continue msg_type frame[0] if msg_type 2: # CALL需要回 CALLRESULT uid, action, payload frame[1], frame[2], frame[3] print(f[RECV] {action}: {payload}) resp await self._handle_action(action, payload) reply [3, uid, resp] await self.ws.send(json.dumps(reply)) elif msg_type 3: # CALLRESULT匹配 pending uid, payload frame[1], frame[2] future self.pending.pop(uid, None) if future and not future.done(): future.set_result(payload) elif msg_type 4: # CALLERROR uid frame[1] future self.pending.pop(uid, None) if future and not future.done(): future.set_exception(Exception(fCALLERROR: {frame[2]})) async def _handle_action(self, action, payload): 根据 action 返回对应的响应 payload if action RemoteStartTransaction: # 模拟接受远程启动返回 Accepted return {status: Accepted} elif action Reset: # 模拟接受复位返回 Accepted return {status: Accepted} elif action UnlockConnector: return {status: Unlocked} else: return {}recv_loop里用async for持续读 WebSocket 消息。类型 2 是 CALL解析出action后调用_handle_action生成响应再按[3, uid, resp]格式回过去。类型 3 是 CALLRESULT从pending字典里取出对应的Future并设置结果这样_send_call里的await就能拿到返回值。类型 4 是 CALLERROR把Future设成异常让调用方感知错误。_handle_action是个简化版的分发器实际项目里要根据业务逻辑做更细的处理比如RemoteStartTransaction要检查枪口是否可用、卡片是否有效再决定返回Accepted还是Rejected。3.5 主循环与断线重连把上面这些拼起来加一个外层重连循环async def run(self): while True: try: async with websockets.connect( self.csms_url, subprotocols[ocpp1.6] ) as ws: self.ws ws print(f[CONN] connected to {self.csms_url}) await self.boot() await self.status_notification(0, Available) await self.status_notification(1, Available) # 并发跑心跳和接收循环 await asyncio.gather( self.heartbeat_loop(), self.recv_loop() ) except Exception as e: print(f[CONN] disconnected: {e}, retry in 10s) await asyncio.sleep(10) if __name__ __main__: cp OCPPChargePoint(ws://127.0.0.1:9000/CP001, CP001) asyncio.run(cp.run())websockets.connect里指定subprotocols[ocpp1.6]这是 OCPP1.6 的硬性要求。连接成功后先boot再上报桩体和枪口状态为Available。然后asyncio.gather并发跑心跳和接收循环。如果任何一步抛异常外层except捕获后等 10 秒重连。这个重连间隔可以按指数退避调整避免频繁重连打爆 CSMS。跑起来后如果 CSMS 那边正常你会看到[BOOT] Accepted、[HEARTBEAT] server time: ...这些日志。如果 CSMS 返回Rejected检查chargePointModel和chargePointVendor是否在后台白名单里。如果连接直接被拒检查子协议和 URL 路径。4. 避坑与排查OCPP1.6 联调中最容易翻车的五个点这一章是我和几个做桩端的朋友踩过的坑按「现象 → 原因 → 解决」整理。有些坑不涉及代码但排查起来比代码还费时间。4.1 现象CSMS 一直不回 CALLRESULT桩端超时原因uniqueId格式不对或者 CALL 帧的 JSON 数组元素顺序错了。OCPP1.6 要求 CALL 必须是[2, uniqueId, action, payload]少一个元素或者把action和payload顺序颠倒CSMS 解析失败就直接丢弃不会回 CALLERROR。解决抓 WebSocket 原始帧确认数组长度是 4uniqueId是字符串action是字符串payload是对象。用json.dumps时注意ensure_asciiFalse避免中文被转义后 CSMS 不识别。4.2 现象BootNotification 返回 Rejected但字段都填了原因chargePointSerialNumber和后台注册的序列号不一致或者chargePointVendor大小写不匹配。有些 CSMS 做严格字符串比对VendorX和vendorx会被当成两个不同的厂商。解决和后台确认注册信息逐字段比对。如果后台支持模糊匹配也要确认是否开启了大小写忽略。我一般会在桩端加一个配置项把chargePointVendor和chargePointModel做成可配置的方便现场调试。4.3 现象Heartbeat 正常发但 CSMS 过一会儿就判定离线原因WebSocket 层面的 Ping/Pong 没回。有些 CSMS 用 WebSocket Ping 做链路检测如果桩端的 WebSocket 库没有自动回 PongCSMS 会在几个 Ping 周期后断开连接。解决确认使用的 WebSocket 库是否自动处理 Ping/Pong。websockets库默认会自动回 Pong但如果自己实现了底层 socket就要手动处理。另外检查防火墙是否拦截了 Ping 帧。4.4 现象StartTransaction 返回的 transactionId 是 0 或负数原因CSMS 那边事务创建失败但返回了默认值。常见于idTag无效、connectorId不存在、或者meterStart超出范围。解决先检查idTagInfo.status如果是Invalid或Blocked说明卡片问题。如果idTagInfo正常但transactionId异常联系后台查事务表。桩端应该在拿到transactionId后做校验小于等于 0 就视为失败不要继续发MeterValues。4.5 现象MeterValues 上报后 CSMS 报 FormationViolation原因meterValue数组里的sampledValue字段格式不对。OCPP1.6 要求sampledValue是数组每个元素包含value、measurand、unit等字段value必须是字符串不能是数字。解决检查 JSON 序列化时有没有把value写成数字。Python 里json.dumps({value: 100})会输出{value: 100}但 OCPP 要求{value: 100}。用str()转一下。另外measurand的取值要符合规范比如Energy.Active.Import.Register、Power.Active.Import拼错一个字母就报错。5. 二次开发改造从模拟器到真实桩端固件的迁移思路模拟器跑通之后下一步就是把它移植到真实桩端的嵌入式环境里。这一步不是简单地把 Python 代码翻译成 C而是要考虑资源限制、实时性、异常恢复这些工程问题。我一般会按下面的顺序做改造。5.1 消息层与业务层分离模拟器里_handle_action把业务逻辑和协议处理混在一起真实项目里要拆开。协议层只负责 JSON 的序列化/反序列化和 WebSocket 收发业务层用状态机管理充电流程。比如RemoteStartTransaction到达时协议层解析出idTag和connectorId交给业务层判断是否允许启动业务层返回结果后再由协议层封装 CALLRESULT。// 伪代码协议层与业务层接口 typedef struct { int connector_id; char id_tag[32]; } remote_start_req_t; typedef enum { REMOTE_START_ACCEPTED, REMOTE_START_REJECTED, REMOTE_START_OCCUPIED } remote_start_resp_t; // 业务层实现 remote_start_resp_t business_handle_remote_start(remote_start_req_t *req) { if (is_connector_occupied(req-connector_id)) { return REMOTE_START_OCCUPIED; } if (!is_id_tag_valid(req-id_tag)) { return REMOTE_START_REJECTED; } start_charging(req-connector_id, req-id_tag); return REMOTE_START_ACCEPTED; }这样拆分后协议层可以独立测试业务层也可以在没有 CSMS 的情况下用 mock 数据验证。移植到 C 的时候JSON 解析建议用cJSON或janssonWebSocket 用libwebsockets。注意libwebsockets的回调模型和 Python 的async不一样需要把状态机改成事件驱动。5.2 持久化 transactionId 和配置项真实桩端会断电重启transactionId、heartbeat_interval、csms_url这些不能只放内存。我一般用 SQLite 或者简单的键值文件存。transactionId在充电过程中必须持久化否则重启后StopTransaction带不上正确的 ID后台会认为事务异常。-- 用 SQLite 存事务信息 CREATE TABLE transaction ( transaction_id INTEGER PRIMARY KEY, connector_id INTEGER NOT NULL, id_tag TEXT NOT NULL, meter_start INTEGER NOT NULL, start_time TEXT NOT NULL, status TEXT DEFAULT active );充电开始时插入一条记录StopTransaction成功后把status改成finished。如果重启后发现还有active的事务说明上次异常断电需要补发StopTransaction或者走后台的异常事务处理流程。5.3 用真实报文样本做回归测试改造过程中最容易引入的 bug 是 JSON 字段拼写错误或者类型错误。我习惯把联调时抓到的真实报文存成文件每次改完代码跑一遍回归测试对比输出和预期。比如把BootNotification的请求和响应各存一个.json文件测试脚本读入请求调用桩端处理函数把输出和响应文件做 diff。import json def test_boot_notification(): with open(samples/boot_req.json) as f: req json.load(f) with open(samples/boot_resp.json) as f: expected json.load(f) # 调用桩端处理逻辑得到 actual actual handle_boot_notification(req[3]) assert actual expected[2], fmismatch: {actual} ! {expected[2]}这个习惯帮我省了很多时间。有一次改StatusNotification的timestamp格式从%Y-%m-%dT%H:%M:%S改成带毫秒的%Y-%m-%dT%H:%M:%S.%fZ回归测试直接报错因为预期文件里没有毫秒。如果没有这个测试可能要到现场联调才发现后台不认。5.4 日志与抓包把 WebSocket 帧落到文件真实环境出问题时光看应用日志不够得看 WebSocket 原始帧。我一般在桩端加一个可开关的抓包功能把收发的每一帧按时间戳写到文件里格式就是[方向] [时间] [帧内容]。出问题时把文件拉回来用脚本解析成可读的 CALL/CALLRESULT 序列比在设备上调试方便得多。# 抓包日志示例 # 2024-01-01T12:00:00Z [2,uid1,BootNotification,{...}] # 2024-01-01T12:00:01Z [3,uid1,{status:Accepted,...}]解析脚本可以按uniqueId把请求和响应配对算出每条消息的往返延迟。如果某条 CALLRESULT 延迟超过 5 秒就要查 CSMS 那边的处理逻辑或者网络质量。从那以后我每次改 OCPP 报文相关的代码都强制走一遍「样本回归 抓包对比」哪怕只改了一个字段名。这个习惯让我在两次现场联调里提前发现了字段拼写和类型问题省了至少两天的排查时间。希望帮到你。本文还有配套的精品资源点击获取
