|
<p>.. image:: <a href="https://github.com/mobilityhouse/ocpp/actions/workflows/pull-request.yml/badge.svg?style=svg" rel="nofollow">https://github.com/mobilityhouse/ocpp/actions/workflows/pull-request.yml/badge.svg?style=svg</a>
:target: <a href="https://github.com/mobilityhouse/ocpp/actions/workflows/pull-request.yml" rel="nofollow">https://github.com/mobilityhouse/ocpp/actions/workflows/pull-request.yml</a></p>
<p>.. image:: <a href="https://img.shields.io/pypi/pyversions/ocpp.svg" rel="nofollow">https://img.shields.io/pypi/pyversions/ocpp.svg</a>
:target: <a href="https://pypi.org/project/ocpp/" rel="nofollow">https://pypi.org/project/ocpp/</a></p>
<p>.. image:: <a href="https://img.shields.io/readthedocs/ocpp.svg" rel="nofollow">https://img.shields.io/readthedocs/ocpp.svg</a>
:target: <a href="https://ocpp.readthedocs.io/en/latest/" rel="nofollow">https://ocpp.readthedocs.io/en/latest/</a></p>
<h2>OCPP</h2>
<p>Python package implementing the JSON version of the Open Charge Point Protocol
(OCPP). Currently OCPP 1.6 (errata v4), OCPP 2.0 and OCPP 2.0.1 (Final Version)
are supported.</p>
<p>You can find the documentation on <code>rtd</code>_.</p>
<h2>Installation</h2>
<p>You can either the project install from Pypi:</p>
<p>.. code-block:: bash</p>
<p>$ pip install ocpp</p>
<p>Or clone the project and install it manually using:</p>
<p>.. code-block:: bash</p>
<p>$ pip install .</p>
<h2>Quick start</h2>
<p>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.</p>
<p>.. note::</p>
<p>To run these examples the dependency websockets_ is required! Install it by running:</p>
<p>.. code-block:: bash</p>
<pre><code> $ pip install websockets
</code></pre>
<p>Central system
~~~~~~~~~~~~~~</p>
<p>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 <code>Central System documentation</code>_.</p>
<p>.. code-block:: python</p>
<pre><code>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 RegistrationStatusType
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=RegistrationStatusType.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")
return await websocket.close()
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)
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())
</code></pre>
<p>Charge point
~~~~~~~~~~~~</p>
<p>.. code-block:: python</p>
<pre><code>import asyncio
from ocpp.v201.enums import RegistrationStatusType
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 send_boot_notification(self):
request = call.BootNotificationPayload(
charging_station={
'model': 'Wallbox XYZ',
'vendor_name': 'anewone'
},
reason="PowerUp"
)
response = await self.call(request)
if response.status == RegistrationStatusType.accepted:
print("Connected to central system.")
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.send_boot_notification())
if __name__ == '__main__':
asyncio.run(main())
</code></pre>
<h2>Debugging</h2>
<p>Python's default log level is <code>logging.WARNING</code>. As result most of the logs
generated by this package are discarded. To see the log output of this package
lower the log level to <code>logging.DEBUG</code>.</p>
<p>.. code-block:: python</p>
<p>import logging
logging.basicConfig(level=logging.DEBUG)</p>
<p>However, this approach defines the log level for the complete logging system.
In other words: the log level of all dependencies is set to <code>logging.DEBUG</code>.</p>
<p>To lower the logs for this package only use the following code:</p>
<p>.. code-block:: python</p>
<p>import logging
logging.getLogger('ocpp').setLevel(level=logging.DEBUG)
logging.getLogger('ocpp').addHandler(logging.StreamHandler())</p>
<h2>License</h2>
<p>Except from the documents in <code>docs/v16</code> and <code>docs/v201</code> everything is licensed under MIT<em>.
© <code>The Mobility House</code></em></p>
<p>The documents in <code>docs/v16</code> and <code>docs/v201</code> are licensed under Creative Commons
Attribution-NoDerivatives 4.0 International Public License.</p>
<p>.. _Central System documentation: <a href="https://ocpp.readthedocs.io/en/latest/central_system.html" rel="nofollow">https://ocpp.readthedocs.io/en/latest/central_system.html</a>
.. _MIT: <a href="https://github.com/mobilityhouse/ocpp/blob/master/LICENSE" rel="nofollow">https://github.com/mobilityhouse/ocpp/blob/master/LICENSE</a>
.. _rtd: <a href="https://ocpp.readthedocs.io/en/latest/index.html" rel="nofollow">https://ocpp.readthedocs.io/en/latest/index.html</a>
.. _The Mobility House: <a href="https://www.mobilityhouse.com/int_en/" rel="nofollow">https://www.mobilityhouse.com/int_en/</a>
.. _websockets: <a href="https://pypi.org/project/websockets/" rel="nofollow">https://pypi.org/project/websockets/</a></p>
|
