Skip to main content

Python package implementing the JSON version of the Open Charge Point Protocol (OCPP).

Project description

https://circleci.com/gh/mobilityhouse/ocpp/tree/master.svg?style=svg https://img.shields.io/pypi/pyversions/ocpp.svg https://img.shields.io/readthedocs/ocpp.svg

OCPP

Python package implementing the JSON version of the Open Charge Point Protocol (OCPP). Currently only OCPP 1.6 is 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 charge point as well as a charge point.

Central system

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

import asyncio
import websockets
from datetime import datetime

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


class ChargePoint(cp):
    @on(Action.BootNotification)
    def on_boot_notification(self, charge_point_vendor, charge_point_model, **kwargs):
        return call_result.BootNotificationPayload(
            current_time=datetime.utcnow().isoformat(),
            interval=10,
            status=RegistrationStatus.accepted
        )

   @after(Action.BootNotification)
   def after_boot_notification(self, charge_point_vendor, charge_point_model, **kwargs):
        print("ChargePoint Vendor is: %s", charge_point_vendor)
        print("ChargePoint Model is: %s",charge_point_model)


async def on_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)

    await cp.start()


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

    await server.wait_closed()


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

Charge point

import asyncio
import websockets

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


class ChargePoint(cp):
    async def send_boot_notification(self):
        request = call.BootNotificationPayload(
            charge_point_model="Optimus",
            charge_point_vendor="The Mobility House"
        )

        response = 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:

        cp = ChargePoint('CP_1', ws)

        await asyncio.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.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

ocpp-0.3.2.tar.gz (22.8 kB view hashes)

Uploaded Source

Built Distribution

ocpp-0.3.2-py3-none-any.whl (42.9 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page