OCPP v2.0.1中央系统向充电站发起请求及响应方式咨询

核心逻辑 #

OCPP v2.0.1中，中央系统（CSMS）向充电站（CP）发起的是Call类型主动请求，充电站收到后需返回对应的CallResult（成功响应）或CallError（异常响应）。框架已封装底层异步交互逻辑，只需按规范实现请求发起与响应处理即可。

基于框架的实现步骤 #

1. 初始化中央系统服务端 #

先搭建CSMS基础服务，通过OCPPManager管理连接的充电站实例：

from ocpp.v201 import ChargePoint as cp
from ocpp.routing import on
from ocpp.v201.enums import Action
from ocpp.v201 import call
import datetime

class ChargePoint(cp):
    # 保留充电站主动请求的处理逻辑（如BootNotification）
    @on(Action.BootNotification)
    def on_boot_notification(self, charging_station, reason, **kwargs):
        return call.BootNotificationPayload(
            current_time=datetime.utcnow().isoformat(),
            interval=30,
            status="Accepted"
        )

# 初始化管理器
manager = OCPPManager(
    charge_point_class=ChargePoint,
    host="0.0.0.0",
    port=9000
)

2. 中央系统发起主动请求 #

通过OCPPManager获取目标充电站实例，调用框架提供的请求构造器发起指令，异步等待响应：

示例：发起GetVariables请求获取充电站数据

import asyncio
from ocpp.v201 import call

async def send_get_variables(charge_point_id):
    # 获取已连接的充电站实例
    cp_instance = manager.get_charge_point(charge_point_id)
    if not cp_instance:
        print(f"充电站 {charge_point_id} 未连接")
        return

    # 构造请求参数（按OCPP v2.0.1规范定义）
    request = call.GetVariablesPayload(
        variable=[
            {
                "name": "AvailableEnergy",
                "instance": "1",
                "component": {"name": "EnergyMeter", "instance": "1"}
            }
        ]
    )

    try:
        # 发送请求并等待响应
        response = await cp_instance.call(request)
        print(f"响应结果: {response}")
        # 解析响应数据
        for var in response.variable:
            print(f"变量 {var.name} 值: {var.value.value} {var.value.unit}")
    except Exception as e:
        print(f"请求失败: {str(e)}")

# 执行异步请求
asyncio.run(send_get_variables("cp_001"))

3. 充电站侧响应处理 #

充电站通过@on(Action.xxx)装饰器绑定对应请求的处理逻辑，返回规范的响应结构：

示例：处理GetVariables请求

from ocpp.v201 import call_result
from ocpp.v201.enums import Action, VariableStatusType

@on(Action.GetVariables)
def on_get_variables(self, variable, **kwargs):
    variable_results = []
    for var in variable:
        # 模拟从硬件获取数据
        if var.component.name == "EnergyMeter" and var.name == "AvailableEnergy":
            variable_results.append({
                "name": var.name,
                "instance": var.instance,
                "component": var.component,
                "value": {"value": "4800", "unit": "Wh"},
                "status": VariableStatusType.accepted
            })
        else:
            variable_results.append({
                "name": var.name,
                "instance": var.instance,
                "component": var.component,
                "status": VariableStatusType.unknown_variable
            })
    
    return call_result.GetVariablesPayload(variable=variable_results)

异常响应处理

若请求无法处理，直接抛出OCPPError返回错误信息：

from ocpp.exceptions import OCPPError

@on(Action.RemoteStartTransaction)
def on_remote_start(self, id_token, **kwargs):
    if id_token.id_token != "valid_123":
        raise OCPPError(
            error_code="InvalidIdToken",
            error_description="令牌无效"
        )
    return call_result.RemoteStartTransactionPayload(status="Accepted")

关键注意事项 #

• 所有主动请求均为异步调用，必须用await等待响应
• 需确保目标充电站已成功连接，否则manager.get_charge_point()会返回None
• 请求与响应的Payload结构需严格遵循OCPP v2.0.1官方规范
• 需覆盖请求超时、充电站离线、参数错误等异常场景

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