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