tests/ocpp-server/server.py

   1 import asyncio
   2 import logging
   3 from datetime import datetime, timezone
   4 from threading import Timer
   5
   6 import ocpp.v201
   7 import websockets
   8 from ocpp.routing import on
   9 from ocpp.v201.enums import (
  10     Action,
  11     AuthorizationStatusType,
  12     ClearCacheStatusType,
  13     RegistrationStatusType,
  14     TransactionEventType,
  15 )
  16 from websockets import ConnectionClosed
  17
  18 # Setting up the logging configuration to display debug level messages.
  19 logging.basicConfig(level=logging.DEBUG)
  20
  21 ChargePoints = set()
  22
  23
  24 class RepeatTimer(Timer):
  25     """Class that inherits from the Timer class. It will run a
  26     function at regular intervals."""
  27
  28     def run(self):
  29         while not self.finished.wait(self.interval):
  30             self.function(*self.args, **self.kwargs)
  31
  32
  33 # Define a ChargePoint class inheriting from the OCPP 2.0.1 ChargePoint class.
  34 class ChargePoint(ocpp.v201.ChargePoint):
  35     # Message handlers to receive OCPP messages.
  36     @on(Action.BootNotification)
  37     async def on_boot_notification(self, charging_station, reason, **kwargs):
  38         logging.info("Received %s", Action.BootNotification)
  39         # Create and return a BootNotification response with the current time,
  40         # an interval of 60 seconds, and an accepted status.
  41         return ocpp.v201.call_result.BootNotification(
  42             current_time=datetime.now(timezone.utc).isoformat(),
  43             interval=60,
  44             status=RegistrationStatusType.accepted,
  45         )
  46
  47     @on(Action.Heartbeat)
  48     async def on_heartbeat(self, **kwargs):
  49         logging.info("Received %s", Action.Heartbeat)
  50         return ocpp.v201.call_result.Heartbeat(
  51             current_time=datetime.now(timezone.utc).isoformat()
  52         )
  53
  54     @on(Action.StatusNotification)
  55     async def on_status_notification(
  56         self, timestamp, evse_id: int, connector_id: int, connector_status, **kwargs
  57     ):
  58         logging.info("Received %s", Action.StatusNotification)
  59         return ocpp.v201.call_result.StatusNotification()
  60
  61     @on(Action.Authorize)
  62     async def on_authorize(self, id_token, **kwargs):
  63         logging.info("Received %s", Action.Authorize)
  64         return ocpp.v201.call_result.Authorize(
  65             id_token_info={"status": AuthorizationStatusType.accepted}
  66         )
  67
  68     @on(Action.TransactionEvent)
  69     async def on_transaction_event(
  70         self,
  71         event_type: TransactionEventType,
  72         timestamp,
  73         trigger_reason,
  74         seq_no: int,
  75         transaction_info,
  76         **kwargs,
  77     ):
  78         match event_type:
  79             case TransactionEventType.started:
  80                 logging.info("Received %s Started", Action.TransactionEvent)
  81                 return ocpp.v201.call_result.TransactionEvent(
  82                     id_token_info={"status": AuthorizationStatusType.accepted}
  83                 )
  84             case TransactionEventType.updated:
  85                 logging.info("Received %s Updated", Action.TransactionEvent)
  86                 return ocpp.v201.call_result.TransactionEvent(total_cost=10)
  87             case TransactionEventType.ended:
  88                 logging.info("Received %s Ended", Action.TransactionEvent)
  89                 return ocpp.v201.call_result.TransactionEvent()
  90
  91     @on(Action.MeterValues)
  92     async def on_meter_values(self, evse_id: int, meter_value, **kwargs):
  93         logging.info("Received %s", Action.MeterValues)
  94         return ocpp.v201.call_result.MeterValues()
  95
  96     # Request handlers to emit OCPP messages.
  97     async def send_clear_cache(self):
  98         request = ocpp.v201.call.ClearCache()
  99         response = await self.call(request)
 100
 101         if response.status == ClearCacheStatusType.accepted:
 102             logging.info("%s successful", Action.ClearCache)
 103         else:
 104             logging.info("%s failed", Action.ClearCache)
 105
 106
 107 # Function to handle new WebSocket connections.
 108 async def on_connect(websocket, path):
 109     """For every new charge point that connects, create a ChargePoint instance and start
 110     listening for messages."""
 111     try:
 112         requested_protocols = websocket.request_headers["Sec-WebSocket-Protocol"]
 113     except KeyError:
 114         logging.info("Client hasn't requested any Subprotocol. Closing Connection")
 115         return await websocket.close()
 116
 117     if websocket.subprotocol:
 118         logging.info("Protocols Matched: %s", websocket.subprotocol)
 119     else:
 120         logging.warning(
 121             "Protocols Mismatched | Expected Subprotocols: %s,"
 122             " but client supports %s | Closing connection",
 123             websocket.available_subprotocols,
 124             requested_protocols,
 125         )
 126         return await websocket.close()
 127
 128     charge_point_id = path.strip("/")
 129     cp = ChargePoint(charge_point_id, websocket)
 130     ChargePoints.add(cp)
 131     try:
 132         await cp.start()
 133     except ConnectionClosed:
 134         logging.info("ChargePoint %s closed connection", cp.id)
 135         ChargePoints.remove(cp)
 136         logging.debug("Connected ChargePoint(s): %d", len(ChargePoints))
 137
 138
 139 # Main function to start the WebSocket server.
 140 async def main():
 141     # Create the WebSocket server and specify the handler for new connections.
 142     server = await websockets.serve(
 143         on_connect,
 144         "127.0.0.1",  # Listen on loopback.
 145         9000,  # Port number.
 146         subprotocols=["ocpp2.0", "ocpp2.0.1"],  # Specify OCPP 2.0.1 subprotocols.
 147     )
 148     logging.info("WebSocket Server Started")
 149     # Wait for the server to close (runs indefinitely).
 150     await server.wait_closed()
 151
 152
 153 # Entry point of the script.
 154 if __name__ == "__main__":
 155     # Run the main function to start the server.
 156     asyncio.run(main())