OCPP协议中基于RFID的EV驱动授权

Question

我是OCPP协议的新手，我正在构建一个Python服务器，它可以使用OCPP协议与EV充电器通信。该服务器具有“通过RFID认证用户”的功能。我已经创建了2个Python文件，它们是Charge_Stattion.py：

# Charge_Stattion.py


import asyncio
import logging
import websockets

from ocpp.v201 import call
from ocpp.v201 import ChargePoint as cp

logging.basicConfig(level=logging.INFO)


class ChargePoint(cp):

    
    async def authentication(self):
        request = call.AuthorizePayload(
            id_token={'id_token':'AA12345',
                    'type': 'ISO14443'})
        response = await self.call(request)
        print(response)



async def main():
   async with websockets.connect(
       'ws://localhost:9000/CP_1',
        subprotocols=['ocpp2.0.1']
   ) as ws:

       cp = ChargePoint('CP_1', ws)

       await asyncio.gather(cp.start(), cp.authentication())


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

和Central_System.py：

#Central_System.py


import asyncio
import logging
import websockets
from datetime import datetime

from ocpp.routing import on
from ocpp.v201 import ChargePoint as cp
from ocpp.v201 import call_result
from ocpp.v201.enums import AuthorizationStatusType, Action

logging.basicConfig(level=logging.INFO)


class ChargePoint(cp):
   @on('BootNotification')
   async def on_boot_notification(self, charging_station, reason, **kwargs):
      return call_result.BootNotificationPayload(
         current_time=datetime.utcnow().isoformat(),
         interval=10,
         status='Accepted'
      )
   
   @on(Action.Authorize)
   async def on_authorize(self, id_token):
       return call_result.AuthorizePayload(id_token_info={"status": AuthorizationStatusType.accepted})


async def on_connect(websocket, path):
    """ For every new charge point that connects, create a ChargePoint
    instance and start listening for messages.
    """
    try:
        requested_protocols = websocket.request_headers[
            'Sec-WebSocket-Protocol']
    except KeyError:
        logging.info("Client hasn't requested any Subprotocol. "
                 "Closing Connection")
    if websocket.subprotocol:
        logging.info("Protocols Matched: %s", websocket.subprotocol)
    else:
        # In the websockets lib if no subprotocols are supported by the
        # client and the server, it proceeds without a subprotocol,
        # so we have to manually close the connection.
        logging.warning('Protocols Mismatched | Expected Subprotocols: %s,'
                        ' but client supports  %s | Closing connection',
                        websocket.available_subprotocols,
                        requested_protocols)
        return await websocket.close()

    charge_point_id = path.strip('/')
    cp = ChargePoint(charge_point_id, websocket)
    logging.info("abcxyz: %s", charge_point_id)
    await cp.start()


async def main():
    server = await websockets.serve(
        on_connect,
        '0.0.0.0',
        9000,
        subprotocols=['ocpp2.0.1']
    )
    logging.info("WebSocket Server Started")
    await server.wait_closed()

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

根据文件这里，我知道用户必须先出示射频识别卡，然后充电站将包含idToken的AuthorizeRequest发送到中央系统，然后中央系统将发送，AuthorizeResponse发送到收费站。在上面的两个python文件中，我实现了进程收费站将andAuthorizeRequest发送到中央系统，中央系统将AuthorizeResponse发送回充电站。这个图片演示了这些过程

我的问题是：

• 如何实现电动车驱动程序向充电站提交RFID卡。我应该创建另外两个python文件，代表EV驱动程序和RFID卡吗？
• 如何知道中心系统是否接受此身份验证以及如何实现此身份验证？

任何帮助都将不胜感激。

Answer 1

这是一个简单的流程

1. EV所有者将自己注册为某个服务器上的EV客户端，其中服务器提供了一个惟一的id，比如" unique - client -id“，并将此值作为idTag存储在数据库中。
2. 当该客户端到某个充电站充电时，他将该唯一的id输入到充电设备中，该设备通过websocket连接以下列形式发送id：

[3, "unique-id-representing-the-current-msg", "Authorize", {"idTag": "unique-client-id"}]

1. OCPP服务器接收到该消息，并在数据库中查找接收到的idTag，如果它存在，它将返回响应如下：

[4, "unique-id-representing-the-current-msg", {"idTagInfo": {"status": "Accepted"}}]

我建议使用sanic框架，因为它在默认情况下具有websocket和http支持。