ocpp Python Library

To implement the actual OCPP protocol, you must subclass ocpp.v16.ChargePoint (or the version appropriate for your needs). This class handles message routing and validation.

Key Implementation Steps:

1. Subclass ChargePoint: Create a class that inherits from ocpp.v16.ChargePoint.
2. Register Handlers: Use the @on(Action.ACTION_NAME) decorator to register methods that handle specific OCPP actions (e.g., Action.boot_notification).
3. Handle Arguments: The library automatically converts OCPP camelCase JSON keys to Pythonic snake_case. Required arguments from the OCPP payload should be defined as named parameters in your handler, while optional arguments can be captured using **kwargs.
4. Return Results: Handlers must return an instance of a call_result class (e.g., call_result.BootNotification) to send a valid response back to the client.
5. Connection Lifecycle: In your WebSocket on_connect handler, extract the charge_point_id from the connection path, instantiate your custom ChargePoint class, and call await charge_point.start().

Example Implementation:

from datetime import datetime, timezone
import websockets

from ocpp.routing import on
from ocpp.v16 import ChargePoint as cp
from ocpp.v16 import call_result
from ocpp.v16.enums import Action, RegistrationStatus


class MyChargePoint(cp):
    @on(Action.boot_notification)
    async def on_boot_notification(
        self, charge_point_vendor, charge_point_model, **kwargs
    ):
        return call_result.BootNotification(
            current_time=datetime.now(tz=timezone.utc).isoformat(),
            interval=10,
            status=RegistrationStatus.accepted,
        )


async def on_connect(connection: websockets.ServerConnection):
    # Extract ID from path, e.g., /cp_id_1 -> cp_id_1
    charge_point_id = connection.request.path.split("/")[-1]
    charge_point = MyChargePoint(charge_point_id, connection)

    await charge_point.start()

To create a charge point that connects to a central system, you must subclass the version-specific ChargePoint class (e.g., ocpp.v16.ChargePoint) and provide a websocket connection.

Connection Steps:

1. Establish a WebSocket connection: Use websockets.connect() with a URL that includes the charge point identifier in the path (e.g., ws://localhost:9000/CP_1).
2. Specify the Subprotocol: You must pass the OCPP version as a subprotocol in the websockets.connect call (e.g., subprotocols=['ocpp1.6'] or subprotocols=['ocpp2.0.1']).
3. Initialize the ChargePoint: Pass the charge point ID and the websocket object to the ChargePoint constructor.
4. Run the Client: Use asyncio.gather() to run the charge_point.start() coroutine (which listens for incoming messages) concurrently with your business logic coroutines.

Sending Requests:

Use the call() method on your ChargePoint instance. This method accepts a request object from the ocpp.v16.call module, serializes it, and returns a corresponding call_result dataclass.

import asyncio
import websockets
from ocpp.v16 import ChargePoint as cp
from ocpp.v16 import call, call_result
from ocpp.v16.enums import RegistrationStatus

class ChargePoint(cp):
    async def send_boot_notification(self):
        request = call.BootNotification(
            charge_point_model="Wallbox XYZ",
            charge_point_vendor="acme",
        )
        response: call_result.BootNotification = await self.call(request)

        if response.status == RegistrationStatus.accepted:
            print("Connected to central system.")

async def main():
    async with websockets.connect(
        "ws://localhost:9000/CP_1", subprotocols=["ocpp1.6"]
    ) as ws:
        charge_point = ChargePoint("CP_1", ws)

        await asyncio.gather(
            charge_point.start(),
            charge_point.send_boot_notification(),
        )

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

The ChargePoint class uses a routing mechanism to map incoming OCPP Action names to specific handler functions (hooks).

The Hook Lifecycle:

1. _on_action: The primary handler for the incoming CALL. It receives the payload as snake_case keyword arguments. If this handler returns a value, it is used to construct the CALLRESULT.
2. _after_action: An optional hook executed after the response has been sent. It can be used for side effects. If decorated with inject_response=True, it also receives the call_response payload.

Key Features:

• Automatic Translation: The class automatically converts incoming camelCase payloads to snake_case for your handlers, and converts your returned snake_case dicts back to camelCase for the wire.
• Schema Validation: By default, incoming CALL payloads and outgoing CALLRESULT payloads are validated against the OCPP schema unless _skip_schema_validation is set in the route map.