OCPP1.6 BootNotification异常与RemoteStartTransaction方案咨询

DevOps 小满

2026-8-30

OCPP 1.6充电桩中央系统对接问题排查与实现方案

基础上下文

核心服务：基于Python OCPP 1.6库实现的CentralSystem.py，通过WebSocket对外提供充电桩接入服务
接入规则：充电桩连接地址格式为 ws://HostName:Port/充电桩序列号，服务端监听9000端口，使用ocpp1.6子协议
现有异常日志

Charge point /D5200372001000010101 connected
INFO:ocpp:D5200372001000010101: receive message [2,"530","Heartbeat",{}]
Heartbeat
INFO:ocpp:D5200372001000010101: send [3,"530",{"currentTime":"2022-06-10T10:43:26Z"}]
INFO:ocpp:D5200372001000010101: receive message [2,"531","Heartbeat",{}]
Heartbeat

待解决问题
- 问题1：充电桩上线后每2分钟正常上报心跳，但无车辆连接时始终不上报BootNotification、StatusNotification报文，仅插枪后才会上报StatusNotification，需排查根因与解决方法
- 问题2：Flutter端需实现远程启动充电（Remote Start Transaction）功能，确认移动端是否需要直连充电桩使用的WebSocket端点调用接口，求最优实现方案
现有核心实现代码

import asyncio
import logging
from asyncio import constants

import websockets
from datetime import datetime, timezone
from ocpp.routing import on
from ocpp.v16 import ChargePoint as cp
from ocpp.v16.enums import Action, RegistrationStatus, RemoteStartStopStatus
import ocpp.v16.enums as enums
from ocpp.v16 import call_result, call
import requests

logging.basicConfig(level=logging.INFO)


class ChargePoint(cp):
  chargingProfile = {}

  @on(Action.BootNotification)
  async def on_boot_notification(self, charge_point_vendor, charge_point_model, **kwargs):
    print("Boot Notification")
    return call_result.BootNotificationPayload(
      current_time=datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S') + "Z",
      interval=100,
      status=RegistrationStatus.accepted
    )

  @on(Action.StatusNotification)
  async def on_status_notification(self, connector_id, error_code, status, **kwargs):
    print("Status Notification")
    return call_result.StatusNotificationPayload()

  @on(Action.Heartbeat)
  async def on_heartbeat(self):
    print("Heartbeat")
    return call_result.HeartbeatPayload(
      current_time=datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S') + "Z"
    )

  @on(Action.Authorize)
  async def on_authorize(self, id_tag):
    print("Authorize")
    query = {'access_token': 'masterKey'}

    response = requests.get('http://内部业务服务地址/chargersOnboarding/',
                            params=query)
    data = response.json()
    isFound = "false"
    status = ""

    for item in data['rows']:
      if item['RFID'] == id_tag:
        isFound = "true"
        if item['status'] == 0:
          status = "Accepted"
        else:
          status = "Expired"
        break
      else:
        continue
    if isFound == "true":
      print("Authorized")
      return call_result.AuthorizePayload(
        id_tag_info={
          "expiryDate": "2023-05-19T13:00:00.21Z",
          "parentIdTag": id_tag,
          "status": status
        },
      )
    else:
      print("Not Authorized")
      return call_result.AuthorizePayload(
        id_tag_info={
          "expiryDate": "",
          "parentIdTag": id_tag,
          "status": "Invalid"
        },
      )

  @on(Action.StartTransaction)
  async def on_start_transaction(self, connector_id, id_tag, meter_start, timestamp, **kwargs):
    print("START TRANSACTION COMING FROM CHARGER")
    query = {'access_token': 'masterKey'}

    response = requests.get('http://内部业务服务地址/chargersOnboarding/',
                            params=query)
    data = response.json()
    isFound = "false"
    status = ""

    for item in data['rows']:
      if item['RFID'] == id_tag:
        isFound = "true"
        if item['status'] == 0:
          status = "Accepted"
        else:
          status = "Expired"
        break
      else:
        continue
    if isFound == "true":
      return call_result.StartTransactionPayload(
        id_tag_info={
          "expiryDate": "2023-05-19T13:00:00.21Z",
          "parentIdTag": id_tag,
          "status": status
        },
        transaction_id=int(1)
      )
    else:
      print("Not Authorized")
      return call_result.StartTransactionPayload(
        id_tag_info={
          "expiryDate": "",
          "parentIdTag": id_tag,
          "status": "Invalid"
        },
        transaction_id=int(1)
      )

  @on(Action.StopTransaction)
  async def on_stop_transaction(self, transaction_id, timestamp, meter_stop, id_tag, **kwargs):
    query = {'access_token': 'masterKey'}
    response = requests.get('http://内部业务服务地址/chargersOnboarding/',
                            params=query)
    data = response.json()
    isFound = "false"
    status = ""

    for item in data['rows']:
      if item['RFID'] == id_tag:
        isFound = "true"
        if item['status'] == 0:
          status = "Accepted"
        else:
          status = "Expired"
        break
      else:
        continue
    if isFound == "true":
      return call_result.StopTransactionPayload(
        id_tag_info={
          "expiryDate": "2023-05-19T13:00:00.21Z",
          "parentIdTag": id_tag,
          "status": status
        },
      )
    else:
      print("Not Authorized")
      return call_result.StopTransactionPayload(
        id_tag_info={
          "expiryDate": "",
          "parentIdTag": id_tag,
          "status": "Invalid"
        },
        transaction_id=int(1)
      )

  @on(Action.MeterValues)
  async def on_meter_value(self, **kwargs):
    return call_result.MeterValuesPayload()

  @on(Action.DataTransfer)
  async def on_data_transfer(self, vendor_id, message_id, data):
    return call_result.DataTransferPayload(
      status='Accepted'
    )

  @on(Action.ChangeAvailability)
  async def on_change_availabilty(self, connector_id, type):
    return call_result.ChangeAvailabilityPayload(
      status='Accepted'
    )

  async def send_limitation(self, limit):
    response = await self.call(call.SetChargingProfilePayload(
      connector_id=0,
      cs_charging_profiles={
        'chargingProfileId': 1,
        'stackLevel': 0,
        'chargingProfilePurpose': enums.ChargingProfilePurposeType.chargepointmaxprofile,
        'chargingProfileKind': enums.ChargingProfileKindType.absolute,
        'chargingSchedule': {
          'startSchedule': datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S') + "Z",
          'chargingRateUnit': enums.ChargingRateUnitType.amps,
          'chargingSchedulePeriod': [{
            'startPeriod': 0,
            'limit': limit
          }]
        }
      }
    ))
    print("SEND Limitation")
    print(response)

  async def remote_start_transaction(self, id_tag, connector_id, limit=8.0):
    obj = {
      'chargingProfileId': 1,
      'stackLevel': 0,
      'chargingProfilePurpose': enums.ChargingProfilePurposeType.chargepointmaxprofile,
      'chargingProfileKind': enums.ChargingProfileKindType.absolute,
      'chargingSchedule': {
        'startSchedule': datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S') + "Z",
        'chargingRateUnit': enums.ChargingRateUnitType.amps,
        'chargingSchedulePeriod': [{
          'startPeriod': 0,
          'limit': limit
        }]
      },

    }
    print("REMOTE START!!!")
    request = call.RemoteStartTransactionPayload(
      id_tag=id_tag,
      charging_profile=obj,
      connector_id=connector_id
    )
    response = await self.call(request)
    print(response)
    if response.status == RemoteStartStopStatus.accepted:
      print("Transaction Started!!!")
    else:
      print("Transaction Failed to Start!!!")
      print(response.status)
    return response

  async def remote_stop_transaction(self, transaction_id):
    print("REMOTE STOP!!!")
    request = call.RemoteStopTransactionPayload(
      transaction_id=transaction_id
    )
    response = await self.call(request)

    if response.status == RemoteStartStopStatus.accepted:
      print("Stopping transaction")
    return response


# 全局在线充电桩实例存储
online_chargers = {}

async def on_connect(websocket, path):

  charge_point_id = path.strip('/')
  cp_instance = ChargePoint(charge_point_id, websocket)
  online_chargers[charge_point_id] = cp_instance
  try:
    print(f'Charge point {path} connected')
    await asyncio.gather(cp_instance.start())
  except websockets.exceptions.ConnectionClosed:
    print(f"Charge Point {path} disconnected")
    if charge_point_id in online_chargers:
      del online_chargers[charge_point_id]


async def main():
  server = await websockets.serve(
    on_connect,
    '0.0.0.0',
    9000,
    subprotocols=['ocpp1.6'],
    ping_interval=None,
    ping_timeout=None

  )

  logging.info("Server Started listening to new connections...")
  await server.wait_closed()


if __name__ == '__main__':
  asyncio.run(main())

问题1：初始化报文缺失排查方案

按照OCPP 1.6协议强制规范，充电桩完成WebSocket握手后，第一帧必须上报BootNotification，中央系统返回Accepted状态后，充电桩需要立即上报所有连接器的当前状态（每个连接器对应一帧StatusNotification），完成初始化流程后才会按配置间隔上报心跳。当前能收到心跳但收不到初始化报文，按以下优先级排查：

桩端配置排查
- 确认桩端OCPP版本配置与服务端一致，必须为OCPP 1.6，版本不匹配时部分厂商桩会出现报文逻辑异常，跳过初始化流程
- 检查桩端初始化触发配置，确认开启“连接建立立即上报BootNotification”开关，部分厂商桩存在“仅插枪触发初始化”的错误配置项，直接修改即可
- 核对桩端心跳间隔、BootNotification重试间隔参数，心跳间隔建议配置为30-120秒，BootNotification重试间隔不超过60秒
服务端逻辑排查
- 原代码中除on_remote_start外所有@on装饰的报文回调都是同步方法，Python OCPP库运行在asyncio事件循环中，同步方法会阻塞事件循环，可能导致初始化报文回调不触发，需将所有回调方法统一改为async def异步实现
- 在连接建立后增加主动触发逻辑：桩连接成功后，主动下发TriggerMessage指令，要求桩端立即上报BootNotification和所有连接器的StatusNotification，符合OCPP协议规范
- 检查前置反向代理（如果有）的WebSocket配置，确认Sec-WebSocket-Protocol头正确透传，子协议协商结果为ocpp1.6，子协议协商失败时部分桩会认为OCPP连接未建立，仅发送心跳维持链路
链路验证
- 在服务端9000端口抓包，确认桩端是否真的未发送BootNotification、StatusNotification报文，排除服务端路由逻辑漏处理的问题

问题2：远程启动充电实现方案

结论：移动端绝对不需要直连充电桩使用的OCPP WebSocket端点。OCPP长连接是充电桩与中央系统之间的专属通信链路，仅传输OCPP协议报文，不对终端用户客户端开放。
最优实现采用三层架构：

中央系统层

维护全局在线充电桩实例映射（参考代码中新增的online_chargers字典），存储每个在线充电桩序列号对应的ChargePoint实例
对外提供标准HTTP/HTTPS业务接口（例如/api/charger/{charger_sn}/remote-start），接口层做用户鉴权、参数校验、权限判断
收到业务接口的远程启动请求后，从online_chargers中找到对应桩的实例，调用调整后的remote_start_transaction方法下发OCPP指令，等桩返回结果后，将结果通过HTTP响应返回给调用方
充电过程中的状态变更（启动成功、充电进度、故障、停止）通过消息推送通道主动推送给移动端，无需移动端轮询

移动端层

Flutter端仅需调用中央系统暴露的HTTP业务接口，传入目标充电桩ID、连接器ID、用户授权ID、充电参数即可，不需要感知任何OCPP协议细节

现有代码调整点

删除原代码中@on(Action.RemoteStartTransaction)装饰的回调方法：OCPP 1.6协议中RemoteStartTransaction是中央系统主动下发给充电桩的指令，充电桩不会主动上报该报文，该装饰器无实际作用反而可能引发路由冲突
将原代码硬编码的id_tag、connector_id、限流值改为动态传入参数，适配不同用户、不同充电桩的启动需求
抽离重复的RFID鉴权逻辑为公共方法，增加本地缓存减少重复接口调用，避免阻塞事件循环
全局维护在线充电桩实例映射，供业务层调用下发控制指令

内容的提问来源于stack exchange，提问作者user1288906

OCPP1.6 BootNotification异常与RemoteStartTransaction方案咨询

基础上下文 #

问题1：初始化报文缺失排查方案 #

问题2：远程启动充电实现方案 #

基础上下文

问题1：初始化报文缺失排查方案

问题2：远程启动充电实现方案