OCPP

Python package implementing the JSON version of the Open Charge Point Protocol (OCPP). Currently OCPP 1.6 (errata v4) and OCPP 2.0 are supported.

You can find the documentation on rtd.

Installation

You can either the project install from Pypi:

$ pip install ocpp

Or clone the project and install it manually using:

$ pip install .

Quick start

Below you can find examples on how to create a simple OCPP 2.0 central system as well as an OCPP 2.0 charge point.

Note

To run these examples the dependency websockets is required! Install it by running:

$ pip install websockets

Central system

The code snippet below creates a simple OCPP 2.0 central system which is able to handle BootNotification calls. You can find a detailed explanation of the code in the Central System documentation_.

importasyncioimportloggingimportwebsocketsfromdatetimeimportdatetimefromocpp.routingimportonfromocpp.v20importChargePointascpfromocpp.v20importcall_resultlogging.basicConfig(level=logging.INFO)
classChargePoint(cp):
@on('BootNotification')defon_boot_notitication(self, charging_station, reason, **kwargs):
returncall_result.BootNotificationPayload(
current_time=datetime.utcnow().isoformat(),
interval=10,
status='Accepted'
)
asyncdefon_connect(websocket, path):
""" For every new charge point that connects, create a ChargePoint instance and start listening for messages. """charge_point_id=path.strip('/')
cp=ChargePoint(charge_point_id, websocket)
awaitcp.start()
asyncdefmain():
server=awaitwebsockets.serve(
on_connect,
'0.0.0.0',
9000,
subprotocols=['ocpp2.0']
)
awaitserver.wait_closed()
if__name__=='__main__':
asyncio.run(main())

Charge point

importasyncioimportloggingimportwebsocketsfromocpp.v20importcallfromocpp.v20importChargePointascplogging.basicConfig(level=logging.INFO)
classChargePoint(cp):
asyncdefsend_boot_notification(self):
request=call.BootNotificationPayload(
charging_station={
'model': 'Wallbox XYZ',
'vendor_name': 'anewone'
},
reason="PowerUp"
)
response=awaitself.call(request)
ifresponse.status=='Accepted':
print("Connected to central system.")
asyncdefmain():
asyncwithwebsockets.connect(
'ws://localhost:9000/CP_1',
subprotocols=['ocpp2.0']
) asws:
cp=ChargePoint('CP_1', ws)
awaitasyncio.gather(cp.start(), cp.send_boot_notification())
if__name__=='__main__':
asyncio.run(main())

License

Except from the documents in docs/v16/specification/ everything is licensed under MIT. © The Mobility House

The documents in docs/v16/specification/ are licensed under Creative Commons Attribution-NoDerivatives 4.0 International Public License.