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